diff --git a/pcapkit/protocols/data/internet/ipv4.py b/pcapkit/protocols/data/internet/ipv4.py index ad5f5aae99..715e0cdbf8 100644 --- a/pcapkit/protocols/data/internet/ipv4.py +++ b/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 diff --git a/pcapkit/protocols/internet/ipv4.py b/pcapkit/protocols/internet/ipv4.py index e1ece1e837..0cd3e6bccb 100644 --- a/pcapkit/protocols/internet/ipv4.py +++ b/pcapkit/protocols/internet/ipv4.py @@ -798,8 +798,10 @@ 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, @@ -807,6 +809,7 @@ def _read_opt_sec(self, schema: 'Schema_SECOption', *, options: 'Option') -> 'Da 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, *, diff --git a/pcapkit/protocols/schema/internet/ipv4.py b/pcapkit/protocols/schema/internet/ipv4.py index a515e10457..2ac7b5eb14 100644 --- a/pcapkit/protocols/schema/internet/ipv4.py +++ b/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 diff --git a/tests/protocols/internet/test_ipv4_ts_sec_roundtrip_unit.py b/tests/protocols/internet/test_ipv4_ts_sec_roundtrip_unit.py new file mode 100644 index 0000000000..a275a1a78d --- /dev/null +++ b/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() diff --git a/tests/protocols/internet/test_ipv4_unit.py b/tests/protocols/internet/test_ipv4_unit.py index dd4e2684e6..46d6da9973 100644 --- a/tests/protocols/internet/test_ipv4_unit.py +++ b/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), ])