dpkt: key IPv6 reassembly on the fragment ID, not the flow label

pcapkit/toolkit/dpkt.py

@@ -194,10 +194,23 @@ 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.flow,                                       # label
+                # NOTE: The reassembly key is the Fragment header's Identification
+                # (:rfc:`8200#section-4.5`), not the IPv6 header's Flow Label. The
+                # label is optional and routinely zero, so keying on it collapses
+                # every datagram between one address pair into a single buffer and
+                # interleaves their fragments; it also disagreed with ``bufid[2]``
+                # in every other engine, which feeds
+                # :attr:`pcapkit.foundation.reassembly.data.ip.DatagramID.id`.
+                ipv6_frag.id,                                    # identification
                 Enum_TransType.get(ipv6_frag.nxt),               # next header field in IPv6 Fragment Header
             ),
             num=count,                                           # original packet range number
+            # NOTE: ``IP6FragmentHeader.frag_off`` is a ``__bit_fields__`` property
+            # over the 13-bit on-wire Fragment Offset, i.e. already shifted out of
+            # the flags word, so it counts 8-octet units (:rfc:`8200#section-4.5`).
+            # The reassembly machinery indexes the datagram buffer with ``fo``, so
+            # the units have to become octets here, exactly as
+            # :func:`pcapkit.toolkit.scapy.ipv6_reassembly` does.
             fo=ipv6_frag.frag_off * 8,                           # fragment offset
             ihl=hdr_len,                                         # header length, only headers before IPv6-Frag
             mf=bool(ipv6_frag.m_flag),                           # more fragment flag

tests/integration/test_reassembly_end_to_end.py

@@ -14,9 +14,6 @@
 and the duplicate were both handled. See the module docstring of
 :file:`examples/generators/legacy.py`.
 
-One expectation here is skipped rather than asserted: see
-:class:`IPv6FragmentPayloadTests`.
-
 """
 from __future__ import annotations
 
@@ -244,9 +241,9 @@ class IPv6FragmentReassemblyTests(EndToEndTestCase):
     """IPv6 fragment reassembly over :file:`ipv6.pcap`.
 
     Frames 13 to 16 are one 4778 octet UDP datagram fragmented into
-    1448/1448/1448/434 octets. What is asserted here is which frames were
-    collected and that the datagram is reported complete -- both of which are
-    right. Its *contents* are not, so they live in the skipped test below.
+    1448/1448/1448/434 octets. Asserted here are which frames were collected and
+    that the datagram is reported complete; its *contents* are asserted in
+    :class:`IPv6FragmentPayloadTests` below.
 
     """
 
@@ -274,39 +271,39 @@ def test_the_datagram_is_reported_complete(self) -> None:
         self.assertEqual(int(datagram.id.proto), 17)
 
 
+@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed')
 class IPv6FragmentPayloadTests(EndToEndTestCase):
-    """What the reassembled IPv6 datagram should contain.
+    """What the reassembled IPv6 datagram contains.
 
-    Kept apart from :class:`IPv6FragmentReassemblyTests` so the skip covers only
-    the payload, and the assertions that are correct today keep running.
+    Kept apart from :class:`IPv6FragmentReassemblyTests` because these are the
+    assertions on the datagram's octets, which are what the two defects recorded
+    below used to get wrong while the frame collection and the ``completed`` flag
+    stayed right.
 
     """
 
-    @unittest.skip('blocked on IPv6 fragment offsets being used as byte offsets while the '
-                   'field is in eight-octet units (pcapkit/protocols/internet/ipv6_frag.py:141)')
     def test_the_four_fragments_reassemble_into_the_whole_datagram(self) -> None:
