From a15053649a7ecc77fec52dce4b3799c3c6d9b275 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Wed, 7 Oct 2026 19:15:06 -0400 Subject: [PATCH] fix(ipv4): the TS option zeroes the timestamp slots beyond the pointer on rebuild `ts_data` now models every 32-bit slot of the option; the slots at or beyond the pointer are the new data field `remaining` (raw words) and a make() keyword, as #1319 did for LSR/SSR/RR. A paired flag never splits a pair, and a pre-specified flag keeps every whole pair as before, with an odd trailing slot kept in `remaining` instead of dropped. Closes #1349 --- pcapkit/protocols/data/internet/ipv4.py | 6 +- pcapkit/protocols/internet/ipv4.py | 12 ++- pcapkit/protocols/schema/internet/ipv4.py | 55 +++++----- .../internet/test_ipv4_ts_slots_unit.py | 100 ++++++++++++++++++ tests/protocols/internet/test_ipv4_unit.py | 21 ++-- 5 files changed, 155 insertions(+), 39 deletions(-) create mode 100644 tests/protocols/internet/test_ipv4_ts_slots_unit.py diff --git a/pcapkit/protocols/data/internet/ipv4.py b/pcapkit/protocols/data/internet/ipv4.py index 458acc0103..ad5f5aae99 100644 --- a/pcapkit/protocols/data/internet/ipv4.py +++ b/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 diff --git a/pcapkit/protocols/internet/ipv4.py b/pcapkit/protocols/internet/ipv4.py index f652962d87..e1ece1e837 100644 --- a/pcapkit/protocols/internet/ipv4.py +++ b/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,6 +1580,7 @@ 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. @@ -1586,7 +1588,9 @@ def _make_opt_ts(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_TSOptio 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 diff --git a/pcapkit/protocols/schema/internet/ipv4.py b/pcapkit/protocols/schema/internet/ipv4.py index 9e0cabfcee..a515e10457 100644 --- a/pcapkit/protocols/schema/internet/ipv4.py +++ b/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 diff --git a/tests/protocols/internet/test_ipv4_ts_slots_unit.py b/tests/protocols/internet/test_ipv4_ts_slots_unit.py new file mode 100644 index 0000000000..b7cae3c760 --- /dev/null +++ b/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() diff --git a/tests/protocols/internet/test_ipv4_unit.py b/tests/protocols/internet/test_ipv4_unit.py index b794912fe1..dd4e2684e6 100644 --- a/tests/protocols/internet/test_ipv4_unit.py +++ b/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,7 +1940,7 @@ 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=[ @@ -1940,9 +1948,10 @@ def test_ipv4_schema_helpers_and_post_process_branches(self) -> None: 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)