fix(ipv6-route): rebuild RPL source routes byte for byte (#1456, #1457)

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,15 +855,21 @@ 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):
                     test_list.append(ip[-1])
                 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

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

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
+<pcapkit.protocols.internet.ipv6_route.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()