-        """The reassembled payload should be the four fragments, in order.
-
-        It is not. ``pcapkit/protocols/internet/ipv4.py:274`` scales the IPv4
-        fragment offset into octets (``int(schema.flags['offset']) * 8``), but
-        ``pcapkit/protocols/internet/ipv6_frag.py:141`` passes the IPv6 one
-        through unscaled (``offset=schema.flags['offset']``), and
-        ``pcapkit/toolkit/pcap.py:115`` then hands it to the reassembly
-        machinery as ``fo``, which is a byte offset. The four fragments are
-        therefore placed at octets 0, 181, 362 and 543 instead of 0, 1448, 2896
-        and 4344, so they overwrite one another and the datagram comes out 977
-        octets long -- ``543 + 434`` -- while still reporting
-        ``completed=True``.
-
-        Measured on this fixture: ``len(datagram.payload)`` is 977, and the
-        assertion below expects 4778.
-
-        A second defect is visible in the same call and is left alone here:
-        ``pcapkit/toolkit/pcap.py:111`` keys the reassembly buffer on the IPv6
-        header's flow label rather than on the fragment header's identification,
-        so ``datagram.id.id`` is 0 where the fixture's identification is 110308.
-        One fragmented datagram cannot show the consequence, so no assertion is
-        made about it either way.
+        """The reassembled payload is the four fragments, in order.
+
+        This is a regression test for two fixed defects, which is why it asserts
+        the exact octet count rather than only that the payload is non-empty.
+
+        The first was the fragment offset scaling.
+        ``pcapkit/protocols/internet/ipv6_frag.py`` passed the IPv6 fragment
+        offset through unscaled where the IPv4 path scaled it into octets, and
+        ``pcapkit/toolkit/pcap.py`` then handed it to the reassembly machinery as
+        ``fo``, which is an octet offset. The four fragments landed at octets 0,
+        181, 362 and 543 instead of 0, 1448, 2896 and 4344, overwrote one
+        another, and the datagram came out 977 octets long -- ``543 + 434`` --
+        while still reporting ``completed=True``. That is what the assertion on
+        ``len(datagram.payload)`` pins.
+
+        The second was the buffer identifier. ``pcapkit/toolkit/pcap.py`` keyed
+        the reassembly buffer on the IPv6 header's flow label rather than on the
+        fragment header's identification, so ``datagram.id.id`` came out 0 where
+        this fixture's identification is 110308. One fragmented datagram cannot
+        show the merging that the wrong key causes, but it does show the wrong
+        value, so the identification is asserted here.
 
         """
         extractor = self.extract(fin=sample_path('ipv6.pcap'), nofile=True, store=True,
@@ -320,6 +317,8 @@ def test_the_four_fragments_reassemble_into_the_whole_datagram(self) -> None:
         self.assertEqual([len(fragment) for fragment in fragments], [1448, 1448, 1448, 434])
         self.assertEqual(len(datagram.payload), 4778)
         self.assertEqual(bytes(datagram.payload), b''.join(fragments))
+        # the fragment header's identification, not the flow label, which is 0
+        self.assertEqual(datagram.id.id, 110308)
 
 
 if __name__ == '__main__':

tests/toolkit/test_dpkt_unit.py

@@ -50,6 +50,9 @@ class FakeFragment:
     #: Fragment offset, in 8-octet units, so the byte offset is ``2 * 8``.
     frag_off = 2
     m_flag = 1
+    #: Identification, deliberately different from :attr:`FakeIPv6.flow`, so a
+    #: buffer identifier keyed on the wrong one of the two is visible.
+    id = 4321
 
     def __len__(self) -> int:
         return 8
@@ -251,7 +254,10 @@ def test_ipv6_header_length_and_reassembly_with_fragment_fake(self) -> None:
         self.assertEqual(reassembled.num, 5)
         self.assertEqual(reassembled.bufid[0], ip_address('2001:db8::1'))
         self.assertEqual(reassembled.bufid[1], ip_address('2001:db8::2'))
-        self.assertEqual(reassembled.bufid[2], 7)
+        # the fragment header's Identification, not the IPv6 header's Flow Label
+        # -- ``bufid[2]`` feeds ``DatagramID.id``
+        self.assertEqual(reassembled.bufid[2], frag.id)
+        self.assertNotEqual(reassembled.bufid[2], ipv6.flow)
         # the buffer identifier carries the Next Header field of the fragment
         # header, as a registry enum rather than its name
         self.assertEqual(reassembled.bufid[3], TransType.get(frag.nxt))
@@ -344,27 +350,32 @@ def _make_ipv4_fragment(*, offset_units: int, mf: bool, body: bytes):
 
 
 def _ipv6_fragment_bytes(*, offset_units: int, mf: bool, body: bytes,
-                         ident: int = 110308, nxt: int = 17) -> bytes:
+                         ident: int = 110308, nxt: int = 17, flow: int = 0) -> bytes:
     """Serialise an IPv6 packet carrying a Fragment header, as wire octets.
 
     Built by hand rather than through :mod:`dpkt`'s constructors so the Fragment
