Fix four IPv6 defects (#352, #353, #360, #369)

pcapkit/protocols/internet/ipv6_frag.py

@@ -136,9 +136,15 @@ def read(self, length: 'Optional[int]' = None, *, extension: 'bool' = False,  #
             length = len(self)
         schema = self.__header__
 
+        # NOTE: The on-wire Fragment Offset is a 13-bit count of 8-octet units
+        # (:rfc:`8200#section-4.5`), but ``Data_IPv6_Frag.offset`` carries octets
+        # so that it matches ``Data_IPv4.offset`` and can be handed straight to
+        # the reassembly machinery, which indexes the datagram buffer with it
+        # (:meth:`pcapkit.foundation.reassembly.ip.IP.reassembly`). Scale here,
+        # exactly as :meth:`pcapkit.protocols.internet.ipv4.IPv4.read` does.
         ipv6_frag = Data_IPv6_Frag(
             next=schema.next,
-            offset=schema.flags['offset'],
+            offset=int(schema.flags['offset']) * 8,
             mf=bool(schema.flags['mf']),
             id=schema.id,
         )
@@ -164,7 +170,10 @@ def make(self,
             next_default: Default value of next header.
             next_namespace: Namespace of next header.
             next_reversed: If the namespace of next header is reversed.
-            offset: Fragment offset.
+            offset: Fragment offset, in on-wire 8-octet units (:rfc:`8200#section-4.5`).
+                Note that :attr:`Data_IPv6_Frag.offset
+                <pcapkit.protocols.data.internet.ipv6_frag.IPv6_Frag.offset>` is in
+                octets, so it must be divided by 8 before being passed here.
             mf: More fragments flag.
             id: Identification.
             payload: Payload of current instance.
@@ -249,9 +258,12 @@ def _make_data(cls, data: 'Data_IPv6_Frag') -> 'dict[str, Any]':  # type: ignore
             Key-value pairs for protocol construction.
 
         """
+        # NOTE: ``make`` takes ``offset`` in on-wire 8-octet units, while
+        # ``data.offset`` is in octets (see :meth:`read`), so scale back down to
+        # keep the data-to-schema round trip exact.
         return {
             'next': data.next,
-            'offset': data.offset,
+            'offset': data.offset // 8,
             'mf': data.mf,
             'id': data.id,
             'payload': cls._make_payload(data),

pcapkit/protocols/schema/internet/ipv6.py

@@ -38,10 +38,18 @@ class IPv6(Schema):
     """Header schema for IPv6 packet."""
 
     #: Version, traffic class and flow label.
+    #:
+    #: The :class:`~pcapkit.corekit.fields.strings.BitField` namespace maps each
+    #: subfield to a ``(start_bit, length_in_bits)`` pair, *not* to
+    #: ``(start_bit, end_bit)``. Per :rfc:`8200#section-3` the first 32 bits of
+    #: the IPv6 header are Version (bits 0-3), Traffic Class (bits 4-11) and
+    #: Flow Label (bits 12-31), so the flow label starts at bit 12 -- reading it
+    #: as ``(8, 20)`` would overlap the low nibble of the traffic class and drop
+    #: the low nibble of the label.
     hextet: 'IPv6Hextet' = BitField(length=4, namespace={
         'version': (0, 4),
         'class': (4, 8),
-        'label': (8, 20),
+        'label': (12, 20),
     })
     #: Payload length.
     length: int = UInt16Field()

pcapkit/protocols/schema/internet/ipv6_opts.py

@@ -489,6 +489,13 @@ def post_process(self, packet: 'dict[str, Any]') -> 'QuickStartOption':
 class QuickStartOption(Option, EnumSchema[Enum_QSFunction]):
     """Header schema for IPv6-Opts quick start options."""
 
+    # NOTE: This declaration scopes the QS-function registry to this class. Without
+    # it the subclasses below register into the *parent* ``Option`` registry, whose
+    # keys are IPv6-Opts option *types* -- and since ``Quick_Start_Request`` is 0
+    # and ``Report_of_Approved_Rate`` is 8, they would clobber option types 0
+    # (``Pad1``) and 8 (``SMF_DPD``). ``hopopt.py`` carries the same declaration.
+    __enum__: 'DefaultDict[Enum_QSFunction, Type[QuickStartOption]]' = collections.defaultdict(lambda: None)  # type: ignore[arg-type,return-value]
+
     #: Flags.
     flags: 'QuickStartFlags' = BitField(length=1, namespace={
         'func': (0, 4),

pcapkit/toolkit/pcap.py

@@ -108,7 +108,7 @@ def ipv6_reassembly(frame: 'Frame') -> 'IP_Packet[IPv6Address] | None':
             bufid=(
                 ipv6_info.src,                              # source IP address
                 ipv6_info.dst,                              # destination IP address
-                ipv6_info.label,                            # label
+                ipv6_frag_info.id,                          # identification
                 ipv6_frag_info.next,                        # next header field in IPv6 Fragment Header
             ),
             num=frame.info.number,                          # original packet range number

pcapkit/toolkit/pcapng.py

@@ -112,7 +112,7 @@ def ipv6_reassembly(frame: 'PCAPNG') -> 'IP_Packet[IPv6Address] | None':
             bufid=(
                 ipv6_info.src,                              # source IP address
                 ipv6_info.dst,                              # destination IP address
-                ipv6_info.label,                            # label
+                ipv6_frag_info.id,                          # identification
                 ipv6_frag_info.next,                        # next header field in IPv6 Fragment Header
             ),
             num=frame_info.number,                          # original packet range number

pcapkit/toolkit/scapy.py

@@ -190,11 +190,14 @@ def ipv6_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv6Ad
                      ipaddress.ip_address(ipv6.src)),     # source IP address
                 cast('IPv6Address',
                      ipaddress.ip_address(ipv6.dst)),     # destination IP address
-                ipv6.fl,                                  # label
+                ipv6_frag.id,                             # identification
                 Enum_TransType.get(ipv6_frag.nh),      # next header field in IPv6 Fragment Header
             ),
             num=count,                                    # original packet range number
-            fo=ipv6_frag.offset,                          # fragment offset
+            # NOTE: Scapy reports ``IPv6ExtHdrFragment.offset`` in on-wire 8-octet
+            # units (:rfc:`8200#section-4.5`), but the reassembly machinery indexes
+            # the datagram buffer with ``fo``, so it must be scaled into octets.
+            fo=ipv6_frag.offset * 8,                      # fragment offset
             ihl=len(ipv6) - len(ipv6_frag),               # header length, only headers before IPv6-Frag
             mf=bool(ipv6_frag.m),                         # more fragment flag
             tl=len(ipv6),                                 # total length, header includes

tests/protocols/internet/test_ipv6_extension_runtime.py

@@ -112,7 +112,11 @@ def test_ipv6_fragment_extension_forbids_direct_payload_accessors(self) -> None:
         frag = list(ipv6.extension_headers.values())[0]
 
         self.assertEqual(str(frame.protochain), 'Ethernet:IPv6:IPv6-Frag:UDP:Raw')
-        self.assertEqual(frag.info.offset, 543)
+        # ``Data_IPv6_Frag.offset`` is in octets, matching ``Data_IPv4.offset``: the
+        # on-wire field is 543 counts of 8 octets (:rfc:`8200#section-4.5`), i.e. the
+        # 4344th octet of the fragmentable part, which with this fragment's 434
+        # octets of payload accounts for the full 4778 octet datagram.
+        self.assertEqual(frag.info.offset, 4344)
         self.assertFalse(frag.info.mf)
         self.assertEqual(frag.info.id, 110308)
         self.assertEqual(ipv6.info.raw_len, 434)

tests/protocols/internet/test_ipv6_extension_unit.py

@@ -111,7 +111,9 @@ def test_ipv6_frag_index_length_and_make_data(self) -> None:
         self.assertEqual(proto.__length_hint__(), 8)
         values = IPv6_Frag._make_data(data)
         self.assertEqual(values['next'], TransType.TCP)
-        self.assertEqual(values['offset'], 16)
+        # ``data.offset`` is in octets while ``make`` takes on-wire 8-octet units,
+        # so 16 octets is written back as 2 units
+        self.assertEqual(values['offset'], 2)
         self.assertEqual(values['mf'], True)
         self.assertEqual(values['id'], 99)
         self.assertIn('payload', values)
@@ -198,9 +200,15 @@ def test_ipv6_frag_read_make_and_properties(self) -> None:
         self.assertEqual(proto.protochain, ['TCP'])
         data = proto.read(extension=True)
         self.assertEqual(data.next, TransType.TCP)
-        self.assertEqual(data.offset, 12)
+        # the schema's ``offset`` is the raw 13-bit on-wire field, a count of
+        # 8-octet units (:rfc:`8200#section-4.5`); ``Data_IPv6_Frag.offset`` is in
+        # octets, like ``Data_IPv4.offset``, so 12 units becomes 96 octets
+        self.assertEqual(proto.__header__.flags['offset'], 12)
+        self.assertEqual(data.offset, 96)
         self.assertTrue(data.mf)
         self.assertEqual(data.id, 0x12345678)
+        # ``_make_data`` must undo the scaling, so that data -> make round trips
+        self.assertEqual(IPv6_Frag._make_data(data)['offset'], 12)
 
         with mock.patch.object(IPv6_Frag, '_decode_next_layer', return_value='decoded') as decode:
             self.assertEqual(proto.read(length=12), 'decoded')
@@ -1276,6 +1284,80 @@ def test_ipv6_route_schema_selector_and_rpl_post_process_branches(self) -> None:
         with_dst.post_process({'dst': ip_address('2001:db8::ffff')})
         self.assertEqual([str(item) for item in with_dst.ip], ['2001:db8::1', '2001:db8::2'])
 
+    def test_option_registries_are_not_clobbered_by_a_nested_enum_registry(self) -> None:
+        """Option type 0 must resolve to the padding option in every module.
+
+        An :class:`~pcapkit.protocols.schema.schema.EnumSchema` subclass that
+        does not declare its own ``__enum__`` shares its parent's registry, so
+        registering *its* subclasses writes into the parent's key space. The
+        quick-start options are keyed by
+        :class:`~pcapkit.const.ipv6.qs_function.QSFunction`, whose members are
+        ``0`` and ``8``, so a shared registry silently overwrites option types
+        ``0`` (``Pad1``) and ``8`` (``SMF_DPD``) -- and the failure is silent
+        because a missing ``__enum__`` is not an error, it just merges two
+        registries whose key spaces happen to overlap.
+
+        :rfc:`8200#section-4.2` defines ``Pad1`` for both the Hop-by-Hop Options
+        and the Destination Options header, so it must resolve in both.
+
+        """
+        from pcapkit.const.ipv6.option import Option as Enum_Option
+        from pcapkit.const.ipv6.qs_function import QSFunction as Enum_QSFunction
+        from pcapkit.protocols.schema.internet import hopopt as hopopt_schema
+        from pcapkit.protocols.schema.internet import ipv6_opts as opts_schema
+
+        # the two enum values that collide with option types when unscoped
+        self.assertEqual(Enum_QSFunction.Quick_Start_Request.value, 0)
+        self.assertEqual(Enum_QSFunction.Report_of_Approved_Rate.value, 8)
+        self.assertEqual(Enum_Option.Pad1.value, 0)
+        self.assertEqual(Enum_Option.PadN.value, 1)
+
+        for module in (hopopt_schema, opts_schema):
+            with self.subTest(module=module.__name__.rsplit('.', 1)[-1]):
+                option = module.Option
+                quick_start = module.QuickStartOption
+
+                # the quick-start registry must be its own, not the option one
+                self.assertIsNot(quick_start.__enum__, option.__enum__)
+                self.assertIn('__enum__', vars(quick_start))
+
+                # snapshot the registries: they are ``defaultdict``s, so reading a
+                # missing key through them would insert it
+                options = dict(option.registry)
+                functions = dict(quick_start.registry)
+
+                # option type 0 (Pad1) and 1 (PadN) both resolve to PadOption
+                self.assertIs(options[Enum_Option.Pad1], module.PadOption)
+                self.assertIs(options[Enum_Option.PadN], module.PadOption)
+                # option type 8 (SMF_DPD) must not be the quick-start report either
+                self.assertIsNot(options[Enum_Option.SMF_DPD], module.QuickStartReportOption)
+
+                # and the quick-start functions resolve inside their own registry
+                self.assertIs(functions[Enum_QSFunction.Quick_Start_Request],
+                              module.QuickStartRequestOption)
+                self.assertIs(functions[Enum_QSFunction.Report_of_Approved_Rate],
+                              module.QuickStartReportOption)
+
+    def test_ipv6_opts_and_hopopt_option_registries_agree(self) -> None:
+        """The two modules describe the same option space and must map it alike.
+
+        ``hopopt`` and ``ipv6_opts`` are near-identical by construction, so the
+        cheapest guard against one drifting from the other is to compare their
+        registries key for key. The registry collision this catches came about
+        from exactly one line present in one module and missing in the other.
+
+        """
+        from pcapkit.protocols.schema.internet import hopopt as hopopt_schema
+        from pcapkit.protocols.schema.internet import ipv6_opts as opts_schema
+
+        hopopt_options = dict(hopopt_schema.Option.registry)
+        opts_options = dict(opts_schema.Option.registry)
+        self.assertEqual(set(hopopt_options), set(opts_options))
+
+        for key in sorted(hopopt_options, key=int):
+            with self.subTest(option=int(key)):
+                self.assertEqual(hopopt_options[key].__name__, opts_options[key].__name__)
+
 
 if __name__ == '__main__':
     unittest.main()