fix(ipv6)!: rebuild a cut extension header chain byte-exact (#1471, #1473)

pcapkit/protocols/internet/ipv6.py

@@ -392,6 +392,21 @@ def _make_data(cls, data: 'Data_IPv6') -> 'dict[str, Any]':  # type: ignore[over
             'trailer': data.get('trailer', b''),
         }
 
+    @staticmethod
+    def _exthdr_key(alias: 'str') -> 'str':
+        """Key an extension header's info is written under in :class:`Data_IPv6`.
+
+        Args:
+            alias: :attr:`~pcapkit.protocols.protocol.ProtocolBase.alias` of
+                the extension header's parser.
+
+        Returns:
+            The key; :meth:`str.lstrip` strips a character *set*, which every
+            parser's alias is chosen to suit.
+
+        """
+        return alias.lstrip('IPv6-').lower()
+
     @classmethod
     def _lookup_exthdr(cls, data: 'Data_IPv6') -> 'tuple[tuple[str, ...], tuple[tuple[Type[ProtocolBase], Any], ...]]':  # pylint: disable=line-too-long
         """Recover the extension header chain of ``data`` without ``__exthdr__``.
@@ -412,6 +427,9 @@ def _lookup_exthdr(cls, data: 'Data_IPv6') -> 'tuple[tuple[str, ...], tuple[tupl
         predecessor, and ``data.items(multi=True)`` yields each of them in turn.
 
         """
+        from pcapkit.protocols.internet.ipv6_ext import \
+            IPv6_Ext  # isort: skip # pylint: disable=import-outside-toplevel
+
         names = []  # type: list[str]
         chain = []  # type: list[tuple[Type[ProtocolBase], Any]]
         proto = data.get('next')
@@ -429,8 +447,19 @@ def _lookup_exthdr(cls, data: 'Data_IPv6') -> 'tuple[tuple[str, ...], tuple[tupl
             # carrying ``next`` ends the chain, and so does a fragment header.
             if 'next' not in info:
                 break
+
+            # NOTE: A header whose dedicated parser raised was parsed by
+            # :class:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext` instead,
+            # c.f. :meth:`_import_next_layer`, and written under that class's
+            # key rather than the dedicated one's. Rebuilding it with the
+            # dedicated class would look for fields only that class parses
+            # (:issue:`1473`), so the key decides the parser.
+            if key == cls._exthdr_key(IPv6_Ext.alias.fget(None)):  # type: ignore[attr-defined]
+                parser = IPv6_Ext  # type: Type[ProtocolBase]
+            else:
+                parser = cls._lookup_next_layer(cls.__proto__, int(ex_proto))
             names.append(key)
-            chain.append((cls._lookup_next_layer(cls.__proto__, int(ex_proto)), info))
+            chain.append((parser, info))
 
             proto = info['next']
             if ex_proto == Enum_ExtensionHeader.IPv6_Frag:
