fix(ipv4)!: the TS option zeroes the timestamp slots beyond the pointer on rebuild (#1349)

pcapkit/protocols/data/internet/ipv4.py

@@ -225,11 +225,13 @@ class TSOption(Option):
     overflow: 'int'
     #: Flag.
     flag: 'TSFlag'
-    #: Timestamp data.
+    #: Timestamp data, i.e. the slots before the pointer.
     timestamp: 'tuple[timedelta | int, ...] | OrderedMultiDict[IPv4Address, timedelta | int]'
+    #: Slots at or beyond the pointer, as raw 32-bit words.
+    remaining: 'tuple[int, ...]'
 
     if TYPE_CHECKING:
-        def __init__(self, code: 'OptionNumber', length: 'int', type: 'OptionType', pointer: 'int', overflow: 'int', flag: 'TSFlag', timestamp: 'tuple[timedelta | int, ...] | OrderedMultiDict[IPv4Address, timedelta | int]') -> 'None': ...  # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin
+        def __init__(self, code: 'OptionNumber', length: 'int', type: 'OptionType', pointer: 'int', overflow: 'int', flag: 'TSFlag', timestamp: 'tuple[timedelta | int, ...] | OrderedMultiDict[IPv4Address, timedelta | int]', remaining: 'tuple[int, ...]') -> 'None': ...  # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin
 
 
 @info_final

pcapkit/protocols/internet/ipv4.py

@@ -890,6 +890,7 @@ def _read_opt_ts(self, schema: 'Schema_TSOption', *, options: 'Option') -> 'Data
             overflow=schema.flags['oflw'],
             flag=schema.ts_flag,
             timestamp=schema.timestamp,
+            remaining=schema.remaining,
         )
         return opt
 
@@ -1579,14 +1580,17 @@ def _make_opt_ts(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_TSOptio
                      counts: 'int' = 5,
                      overflow: 'int' = 0,
                      timestamp: 'Optional[list[int | timedelta] | dict[IPv4Address, int | timedelta]]' = None,
+                     remaining: 'Optional[list[int]]' = None,
                      **kwargs: 'Any') -> 'Schema_TSOption':
         """Make IPv4 Timestamp (``TS``) option.
 
         Args:
             kind: option type code
             option: option data
             counts: maximum number of timestamps to record
-            timestamp: list of timestamps
+            timestamp: list of timestamps, i.e. the slots before the pointer
+            remaining: raw 32-bit words for the slots at or beyond the pointer;
+                the slots left over up to ``counts`` are zero
             **kwargs: arbitrary keyword arguments
 
         Returns:
@@ -1615,6 +1619,7 @@ def _make_opt_ts(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_TSOptio
                             ts = ts | 0x80000000
                     ts_list.append(ts)
 
+            ts_list += getattr(option, 'remaining', ())
             length = option.length
             pointer = option.pointer
             overflow = option.overflow
@@ -1660,6 +1665,11 @@ def _make_opt_ts(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_TSOptio
                 raise ProtocolError(f'{self.alias}: [OptNo {kind}] invalid timestamp value: {timestamp}')
             pointer = 5 + len(ts_list) * 4
 
+            # NOTE: Every slot of the option is written, the empty ones as zero.
+            slots = (length - 4) // 4
+            ts_list = (ts_list + list(remaining or ()))[:slots]
+            ts_list += [0] * (slots - len(ts_list))
+
         # NOTE: ``ts_data`` is the field on
         # :class:`~pcapkit.protocols.schema.internet.ipv4.TSOption`; ``data`` is
         # the attribute its ``post_process`` *derives* from ``ts_data``. Passing

pcapkit/protocols/schema/internet/ipv4.py

@@ -334,15 +334,15 @@ class TSOption(Option, code=Enum_OptionNumber.TS):
         'oflw': (0, 4),
         'flag': (4, 4),
     })
-    #: Timestamps and internet addresses.
+    #: Timestamps and internet addresses: every 32-bit slot of the option,
+    #: those before the pointer and those at or beyond it alike.
     ts_data: 'list[int]' = ListField(
-        length=lambda pkt: pkt['pointer'] - 5 if pkt['flags']['flag'] != 3 else pkt['length'] - 4,
+        length=lambda pkt: max(pkt['length'] - 4, 0) // 4 * 4,
         item_type=UInt32Field(),
     )
-    #: Remaining data buffer.
+    #: Trailing octets that do not make up a whole 32-bit slot.
     remainder: 'bytes' = PaddingField(
-        length=lambda pkt: pkt['length'] - pkt['pointer'] + 1 if pkt['flags']['flag'] != 3 else 0,
-        default=bytes(36),  # 36 is the maximum length of the option data field for timestamps
+        length=lambda pkt: max(pkt['length'] - 4, 0) % 4,
     )
 
     def post_process(self, packet: 'dict[str, Any]') -> 'Schema':
@@ -379,8 +379,22 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema':
 
         """
         ts_flag = Enum_TSFlag.get(self.flags['flag'])
