diff --git a/pcapkit/protocols/internet/ipv6_route.py b/pcapkit/protocols/internet/ipv6_route.py index adb4179ef1..1292b024e7 100644 --- a/pcapkit/protocols/internet/ipv6_route.py +++ b/pcapkit/protocols/internet/ipv6_route.py @@ -259,6 +259,7 @@ def make(self, type_reversed: 'bool' = False, seg_left: 'int' = 0, data: 'bytes | Data_IPv6_Route | Schema_RoutingType | dict[str, Any]' = b'\x00\x00\x00\x00', + hdr_ext_len: 'Optional[int]' = None, payload: 'ProtocolBase | Schema | bytes' = b'', **kwargs: 'Any') -> 'Schema_IPv6_Route': """Make (construct) packet data. @@ -275,6 +276,8 @@ def make(self, type_reversed: If the namespace of routing type is reversed. seg_left: Segments left. data: Routing data. + hdr_ext_len: ``Hdr Ext Len`` field value; computed from ``data`` + when omitted. payload: Payload of current instance. **kwargs: Arbitrary keyword arguments. @@ -336,6 +339,13 @@ def make(self, else: raise ProtocolError(f'{self.alias}: invalid routing data type: {data.__class__}') + # NOTE: An explicit ``hdr_ext_len`` is written as given, so that a + # ``from_data`` rebuild of a header the capture cut short keeps the + # length the sender declared instead of the one that was recorded + # (:issue:`1457`), the same as IPv6's ``payload_length``. + if hdr_ext_len is not None: + length = hdr_ext_len + return Schema_IPv6_Route( next=next_val, length=length, @@ -450,6 +460,11 @@ def _make_data(cls, data: 'Data_IPv6_Route') -> 'dict[str, Any]': # type: ignor 'type': data.type, 'seg_left': data.seg_left, 'data': data, + # NOTE: ``data.length`` is the parsed header length in octets, + # passed through as ``Hdr Ext Len`` rather than left to ``make`` to + # recompute, so that a truncated capture rebuilds with the length + # that was on the wire (:issue:`1457`). + 'hdr_ext_len': (data.length - 8) // 8, 'payload': cls._make_payload(data), } @@ -784,6 +799,7 @@ def _make_data_type_rpl(self, type: 'Enum_Routing', route: 'Optional[Data_RPL]' Raises: FieldValueError: If an entry of ``ip`` is a :obj:`bool` (c.f. :func:`~pcapkit.corekit.fields.ipaddress.parse_ip_address`). + ProtocolError: If ``ip`` is empty. """ if route is not None: @@ -792,8 +808,20 @@ def _make_data_type_rpl(self, type: 'Enum_Routing', route: 'Optional[Data_RPL]' pad = route.pad reserved = route.reserved padding = getattr(route, 'padding', b'') + + # NOTE: A parsed address is decoded to its full 16 octets whenever + # the header sat inside IPv6, the ``dst`` prefix restored, so it is + # compressed again here: ``Addresses[n]`` by ``CmprE``, every other + # one by ``CmprI``. Which item is ``Addresses[n]`` comes from the + # count ``Hdr Ext Len`` declares (:rfc:`6554#section-4.2`), since a + # capture cut inside the list holds fewer; one it cut short is left + # as the raw :obj:`bytes` read and written back as is. + area = route.length - 8 - pad + count = max(area - (16 - cmpr_e), 0) // (16 - cmpr_i) + 1 + last = max(len(route.ip), count) - 1 ip_val = [ - addr if isinstance(addr, bytes) else addr.packed for addr in route.ip + addr if isinstance(addr, bytes) else addr.packed[cmpr_e if index == last else cmpr_i:] + for index, addr in enumerate(route.ip) ] else: ip = [] if ip is None else ip @@ -806,6 +834,12 @@ def _make_data_type_rpl(self, type: 'Enum_Routing', route: 'Optional[Data_RPL]' # address of ``00000001`` with ``cmpr_e=0``. descr = f'{self.alias}: invalid RPL source address' + # NOTE: :rfc:`6554#section-4.2` counts ``n`` as ``(...) + 1``, so + # a header always carries ``Addresses[n]``; one with none would not + # parse back. + if not ip: + raise ProtocolError(f'{self.alias}: [TypeNo {type}] RPL source route needs at least one address') + if dst is None: pad = 0 cmpr_i = 0 @@ -821,7 +855,13 @@ def _make_data_type_rpl(self, type: 'Enum_Routing', route: 'Optional[Data_RPL]' else: test_list.append(cast('IPv6Address', parse_ip_address(item, descr, version=6)).packed) prefix_i = os_path.commonprefix(test_list) - cmpr_i = len(prefix_i) + + # NOTE: ``CmprI`` and ``CmprE`` are 4-bit fields + # (:rfc:`6554#section-3`), so a prefix of all 16 octets -- an + # address equal to ``dst``, or no ``Addresses[1..n-1]`` at all, + # which leaves ``dst`` alone in the list -- is elided as 15. With + # no ``Addresses[1..n-1]``, ``CmprI`` governs nothing and is 0. + cmpr_i = min(len(prefix_i), 15) if len(ip) > 1 else 0 test_list = [dst.packed] if isinstance(ip[-1], bytes): @@ -829,7 +869,7 @@ def _make_data_type_rpl(self, type: 'Enum_Routing', route: 'Optional[Data_RPL]' else: test_list.append(cast('IPv6Address', parse_ip_address(ip[-1], descr, version=6)).packed) prefix_e = os_path.commonprefix(test_list) - cmpr_e = len(prefix_e) + cmpr_e = min(len(prefix_e), 15) # NOTE: the outer ``% 8`` keeps an already 8-octet-aligned # vector from being handed a *full* 8 octets of padding diff --git a/pcapkit/protocols/schema/internet/ipv6_route.py b/pcapkit/protocols/schema/internet/ipv6_route.py index 0027c8d7eb..0ef04791dd 100644 --- a/pcapkit/protocols/schema/internet/ipv6_route.py +++ b/pcapkit/protocols/schema/internet/ipv6_route.py @@ -215,6 +215,24 @@ class RPL(RoutingType, code=Enum_Routing.RPL_Source_Route_Header): #: Padding. padding: 'bytes' = PaddingField(length=lambda pkt: pkt['pad']['pad_len']) + @classmethod + def pre_unpack(cls, packet: 'dict[str, Any]') -> 'None': + """Prepare ``packet`` data for unpacking process. + + Args: + packet: packet data + + """ + # NOTE: The declared span -- ``4 + Hdr Ext Len * 8`` octets inside + # IPv6-Route -- is kept for :meth:`post_process`, which splits + # ``Addresses[1..n]`` by it. By then ``__length__`` has been counted + # down to zero, and the octets actually read are fewer than declared + # whenever the capture ends inside the address list. :meth:`pack` calls + # this too, before it has set ``__length__``; the packing path does not + # need the span, so nothing is kept there. + if '__length__' in packet: + packet['__rpl_length__'] = packet['__length__'] + def post_process(self, packet: 'dict[str, Any]') -> 'Schema': """Revise ``schema`` data after unpacking process. @@ -270,39 +288,43 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': ilen = 16 - cmpr_i elen = 16 - cmpr_e + + # NOTE: The address list is split by the span ``Hdr Ext Len`` declares, + # not by the octets actually read: once a capture ends inside the list + # the two differ, and splitting by the latter decodes the octets of + # ``Addresses[1]`` as ``Addresses[n]`` (or a short run as an IPv4 + # address). The declared span is what :attr:`addresses`'s own + # ``length`` callback asked for -- ``pad_len`` already subtracted, the + # trailing padding octets being read by :attr:`padding` -- and a + # malformed one (no whole number of addresses) is rejected by + # :meth:`~pcapkit.protocols.internet.ipv6_route.IPv6_Route._read_data_type_rpl`, + # so it is only clamped here. + declared = packet.get('__rpl_length__', len(buffer) + 4 + self.pad['pad_len']) + area = max(declared - 4 - self.pad['pad_len'], 0) + widths = [ilen] * (max(area - elen, 0) // ilen) + [elen] + + # NOTE: Only an item read at its full width is decoded. One the capture + # cut short stays the raw :obj:`bytes` that were read -- not prefixed + # with ``dst``, so a rebuild writes back exactly those octets -- and + # one cut off entirely is not listed at all. addr = [] # type: list[IPv6Address | bytes] counter = 0 + for index, width in enumerate(widths): + buf = buffer[counter:counter + width] + counter += width + if not buf: + break + if len(buf) < width: + addr.append(buf) + break - # Addresses[1..n-1] - # - # NOTE: ``buffer`` is ``self.addresses``, whose own ``length`` callback - # above already subtracted ``pad_len`` -- the trailing padding octets - # are read by :attr:`padding`, not by this field. Subtracting - # ``pad_len`` a *second* time here would drop one address for every - # ``ilen`` octets of padding, so a padded (i.e. compressed) header - # would parse one address short. - for _ in range((len(buffer) - elen) // ilen): - buf = buffer[counter:counter + ilen] - if dst is None: - if cmpr_i == 0: - addr.append(cast('IPv6Address', ipaddress.ip_address(buf))) - else: - addr.append(buf) - else: - buf = dst[:cmpr_i] + buf - addr.append(cast('IPv6Address', ipaddress.ip_address(buf))) - counter += ilen - - # Addresses[n] - buf = buffer[counter:counter + elen] - if dst is None: - if cmpr_e == 0: + cmpr = cmpr_e if index == len(widths) - 1 else cmpr_i + if dst is not None: + addr.append(cast('IPv6Address', ipaddress.ip_address(dst[:cmpr] + buf))) + elif cmpr == 0: addr.append(cast('IPv6Address', ipaddress.ip_address(buf))) else: addr.append(buf) - else: - buf = dst[:cmpr_e] + buf - addr.append(cast('IPv6Address', ipaddress.ip_address(buf))) self.ip = addr return self diff --git a/tests/protocols/internet/test_ipv6_route_rpl_roundtrip_unit.py b/tests/protocols/internet/test_ipv6_route_rpl_roundtrip_unit.py new file mode 100644 index 0000000000..7731430764 --- /dev/null +++ b/tests/protocols/internet/test_ipv6_route_rpl_roundtrip_unit.py @@ -0,0 +1,153 @@ +# -*- coding: utf-8 -*- +"""An RPL source route header (IPv6-Route type 3) rebuilds byte for byte. + +GitHub issue #1456: inside IPv6, the parser restores the ``dst`` prefix of each +compressed address, and :meth:`IPv6_Route._make_data_type_rpl +` wrote +the decoded address back at its full 16 octets, so a header with ``CmprI`` or +``CmprE`` above zero grew on rebuild. + +GitHub issue #1457: a header the capture cut inside its address list was split +by the octets read rather than by ``Hdr Ext Len``, so a short run decoded as an +IPv4 address or raised a builtins :exc:`ValueError`, and the rebuild wrote a +smaller ``Hdr Ext Len``. + +A cut that leaves ``Pad`` octets unread still rebuilds with them zero-filled; +that is the corekit short-read defect, #1458, so those cases only check that +the octets read come back first and ``Hdr Ext Len`` is kept. + +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 ipaddress +import unittest +import warnings + +from tests._support import reimport_once_per_class + +SRC = ipaddress.IPv6Address('2001:db8::1').packed +DST = ipaddress.IPv6Address('2001:db8:aa:bb:cc:dd:ee:ff').packed + + +def ipv6(route: bytes) -> bytes: + """Wrap ``route`` in an IPv6 header from ``SRC`` to ``DST``.""" + return (bytes([0x60, 0, 0, 0]) + len(route).to_bytes(2, 'big') + bytes([43, 64]) + + SRC + DST + route) + + +def rpl(cmpr_i: int, cmpr_e: int, count: int, extra_pad: int = 0) -> bytes: + """Build an RPL header carrying ``count`` addresses.""" + area = (count - 1) * (16 - cmpr_i) + (16 - cmpr_e) + pad = (-area) % 8 + extra_pad + body = bytes((i * 7 + 3) % 256 for i in range(area)) + hdr_ext_len = (area + pad) // 8 + return bytes([59, hdr_ext_len, 3, count, (cmpr_i << 4) | cmpr_e, pad << 4, 0, 0]) + body + bytes(pad) + + +def headers(): + """Every ``CmprI``/``CmprE`` pair, one to three addresses, minimal and widened ``Pad``.""" + for cmpr_i in range(16): + for cmpr_e in range(16): + for count in (1, 2, 3): + minimal = (-((count - 1) * (16 - cmpr_i) + (16 - cmpr_e))) % 8 + for extra in (0, 8): + if minimal + extra <= 15: + yield (cmpr_i, cmpr_e, count, extra), rpl(cmpr_i, cmpr_e, count, extra) + + +class TestIPv6RouteRPLRoundTrip(unittest.TestCase): + """Pin the rebuild of RPL source route headers, on their own and inside IPv6.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + warnings.simplefilter('ignore') + self.addCleanup(warnings.resetwarnings) + + def rebuild(self, cls, raw: bytes) -> bytes: + return cls.from_data(cls(raw, len(raw)).info).data + + def test_issue_1456_compressed_header_inside_ipv6(self) -> None: + from pcapkit.protocols.internet.ipv6 import IPv6 + + raw = ipv6(bytes([59, 2, 3, 1, 0x88, 0, 0, 0]) + bytes(range(1, 17))) + self.assertEqual(len(raw), 64) + self.assertEqual(self.rebuild(IPv6, raw), raw) + + def test_issue_1457_cut_address_is_kept_as_read(self) -> None: + from pcapkit.protocols.internet.ipv6_route import IPv6_Route + + route = bytes([59, 2, 3, 1, 0, 0, 0, 0]) + bytes(range(1, 17)) + for cut in (12, 23): + with self.subTest(cut=cut): + info = IPv6_Route(route[:cut], cut).info + self.assertEqual(info.ip, (route[8:cut],)) + self.assertEqual(self.rebuild(IPv6_Route, route[:cut]), route[:cut]) + + def test_issue_1457_cut_keeps_hdr_ext_len(self) -> None: + from pcapkit.protocols.internet.ipv6_route import IPv6_Route + + route = bytes([59, 2, 3, 1, 0x88, 0, 0, 0]) + bytes(range(1, 17)) + self.assertEqual(self.rebuild(IPv6_Route, route[:12]), route[:12]) + + def test_every_compression_rebuilds(self) -> None: + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.internet.ipv6_route import IPv6_Route + + for label, route in headers(): + with self.subTest(case=label): + self.assertEqual(self.rebuild(IPv6_Route, route), route) + self.assertEqual(self.rebuild(IPv6, ipv6(route)), ipv6(route)) + + def test_every_cut_rebuilds(self) -> None: + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.internet.ipv6_route import IPv6_Route + + # Two addresses put a cut in both ``Addresses[1..n-1]`` and ``Addresses[n]``. + for label, route in headers(): + if label[2:] != (2, 0): + continue + pad = route[5] >> 4 + for cut in range(8, len(route)): + with self.subTest(case=label, cut=cut): + alone = self.rebuild(IPv6_Route, route[:cut]) + inner = self.rebuild(IPv6, ipv6(route)[:40 + cut]) + if pad == 0: + self.assertEqual(alone, route[:cut]) + self.assertEqual(inner, ipv6(route)[:40 + cut]) + else: # unread ``Pad`` octets are zero-filled (#1458) + self.assertEqual(alone[:cut], route[:cut]) + self.assertEqual(inner[:40 + cut], ipv6(route)[:40 + cut]) + self.assertEqual(alone[1], route[1]) + + def test_make_parse_rebuilds(self) -> None: + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.internet.ipv6_route import IPv6_Route + + dst = ipaddress.IPv6Address(DST) + for shared in (0, 4, 8, 15, 16): + for count in (1, 2, 3): + addrs = [ipaddress.IPv6Address(DST[:shared] + bytes([0x40 + j]) * (16 - shared)) + for j in range(count)] + with self.subTest(shared=shared, count=count): + made = IPv6_Route(type=3, dst=dst, data={'ip': addrs}, seg_left=count).data + self.assertEqual(self.rebuild(IPv6_Route, made), made) + packet = ipv6(made) + self.assertEqual(self.rebuild(IPv6, packet), packet) + route = list(IPv6(packet, len(packet)).extension_headers.values())[0].info + self.assertEqual(list(route.ip), addrs) + self.assertLessEqual(route.cmpr_e, 15) + + def test_make_without_addresses_is_rejected(self) -> None: + from pcapkit.protocols.internet.ipv6_route import IPv6_Route + from pcapkit.utilities.exceptions import ProtocolError + + for dst in (None, ipaddress.IPv6Address(DST)): + with self.subTest(dst=dst): + with self.assertRaisesRegex(ProtocolError, 'needs at least one address'): + IPv6_Route(type=3, dst=dst, data={'ip': []}) + + +if __name__ == '__main__': + unittest.main()