@@ -528,7 +557,7 @@ def _decode_next_layer(self, ipv6: 'Data_IPv6', proto: 'Optional[int]' = None,
             # keeps its predecessor's info rather than overwriting it, and
             # :meth:`Info.to_dict <pcapkit.corekit.infoclass.Info.to_dict>`
             # writes every one of them, in wire order (:issue:`1453`).
-            name = next_.alias.lstrip('IPv6-').lower()
+            name = self._exthdr_key(next_.alias)
             ipv6.__update__(OrderedMultiDict([(name, info)]))
 
             # record protocol name

pcapkit/utilities/decorators.py

@@ -149,7 +149,16 @@ def behold(*args: 'P.args', **kwargs: 'P.kwargs') -> 'R_beholder':
             # turning a next-layer parse failure that should have degraded to
             # Raw into a crash. Reachable once something is registered on an
             # SCTP payload protocol identifier, which NGAP is.
-            file_ = self._get_payload()
+            #
+            # A ``payload`` passed to the decorated method is the octets that
+            # method was asked to parse, and wins: IPv6 hands its upper layer
+            # the payload *past* the extension headers, so falling back to the
+            # whole IPv6 payload would hold those headers twice, once in the
+            # extension header chain and once more in this ``Raw``
+            # (:issue:`1471`).
+            file_ = cast('Optional[bytes]', kwargs.get('payload'))
+            if file_ is None:
+                file_ = self._get_payload()
 
             # NOTE: ``alias=proto`` matches what ``_import_next_layer`` passes, so
             # a payload that failed to parse still reports the code it arrived

tests/protocols/internet/test_ipv6_exthdr_cut_rebuild_unit.py

@@ -0,0 +1,118 @@
+# -*- coding: utf-8 -*-
+"""A cut IPv6 extension header chain rebuilds byte-exact, from info and dict.
+
+GitHub issue #1471: an upper layer cut short after an extension header fell
+back to :class:`~pcapkit.protocols.misc.raw.Raw` over the *whole* IPv6 payload,
+since :func:`~pcapkit.utilities.decorators.beholder` ignored the ``payload`` it
+was handed, so the rebuild wrote the extension headers twice -- Destination
+Options, TCP cut at 57 octets rebuilt 65.
+
+GitHub issue #1473: a header whose dedicated parser raised is parsed by
+:class:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext` and written under its
+``ext`` key, but the ``to_dict`` rebuild chose the dedicated class from the
+``next`` code, which then missed fields only it parses -- ``AttributeError:
+'HOPOPT' object has no attribute 'options'``.
+
+Every case builds its own octets and reads no capture. Classes are imported
+inside each test, after :func:`~tests._support.reimport_once_per_class`.
+
+"""
+
+import struct
+import unittest
+import warnings
+
+from tests._support import reimport_once_per_class
+
+HOP, TCP, DST, AH, NONXT, HIP = 0, 6, 60, 51, 59, 139
+
+#: A TCP header and one octet of data.
+SEGMENT = struct.pack('!HHIIBBHHH', 1, 2, 0, 0, 0x50, 2, 0, 0, 0) + b'x'
+
+
+def opts(nxt: int) -> bytes:
+    """An 8-octet Hop-by-Hop or Destination Options header: one PadN."""
+    return bytes([nxt, 0, 1, 4, 0, 0, 0, 0])
+
+
+def auth(nxt: int) -> bytes:
+    """A 24-octet Authentication Header with a 12-octet ICV."""
+    return bytes([nxt, 4, 0, 0]) + bytes(20)
+
+
+def hip(nxt: int) -> bytes:
+    """A 40-octet HIPv2 header with no parameters."""
+    return bytes([nxt, 4, 1, 0x21, 0, 0, 0, 0]) + bytes(32)
+
+
+#: First next-header code and IPv6 payload of each chain.
+CHAINS = {
+    'dst-tcp': (DST, opts(TCP) + SEGMENT),
+    'hop-tcp': (HOP, opts(TCP) + SEGMENT),
+    'hop-dst-tcp': (HOP, opts(DST) + opts(TCP) + SEGMENT),
+    'dst-ah-tcp': (DST, opts(AH) + auth(TCP) + SEGMENT),
+    'hop-nonxt': (HOP, bytes([NONXT, 1, 1, 12]) + bytes(12)),
+    'hip': (HIP, hip(NONXT)),
+    'dst-hip': (DST, opts(HIP) + hip(NONXT)),
+}
+
+
+def packet(name: str) -> bytes:
+    """The whole IPv6 packet of ``CHAINS[name]``."""
+    code, body = CHAINS[name]
+    return struct.pack('!IHBB', 6 << 28, len(body), code, 64) + bytes(32) + body
+
+
+class TestIPv6ExtensionHeaderCutRebuild(unittest.TestCase):
+    """Pin the rebuild of an extension header chain cut at every octet."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+        quiet = warnings.catch_warnings()
+        quiet.__enter__()  # pylint: disable=unnecessary-dunder-call
+        self.addCleanup(quiet.__exit__, None, None, None)
+        warnings.simplefilter('ignore')
+
+    def test_every_cut_rebuilds_from_info_and_dict(self) -> None:
+        from pcapkit.protocols.internet.ipv6 import IPv6
+        from pcapkit.protocols.link.ethernet import Ethernet
+
+        for name in CHAINS:
+            full = packet(name)
+            for cut in range(40, len(full) + 1):
+                for cls, data in ((IPv6, full[:cut]), (Ethernet, bytes(12) + b'\x86\xdd' + full[:cut])):
+                    proto = cls(data, len(data))
+                    with self.subTest(chain=name, cut=cut, layer=cls.__name__, form='parse'):
+                        self.assertEqual(proto.data, data)
+                    with self.subTest(chain=name, cut=cut, layer=cls.__name__, form='info'):
+                        self.assertEqual(cls.from_data(proto.info).data, data)
+                    with self.subTest(chain=name, cut=cut, layer=cls.__name__, form='dict'):
+                        self.assertEqual(cls.from_data(proto.info.to_dict()).data, data)
+
+    def test_raw_upper_layer_starts_past_the_extension_headers(self) -> None:
+        from pcapkit.protocols.internet.ipv6 import IPv6
+        from pcapkit.protocols.misc.raw import Raw
+
+        data = packet('dst-tcp')[:57]
+        ip6 = IPv6(data, len(data))
+        self.assertIsInstance(ip6.payload, Raw)
+        self.assertEqual(ip6.payload.data, data[48:])
+        self.assertEqual(len(IPv6.from_data(ip6.info).data), 57)
+
+    def test_dict_rebuild_keeps_the_generic_parser(self) -> None:
+        from pcapkit.const.ipv6.extension_header import ExtensionHeader
+        from pcapkit.protocols.internet.ipv6 import IPv6
+        from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext
+
+        for cut in (48, 50, 52, 55):
+            data = packet('hop-nonxt')[:cut]
+            with self.subTest(cut=cut):
+                ip6 = IPv6(data, len(data))
+                self.assertIs(type(ip6._exthdr[ExtensionHeader.HOPOPT]), IPv6_Ext)
+                _, chain = IPv6._lookup_exthdr(ip6.info.to_dict())
+                self.assertEqual([parser for parser, _ in chain], [IPv6_Ext])
+                self.assertEqual(IPv6.from_data(ip6.info.to_dict()).data, data)
+
+
+if __name__ == '__main__':
+    unittest.main()