+
+        # NOTE: Only the slots wholly before the pointer have been filled in, so
+        # the rest are kept apart, as the raw 32-bit words, in ``remaining``. A
+        # pre-specified address is part of the option's content whether or not
+        # its timestamp has been filled in yet, so that flag consumes every whole
+        # address and timestamp pair. A paired flag never splits a pair.
+        if ts_flag == Enum_TSFlag.Prespecified_IP_with_Timestamp:
+            used = len(self.ts_data)
+        else:
+            used = min(max(self.pointer - 5, 0) // 4, len(self.ts_data))
+        if ts_flag in (Enum_TSFlag.IP_with_Timestamp, Enum_TSFlag.Prespecified_IP_with_Timestamp):
+            used -= used % 2
+        self.remaining = tuple(self.ts_data[used:])
+
         if ts_flag == Enum_TSFlag.Timestamp_Only:
-            ts_data = self.ts_data
+            ts_data = self.ts_data[:used]
             self.data = []
             ts_list = []  # type: list[int | timedelta]
 
@@ -396,7 +410,7 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema':
                 ts_list.append(ts_val)
             timestamp = tuple(ts_list)  # type: tuple[int | timedelta, ...] | OrderedMultiDict[IPv4Address, int | timedelta]
         elif ts_flag == Enum_TSFlag.IP_with_Timestamp:
-            ts_data = self.ts_data
+            ts_data = self.ts_data[:used]
             self.data = OrderedMultiDict()
             timestamp = OrderedMultiDict()
 
@@ -411,7 +425,7 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema':
                     ts_val = datetime.timedelta(milliseconds=ts)
                 timestamp.add(ip_val, ts_val)
         elif ts_flag == Enum_TSFlag.Prespecified_IP_with_Timestamp:
-            ts_data = self.ts_data
+            ts_data = self.ts_data[:used]
             self.data = OrderedMultiDict()
             timestamp = OrderedMultiDict()
 
@@ -425,30 +439,10 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema':
                 else:
                     ts_val = datetime.timedelta(milliseconds=ts)
                 timestamp.add(ip_val, ts_val)
-
-            # extract also the prespecified IP addresses
-            # but set the timestamp to 0
-            #
-            # NOTE: Through ``parse_ip_address`` like the two conversions above,
-            # even though ``remainder`` is a
-            # :class:`~pcapkit.corekit.fields.strings.PaddingField`, so what it
-            # holds is octets rather than anything the caller named, and cannot
-            # launder a :obj:`bool`. It is routed through anyway because a bare
-            # :func:`ipaddress.ip_address` here still raises a plain
-            # :exc:`ValueError` for a tail that is not a whole number of 8-octet
-            # pairs, which no ``except BaseError`` can catch, and because leaving
-            # one of this method's three conversions unguarded invites the same
-            # defect back.
-            pad = self.remainder
-            for index in range(0, len(pad), 8):
-                buf_ip = pad[index:index + 4]
-                self.data.add(parse_ip_address(  # type: ignore[arg-type]
-                    buf_ip, f'IPv4: [OptNo {self.type}] invalid prespecified address',
-                    version=4), 0)
         else:
             warn(f'IPv4: [OptNo {self.type}] invalid format: unknown timestmap flag: {ts_flag}', ProtocolWarning)
-            self.data = self.ts_data
-            timestamp = tuple(self.ts_data)
+            self.data = self.ts_data[:used]
+            timestamp = tuple(self.data)
 
         self.ts_flag = ts_flag
         self.timestamp = timestamp
@@ -457,6 +451,7 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema':
     if TYPE_CHECKING:
         ts_flag: 'Enum_TSFlag'
         data: 'list[int] | OrderedMultiDict[IPv4Address, int]'
+        remaining: 'tuple[int, ...]'
         timestamp: 'tuple[int | timedelta] | OrderedMultiDict[IPv4Address, int | timedelta]'
 
         # NOTE: The keyword is ``ts_data``, the name of the field above, not

tests/protocols/internet/test_ipv4_ts_slots_unit.py

@@ -0,0 +1,100 @@
+# -*- coding: utf-8 -*-
+"""The IPv4 Timestamp option keeps the slots at or beyond its pointer.
+
+GitHub issue #1349: the ``TS`` option's ``ts_data`` covered only the
+``pointer - 5`` octets before the pointer, and the rest of the option was a
+:class:`~pcapkit.corekit.fields.strings.PaddingField` that packed zeros, so
+``from_data`` over a parsed header zeroed every slot not yet filled in. Those
+slots are now the ``remaining`` data field and a ``make()`` keyword, the same
+shape #1319 gave LSR, SSR and RR.
+
+Every case builds its own octets in memory and reads no capture. Classes are
+imported inside each test, after :func:`~tests._support.reimport_once_per_class`.
+
+"""
+
+import io
+import unittest
+
+from tests._support import reimport_once_per_class
+
+
+def header(options: 'str') -> 'bytes':
+    """An IPv4 header from 10.0.0.1 to 10.0.0.2, protocol 253, with ``options`` in hex."""
+    opts = bytes.fromhex(options)
+    ihl = 5 + len(opts) // 4
+    return (bytes([0x40 | ihl, 0]) + (ihl * 4).to_bytes(2, 'big') + bytes(2)
+            + bytes.fromhex('000040fd0000') + bytes([10, 0, 0, 1, 10, 0, 0, 2]) + opts)
+
+
+#: Wire headers that must rebuild through ``from_data`` unchanged.
+HEADERS = {
+    # The two repros of the issue.
+    'timestamp only, pointer 5': header('440c0500' '0000000100000002'),
+    'timestamp only, pointer 9': header('44100900' '000000010000000200000003'),
+    'timestamp only, unaligned pointer': header('440c0600' '0000000100000002'),
+    'address and timestamp, one pair filled': header('44140d01' 'c000020100000005' 'c000020200000006'),
+    'address and timestamp, half a pair filled': header('44140901' 'c000020100000005' 'c000020200000006'),
+    'pre-specified, odd slot count': header('44100503' 'c000020100000000' 'c0000202'),
+}
+
+
+class TestIPv4TimestampSlots(unittest.TestCase):
+    """Pin the TS slots at or beyond the pointer through parse, rebuild and make."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+
+    def parse(self, data: 'bytes') -> 'object':
+        from pcapkit.protocols.internet.ipv4 import IPv4
+
+        return IPv4(io.BytesIO(data), len(data))
+
+    def option(self, data: 'bytes') -> 'object':
+        from pcapkit.const.ipv4.option_number import OptionNumber
+
+        return self.parse(data).info.options[OptionNumber.TS]  # type: ignore[attr-defined]
+
+    def test_from_data_rebuilds_byte_for_byte(self) -> None:
+        from pcapkit.protocols.internet.ipv4 import IPv4
+
+        for name, data in HEADERS.items():
+            with self.subTest(header=name):
+                parsed = self.parse(data)
+                self.assertEqual(bytes(IPv4.from_data(parsed.info)), data)  # type: ignore[attr-defined]
+
+    def test_slots_at_or_beyond_the_pointer_are_data(self) -> None:
+        import datetime
+
+        option = self.option(HEADERS['timestamp only, pointer 9'])
+        self.assertEqual(option.timestamp, (datetime.timedelta(milliseconds=1),))  # type: ignore[attr-defined]
+        self.assertEqual(option.remaining, (2, 3))  # type: ignore[attr-defined]
+
+        option = self.option(HEADERS['timestamp only, pointer 5'])
+        self.assertEqual(option.timestamp, ())  # type: ignore[attr-defined]
+        self.assertEqual(option.remaining, (1, 2))  # type: ignore[attr-defined]
+
+        # A pair is never split: the address of a pair whose timestamp is not
+        # filled in yet stays in ``remaining`` with it.
+        option = self.option(HEADERS['address and timestamp, half a pair filled'])
+        self.assertEqual(len(option.timestamp), 0)  # type: ignore[attr-defined]
+        self.assertEqual(option.remaining, (0xc0000201, 5, 0xc0000202, 6))  # type: ignore[attr-defined]
+
+        # A pre-specified address is content regardless of the pointer; a slot
+        # left over from the pairs is kept rather than dropped.
+        option = self.option(HEADERS['pre-specified, odd slot count'])
+        self.assertEqual(len(option.timestamp), 1)  # type: ignore[attr-defined]
+        self.assertEqual(option.remaining, (0xc0000202,))  # type: ignore[attr-defined]
+
+    def test_make_writes_remaining_after_the_timestamps(self) -> None:
+        from pcapkit.const.ipv4.option_number import OptionNumber
+        from pcapkit.protocols.internet.ipv4 import IPv4
+
+        datagram = IPv4(src='10.0.0.1', dst='10.0.0.2', ttl=64, id=0,
+                        options=[(OptionNumber.TS, {'counts': 3, 'timestamp': [1],
+                                                    'remaining': [2]})], payload=b'')
+        self.assertEqual(bytes(datagram)[20:].hex(), '44100900' '000000010000000200000000')
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/protocols/internet/test_ipv4_unit.py

@@ -804,13 +804,18 @@ def option(ts_data: 'list[int]', flag: 'TSFlag') -> 'TSOption':
                          datetime.timedelta(milliseconds=5))
 
         # And the third conversion in the same method, the one that reads the
-        # prespecified addresses out of the option's padding.
-        prespecified = option([int(ip_address('192.0.2.2')), 6],
-                              TSFlag.Prespecified_IP_with_Timestamp)
-        prespecified.remainder = ip_address('192.0.2.3').packed + bytes(4)
+        # prespecified addresses, those at or beyond the pointer included.
+        prespecified = TSOption(type=OptionNumber.TS, length=20, pointer=13,
+                                flags={'oflw': 0, 'flag': TSFlag.Prespecified_IP_with_Timestamp},
+                                ts_data=[int(ip_address('192.0.2.2')), 6,
+                                         int(ip_address('192.0.2.3')), 0])
         prespecified.pack()
         self.assertEqual(prespecified.data[ip_address('192.0.2.2')], 6)
         self.assertEqual(prespecified.data[ip_address('192.0.2.3')], 0)
+        with self.assertRaises(FieldValueError):
+            TSOption(type=OptionNumber.TS, length=20, pointer=13,
+                     flags={'oflw': 0, 'flag': TSFlag.Prespecified_IP_with_Timestamp},
+                     ts_data=[int(ip_address('192.0.2.2')), 6, True, 0]).pack()
 
     def test_ipv4_quick_start_option_is_eight_octets_wide_on_the_wire(self) -> None:
         """Both Quick-Start suboptions survive the round trip whole. C.f. #552.
@@ -1460,6 +1465,7 @@ def opt_type(code):
             overflow=1,
             flag=TSFlag.Timestamp_Only,
             timestamp=(datetime.timedelta(seconds=1), 0x80000000),
+            remaining=(),
         )
         # An integer is the raw field, its high-order bit marking a non-standard
         # value, so it is written as is and without a warning. C.f. #1320.
