fix(ipv4)!: TS and SEC options do not rebuild byte for byte (#1391, #1392, #1393)

pcapkit/protocols/data/internet/ipv4.py

@@ -195,9 +195,12 @@ class SECOption(Option):
     level: 'ClassificationLevel'
     #: Protection authority flags.
     flags: 'tuple[ProtectionAuthority, ...]'
+    #: Field termination indicator of each protection authority octet, as on
+    #: the wire.
+    termination: 'tuple[bool, ...]'
 
     if TYPE_CHECKING:
-        def __init__(self, code: 'OptionNumber', length: 'int', type: 'OptionType', level: 'ClassificationLevel', flags: 'tuple[ProtectionAuthority, ...]') -> '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', level: 'ClassificationLevel', flags: 'tuple[ProtectionAuthority, ...]', termination: 'tuple[bool, ...]') -> 'None': ...  # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin
 
 
 @info_final
@@ -229,9 +232,11 @@ class TSOption(Option):
     timestamp: 'tuple[timedelta | int, ...] | OrderedMultiDict[IPv4Address, timedelta | int]'
     #: Slots at or beyond the pointer, as raw 32-bit words.
     remaining: 'tuple[int, ...]'
+    #: Trailing octets that do not make up a whole 32-bit slot.
+    remainder: 'bytes'
 
     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]', remaining: 'tuple[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, ...]', remainder: 'bytes') -> 'None': ...  # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin
 
 
 @info_final

pcapkit/protocols/internet/ipv4.py

@@ -798,15 +798,18 @@ def _read_opt_sec(self, schema: 'Schema_SECOption', *, options: 'Option') -> 'Da
 
             if schema.data[-1] & 0x01 == 1:
                 warn(f'{self.alias}: [OptNo {schema.type}] invalid format: field termination indicator not set', ProtocolWarning)
+            termination = tuple(bool(byte & 0x01) for byte in schema.data)
         else:
             flags = []
+            termination = ()
 
         opt = Data_SECOption(
             code=schema.type,
             type=self._read_ipv4_opt_type(schema.type),
             length=schema.length,
             level=schema.level,
             flags=tuple(flags),
+            termination=termination,
         )
 
         return opt
@@ -891,6 +894,7 @@ def _read_opt_ts(self, schema: 'Schema_TSOption', *, options: 'Option') -> 'Data
             flag=schema.ts_flag,
             timestamp=schema.timestamp,
             remaining=schema.remaining,
+            remainder=schema.remainder,
         )
         return opt
 
