From e732e194ae2163d903caf9c7a939765a80a9ec2b Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Tue, 15 Sep 2026 11:44:19 -0400 Subject: [PATCH] dpkt: key IPv6 reassembly on the fragment ID, not the flow label #389 changed the IPv6 reassembly key from the Flow Label to the Fragment header's Identification, but it touched `pcapkit/toolkit/pcap.py`, `pcapng.py`, `scapy.py` and `pcapkit/protocols/internet/ipv6_frag.py` and missed `pcapkit/toolkit/dpkt.py`. The dpkt adapter was the only engine still passing `ipv6.flow` as `bufid[2]`, which feeds `pcapkit.foundation.reassembly.data.ip.DatagramID.id`. That is not cosmetic. The Flow Label is optional and is zero on all four fragments of the `ipv6.pcap` fixture, while their Identification is 110308. Keying on the label therefore collapses every fragmented datagram between one address pair carrying the same next-header into a single reassembly buffer and interleaves their fragments. Two such datagrams reach the machinery as one and it fails with `TypeError: 'bytes' object cannot be interpreted as an integer`. Also lifted the `@unittest.skip` at `tests/integration/test_reassembly_end_to_end.py`, which named the IPv6 offset scaling that #389 fixed. The test passes with no assertions weakened, and it now additionally asserts `datagram.id.id == 110308` -- the check the old docstring had explicitly declined to make while the key was wrong. That class was also the only one in the file missing its `@unittest.skipUnless(HAS_RUNTIME)` guard, which would have errored rather than skipped on a machine without the optional runtime dependencies once un-skipped. Investigated and found *correct*, so left alone with an explanatory comment: `fo=ipv6_frag.frag_off * 8`. dpkt declares the fragment header's flags word via `__bit_fields__`, so `frag_off` is the 13-bit offset already shifted out of it -- a count of 8-octet units -- and scaling once is right. On `ipv6.pcap` the four fragments read 0, 181, 362, 543 units, giving 0, 1448, 2896, 4344 octets, exactly the 1448/1448/1448/434 fragment boundaries. Had `frag_off` been the raw word, frame 14 would have read 1449. Verified: reverting the one-line fix fails three tests, including both new ones (`AssertionError: 74565 != 110308` and the `TypeError` above), so they genuinely catch the defect. Post-fix the dpkt engine agrees with the default engine octet-for-octet on `ipv6.pcap` -- same id, proto, length and payload digest. Integration 70 passed / 3 skipped; dpkt toolkit 17 passed; fixture-free unit tier 545 passed / 14 skipped. --- pcapkit/toolkit/dpkt.py | 15 ++- .../integration/test_reassembly_end_to_end.py | 65 +++++----- tests/toolkit/test_dpkt_unit.py | 118 +++++++++++++++++- 3 files changed, 158 insertions(+), 40 deletions(-) diff --git a/pcapkit/toolkit/dpkt.py b/pcapkit/toolkit/dpkt.py index efc652d472..793f17b1cc 100644 --- a/pcapkit/toolkit/dpkt.py +++ b/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 diff --git a/tests/integration/test_reassembly_end_to_end.py b/tests/integration/test_reassembly_end_to_end.py index 0e73f64aab..657d4d0d23 100644 --- a/tests/integration/test_reassembly_end_to_end.py +++ b/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__': diff --git a/tests/toolkit/test_dpkt_unit.py b/tests/toolkit/test_dpkt_unit.py index d7e0ea5b6b..5c8ebdb6f9 100644 --- a/tests/toolkit/test_dpkt_unit.py +++ b/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