@@ -1481,6 +1487,7 @@ def opt_type(code):
             overflow=0,
             flag=TSFlag.Prespecified_IP_with_Timestamp,
             timestamp=ts_map,
+            remaining=(),
         )
         with mock.patch('pcapkit.protocols.internet.ipv4.warn') as warn:
             self.assertEqual(proto._make_opt_ts(OptionNumber.TS, ts_prespecified).flags['flag'],
@@ -1720,6 +1727,7 @@ def assert_bad(reader, schema) -> None:
             datetime.timedelta(seconds=1),
             datetime.timedelta(seconds=2),
         ))
+        object.__setattr__(ts_schema, 'remaining', ())
         ts = proto._read_opt_ts(ts_schema, options=options)
         self.assertEqual(ts.flag, TSFlag.Timestamp_Only)
         self.assertEqual(ts.timestamp[0], datetime.timedelta(seconds=1))
@@ -1932,17 +1940,18 @@ def test_ipv4_schema_helpers_and_post_process_branches(self) -> None:
 
         pre_ts = ipv4_schema.TSOption(
             type=OptionNumber.TS,
-            length=20,
+            length=28,
             pointer=21,
             flags={'oflw': 0, 'flag': TSFlag.Prespecified_IP_with_Timestamp},
             ts_data=[
                 int(ip_address('192.0.2.2')),
                 0x80000006,
                 int(ip_address('192.0.2.5')),
                 3000,
+                int(ip_address('192.0.2.3')),
+                0,
             ],
         )
-        pre_ts.remainder = ip_address('192.0.2.3').packed + b'\x00\x00\x00\x00'
         with mock.patch('pcapkit.protocols.schema.internet.ipv4.warn') as warn:
             pre_ts.post_process({})
         self.assertEqual(pre_ts.timestamp[ip_address('192.0.2.2')], 0x80000006)