diff --git a/pcapkit/protocols/data/internet/hip.py b/pcapkit/protocols/data/internet/hip.py index a2ea0c1376..0dbec4b870 100644 --- a/pcapkit/protocols/data/internet/hip.py +++ b/pcapkit/protocols/data/internet/hip.py @@ -66,9 +66,12 @@ class Control(Data): #: Anonymous flag. anonymous: 'bool' + #: Reserved bits -- the 15 high-order bits of ``Controls``, carried + #: verbatim so that re-serialising reproduces them. + reserved: 'int' if TYPE_CHECKING: - def __init__(self, anonymous: 'bool') -> 'None': ... # pylint: disable=super-init-not-called,unused-argument,multiple-statements + def __init__(self, anonymous: 'bool', reserved: 'int') -> 'None': ... # pylint: disable=super-init-not-called,unused-argument,multiple-statements class Parameter(Data): @@ -125,6 +128,8 @@ def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', cont class ESPInfoParameter(Parameter): """Data model for HIP ``ESP_INFO`` parameter.""" + #: Reserved octets, carried verbatim so that re-serialising reproduces them. + reserved: 'int' #: KEYMAT index. index: 'int' #: Old SDI. @@ -133,7 +138,7 @@ class ESPInfoParameter(Parameter): new_spi: 'int' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', index: 'int', old_spi: 'int', new_spi: 'int') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long + def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', reserved: 'int', index: 'int', old_spi: 'int', new_spi: 'int') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long @info_final @@ -198,8 +203,16 @@ class PuzzleParameter(Parameter): #: Numeric index. index: 'int' - #: Lifetime. - lifetime: 'timedelta' + #: The ``Lifetime`` octet as it appears on the wire. :rfc:`7401#section-5.2.4` + #: defines the lifetime as ``2^(value - 32)`` seconds, and this is ``value``. + #: It is what re-serialising writes back, since most of the 256 octets have + #: no exact :class:`~datetime.timedelta`. + lifetime_exponent: 'int' + #: Lifetime, derived from :attr:`lifetime_exponent` to + #: :class:`~datetime.timedelta`'s microsecond resolution, so octets below + #: ``12`` read as ``timedelta(0)``. :obj:`None` for octets of ``79`` and above, + #: whose duration exceeds :attr:`datetime.timedelta.max`. + lifetime: 'Optional[timedelta]' #: Puzzle data. opaque: 'bytes' #: Random number. @@ -213,7 +226,7 @@ class PuzzleParameter(Parameter): rhash_len: 'int' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', index: 'int', lifetime: 'timedelta', opaque: 'bytes', random: 'int', rhash_len: 'int') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long + def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', index: 'int', lifetime_exponent: 'int', lifetime: 'Optional[timedelta]', opaque: 'bytes', random: 'int', rhash_len: 'int') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long @info_final diff --git a/pcapkit/protocols/internet/hip.py b/pcapkit/protocols/internet/hip.py index f36b7d2aad..a1b5fa0a18 100644 --- a/pcapkit/protocols/internet/hip.py +++ b/pcapkit/protocols/internet/hip.py @@ -583,6 +583,7 @@ def read(self, length: 'Optional[int]' = None, *, extension: bool = False, **kwa chksum=schema.checksum, control=Data_Control( anonymous=bool(schema.control['anonymous']), + reserved=schema.control['reserved'], ), shit=schema.shit, rhit=schema.rhit, @@ -615,6 +616,7 @@ def make(self, version: 'int' = 2, checksum: 'bytes' = b'\x00\x00', controls_anonymous: 'bool' = False, + controls_reserved: 'int' = 0, shit: 'int' = 0, rhit: 'int' = 0, parameters: 'Optional[list[Schema_Parameter | tuple[Enum_Parameter, dict[str, Any]] | bytes] | Parameter]' = None, # pylint: disable=line-too-long @@ -634,6 +636,7 @@ def make(self, version: HIP version. checksum: Checksum. controls_anonymous: If the sender is anonymous. + controls_reserved: The 15 reserved high-order bits of ``Controls``. shit: Sender's host identity tag (HIT). rhit: Receiver's host identity tag (HIT). parameters: HIP parameters. @@ -651,7 +654,9 @@ def make(self, if parameters is not None: parameters_value, total_length = self._make_hip_param(parameters, version=version) - length = total_length // 8 + 4 + # Ceiling division, so that Header Length never declares fewer + # octets than are packed. + length = -(-total_length // 8) + 4 else: # NOTE: Header Length excludes the first 8 octets but always # counts the two 16-octet HITs, hence the minimum of 4. @@ -670,6 +675,7 @@ def make(self, }, checksum=checksum, control = { + 'reserved': controls_reserved, 'anonymous': controls_anonymous, }, shit=shit, @@ -759,9 +765,11 @@ def _make_data(cls, data: 'Data_HIP') -> 'dict[str, Any]': # type: ignore[overr 'version': data.version, 'checksum': data.chksum, 'controls_anonymous': data.control.anonymous, + 'controls_reserved': data.control.reserved, 'shit': data.shit, 'rhit': data.rhit, - 'parameters': data.parameters, + # ``read`` sets ``parameters`` only when the packet carries some. + 'parameters': getattr(data, 'parameters', None), 'payload': cls._make_payload(data), } @@ -878,6 +886,7 @@ def _read_param_esp_info(self, schema: 'Schema_ESPInfoParameter', *, version: 'i type=schema.type, critical=bool(schema.type & 0b1), length=parameter_total_len(schema.len), + reserved=schema.reserved, index=schema.index, old_spi=schema.old_spi, new_spi=schema.new_spi, @@ -1080,7 +1089,8 @@ def _read_param_puzzle(self, schema: 'Schema_PuzzleParameter', *, version: 'int' critical=bool(schema.type & 0b1), length=parameter_total_len(schema.len), index=_numk, - lifetime=datetime.timedelta(seconds=2 ** (_time - 32)), + lifetime_exponent=_time, + lifetime=self._read_puzzle_lifetime(_time), opaque=_opak, random=_rand, # Keep the field's on-wire width, which ``_rand`` alone cannot carry: @@ -1091,6 +1101,28 @@ def _read_param_puzzle(self, schema: 'Schema_PuzzleParameter', *, version: 'int' ) return puzzle + @staticmethod + def _read_puzzle_lifetime(octet: 'int') -> 'Optional[timedelta]': + """Decode a ``PUZZLE`` ``Lifetime`` octet into a duration. + + :rfc:`7401#section-5.2.4` gives the lifetime as ``2^(value - 32)`` + seconds. The result is informational only: :meth:`_make_param_puzzle` + writes back the octet itself, since :class:`~datetime.timedelta` rounds + octets below ``12`` to zero and cannot hold those of ``79`` and above. + + Args: + octet: the ``Lifetime`` octet + + Returns: + The lifetime, or :obj:`None` if it exceeds + :attr:`datetime.timedelta.max`. + + """ + try: + return datetime.timedelta(seconds=2 ** (octet - 32)) + except OverflowError: + return None + def _read_param_solution(self, schema: 'Schema_SolutionParameter', *, version: 'int', options: 'Parameter') -> 'Data_SolutionParameter': # pylint: disable=unused-argument """Read HIP ``SOLUTION`` parameter. @@ -2966,6 +2998,10 @@ def _make_hip_param(self, parameters: 'list[Schema_Parameter | tuple[Enum_Parame parameters_list = [] # type: list[Schema_Parameter | bytes] for schema in parameters: if isinstance(schema, bytes): + # :rfc:`7401#section-5.2.1` pads every parameter to a + # multiple of 8 octets, and Header Length counts in those + # units; a raw parameter is taken to omit its padding. + schema += bytes(-len(schema) % 8) parameters_list.append(schema) total_length += len(schema) elif isinstance(schema, Schema): @@ -3035,6 +3071,7 @@ def _make_param_unassigned(self, code: 'Enum_Parameter', param: 'Optional[Data_U def _make_param_esp_info(self, code: 'Enum_Parameter', param: 'Optional[Data_ESPInfoParameter]' = None, *, # pylint: disable=unused-argument version: 'int', + reserved: 'int' = 0, index: 'int' = 0, old_spi: 'int' = 0, new_spi: 'int' = 0, @@ -3045,6 +3082,7 @@ def _make_param_esp_info(self, code: 'Enum_Parameter', param: 'Optional[Data_ESP code: parameter code param: parameter data version: HIP protocol version + reserved: reserved octets, zero when sent (:rfc:`7402#section-5.1.1`) index: KEYMAT index old_spi: old SPI new_spi: new SPI @@ -3055,6 +3093,7 @@ def _make_param_esp_info(self, code: 'Enum_Parameter', param: 'Optional[Data_ESP """ if param is not None: + reserved = param.reserved index = param.index old_spi = param.old_spi new_spi = param.new_spi @@ -3062,6 +3101,7 @@ def _make_param_esp_info(self, code: 'Enum_Parameter', param: 'Optional[Data_ESP return Schema_ESPInfoParameter( type=code, len=12, + reserved=reserved, index=index, old_spi=old_spi, new_spi=new_spi, @@ -3248,10 +3288,7 @@ def _make_puzzle_lifetime(code: 'Enum_Parameter', version: 'int', # ``math.log2`` raises a bare :exc:`ValueError` at zero and below. That is # not a :class:`~pcapkit.utilities.exceptions.BaseError`, so it would # escape the library's error handling with a message naming neither HIP - # nor the field. Conformant input reaches it: a ``Lifetime`` octet of - # ``0x00`` means ``2^-32`` seconds, below - # :class:`~datetime.timedelta`'s microsecond resolution, so parsing one - # yields ``timedelta(0)`` and re-serialising it lands here. + # nor the field. if seconds <= 0: raise ProtocolError(f'HIPv{version}: [ParamNo {code}] invalid lifetime: ' f'{seconds} is not a positive number of seconds') @@ -3347,6 +3384,7 @@ def _make_param_puzzle(self, code: 'Enum_Parameter', param: 'Optional[Data_Puzzl version: 'int', index: 'int' = 0, lifetime: 'timedelta | int' = 0, + lifetime_exponent: 'Optional[int]' = None, opaque: 'bytes' = b'', random: 'int' = 0, rhash_len: 'Optional[int]' = None, @@ -3358,7 +3396,10 @@ def _make_param_puzzle(self, code: 'Enum_Parameter', param: 'Optional[Data_Puzzl param: parameter data version: HIP protocol version index: #K index - lifetime: lifetime + lifetime: lifetime, encoded to the nearest lower ``Lifetime`` octet; + ignored when ``lifetime_exponent`` is given + lifetime_exponent: the ``Lifetime`` octet itself, ``value`` in + :rfc:`7401#section-5.2.4`'s ``2^(value - 32)`` seconds opaque: opaque data random: random #I value rhash_len: on-wire width of ``Random #I``, in bits; defaults to the @@ -3369,16 +3410,24 @@ def _make_param_puzzle(self, code: 'Enum_Parameter', param: 'Optional[Data_Puzzl Returns: HIP parameter schema. + Raises: + ProtocolError: If ``lifetime_exponent`` does not fit one octet, or + ``lifetime`` cannot be encoded; see :meth:`_make_puzzle_lifetime`. + """ if param is not None: index = param.index - lifetime = self._make_puzzle_lifetime(code, version, param.lifetime) + lifetime_exponent = param.lifetime_exponent opaque = param.opaque random = param.random if rhash_len is None: rhash_len = param.rhash_len - else: - lifetime = self._make_puzzle_lifetime(code, version, lifetime) + + if lifetime_exponent is None: + lifetime_exponent = self._make_puzzle_lifetime(code, version, lifetime) + elif not 0 <= lifetime_exponent <= 0xFF: + raise ProtocolError(f'HIPv{version}: [ParamNo {code}] invalid lifetime: ' + f'{lifetime_exponent} is outside the one-octet Lifetime field') return Schema_PuzzleParameter( type=code, @@ -3387,7 +3436,7 @@ def _make_param_puzzle(self, code: 'Enum_Parameter', param: 'Optional[Data_Puzzl # spells the same quantity ``4 + RHASH_len / 8``. len=4 + self._make_puzzle_field_width(code, version, rhash_len, random), index=index, - lifetime=lifetime, + lifetime=lifetime_exponent, opaque=opaque, random=random, ) diff --git a/pcapkit/protocols/schema/internet/hip.py b/pcapkit/protocols/schema/internet/hip.py index 677a9ad81a..f50391f7a1 100644 --- a/pcapkit/protocols/schema/internet/hip.py +++ b/pcapkit/protocols/schema/internet/hip.py @@ -90,6 +90,8 @@ class ControlsType(TypedDict): #: Anonymous. anonymous: bool + #: Reserved bits. + reserved: int class LocatorFlags(TypedDict): """Locator flags.""" @@ -517,8 +519,10 @@ def __init__(self, type: 'Enum_Parameter', len: 'int', value: 'bytes') -> 'None' class ESPInfoParameter(Parameter, code=Enum_Parameter.ESP_INFO): """Header schema for HIP ``ESP_INFO`` parameters.""" - #: Reserved. - reserved: 'bytes' = PaddingField(length=2) + #: Reserved, "zero when sent, ignored when received" (:rfc:`7402#section-5.1.1`). + #: Read as a value rather than as padding, so that re-serialising a parsed + #: parameter reproduces the octets it arrived with. + reserved: 'int' = UInt16Field() #: Key management index. index: 'int' = UInt16Field() #: Old SPI. @@ -529,7 +533,7 @@ class ESPInfoParameter(Parameter, code=Enum_Parameter.ESP_INFO): padding: 'bytes' = PaddingField(length=parameter_padding_len) if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', len: 'int', index: 'int', + def __init__(self, type: 'Enum_Parameter', len: 'int', reserved: 'int', index: 'int', old_spi: 'int', new_spi: 'int') -> 'None': ... @@ -1567,6 +1571,7 @@ class HIP(Schema): control: 'ControlsType' = BitField( length=2, namespace={ + 'reserved': (0, 15), 'anonymous': (15, 1), }, ) diff --git a/tests/protocols/internet/test_hip_roundtrip_unit.py b/tests/protocols/internet/test_hip_roundtrip_unit.py new file mode 100644 index 0000000000..25e477f67f --- /dev/null +++ b/tests/protocols/internet/test_hip_roundtrip_unit.py @@ -0,0 +1,108 @@ +# -*- coding: utf-8 -*- +"""HIP rebuilds parsed packets byte for byte, and sizes raw parameters correctly. + +GitHub issues from the internet-v4 round-trip audit (#1202): + +* #1313 -- :meth:`HIP.make ` neither + padded nor validated a raw ``bytes`` parameter, so a 12-octet one gave Header + Length ``5`` over 52 packed octets. +* #1314 -- a packet with no parameters has no ``parameters`` attribute, and + :meth:`~pcapkit.protocols.protocol.ProtocolBase.from_data` raised + :exc:`AttributeError` on it. +* #1323 -- the ``PUZZLE`` ``Lifetime`` octet was kept only as a + :class:`~datetime.timedelta`, so 197 of the 256 octets could not be rebuilt. +* #1328 -- the reserved bits of ``Controls`` and of ``ESP_INFO`` were written + back as zero. + +Every case builds its own octets in memory and reads no capture. :class:`HIP` +is imported inside each test, after :func:`~tests._support.reimport_once_per_class`, +for the reason :mod:`tests.protocols.link.test_ethernet_mac_roundtrip_unit` gives. + +""" + +import io +import unittest + +from tests._support import reimport_once_per_class + + +def packet(params: bytes = b'', controls: bytes = b'\x00\x00') -> bytes: + """Return a HIPv2 I1 packet carrying ``params`` (already 8-octet aligned).""" + return (bytes([59, 4 + len(params) // 8, 0x01, 0x21]) + b'\x00\x00' + + controls + bytes(32) + params) + + +class TestHIPRoundTrip(unittest.TestCase): + """Pin the rebuilt octets of parsed HIP packets.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def assertRoundTrips(self, wire: bytes) -> None: # pylint: disable=invalid-name + from pcapkit.protocols.internet.hip import HIP + + parsed = HIP(io.BytesIO(wire), len(wire), extension=True) + self.assertEqual(HIP.from_data(parsed.info).data.hex(), wire.hex()) + + def test_raw_parameter_is_padded_and_counted(self) -> None: + """#1313: a raw parameter is padded to 8 octets, and Header Length counts it.""" + from pcapkit.protocols.internet.hip import HIP + + for raw, padded in ((b'\x00\x41\x00\x08' + bytes(8), 16), + (b'\x00\x41\x00\x03' + bytes(3), 8), + (b'\x00\x41\x00\x04' + bytes(4), 8)): + with self.subTest(raw=raw.hex()): + wire = HIP.__new__(HIP).make(parameters=[raw]).pack() + self.assertEqual(len(wire), 40 + padded) + self.assertEqual((wire[1] + 1) * 8, len(wire)) + self.assertEqual(wire[40:], raw + bytes(padded - len(raw))) + + def test_parameterless_packet_round_trips(self) -> None: + """#1314: Header Length 4 and no parameters rebuilds unchanged.""" + self.assertRoundTrips(packet()) + + def test_every_puzzle_lifetime_octet_round_trips(self) -> None: + """#1323: all 256 ``Lifetime`` octets survive, including 0-11 and 79-255.""" + for octet in range(256): + with self.subTest(octet=octet): + self.assertRoundTrips(packet( + bytes.fromhex('0101000c01') + bytes([octet]) + b'op' + bytes(8))) + + def test_puzzle_lifetime_is_derived_from_the_octet(self) -> None: + """#1323: ``lifetime`` is informational, ``None`` beyond ``timedelta.max``.""" + import datetime + + from pcapkit.const.hip.parameter import Parameter + from pcapkit.protocols.internet.hip import HIP + from pcapkit.utilities.exceptions import ProtocolError + + for octet, lifetime in ((32, datetime.timedelta(seconds=1)), + (0, datetime.timedelta(0)), + (78, datetime.timedelta(seconds=2 ** 46)), + (79, None), (255, None)): + with self.subTest(octet=octet): + wire = packet(bytes.fromhex('0101000c01') + bytes([octet]) + b'op' + bytes(8)) + param = HIP(io.BytesIO(wire), len(wire), extension=True).info.parameters[Parameter.PUZZLE] + self.assertEqual(param.lifetime_exponent, octet) + self.assertEqual(param.lifetime, lifetime) + + proto = HIP.__new__(HIP) + self.assertEqual(proto._make_param_puzzle( + Parameter.PUZZLE, version=2, lifetime_exponent=7, random=1 << 63).lifetime, 7) + with self.assertRaises(ProtocolError): + proto._make_param_puzzle(Parameter.PUZZLE, version=2, lifetime_exponent=256) + + def test_controls_reserved_bits_round_trip(self) -> None: + """#1328: Controls ``fffe`` and ``ffff`` rebuild unchanged.""" + for controls in (b'\xff\xfe', b'\xff\xff', b'\x80\x01'): + with self.subTest(controls=controls.hex()): + self.assertRoundTrips(packet(controls=controls)) + + def test_esp_info_reserved_round_trips(self) -> None: + """#1328: the two ``ESP_INFO`` reserved octets rebuild unchanged.""" + self.assertRoundTrips(packet(bytes.fromhex('0041000c' 'abcd' '0102') + + bytes.fromhex('0a0b0c0d' '01020304'))) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/internet/test_hip_unit.py b/tests/protocols/internet/test_hip_unit.py index c83e5cae77..0176e96449 100644 --- a/tests/protocols/internet/test_hip_unit.py +++ b/tests/protocols/internet/test_hip_unit.py @@ -32,7 +32,7 @@ def test_hip_index_length_alias_and_make_data(self) -> None: type=Packet.I1, version=2, chksum=b'\x12\x34', - control=DummyDict(anonymous=True), + control=DummyDict(anonymous=True, reserved=0), shit=b'source-hit', rhit=b'recv-hit', parameters=['param'], @@ -88,7 +88,7 @@ def test_hip_read_make_post_init_and_parameter_list_edges(self) -> None: pkt={'bit_0': 0, 'type': Packet.I1}, ver={'bit_1': 1, 'version': 2}, checksum=b'\x12\x34', - control={'anonymous': 1}, + control={'anonymous': 1, 'reserved': 0}, shit=1, rhit=2, param=[], @@ -178,6 +178,7 @@ def test_hip_read_make_post_init_and_parameter_list_edges(self) -> None: type=Parameter.ESP_INFO, critical=True, length=16, + reserved=0, index=8, old_spi=9, new_spi=10, @@ -204,6 +205,7 @@ def test_hip_read_make_post_init_and_parameter_list_edges(self) -> None: type=Parameter.ESP_INFO, critical=True, length=16, + reserved=0, index=1, old_spi=2, new_spi=3, @@ -215,7 +217,7 @@ def test_hip_read_make_post_init_and_parameter_list_edges(self) -> None: pkt={'bit_0': 0, 'type': Packet.I1}, ver={'bit_1': 1, 'version': 2}, checksum=b'\x12\x34', - control={'anonymous': 1}, + control={'anonymous': 1, 'reserved': 0}, shit=1, rhit=2, param=[], @@ -1425,8 +1427,9 @@ def test_hip_parameter_constructors_cover_data_model_and_default_paths(self) -> self.assertEqual(proto._make_param_puzzle( Parameter.PUZZLE, hip_data.PuzzleParameter(type=Parameter.PUZZLE, critical=False, - length=16, index=1, lifetime=dt1, - opaque=b'op', random=5, rhash_len=64), + length=16, index=1, lifetime_exponent=32, + lifetime=dt1, opaque=b'op', random=5, + rhash_len=64), version=2, ).random, 5) self.assertEqual(proto._make_param_solution( @@ -2837,10 +2840,10 @@ def test_hip_puzzle_lifetime_guard_raises_an_in_library_error(self) -> None: :class:`~pcapkit.utilities.exceptions.EnumError` would not do: it is ``EnumError(BaseError, TypeError)``, so it would silently stop being caught. - Reachable from conformant input rather than only from a crafted one: a - ``Lifetime`` octet of ``0x00`` is a legal encoding of ``2^-32`` seconds, - which :class:`~datetime.timedelta` rounds to zero, so parsing an ordinary - PUZZLE and re-emitting it lands here. + A parsed PUZZLE no longer reaches the guard: its ``Lifetime`` octet is + kept as ``lifetime_exponent`` and written back as is (#1323), so an + octet of ``0x00`` -- ``2^-32`` seconds, which :class:`~datetime.timedelta` + rounds to zero -- re-serialises unchanged. """ from pcapkit.const.hip.parameter import Parameter @@ -2882,16 +2885,14 @@ def test_hip_puzzle_lifetime_guard_raises_an_in_library_error(self) -> None: self.assertIn('invalid lifetime', str(caught.exception)) # A PUZZLE whose Lifetime octet is 0x00, parsed then re-serialised: the - # conformant-input path. On the unfixed tree this was a bare ValueError. + # octet is carried, not re-encoded from the zero timedelta (#1323). wire = bytes.fromhex('0101' '000c' '01' '00' '6f70' '8000000000000000' '00000000') parsed = proto._read_param_puzzle( hip_schema.PuzzleParameter.unpack(wire), version=2, options=options, ) self.assertEqual(parsed.lifetime, datetime.timedelta(0)) - with self.assertRaises(ProtocolError) as caught: - proto._make_param_puzzle(Parameter.PUZZLE, parsed, version=2) - self.assertIsInstance(caught.exception, BaseError) + self.assertEqual(proto._make_param_puzzle(Parameter.PUZZLE, parsed, version=2).lifetime, 0) # A positive lifetime still encodes exactly as it did before. self.assertEqual(proto._make_param_puzzle(