diff --git a/pcapkit/protocols/data/internet/hip.py b/pcapkit/protocols/data/internet/hip.py index 6eb8490c50..45585dd306 100644 --- a/pcapkit/protocols/data/internet/hip.py +++ b/pcapkit/protocols/data/internet/hip.py @@ -97,6 +97,9 @@ class HIP(Protocol): type: 'Packet' #: Version. version: 'int' + #: Reserved bits -- the three bits between ``Version`` and the fixed + #: low-order bit, carried verbatim so that re-serialising reproduces them. + reserved: 'int' #: Checksum. chksum: 'bytes' #: Control @@ -110,7 +113,7 @@ class HIP(Protocol): #: HIP parameters. parameters: 'OrderedMultiDict[Enum_Parameter, Parameter]' - def __init__(self, next: 'TransType', length: 'int', type: 'Packet', version: 'int', chksum: 'bytes', control: 'Control', shit: 'int', rhit: 'int') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', version: 'int', reserved: 'int', chksum: 'bytes', control: 'Control', shit: 'int', rhit: 'int') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long @info_final @@ -641,9 +644,12 @@ class Flags(Data): symmetric: 'bool' #: Must follow flag. must_follow: 'bool' + #: Reserved bits -- the 14 low-order bits of ``Flags``, carried verbatim + #: so that re-serialising reproduces them. + reserved: 'int' if TYPE_CHECKING: - def __init__(self, symmetric: 'bool', must_follow: 'bool') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long + def __init__(self, symmetric: 'bool', must_follow: 'bool', reserved: '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 8935ebd50e..b810dd35b6 100644 --- a/pcapkit/protocols/internet/hip.py +++ b/pcapkit/protocols/internet/hip.py @@ -18,7 +18,7 @@ 2 16 Reserved (``\\x00``) 2 17 ``hip.type`` Packet Type 3 24 ``hip.version`` Version - 3 28 Reserved + 3 28 ``hip.reserved`` Reserved 3 31 Reserved (``\\x01``) 4 32 ``hip.chksum`` Checksum 6 48 ``hip.control`` Controls @@ -582,6 +582,7 @@ def read(self, length: 'Optional[int]' = None, *, extension: bool = False, **kwa length=schema.len * 8 + 8, type=Enum_Packet(schema.pkt['type']), version=schema.ver['version'], + reserved=schema.ver['reserved'], chksum=schema.checksum, control=Data_Control( anonymous=bool(schema.control['anonymous']), @@ -616,6 +617,7 @@ def make(self, packet_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long packet_reversed: 'bool' = False, version: 'int' = 2, + version_reserved: 'int' = 0, checksum: 'bytes' = b'\x00\x00', controls_anonymous: 'bool' = False, controls_reserved: 'int' = 0, @@ -636,6 +638,8 @@ def make(self, packet_namespace: Namespace for HIP packet type. packet_reversed: If the HIP packet type is reversed. version: HIP version. + version_reserved: The 3 reserved bits between ``Version`` and the + fixed low-order bit. checksum: Checksum. controls_anonymous: If the sender is anonymous. controls_reserved: The 15 reserved high-order bits of ``Controls``. @@ -674,6 +678,7 @@ def make(self, ver = { 'bit_1': 1, 'version': version, + 'reserved': version_reserved, }, checksum=checksum, control = { @@ -775,6 +780,7 @@ def _make_data(cls, data: 'Data_HIP') -> 'dict[str, Any]': # type: ignore[overr 'next': data.next, 'packet': data.type, 'version': data.version, + 'version_reserved': data.reserved, 'checksum': data.chksum, 'controls_anonymous': data.control.anonymous, 'controls_reserved': data.control.reserved, @@ -2378,6 +2384,7 @@ def _read_param_route_dst(self, schema: 'Schema_RouteDstParameter', *, version: flags=Data_Flags( symmetric=bool(schema.flags['symmetric']), must_follow=bool(schema.flags['must_follow']), + reserved=schema.flags['reserved'], ), hit=tuple(schema.hit), ) @@ -2827,6 +2834,7 @@ def _read_param_route_via(self, schema: 'Schema_RouteViaParameter', *, version: flags=Data_Flags( symmetric=bool(schema.flags['symmetric']), must_follow=bool(schema.flags['must_follow']), + reserved=schema.flags['reserved'], ), hit=tuple(schema.hit), ) @@ -4668,6 +4676,7 @@ def _make_param_route_dst(self, code: 'Enum_Parameter', param: 'Optional[Data_Ro version: 'int', symmetric: 'bool' = False, must_follow: 'bool' = False, + flags_reserved: 'int' = 0, hit: 'Optional[list[bytes | str | int | IPv6Address]]' = None, **kwargs: 'Any') -> 'Schema_RouteDstParameter': """Make HIP ``ROUTE_DST`` parameter. @@ -4678,6 +4687,7 @@ def _make_param_route_dst(self, code: 'Enum_Parameter', param: 'Optional[Data_Ro version: HIP protocol version symmetric: symmetric flag must_follow: must-follow flag + flags_reserved: the 14 reserved low-order bits of ``Flags`` hit: list of HITs Returns: @@ -4687,6 +4697,7 @@ def _make_param_route_dst(self, code: 'Enum_Parameter', param: 'Optional[Data_Ro if param is not None: symmetric = param.flags.symmetric must_follow = param.flags.must_follow + flags_reserved = param.flags.reserved hit_list = cast('list[bytes | str | int | IPv6Address]', param.hit) else: hit_list = hit if hit is not None else [] @@ -4697,6 +4708,7 @@ def _make_param_route_dst(self, code: 'Enum_Parameter', param: 'Optional[Data_Ro flags={ 'symmetric': int(symmetric), 'must_follow': int(must_follow), + 'reserved': flags_reserved, }, hit=hit_list, ) @@ -5042,6 +5054,7 @@ def _make_param_route_via(self, code: 'Enum_Parameter', param: 'Optional[Data_Ro version: 'int', symmetric: 'bool' = False, must_follow: 'bool' = False, + flags_reserved: 'int' = 0, hit: 'Optional[list[IPv6Address | bytes | str | int]]' = None, **kwargs: 'Any') -> 'Schema_RouteViaParameter': """Make HIP ``ROUTE_VIA`` parameter. @@ -5052,6 +5065,7 @@ def _make_param_route_via(self, code: 'Enum_Parameter', param: 'Optional[Data_Ro version: HIP protocol version symmetric: symmetric flag must_follow: must-follow flag + flags_reserved: the 14 reserved low-order bits of ``Flags`` hit: list of HITs Returns: @@ -5061,6 +5075,7 @@ def _make_param_route_via(self, code: 'Enum_Parameter', param: 'Optional[Data_Ro if param is not None: symmetric = param.flags.symmetric must_follow = param.flags.must_follow + flags_reserved = param.flags.reserved hit_list = cast('list[IPv6Address | bytes | str | int]', param.hit) else: hit_list = hit if hit is not None else [] @@ -5071,6 +5086,7 @@ def _make_param_route_via(self, code: 'Enum_Parameter', param: 'Optional[Data_Ro flags={ 'symmetric': int(symmetric), 'must_follow': int(must_follow), + 'reserved': flags_reserved, }, hit=hit_list, ) diff --git a/pcapkit/protocols/schema/internet/hip.py b/pcapkit/protocols/schema/internet/hip.py index 5f3c93fe6f..c1755e3cb4 100644 --- a/pcapkit/protocols/schema/internet/hip.py +++ b/pcapkit/protocols/schema/internet/hip.py @@ -82,6 +82,8 @@ class VersionType(TypedDict): #: Version. version: int + #: Reserved bits. + reserved: int #: Reversed bit. bit_1: Literal[1] @@ -116,6 +118,8 @@ class RouteFlags(TypedDict): symmetric: int #: Must-follow flag. must_follow: int + #: Reserved bits. + reserved: int def locator_value_selector(pkt: 'dict[str, Any]') -> 'Field': @@ -1300,8 +1304,9 @@ class RouteDstParameter(Parameter, code=Enum_Parameter.ROUTE_DST): #: Flags. flags: 'RouteFlags' = BitField(length=2, namespace={ - 'symmetric': (0, 1), - 'must_follow': (1, 1), + 'symmetric': (0, 1), + 'must_follow': (1, 1), + 'reserved': (2, 14), }) #: Reserved. reserved: 'bytes' = PaddingField(length=2) @@ -1474,6 +1479,7 @@ class RouteViaParameter(Parameter, code=Enum_Parameter.ROUTE_VIA): flags: 'RouteFlags' = BitField(length=2, namespace={ 'symmetric': (0, 1), 'must_follow': (1, 1), + 'reserved': (2, 14), }) #: Reserved. reserved: 'bytes' = PaddingField(length=2) @@ -1565,6 +1571,7 @@ class HIP(Schema): length=1, namespace={ 'version': (0, 4), + 'reserved': (4, 3), 'bit_1': (7, 1), }, ) diff --git a/tests/protocols/internet/test_hip_reserved_bits_unit.py b/tests/protocols/internet/test_hip_reserved_bits_unit.py new file mode 100644 index 0000000000..f7ab4f417c --- /dev/null +++ b/tests/protocols/internet/test_hip_reserved_bits_unit.py @@ -0,0 +1,118 @@ +# -*- coding: utf-8 -*- +"""HIP keeps the reserved bits of the ``Version`` octet and the route ``Flags``. + +GitHub issue #1417. Three HIP ``BitField`` namespaces left bits unnamed, so those +bits were packed as zero on rebuild (start, length): + +* ``HIP.ver``: (4, 3) -- the ``RES`` bits between ``Version`` and the fixed + low-order ``1`` (:rfc:`7401#section-5.1`); +* ``RouteDstParameter.flags`` and ``RouteViaParameter.flags``: (2, 14) -- the + reserved bits after ``S`` and ``M`` (:rfc:`6028#section-5`). + +The fix names a ``reserved`` sub-field at each site, per the #654 per-site +policy, and carries it through the data model and ``make()``. + +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 unittest + +from tests._support import reimport_once_per_class + +#: Header fields shared by every HIP packet built here. +HIP_BASE = { + 'next': 59, 'packet': 1, 'version': 2, 'checksum': b'\x00\x00', + 'controls_anonymous': False, 'shit': 0, 'rhit': 0, 'payload': b'', +} + +#: Offset of the ``Version`` octet in the HIP header. +VER_OFFSET = 3 + +#: Offset of the first parameter's two-octet ``Flags``: the 40-octet HIP header, +#: then the parameter's ``Type`` and ``Length``. +FLAGS_OFFSET = 40 + 4 + + +class TestHIPReservedBits(unittest.TestCase): + """Pin the reserved bits of ``Version`` and route ``Flags`` across rebuilds.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _route_codes(self): # type: ignore[no-untyped-def] + from pcapkit.const.hip.parameter import Parameter + + return (Parameter.ROUTE_DST, Parameter.ROUTE_VIA) + + def _frame(self, code=None, *, ver=None, flags=None): # type: ignore[no-untyped-def] + """Build a HIP packet, then overwrite its ``Version`` or route ``Flags``.""" + from pcapkit.protocols.internet.hip import HIP + + params = None if code is None else [(code, {'hit': ['2001:db8::1']})] + octets = bytearray(HIP(parameters=params, extension=True, **HIP_BASE).data) + if ver is not None: + self.assertEqual(octets[VER_OFFSET], 0x21) + octets[VER_OFFSET] = ver + if flags is not None: + self.assertEqual(octets[FLAGS_OFFSET:FLAGS_OFFSET + 2], b'\x00\x00') + octets[FLAGS_OFFSET:FLAGS_OFFSET + 2] = flags.to_bytes(2, 'big') + return bytes(octets) + + def test_from_data_rebuilds_version_reserved(self) -> None: + from pcapkit.protocols.internet.hip import HIP + + for ver in (0x21, 0x23, 0x2f, 0x2b, 0xff): + with self.subTest(ver=hex(ver)): + frame = self._frame(ver=ver) + parsed = HIP(frame, len(frame), extension=True) + self.assertEqual(parsed.info.version, ver >> 4) + self.assertEqual(parsed.info.reserved, (ver >> 1) & 0b111) + self.assertEqual(HIP.from_data(parsed.info).data, frame) + + def test_make_writes_version_reserved(self) -> None: + from pcapkit.protocols.internet.hip import HIP + + built = HIP(version_reserved=0b101, extension=True, **HIP_BASE).data + self.assertEqual(built[VER_OFFSET], 0x2b) + parsed = HIP(built, len(built), extension=True) + self.assertEqual(parsed.info.reserved, 0b101) + self.assertEqual(HIP.from_data(parsed.info).data, built) + + default = HIP(extension=True, **HIP_BASE).data + self.assertEqual(default[VER_OFFSET], 0x21) + + def test_from_data_rebuilds_route_flags_reserved(self) -> None: + from pcapkit.protocols.internet.hip import HIP + + for code in self._route_codes(): + for flags in (0x0000, 0x3fff, 0x8001, 0x4000, 0xffff, 0x2aaa): + with self.subTest(code=code.name, flags=hex(flags)): + frame = self._frame(code, flags=flags) + parsed = HIP(frame, len(frame), extension=True) + data = parsed.info.parameters[code].flags + self.assertEqual(data.symmetric, bool(flags & 0x8000)) + self.assertEqual(data.must_follow, bool(flags & 0x4000)) + self.assertEqual(data.reserved, flags & 0x3fff) + self.assertEqual(HIP.from_data(parsed.info).data, frame) + + def test_make_writes_route_flags_reserved(self) -> None: + from pcapkit.protocols.internet.hip import HIP + + for code in self._route_codes(): + with self.subTest(code=code.name): + built = HIP(parameters=[(code, { + 'symmetric': True, 'flags_reserved': 0x1234, 'hit': ['2001:db8::1'], + })], extension=True, **HIP_BASE).data + self.assertEqual(built[FLAGS_OFFSET:FLAGS_OFFSET + 2], b'\x92\x34') + parsed = HIP(built, len(built), extension=True) + self.assertEqual(parsed.info.parameters[code].flags.reserved, 0x1234) + self.assertEqual(HIP.from_data(parsed.info).data, built) + + default = HIP(parameters=[(code, {'hit': ['2001:db8::1']})], + extension=True, **HIP_BASE).data + self.assertEqual(default[FLAGS_OFFSET:FLAGS_OFFSET + 2], b'\x00\x00') + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/internet/test_hip_unit.py b/tests/protocols/internet/test_hip_unit.py index b6006a2869..e33b6edce4 100644 --- a/tests/protocols/internet/test_hip_unit.py +++ b/tests/protocols/internet/test_hip_unit.py @@ -31,6 +31,7 @@ def test_hip_index_length_alias_and_make_data(self) -> None: next=TransType.TCP, type=Packet.I1, version=2, + reserved=0, chksum=b'\x12\x34', control=DummyDict(anonymous=True, reserved=0), shit=b'source-hit', @@ -57,6 +58,7 @@ def test_hip_index_length_alias_and_make_data(self) -> None: self.assertEqual(values['next'], TransType.TCP) self.assertEqual(values['packet'], Packet.I1) self.assertEqual(values['version'], 2) + self.assertEqual(values['version_reserved'], 0) self.assertEqual(values['checksum'], b'\x12\x34') self.assertEqual(values['controls_anonymous'], True) self.assertEqual(values['shit'], b'source-hit') @@ -86,7 +88,7 @@ def test_hip_read_make_post_init_and_parameter_list_edges(self) -> None: next=TransType.UDP, len=4, pkt={'bit_0': 0, 'type': Packet.I1}, - ver={'bit_1': 1, 'version': 2}, + ver={'bit_1': 1, 'version': 2, 'reserved': 0}, checksum=b'\x12\x34', control={'anonymous': 1, 'reserved': 0}, shit=1, @@ -110,7 +112,7 @@ def test_hip_read_make_post_init_and_parameter_list_edges(self) -> None: next=TransType.UDP, len=4, pkt={'bit_0': 1, 'type': Packet.I1}, - ver={'bit_1': 1, 'version': 2}, + ver={'bit_1': 1, 'version': 2, 'reserved': 0}, checksum=b'\x00\x00', control={'anonymous': 0}, shit=0, @@ -125,7 +127,7 @@ def test_hip_read_make_post_init_and_parameter_list_edges(self) -> None: next=TransType.UDP, len=4, pkt={'bit_0': 0, 'type': Packet.I1}, - ver={'bit_1': 0, 'version': 2}, + ver={'bit_1': 0, 'version': 2, 'reserved': 0}, checksum=b'\x00\x00', control={'anonymous': 0}, shit=0, @@ -215,7 +217,7 @@ def test_hip_read_make_post_init_and_parameter_list_edges(self) -> None: next=TransType.UDP, len=5, pkt={'bit_0': 0, 'type': Packet.I1}, - ver={'bit_1': 1, 'version': 2}, + ver={'bit_1': 1, 'version': 2, 'reserved': 0}, checksum=b'\x12\x34', control={'anonymous': 1, 'reserved': 0}, shit=1, @@ -763,7 +765,7 @@ def test_hip_parameter_readers_cover_simple_models_and_guards(self) -> None: ).id, 99) self.assertTrue(proto._read_param_route_dst( hip_schema.RouteDstParameter(type=Parameter.ROUTE_DST, len=20, - flags={'symmetric': 1, 'must_follow': 1}, + flags={'symmetric': 1, 'must_follow': 1, 'reserved': 0}, hit=[ip_address('2001:db8::20')]), version=2, options=options, @@ -820,7 +822,7 @@ def test_hip_parameter_readers_cover_simple_models_and_guards(self) -> None: ).ttl, datetime.timedelta(seconds=30)) self.assertTrue(proto._read_param_route_via( hip_schema.RouteViaParameter(type=Parameter.ROUTE_VIA, len=20, - flags={'symmetric': 1, 'must_follow': 0}, + flags={'symmetric': 1, 'must_follow': 0, 'reserved': 0}, hit=[ip_address('2001:db8::40')]), version=2, options=options, @@ -889,7 +891,7 @@ def test_hip_parameter_readers_cover_simple_models_and_guards(self) -> None: hip_schema.AckDataParameter(type=Parameter.ACK_DATA, len=5, ack=[1])), (proto._read_param_route_dst, hip_schema.RouteDstParameter(type=Parameter.ROUTE_DST, len=5, - flags={'symmetric': 1, 'must_follow': 1}, + flags={'symmetric': 1, 'must_follow': 1, 'reserved': 0}, hit=[])), (proto._read_param_hip_transport_mode, hip_schema.HIPTransportModeParameter(type=Parameter.HIP_TRANSPORT_MODE, @@ -907,7 +909,7 @@ def test_hip_parameter_readers_cover_simple_models_and_guards(self) -> None: hip_schema.OverlayTTLParameter(type=Parameter.OVERLAY_TTL, len=3, ttl=30)), (proto._read_param_route_via, hip_schema.RouteViaParameter(type=Parameter.ROUTE_VIA, len=5, - flags={'symmetric': 1, 'must_follow': 0}, + flags={'symmetric': 1, 'must_follow': 0, 'reserved': 0}, hit=[])), (proto._read_param_from, hip_schema.FromParameter(type=Parameter.FROM, len=15, @@ -1702,7 +1704,7 @@ def test_hip_parameter_constructors_cover_data_model_and_default_paths(self) -> critical=False, length=8, id=99), version=2, ).id, 99) - flags = hip_data.Flags(symmetric=True, must_follow=True) + flags = hip_data.Flags(symmetric=True, must_follow=True, reserved=0) self.assertTrue(proto._make_param_route_dst( Parameter.ROUTE_DST, hip_data.RouteDstParameter(type=Parameter.ROUTE_DST,