fix(protocols): dissect an IPv6 extension header behind a first fragment as without it (#1539)

pcapkit/protocols/internet/ipv6.py

@@ -45,7 +45,7 @@
 if TYPE_CHECKING:
     from enum import IntEnum as StdlibEnum
     from ipaddress import IPv6Address
-    from typing import Any, Optional, Type
+    from typing import Any, Mapping, Optional, Type
 
     from aenum import IntEnum as AenumEnum
     from typing_extensions import Literal
@@ -436,15 +436,19 @@ def _lookup_exthdr(cls, data: 'Data_IPv6') -> 'tuple[tuple[str, ...], tuple[tupl
         nested = cls._lookup_nested(data)
         pairs = data.items(multi=True) if isinstance(data, (Info, MultiDict)) else data.items()
         for (key, info) in pairs:
-            if key not in nested:
+            # NOTE: ``fragment`` is declared only for type checkers, so it reads
+            # as nested, but it is the packet split at the Fragment header and
+            # sits between that header and the next one.
+            if key not in nested or key == 'fragment':
                 continue
             try:
                 ex_proto = Enum_ExtensionHeader(proto)
             except ValueError:
                 break
 
             # NOTE: As in :meth:`_decode_next_layer`, a code with no parser
-            # carrying ``next`` ends the chain, and so does a fragment header.
+            # carrying ``next`` ends the chain, and so does a Fragment header
+            # other than a first fragment's.
             if 'next' not in info:
                 break
 
@@ -462,10 +466,27 @@ def _lookup_exthdr(cls, data: 'Data_IPv6') -> 'tuple[tuple[str, ...], tuple[tupl
             chain.append((parser, info))
 
             proto = info['next']
-            if ex_proto == Enum_ExtensionHeader.IPv6_Frag:
+            if ex_proto == Enum_ExtensionHeader.IPv6_Frag and not cls._is_first_fragment(info):
                 break
         return tuple(names), tuple(chain)
 
+    @staticmethod
+    def _is_first_fragment(info: 'Mapping[str, Any]') -> 'bool':
+        """Whether a Fragment header's info is that of a first fragment.
+
+        Args:
+            info: info of a Fragment header, as parsed or from :meth:`Info.to_dict
+                <pcapkit.corekit.infoclass.Info.to_dict>`
+
+        Returns:
+            :data:`True` at a Fragment Offset of zero. A header whose dedicated
+            parser raised was parsed by
+            :class:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext` instead and
+            has no offset, so it does not count as one.
+
+        """
+        return info.get('offset') == 0
+
     def _read_ip_hextet(self) -> 'tuple[int, int, int]':
         """Read the first four octets of IPv6.
 
@@ -511,6 +532,7 @@ def _decode_next_layer(self, ipv6: 'Data_IPv6', proto: 'Optional[int]' = None,
         raw_len = ipv6.payload if length is None else length  # payload length
         _protos = []                # ProtoChain buffer
         _exthdr = []                # (parser class, info) per extension header
+        split = None                # type: Optional[tuple[int, int]] # at first Fragment header
 
         # traverse if next header is an extension header
         payload = self.__header__.get_payload()
@@ -582,15 +604,28 @@ def _decode_next_layer(self, ipv6: 'Data_IPv6', proto: 'Optional[int]' = None,
 
             # keep original data after fragment header
             if ex_proto == Enum_ExtensionHeader.IPv6_Frag:
-                ipv6.__update__({
-                    'fragment': self._read_packet(header=hdr_len, payload=raw_len),
-                })
-                break
+                # NOTE: The first Fragment header splits the packet for
+                # reassembly, so ``fragment``, ``hdr_len`` and ``raw_len`` keep
+                # that split however far the walk goes on.
+                if split is None:
+                    split = (hdr_len, raw_len)
+                    ipv6.__update__({
+                        'fragment': self._read_packet(header=hdr_len, payload=raw_len),
+                    })
+
+                # NOTE: Only a first fragment goes on. :rfc:`8200#section-4.5`
+                # puts every header through the upper-layer one in the fragment
+                # at offset zero, so what follows its Fragment header is walked
+                # exactly as it would be without one (:issue:`1539`). A later
+                # fragment carries a slice of the Fragmentable Part from its
+                # offset on, and that slice starts with no header at all.
+                if not self._is_first_fragment(info):
+                    break
 
         # record real header & payload length (headers exclude)
         ipv6.__update__({
-            'hdr_len': hdr_len,
-            'raw_len': raw_len,
+            'hdr_len': hdr_len if split is None else split[0],
+            'raw_len': raw_len if split is None else split[1],
 
             # update next header
             'protocol': proto,

tests/protocols/internet/test_ipv6_ext_after_fragment_unit.py

@@ -0,0 +1,166 @@
+# -*- coding: utf-8 -*-
+"""An IPv6 extension header behind a first fragment dissects as without it.
+
+GitHub issue #1539: :class:`~pcapkit.protocols.internet.ipv6.IPv6` ended its
+extension header walk at every Fragment header, and handed what followed to the
+upper-layer dispatch. That dispatch neither passes ``extension=True`` nor
+``version=6``, so Shim6 failed :class:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext`'s
+version gate, and a zero-filled HIP header or a Routing header of type 3 raised
+with no :class:`IPv6_Ext` to stand in. Each became a plain
+:class:`~pcapkit.protocols.misc.raw.Raw` leaf, so the TCP behind it was never
+dissected, though it was without the Fragment header.
+
+:rfc:`8200#section-4.5` puts every header through the upper-layer one in the
+first fragment, at offset zero, so the walk now goes on past its Fragment
+header. A later fragment starts with no header at all, and still ends the walk.
+The packet split at the Fragment header, which reassembly takes, is unchanged.
+
+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
+import warnings
+
+from tests._support import reimport_once_per_class
+
+#: Source and destination addresses.
+ADDRS = (bytes.fromhex('20010db8000000000000000000000001')
+         + bytes.fromhex('20010db8000000000000000000000002'))
+
+#: A 20-octet TCP header and one octet of payload.
+TCP = bytes.fromhex('9c40000900000001000000005000ffff0000000078')
+
+#: Extension headers that :class:`IPv6_Ext` ends up parsing, by name: Next
+#: Header code, and the 8-octet header with TCP as its Next Header.
+HEADERS = {
+    'Shim6': (140, bytes.fromhex('0600000112345678')),
+    'HIP (zero-filled)': (139, bytes.fromhex('0600000000000000')),
+    'Routing type 3': (43, bytes.fromhex('0600030000000000')),
+}
+
+#: The two frames of the issue, as IPv6 datagrams: a Shim6 header behind a
+#: first fragment, and the same Shim6 header with no Fragment header.
+ISSUE_FRAGMENTED = bytes.fromhex(
+    '6000000000252c40' + ADDRS.hex() + '8c00000112345678' + '0600000000000000' + TCP.hex())
+ISSUE_DIRECT = bytes.fromhex('60000000001d8c40' + ADDRS.hex() + '0600000000000000' + TCP.hex())
+
+
+def datagram(code: int, payload: bytes) -> bytes:
+    """An IPv6 datagram with Next Header ``code`` and ``payload``."""
+    return (bytes.fromhex('60000000') + len(payload).to_bytes(2, 'big')
+            + bytes((code, 64)) + ADDRS + payload)
+
+
+def fragment(code: int, offset: int, more: bool = False) -> bytes:
+    """A Fragment header: Next Header ``code``, at ``offset`` 8-octet units."""
+    return bytes((code, 0)) + (offset * 8 + more).to_bytes(2, 'big') + bytes.fromhex('9c400009')
+
+
+class TestIPv6ExtAfterFragment(unittest.TestCase):
+    """Pin the walk past a first fragment, and its end at a later one."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+        # the zero-filled HIP header warns as it is parsed and rebuilt
+        quiet = warnings.catch_warnings()
+        quiet.__enter__()  # pylint: disable=unnecessary-dunder-call
+        self.addCleanup(quiet.__exit__, None, None, None)
+        warnings.filterwarnings('ignore', message='packet length < 0')
+
+    def _parse(self, data: bytes) -> 'object':
+        from pcapkit.protocols.internet.ipv6 import IPv6
+
+        ipv6 = IPv6(data)
+        self.assertEqual(ipv6.data, data)
+        return ipv6
+
+    def assert_rebuilds(self, ipv6: 'object') -> None:
+        """``ipv6`` rebuilds byte for byte from its info and its info as a dict."""
+        from pcapkit.protocols.internet.ipv6 import IPv6
+
+        self.assertEqual(IPv6.from_data(ipv6.info).data, ipv6.data)
+        self.assertEqual(IPv6.from_data(ipv6.info.to_dict()).data, ipv6.data)
+
+    def test_issue_frames(self) -> None:
+        direct = self._parse(ISSUE_DIRECT)
+        self.assertEqual(str(direct.protochain), 'IPv6:IPv6-Ext:TCP:Raw')
+
+        fragmented = self._parse(ISSUE_FRAGMENTED)
+        self.assertIn('TCP', fragmented)
+        self.assertEqual(str(fragmented.protochain), 'IPv6:IPv6-Frag:IPv6-Ext:TCP:Raw')
+        self.assert_rebuilds(fragmented)
+
+    def test_direct_forms_reach_tcp(self) -> None:
+        from pcapkit.const.reg.transtype import TransType
+
+        for name, (code, header) in HEADERS.items():
+            with self.subTest(header=name):
+                ipv6 = self._parse(datagram(code, header + TCP))
+                self.assertIn('TCP', ipv6)
+                self.assertEqual(ipv6.info.protocol, TransType.TCP)
+                self.assert_rebuilds(ipv6)
+
+    def test_first_fragment_dissects_as_without_it(self) -> None:
+        from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext
+        from pcapkit.protocols.internet.ipv6_frag import IPv6_Frag
+
+        for name, (code, header) in HEADERS.items():
+            direct = self._parse(datagram(code, header + TCP))
+            for more in (False, True):
+                with self.subTest(header=name, more=more):
+                    ipv6 = self._parse(datagram(44, fragment(code, 0, more) + header + TCP))
+                    self.assertIn('TCP', ipv6)
+                    exthdr = list(ipv6.extension_headers.values())
+                    self.assertEqual([type(ext) for ext in exthdr], [IPv6_Frag, IPv6_Ext])
+                    # ``error`` holds an exception, which compares by identity
+                    self.assertEqual(repr(exthdr[1].info),
+                                     repr(next(iter(direct.extension_headers.values())).info))
+                    self.assertEqual(ipv6.info.protocol, direct.info.protocol)
+                    self.assertEqual(ipv6.payload.info, direct.payload.info)
+                    self.assert_rebuilds(ipv6)
+
+    def test_first_fragment_walks_a_dedicated_header(self) -> None:
+        # Destination Options reached TCP behind a first fragment before, but as
+        # an upper layer holding it, not as an extension header beside it
+        from pcapkit.protocols.internet.ipv6_frag import IPv6_Frag
+        from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts
+
+        opts = bytes.fromhex('0600010400000000')
+        direct = self._parse(datagram(60, opts + TCP))
+        ipv6 = self._parse(datagram(44, fragment(60, 0) + opts + TCP))
+        self.assertEqual([type(ext) for ext in ipv6.extension_headers.values()],
+                         [IPv6_Frag, IPv6_Opts])
+        self.assertEqual(ipv6.info.protocol, direct.info.protocol)
+        self.assertEqual(ipv6.payload.info, direct.payload.info)
+        self.assert_rebuilds(ipv6)
+
+    def test_first_fragment_keeps_the_split(self) -> None:
+        # reassembly takes the split at the Fragment header, c.f.
+        # ``pcapkit.toolkit.pcap.ipv6_reassembly``
+        code, header = HEADERS['Shim6']
+        data = datagram(44, fragment(code, 0, True) + header + TCP)
+        ipv6 = self._parse(data)
+        self.assertEqual(ipv6.info.hdr_len, 48)
+        self.assertEqual(ipv6.info.raw_len, len(header + TCP))
+        self.assertEqual(ipv6.info.fragment.header, data[:48])
+        self.assertEqual(ipv6.info.fragment.payload, header + TCP)
+
+    def test_later_fragment_ends_the_walk(self) -> None:
+        from pcapkit.protocols.internet.ipv6_frag import IPv6_Frag
+
+        for name, (code, header) in HEADERS.items():
+            with self.subTest(header=name):
+                data = datagram(44, fragment(code, 1) + header + TCP)
+                ipv6 = self._parse(data)
+                self.assertNotIn('TCP', ipv6)
+                exthdr = list(ipv6.extension_headers.values())
+                self.assertEqual([type(ext) for ext in exthdr], [IPv6_Frag])
+                self.assertEqual(int(ipv6.info.protocol), code)
+                self.assertEqual(ipv6.info.fragment.payload, header + TCP)
+                self.assert_rebuilds(ipv6)
+
+
+if __name__ == '__main__':
+    unittest.main()