Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 8 additions & 2 deletions pcapkit/protocols/data/internet/hip.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand Down
18 changes: 17 additions & 1 deletion pcapkit/protocols/internet/hip.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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']),
Expand Down Expand Up @@ -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,
Expand All @@ -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``.
Expand Down Expand Up @@ -674,6 +678,7 @@ def make(self,
ver = {
'bit_1': 1,
'version': version,
'reserved': version_reserved,
},
checksum=checksum,
control = {
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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),
)
Expand Down Expand Up @@ -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),
)
Expand Down Expand Up @@ -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.
Expand All @@ -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:
Expand All @@ -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 []
Expand All @@ -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,
)
Expand Down Expand Up @@ -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.
Expand All @@ -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:
Expand All @@ -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 []
Expand All @@ -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,
)
Expand Down
11 changes: 9 additions & 2 deletions pcapkit/protocols/schema/internet/hip.py
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,8 @@ class VersionType(TypedDict):

#: Version.
version: int
#: Reserved bits.
reserved: int
#: Reversed bit.
bit_1: Literal[1]

Expand Down Expand Up @@ -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':
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -1565,6 +1571,7 @@ class HIP(Schema):
length=1,
namespace={
'version': (0, 4),
'reserved': (4, 3),
'bit_1': (7, 1),
},
)
Expand Down
118 changes: 118 additions & 0 deletions tests/protocols/internet/test_hip_reserved_bits_unit.py
Original file line number Diff line number Diff line change
@@ -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()
Loading
Loading