-    header is unambiguously the one :rfc:`8200#section-4.5` describes.
+    header is unambiguously the one :rfc:`8200#section-4.5` describes: a 13-bit
+    Fragment Offset counted in 8-octet units, two reserved bits, then the More
+    Fragments flag, which is why ``offset_units`` is shifted left by three.
+
+    ``flow`` is the IPv6 header's Flow Label, kept separate from ``ident`` so a
+    buffer identifier keyed on the wrong one of the two is visible.
 
     """
     frag_hdr = struct.pack('>BBHI', nxt, 0,
                            (offset_units << 3) | (1 if mf else 0), ident)
     payload = frag_hdr + body
-    header = struct.pack('>IHBB', 6 << 28, len(payload), 44, 64)
+    header = struct.pack('>IHBB', (6 << 28) | flow, len(payload), 44, 64)
     header += socket.inet_pton(socket.AF_INET6, '2001:db8::1')
     header += socket.inet_pton(socket.AF_INET6, '2001:db8::2')
     return header + payload
 
 
 def _ipv6_fragment_frame(*, offset_units: int, mf: bool, body: bytes,
-                         ident: int = 110308, nxt: int = 17) -> bytes:
+                         ident: int = 110308, nxt: int = 17, flow: int = 0) -> bytes:
     """Wrap :func:`_ipv6_fragment_bytes` in an Ethernet frame."""
     return b'\xbb' * 6 + b'\xaa' * 6 + b'\x86\xdd' + _ipv6_fragment_bytes(
-        offset_units=offset_units, mf=mf, body=body, ident=ident, nxt=nxt,
+        offset_units=offset_units, mf=mf, body=body, ident=ident, nxt=nxt, flow=flow,
     )
 
 
@@ -578,6 +589,13 @@ def test_reads_the_real_fragment_header_attributes(self) -> None:
         self.assertFalse(hasattr(frag, 'nh'))
         self.assertEqual(frag.nxt, dpkt.ip.IP_PROTO_UDP)
         self.assertEqual(frag.frag_off, offset_units)
+        # ``frag_off`` is the 13-bit Fragment Offset, already shifted out of the
+        # flags word, and *not* the raw 16-bit ``_frag_off_resv_m`` -- which is
+        # what makes the ``* 8`` below a scaling into octets rather than a second
+        # scaling on top of one DPKT had already applied
+        self.assertEqual(frag._frag_off_resv_m, offset_units << 3)
+        self.assertEqual(frag.frag_off, frag._frag_off_resv_m >> 3)
+        self.assertNotEqual(frag.frag_off, frag._frag_off_resv_m)
 
         data = toolkit.ipv6_reassembly(types.SimpleNamespace(ip6=ipv6), count=3)
         self.assertIsNotNone(data)
@@ -597,6 +615,94 @@ def test_reads_the_real_fragment_header_attributes(self) -> None:
         # ... and ``tl - ihl`` has to be that payload's length, not 8 more
         self.assertEqual(data.tl - data.ihl, len(data.payload))
 
+    def test_offset_is_scaled_once_into_octets(self) -> None:
+        """``fo`` is the octet offset, so it is neither ``frag_off`` nor ``* 64``.
+
+        The three candidate readings of the field are pinned against each other
+        rather than only the right one being asserted, because the failure mode
+        that matters is an offset scaled the wrong number of times: an unscaled
+        ``fo`` overlaps the fragments and a doubly scaled one leaves holes, and
+        both still look like plausible integers.
+
+        """
+        import dpkt
+
+        from pcapkit.toolkit import dpkt as toolkit
+
+        offset_units = 181
+        ipv6 = dpkt.ip6.IP6(_ipv6_fragment_bytes(
+            offset_units=offset_units, mf=True, body=b'C' * 64,
+        ))
+        data = toolkit.ipv6_reassembly(types.SimpleNamespace(ip6=ipv6), count=1)
+        assert data is not None
+
+        self.assertEqual(data.fo, 1448)               # 181 units of 8 octets
+        self.assertNotEqual(data.fo, offset_units)    # not left in 8-octet units
+        self.assertNotEqual(data.fo, offset_units * 64)  # not scaled twice
+
+    def test_buffer_identifier_is_keyed_on_the_fragment_identification(self) -> None:
+        """``bufid[2]`` is the Fragment header's Identification, not the Flow Label.
+
+        The Flow Label is optional and routinely zero, so keying on it merges
+        unrelated datagrams between one address pair. The fixture gives the two a
+        different value so the wrong one cannot pass by coincidence.
+
+        """
+        import dpkt
+
+        from pcapkit.toolkit import dpkt as toolkit
+
+        ident, flow = 110308, 0x12345
+        ipv6 = dpkt.ip6.IP6(_ipv6_fragment_bytes(
+            offset_units=0, mf=True, body=b'D' * 32, ident=ident, flow=flow,
+        ))
+        self.assertEqual(ipv6.flow, flow)
+        self.assertEqual(ipv6.extension_hdrs[44].id, ident)
+
+        data = toolkit.ipv6_reassembly(types.SimpleNamespace(ip6=ipv6), count=1)
+        assert data is not None
+
+        self.assertEqual(data.bufid[2], ident)
+        self.assertNotEqual(data.bufid[2], flow)
+
+    def test_two_datagrams_sharing_a_flow_label_stay_separate(self) -> None:
+        """Distinct identifications must not be merged into one datagram.
+
+        This is the consequence a single fragmented datagram cannot show: both
+        datagrams below share source, destination, Next Header *and* Flow Label,
+        so a buffer identifier keyed on the label puts all four fragments into
+        one buffer and the payloads overwrite one another.
+
+        """
+        import dpkt
+
+        from pcapkit.foundation.reassembly.ipv6 import IPv6
+        from pcapkit.toolkit import dpkt as toolkit
+
+        first, second = b'E' * 64, b'F' * 64
+        third, fourth = b'G' * 64, b'H' * 64
+
+        reasm = IPv6(strict=True)
+        for index, (ident, offset_units, mf, chunk) in enumerate((
+            (1000, 0, True, first),
+            (2000, 0, True, third),
+            (1000, 8, False, second),
+            (2000, 8, False, fourth),
+        ), start=1):
+            ipv6 = dpkt.ip6.IP6(_ipv6_fragment_bytes(
+                offset_units=offset_units, mf=mf, body=chunk, ident=ident, flow=0x12345,
+            ))
+            data = toolkit.ipv6_reassembly(types.SimpleNamespace(ip6=ipv6), count=index)
+            assert data is not None
+            reasm(data)
+
+        datagrams = list(reasm.datagram)
+        self.assertEqual(len(datagrams), 2)
+        payloads = {datagram.id.id: bytes(datagram.payload) for datagram in datagrams}
+        self.assertEqual(sorted(payloads), [1000, 2000])
+        self.assertEqual(payloads[1000], first + second)
+        self.assertEqual(payloads[2000], third + fourth)
+
     def test_reassembly_reconstructs_a_fragmented_datagram(self) -> None:
         import dpkt