@@ -1472,10 +1476,17 @@ def _make_opt_sec(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_SECOpt
             here rather than written: the enumeration names it as structure, and
             a value written there would be dropped on the way back in. See :issue:`537`.
 
+            The indicators themselves are set from position, except that a
+            parsed ``option`` whose ``termination`` covers every octet written
+            gets back the bits it carried, so that a non-conforming indicator
+            rebuilds as read.
+
         """
+        termination = ()  # type: tuple[bool, ...]
         if option is not None:
             level_val = option.level
             authorities = cast('list[Enum_ProtectionAuthority]', option.flags)
+            termination = tuple(getattr(option, 'termination', ()))
         else:
             level_val = self._make_index(level, level_default, namespace=level_namespace,  # type: ignore[assignment]
                                          reversed=level_reversed, pack=False)
@@ -1518,9 +1529,15 @@ def _make_opt_sec(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_SECOpt
 
             # The low-order bit of every octet but the last is ``1``, meaning
             # "another octet follows"; the last keeps its ``0``, which ends the
-            # field.
-            for index in range(int_len - 1):
-                data_list[index * 8 + 7] = b'1'
+            # field. A parsed option instead writes back the indicators it
+            # carried on the wire, conforming or not, so long as they still
+            # cover every octet written.
+            if len(termination) == int_len:
+                for index, bit in enumerate(termination):
+                    data_list[index * 8 + 7] = b'1' if bit else b'0'
+            else:
+                for index in range(int_len - 1):
+                    data_list[index * 8 + 7] = b'1'
 
             data = int(b''.join(data_list), base=2).to_bytes(int_len, 'big', signed=False)
         else:
@@ -1604,7 +1621,7 @@ def _make_opt_ts(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_TSOptio
                     # NOTE: An integer is the raw 32-bit field, written as
                     # is: a set high-order bit marks a non-standard time value.
                     if not isinstance(ts, int):
-                        ts = math.floor(ts.total_seconds() * 1000)
+                        ts = ts // datetime.timedelta(milliseconds=1)
                         if ts.bit_length() > 31:
                             warn(f'{self.alias}: [OptNo {kind}] timestamp value is too large: {ts}', ProtocolWarning)
                             ts = ts | 0x80000000
@@ -1613,19 +1630,21 @@ def _make_opt_ts(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_TSOptio
                 for ip, ts in option.timestamp.items(True):
                     ts_list.append(int(ip))
                     if not isinstance(ts, int):
-                        ts = math.floor(ts.total_seconds() * 1000)
+                        ts = ts // datetime.timedelta(milliseconds=1)
                         if ts.bit_length() > 31:
                             warn(f'{self.alias}: [OptNo {kind}] timestamp value is too large: {ts}', ProtocolWarning)
                             ts = ts | 0x80000000
                     ts_list.append(ts)
 
             ts_list += getattr(option, 'remaining', ())
+            remainder = getattr(option, 'remainder', b'')  # type: bytes
             length = option.length
             pointer = option.pointer
             overflow = option.overflow
             flag = option.flag
         else:
             ts_list = []
+            remainder = b''
             if isinstance(timestamp, list):
                 flag = Enum_TSFlag.Timestamp_Only  # type: ignore[assignment]
                 counts = min(9, counts)  # 9 is the maximum number of timestamps
@@ -1637,7 +1656,7 @@ def _make_opt_ts(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_TSOptio
                         break
 
                     if not isinstance(ts, int):
-                        ts = math.floor(ts.total_seconds() * 1000)
+                        ts = ts // datetime.timedelta(milliseconds=1)
                         if ts.bit_length() > 31:
                             warn(f'{self.alias}: [OptNo {kind}] timestamp value is too large: {ts}', ProtocolWarning)
                             ts = ts | 0x80000000
@@ -1654,7 +1673,7 @@ def _make_opt_ts(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_TSOptio
 
                     ts_list.append(int(ip))
                     if not isinstance(ts, int):
-                        ts = math.floor(ts.total_seconds() * 1000)
+                        ts = ts // datetime.timedelta(milliseconds=1)
                         if ts.bit_length() > 31:
                             warn(f'{self.alias}: [OptNo {kind}] timestamp value is too large: {ts}', ProtocolWarning)
                             ts = ts | 0x80000000
@@ -1686,6 +1705,7 @@ def _make_opt_ts(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_TSOptio
                 'flag': flag,
             },
             ts_data=ts_list,
+            remainder=remainder,
         )
 
     def _make_opt_e_sec(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_ESECOption]' = None, *,

pcapkit/protocols/schema/internet/ipv4.py

@@ -340,9 +340,11 @@ class TSOption(Option, code=Enum_OptionNumber.TS):
         length=lambda pkt: max(pkt['length'] - 4, 0) // 4 * 4,
         item_type=UInt32Field(),
     )
-    #: Trailing octets that do not make up a whole 32-bit slot.
-    remainder: 'bytes' = PaddingField(
+    #: Trailing octets that do not make up a whole 32-bit slot, kept as read
+    #: rather than packed as zeros.
+    remainder: 'bytes' = BytesField(
         length=lambda pkt: max(pkt['length'] - 4, 0) % 4,
+        default=b'',
     )
 
     def post_process(self, packet: 'dict[str, Any]') -> 'Schema':
@@ -463,7 +465,7 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema':
         # field name with an
         # :class:`~pcapkit.utilities.warnings.UnknownFieldWarning` rather than an
         # error.
-        def __init__(self, type: 'Enum_OptionNumber', length: 'int', pointer: 'int', flags: 'TSFlags', ts_data: 'list[int]') -> 'None': ...
+        def __init__(self, type: 'Enum_OptionNumber', length: 'int', pointer: 'int', flags: 'TSFlags', ts_data: 'list[int]', remainder: 'bytes' = ...) -> 'None': ...
 
 
 @schema_final

tests/protocols/internet/test_ipv4_ts_sec_roundtrip_unit.py

@@ -0,0 +1,132 @@
+# -*- coding: utf-8 -*-
+"""The IPv4 Timestamp and Security options rebuild byte for byte.
+
+- GitHub issue #1391: ``_make_opt_ts`` turned a ``timedelta`` into milliseconds
+  with ``math.floor(td.total_seconds() * 1000)``, and the float product is
+  inexact, so a stamp of 1001 ms rebuilt as 1000.
+- GitHub issue #1392: the 1-3 octets a ``TS`` length leaves after its last
+  whole slot were a zero-packing padding field, so they rebuilt as zeros. They
+  are now the ``remainder`` data field, kept as read.
+- GitHub issue #1393: ``_make_opt_sec`` set every field termination indicator
+  from the octet's position, so a non-conforming one was rewritten. The wire
+  bits are now the ``termination`` data field, written back on rebuild.
+
+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
+import warnings
+
+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)
+    opts += bytes(-len(opts) % 4)
+    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, by issue.
+HEADERS = {
+    '#1391 timestamp only, 1001 ms': header('44080900' '000003e9'),
+    '#1391 address and timestamp, 1001 ms': header('440c0d01' 'c0000201' '000003e9'),
+    '#1391 timestamp only, several inexact stamps': header('44100d00' '000003e9' '000003f1' '001f4006'),
+    '#1392 one-octet tail': header('44050500' 'aa'),
+    '#1392 three-octet tail, no slot': header('44070500' 'aabbcc'),
+    '#1392 three-octet tail after a slot': header('440b0900' '00000005' 'aabbcc'),
+    '#1392 two-octet tail, address and timestamp': header('440e0d01' 'c0000201' '00000005' 'aabb'),
+    '#1393 last indicator set': header('8204ab91'),
+    '#1393 last indicator set, low authorities': header('82041181'),
+    '#1393 first indicator clear': header('8205118001'),
+}
+
+
+class TestIPv4TimestampSecurityRoundTrip(unittest.TestCase):
+    """Pin the TS and SEC option bytes 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
+
+        with warnings.catch_warnings():
+            warnings.simplefilter('ignore')
+            return IPv4(io.BytesIO(data), len(data))
+
+    def option(self, data: 'bytes', kind: 'str') -> 'object':
+        from pcapkit.const.ipv4.option_number import OptionNumber
+
+        return self.parse(data).info.options[OptionNumber[kind]]  # 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)
+                with warnings.catch_warnings():
+                    warnings.simplefilter('ignore')
+                    self.assertEqual(bytes(IPv4.from_data(parsed.info)).hex(), data.hex())  # type: ignore[attr-defined]
+
+    def test_make_converts_a_timedelta_exactly(self) -> None:
+        import datetime
+        import ipaddress
+
+        from pcapkit.const.ipv4.option_number import OptionNumber
+        from pcapkit.protocols.internet.ipv4 import IPv4
+
+        # Each of these floors one short through ``total_seconds() * 1000``.
+        for msec in (1001, 1009, 2048006, 2147483644):
+            stamp = datetime.timedelta(milliseconds=msec)
+            with self.subTest(msec=msec, flag='timestamp only'):
+                datagram = IPv4(src='10.0.0.1', dst='10.0.0.2', ttl=64, id=0,
+                                options=[(OptionNumber.TS, {'counts': 1, 'timestamp': [stamp]})],
+                                payload=b'')
+                self.assertEqual(bytes(datagram)[24:28], msec.to_bytes(4, 'big'))
+            with self.subTest(msec=msec, flag='address and timestamp'):
+                datagram = IPv4(src='10.0.0.1', dst='10.0.0.2', ttl=64, id=0,
+                                options=[(OptionNumber.TS, {'counts': 1, 'timestamp': {
+                                    ipaddress.ip_address('192.0.2.1'): stamp,
+                                }})], payload=b'')
+                self.assertEqual(bytes(datagram)[28:32], msec.to_bytes(4, 'big'))
+
+    def test_trailing_partial_slot_is_data(self) -> None:
+        option = self.option(HEADERS['#1392 three-octet tail after a slot'], 'TS')
+        self.assertEqual(option.remainder, b'\xaa\xbb\xcc')  # type: ignore[attr-defined]
+        self.assertEqual(option.remaining, ())  # type: ignore[attr-defined]
+
+        option = self.option(header('44080900' '00000005'), 'TS')
+        self.assertEqual(option.remainder, b'')  # type: ignore[attr-defined]
+
+    def test_termination_indicators_are_data(self) -> None:
+        option = self.option(HEADERS['#1393 first indicator clear'], 'SEC')
+        self.assertEqual(option.termination, (False, True))  # type: ignore[attr-defined]
+
+        option = self.option(header('820301'), 'SEC')
+        self.assertEqual(option.termination, ())  # type: ignore[attr-defined]
+
+    def test_indicators_that_no_longer_cover_the_octets_are_positional(self) -> None:
+        from pcapkit.const.ipv4.classification_level import ClassificationLevel
+        from pcapkit.const.ipv4.option_number import OptionNumber
+        from pcapkit.const.ipv4.protection_authority import ProtectionAuthority
+        from pcapkit.protocols.data.internet.ipv4 import SECOption
+        from pcapkit.protocols.internet.ipv4 import IPv4
+
+        proto = IPv4.__new__(IPv4)
+        # One recorded indicator, but the authorities reach a second octet.
+        option = SECOption(code=OptionNumber.SEC, type=proto._read_ipv4_opt_type(OptionNumber.SEC),
+                           length=4, level=ClassificationLevel.Unclassified,
+                           flags=(ProtectionAuthority.GENSER, ProtectionAuthority(8)),
+                           termination=(False,))
+        self.assertEqual(proto._make_opt_sec(OptionNumber.SEC, option).data, b'\x81\x80')
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/protocols/internet/test_ipv4_unit.py

@@ -1424,6 +1424,7 @@ def opt_type(code):
             # terminator is structure, and ``_read_opt_sec`` never puts it in
             # ``flags`` in the first place.
             flags=(ProtectionAuthority.GENSER, ProtectionAuthority.NSA),
+            termination=(False,),
         )
         self.assertEqual(proto._make_opt_sec(OptionNumber.SEC, sec).level,
                          ClassificationLevel.Unclassified)
@@ -1466,6 +1467,7 @@ def opt_type(code):
             flag=TSFlag.Timestamp_Only,
             timestamp=(datetime.timedelta(seconds=1), 0x80000000),
             remaining=(),
+            remainder=b'',
         )
         # 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.
@@ -1488,6 +1490,7 @@ def opt_type(code):
             flag=TSFlag.Prespecified_IP_with_Timestamp,
             timestamp=ts_map,
             remaining=(),
+            remainder=b'',
         )
         with mock.patch('pcapkit.protocols.internet.ipv4.warn') as warn:
             self.assertEqual(proto._make_opt_ts(OptionNumber.TS, ts_prespecified).flags['flag'],
@@ -1619,6 +1622,7 @@ def opt_type(code):
                 length=5,
                 level=ClassificationLevel.Unclassified,
                 flags=(ProtectionAuthority.GENSER, ProtectionAuthority(8)),
+                termination=(True, False),
             )),
             (OptionNumber.MTUP, mtup),
         ])