From 0e4064fe451abf170f100c772f9b3fff6f7dbf5e Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Mon, 14 Sep 2026 21:57:38 -0400 Subject: [PATCH] dpkt: split headers at their real length, and read the IPv6 fragment API Five defects in pcapkit/toolkit/dpkt.py, two filed and three found while verifying them. Every one made the dpkt engine quietly wrong rather than loud. TCP headers were split at tcp.__hdr_len__, the fixed 20-octet struct size, but dpkt serialises as pack_hdr() + opts + data. With options present the option octets landed at the head of the payload and len disagreed with payload. Worse than a wrong split: the reassembly buffer is sequence-indexed, so the option bytes overwrote payload rather than lengthening it. On test.pcap the dpkt engine returned 10 datagrams instead of 4 - six of them pure SYN option bytes, the four real ones completed=False - and not one was byte-identical to the default engine's. Now tcp.pack()[:tcp.off * 4], with payload taken from tcp.data so len and payload cannot disagree even on a malformed Data Offset. toolkit/scapy.py already did this correctly and was the reference. (#351) IPv6 fragment reassembly read ipv6_frag.nh, which does not exist on dpkt 1.9.8 - the class exposes nxt - so extraction crashed with AttributeError propagating out of the engine, and it passed frag_off unscaled where the reassembler wants octets. (#370) The same fixed-length mistake in the IPv4 path: ipv4.hl * 4 now, not a constant. Not observable on the fixtures because none carries IP options, so this was reported as inferred in the issue and is confirmed here with a constructed packet. Three more, all measured: - fo=ipv4.off used a property dpkt deprecates, and it returns the raw flags-and-offset word: for a first fragment with MF set it gave 8192 instead of 0. Now ipv4.offset * 8, which also silences the DeprecationWarning. - tl=len(ipv6) counted the fragment header, so TL - IHL overshot the payload by 8. Since the reassembler assigns datagram[start:stop] = payload, that silently resized the buffer. Now hdr_len + len(payload). - bufid[3] passed the enum's .name, a string, where BufferID declares TransType and submit() hands it to Protocol.analyze. Internet.__proto__ is a defaultdict so it never raised; it just degraded every reassembled datagram's parsed packet to Raw. The old tests missed all of it because the fakes were unfaithful: the IPv6 mock manufactured the nh attribute whose absence is the bug and asserted the wrong offset, and the TCP mock had no off at all. The fakes now match the real dpkt API, and the new tests pin invariants rather than constants - header length equals dataofs*4, len equals len(payload), header+payload equals tcp.pack(), tl-ihl equals len(payload), bufid[3] is a TransType - plus an end-to-end check that dpkt reassembly of test.pcap matches the default engine byte for byte. Each fixture asserts it actually exercises the defect, so none can pass vacuously. Fail-before: 42 failed across 10 methods. After: 14 passed, 36 subtests. Full suite: 512 passed, 4 skipped, 321 subtests passed. Closes #351, closes #370. --- pcapkit/toolkit/dpkt.py | 28 +- tests/toolkit/test_dpkt_unit.py | 489 +++++++++++++++++++++++++++++++- 2 files changed, 501 insertions(+), 16 deletions(-) diff --git a/pcapkit/toolkit/dpkt.py b/pcapkit/toolkit/dpkt.py index d3d04ebee6..1a4381d974 100644 --- a/pcapkit/toolkit/dpkt.py +++ b/pcapkit/toolkit/dpkt.py @@ -133,6 +133,10 @@ def ipv4_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv4Ad if ipv4 is not None: if ipv4.df: # dismiss not fragmented packet return None + # internet header length, in octets -- ``IP.hl`` counts 32-bit words and + # covers any IP options, whereas ``IP.__hdr_len__`` is the fixed 20-octet + # struct size and would leave option octets at the head of the payload + ihl = ipv4.hl * 4 data = IP_Packet( bufid=( @@ -141,15 +145,15 @@ def ipv4_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv4Ad cast('IPv4Address', ipaddress.ip_address(ipv4.dst)), # destination IP address ipv4.id, # identification - Enum_TransType.get(ipv4.p).name, # payload protocol type + Enum_TransType.get(ipv4.p), # payload protocol type ), num=count, # original packet range number - fo=ipv4.off, # fragment offset - ihl=ipv4.__hdr_len__, # internet header length + fo=ipv4.offset * 8, # fragment offset + ihl=ihl, # internet header length mf=bool(ipv4.mf), # more fragment flag tl=ipv4.len, # total length, header includes - header=ipv4.pack()[:ipv4.__hdr_len__], # raw bytes type header - payload=bytearray(ipv4.pack()[ipv4.__hdr_len__:]), # raw bytearray type payload + header=ipv4.pack()[:ihl], # raw bytes type header + payload=bytearray(ipv4.pack()[ihl:]), # raw bytearray type payload ) return data return None @@ -181,6 +185,8 @@ def ipv6_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv6Ad if ipv6_frag is None: # dismiss not fragmented packet return None hdr_len = ipv6_hdr_len(ipv6) + # payload following the IPv6 Fragment header + payload = ipv6.pack()[hdr_len + ipv6_frag.__hdr_len__:] data = IP_Packet( bufid=( @@ -189,15 +195,15 @@ def ipv6_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv6Ad cast('IPv6Address', ipaddress.ip_address(ipv6.dst)), # destination IP address ipv6.flow, # label - Enum_TransType.get(ipv6_frag.nh).name, # next header field in IPv6 Fragment Header + Enum_TransType.get(ipv6_frag.nxt), # next header field in IPv6 Fragment Header ), num=count, # original packet range number - fo=ipv6_frag.nxt, # fragment offset + 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 - tl=len(ipv6), # total length, header includes + tl=hdr_len + len(payload), # total length, header includes header=ipv6.pack()[:hdr_len], # raw bytearray type header before IPv6-Frag - payload=bytearray(ipv6.pack()[hdr_len + len(ipv6_frag):]), # raw bytearray type payload after IPv6-Frag + payload=bytearray(payload), # raw bytearray type payload after IPv6-Frag ) return data return None @@ -249,8 +255,8 @@ def tcp_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'TCP_Packet | None rst=bool(int(flags[5])), # reset connection flag syn=bool(int(flags[6])), # synchronise flag fin=bool(int(flags[7])), # finish flag - header=tcp.pack()[:tcp.__hdr_len__], # raw bytes type header - payload=bytearray(tcp.pack()[tcp.__hdr_len__:]), # raw bytearray type payload + header=tcp.pack()[:tcp.off * 4], # raw bytes type header + payload=bytearray(bytes(tcp.data)), # raw bytearray type payload first=tcp.seq, # first sequence number of payload last=tcp.seq + raw_len - 1, # last sequence number of payload len=raw_len, # payload length, header excludes diff --git a/tests/toolkit/test_dpkt_unit.py b/tests/toolkit/test_dpkt_unit.py index 7324376bfc..d7e0ea5b6b 100644 --- a/tests/toolkit/test_dpkt_unit.py +++ b/tests/toolkit/test_dpkt_unit.py @@ -2,25 +2,53 @@ import importlib.util from ipaddress import ip_address +import os import socket +import struct +import tempfile import types import unittest import warnings -from tests._support import purge_modules +from tests._support import close_extractor, purge_modules, sample_path HAS_DPKT = importlib.util.find_spec('dpkt') is not None RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) +#: NOP, NOP, Timestamp (kind 8, length 10) -- 12 octets, the option block a +#: modern established TCP segment actually carries. +TCP_TIMESTAMP_OPTS = b'\x01\x01\x08\x0a' + struct.pack('>II', 0x4C0A9AFB, 0xD6175129) +#: NOP, NOP, NOP, End of Option List -- 4 octets of IPv4 options, enough to push +#: the internet header length past the fixed 20-octet struct size. +IPV4_NOP_OPTS = b'\x01\x01\x01\x00' +#: A protocol number reserved for experimentation (:rfc:`3692`), which +#: :mod:`dpkt` has no dissector for. Using it keeps ``IP.data`` as the raw +#: octets that were handed in, so the fixtures stay byte-exact. +IP_PROTO_EXPERIMENTAL = 253 + class FakeHeader: length = 8 class FakeFragment: - nh = 6 - nxt = 2 + """Stand-in for :class:`dpkt.ip6.IP6FragmentHeader`. + + The attribute surface mirrors the real class deliberately: ``nxt`` is the + Next Header field, ``frag_off`` is the fragment offset in 8-octet units and + ``m_flag`` is the More Fragments flag. In particular there is **no** ``nh`` + attribute, because the real :class:`dpkt.ip6.IP6FragmentHeader` has none -- + a fake that invented one is exactly what let the defect behind the + ``AttributeError`` reach a release unnoticed. + + """ + + __hdr_len__ = 8 + #: Next Header (6 == TCP), the value that belongs in the buffer identifier. + nxt = 6 + #: Fragment offset, in 8-octet units, so the byte offset is ``2 * 8``. + frag_off = 2 m_flag = 1 def __len__(self) -> int: @@ -48,6 +76,9 @@ def pack(self) -> bytes: class TCP: __hdr_fields__ = ('sport', 'dport', 'seq', 'ack', 'flags') __hdr_len__ = 4 + #: Data Offset, in 32-bit words. ``off * 4`` has to agree with the header + #: portion of :meth:`pack`, the way it does on a real :class:`dpkt.tcp.TCP`. + off = 1 sport = 2345 dport = 443 seq = 20 @@ -136,7 +167,10 @@ def test_packet_chain_dict_and_ipv4_reassembly_with_real_dpkt_packet(self) -> No self.assertEqual(reassembled.bufid[1], ip_address('198.51.100.1')) self.assertEqual(reassembled.bufid[2], 123) self.assertTrue(reassembled.mf) - self.assertEqual(bytes(reassembled.payload), fragment.ip.pack()[fragment.ip.__hdr_len__:]) + # the split is at the internet header length, which ``hl`` carries in + # 32-bit words -- not at the fixed 20-octet struct size + self.assertEqual(bytes(reassembled.payload), + fragment.ip.pack()[fragment.ip.hl * 4:]) self.assertIsNone(toolkit.ipv4_reassembly(types.SimpleNamespace(), count=1)) self.assertIsNone(toolkit.ipv4_reassembly(self._make_ipv4_tcp_packet(df=True), count=1)) @@ -203,6 +237,7 @@ def test_tcp_helpers_cover_ipv6_and_data_payload_fallback(self) -> None: self.assertEqual(flow.frame['ETHERNET']['FakeTCPIPv6']['TCP']['sport'], 2345) def test_ipv6_header_length_and_reassembly_with_fragment_fake(self) -> None: + from pcapkit.const.reg.transtype import TransType from pcapkit.toolkit import dpkt as toolkit frag = FakeFragment() @@ -217,7 +252,12 @@ def test_ipv6_header_length_and_reassembly_with_fragment_fake(self) -> None: 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) - self.assertEqual(reassembled.fo, 2) + # 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)) + # the fragment offset is stored in 8-octet units on the wire and the + # reassembly machinery indexes the datagram buffer in octets + self.assertEqual(reassembled.fo, frag.frag_off * 8) self.assertTrue(reassembled.mf) self.assertEqual(reassembled.ihl, 48) self.assertEqual(bytes(reassembled.payload), b'PAYLOAD') @@ -226,5 +266,444 @@ def test_ipv6_header_length_and_reassembly_with_fragment_fake(self) -> None: self.assertIsNone(toolkit.ipv6_reassembly(types.SimpleNamespace(ip6=FakeIPv6()), count=1)) +# --------------------------------------------------------------------------- +# regression tests +# --------------------------------------------------------------------------- + + +def _make_tcp_segment_with_options(payload: bytes, *, opts: bytes = TCP_TIMESTAMP_OPTS): + """Build an Ethernet/IPv4/TCP packet whose TCP header carries options. + + The packet is serialised and parsed back, so the object under test comes off + the same code path a captured frame does. + + """ + import dpkt + + tcp = dpkt.tcp.TCP( + sport=1234, dport=80, seq=100, ack=1, + flags=dpkt.tcp.TH_ACK | dpkt.tcp.TH_PUSH, data=payload, + ) + tcp.opts = opts + tcp.off = (tcp.__hdr_len__ + len(opts)) // 4 + + ipv4 = dpkt.ip.IP( + src=socket.inet_aton('192.0.2.1'), dst=socket.inet_aton('198.51.100.1'), + p=dpkt.ip.IP_PROTO_TCP, id=4242, + ) + ipv4.df = 0 + ipv4.data = tcp + ipv4.len = len(ipv4) + + eth = dpkt.ethernet.Ethernet(src=b'\xaa' * 6, dst=b'\xbb' * 6, + type=dpkt.ethernet.ETH_TYPE_IP) + eth.data = ipv4 + return dpkt.ethernet.Ethernet(bytes(eth)) + + +def _make_ipv4_with_options(body: bytes, *, opts: bytes = IPV4_NOP_OPTS): + """Build an Ethernet/IPv4 packet whose internet header carries options.""" + import dpkt + + ipv4 = dpkt.ip.IP( + src=socket.inet_aton('192.0.2.1'), dst=socket.inet_aton('198.51.100.1'), + p=IP_PROTO_EXPERIMENTAL, id=7, + ) + ipv4.df = 0 + ipv4.mf = 0 + ipv4.offset = 0 + ipv4.opts = opts + ipv4._v_hl = (4 << 4) | ((ipv4.__hdr_len__ + len(opts)) // 4) + ipv4.data = body + ipv4.len = ipv4.__hdr_len__ + len(opts) + len(body) + + eth = dpkt.ethernet.Ethernet(src=b'\xaa' * 6, dst=b'\xbb' * 6, + type=dpkt.ethernet.ETH_TYPE_IP) + eth.data = ipv4 + return dpkt.ethernet.Ethernet(bytes(eth)) + + +def _make_ipv4_fragment(*, offset_units: int, mf: bool, body: bytes): + """Build an Ethernet/IPv4 fragment at ``offset_units`` 8-octet units.""" + import dpkt + + ipv4 = dpkt.ip.IP( + src=socket.inet_aton('192.0.2.1'), dst=socket.inet_aton('198.51.100.1'), + p=IP_PROTO_EXPERIMENTAL, id=555, + ) + ipv4.df = 0 + ipv4.data = body + ipv4.len = ipv4.__hdr_len__ + len(body) + ipv4.offset = offset_units + ipv4.mf = int(mf) + + eth = dpkt.ethernet.Ethernet(src=b'\xaa' * 6, dst=b'\xbb' * 6, + type=dpkt.ethernet.ETH_TYPE_IP) + eth.data = ipv4 + return dpkt.ethernet.Ethernet(bytes(eth)) + + +def _ipv6_fragment_bytes(*, offset_units: int, mf: bool, body: bytes, + ident: int = 110308, nxt: int = 17) -> 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. + + """ + 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 += 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: + """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, + ) + + +@unittest.skipUnless(HAS_RUNTIME and HAS_DPKT, 'runtime dependencies not installed') +class DPKTTCPHeaderSplitTests(unittest.TestCase): + """The TCP header is delimited by Data Offset, not by the struct size. + + :class:`dpkt.tcp.TCP` serialises as ``pack_hdr() + opts + data``, and + ``__hdr_len__`` is the fixed 20-octet struct size, so slicing at + ``__hdr_len__`` leaves the option octets at the head of the *payload*. The + invariants below are the ones that hold for any segment whatsoever, which is + why they are asserted rather than a particular octet count. + + """ + + def setUp(self) -> None: + purge_modules(['pcapkit']) + + def test_header_ends_at_data_offset_and_len_matches_payload(self) -> None: + from pcapkit.toolkit import dpkt as toolkit + + body = b'GET /index.html HTTP/1.1\r\n\r\n' + packet = _make_tcp_segment_with_options(body) + tcp = packet.ip.data + + # the fixture has to actually exercise the defect + self.assertGreater(len(tcp.opts), 0) + self.assertGreater(tcp.off * 4, tcp.__hdr_len__) + + data = toolkit.tcp_reassembly(packet, count=1) + self.assertIsNotNone(data) + assert data is not None + + # the header is exactly ``dataofs * 4`` octets ... + self.assertEqual(len(data.header), tcp.off * 4) + # ... and therefore ends with the option block rather than dropping it + self.assertTrue(bytes(data.header).endswith(bytes(tcp.opts))) + + # the reported length and the payload cannot disagree + self.assertEqual(data.len, len(data.payload)) + # the payload starts at the real application bytes + self.assertEqual(bytes(data.payload), body) + self.assertFalse(bytes(data.payload).startswith(bytes(tcp.opts))) + + # header and payload together reconstitute the segment + self.assertEqual(bytes(data.header) + bytes(data.payload), tcp.pack()) + # and the sequence numbers span exactly the payload + self.assertEqual(data.last - data.first + 1, data.len) + + def test_option_only_segment_yields_empty_payload(self) -> None: + """A pure ACK carrying options has no payload at all.""" + from pcapkit.toolkit import dpkt as toolkit + + packet = _make_tcp_segment_with_options(b'') + tcp = packet.ip.data + self.assertGreater(len(tcp.opts), 0) + + data = toolkit.tcp_reassembly(packet, count=1) + assert data is not None + self.assertEqual(data.len, 0) + self.assertEqual(len(data.payload), 0) + self.assertEqual(len(data.header), tcp.off * 4) + + def test_every_tcp_frame_of_the_sample_capture_holds_the_invariants(self) -> None: + """The invariants hold for every TCP frame of a real capture.""" + import dpkt + + from pcapkit.toolkit import dpkt as toolkit + + try: + path = sample_path('test.pcap') + except FileNotFoundError as exc: + self.skipTest(str(exc)) + + with open(path, 'rb') as file: + frames = list(dpkt.pcap.Reader(file)) + + checked = with_options = 0 + for index, (_, buf) in enumerate(frames, start=1): + packet = dpkt.ethernet.Ethernet(buf) + ip = getattr(packet, 'ip', None) or getattr(packet, 'ip6', None) + if ip is None: + continue + tcp = getattr(ip, 'tcp', None) + if tcp is None and type(getattr(ip, 'data', None)).__name__ == 'TCP': + tcp = ip.data + if tcp is None: + continue + + data = toolkit.tcp_reassembly(packet, count=index) + assert data is not None + with self.subTest(frame=index): + self.assertEqual(len(data.header), tcp.off * 4) + self.assertEqual(data.len, len(data.payload)) + self.assertEqual(bytes(data.payload), bytes(tcp.data)) + self.assertEqual(bytes(data.header) + bytes(data.payload), tcp.pack()) + checked += 1 + if len(tcp.opts): + with_options += 1 + + self.assertGreater(checked, 0, 'sample capture carried no TCP frames') + self.assertGreater(with_options, 0, + 'sample capture carried no TCP options, so it cannot ' + 'exercise the Data Offset split') + + +@unittest.skipUnless(HAS_RUNTIME and HAS_DPKT, 'runtime dependencies not installed') +class DPKTIPv4ReassemblyFieldTests(unittest.TestCase): + """The IPv4 path carries the same class of defect as the TCP one.""" + + def setUp(self) -> None: + purge_modules(['pcapkit']) + + def test_header_ends_at_internet_header_length_with_options(self) -> None: + from pcapkit.toolkit import dpkt as toolkit + + body = b'Y' * 100 + packet = _make_ipv4_with_options(body) + ipv4 = packet.ip + + # the fixture has to actually exercise the defect + self.assertEqual(ipv4.hl * 4, ipv4.__hdr_len__ + len(IPV4_NOP_OPTS)) + self.assertGreater(ipv4.hl * 4, ipv4.__hdr_len__) + + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + data = toolkit.ipv4_reassembly(packet, count=1) + self.assertIsNotNone(data) + assert data is not None + + self.assertEqual(data.ihl, ipv4.hl * 4) + self.assertEqual(len(data.header), ipv4.hl * 4) + self.assertTrue(bytes(data.header).endswith(IPV4_NOP_OPTS)) + self.assertEqual(bytes(data.payload), body) + self.assertFalse(bytes(data.payload).startswith(IPV4_NOP_OPTS)) + # the reassembly machinery derives the payload length as ``tl - ihl``, + # so that difference has to be the payload it was handed + self.assertEqual(data.tl - data.ihl, len(data.payload)) + + def test_fragment_offset_is_reported_in_octets(self) -> None: + """``IP.off`` is the raw flags-plus-offset word, not the offset.""" + from pcapkit.const.reg.transtype import TransType + from pcapkit.toolkit import dpkt as toolkit + + offset_units = 179 + packet = _make_ipv4_fragment(offset_units=offset_units, mf=True, body=b'B' * 200) + ipv4 = packet.ip + + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + data = toolkit.ipv4_reassembly(packet, count=1) + assert data is not None + + self.assertEqual(ipv4.offset, offset_units) + # the buffer is indexed in octets, and the wire field is in 8-octet units + self.assertEqual(data.fo, offset_units * 8) + self.assertTrue(data.mf) + # the buffer identifier's protocol slot is a registry enum, which is what + # ``Reassembly.submit`` hands to ``Protocol.analyze`` + self.assertIsInstance(data.bufid[3], TransType) + self.assertEqual(data.bufid[3], TransType.get(ipv4.p)) + + def test_reassembly_reconstructs_a_fragmented_datagram(self) -> None: + """End to end: two IPv4 fragments come back as the original payload.""" + from pcapkit.foundation.reassembly.ipv4 import IPv4 + from pcapkit.toolkit import dpkt as toolkit + + body = bytes(range(256)) * 6 # 1536 octets, a multiple of 8 + first, second = body[:1024], body[1024:] + + reasm = IPv4(strict=True) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + for index, (offset_units, mf, chunk) in enumerate(( + (0, True, first), + (len(first) // 8, False, second), + ), start=1): + packet = _make_ipv4_fragment(offset_units=offset_units, mf=mf, body=chunk) + data = toolkit.ipv4_reassembly(packet, count=index) + assert data is not None + reasm(data) + + datagrams = list(reasm.datagram) + self.assertEqual(len(datagrams), 1) + self.assertTrue(datagrams[0].completed) + self.assertEqual(bytes(datagrams[0].payload), body) + + +@unittest.skipUnless(HAS_RUNTIME and HAS_DPKT, 'runtime dependencies not installed') +class DPKTIPv6ReassemblyTests(unittest.TestCase): + """IPv6 reassembly against a real :class:`dpkt.ip6.IP6FragmentHeader`.""" + + def setUp(self) -> None: + purge_modules(['pcapkit']) + + def test_reads_the_real_fragment_header_attributes(self) -> None: + import dpkt + + from pcapkit.const.reg.transtype import TransType + from pcapkit.toolkit import dpkt as toolkit + + offset_units, body = 175, b'B' * 200 + ipv6 = dpkt.ip6.IP6(_ipv6_fragment_bytes( + offset_units=offset_units, mf=False, body=body, + )) + frag = ipv6.extension_hdrs.get(44) + self.assertIsNotNone(frag) + assert frag is not None + + # the real class exposes ``nxt``/``frag_off``/``m_flag`` and no ``nh`` + self.assertFalse(hasattr(frag, 'nh')) + self.assertEqual(frag.nxt, dpkt.ip.IP_PROTO_UDP) + self.assertEqual(frag.frag_off, offset_units) + + data = toolkit.ipv6_reassembly(types.SimpleNamespace(ip6=ipv6), count=3) + self.assertIsNotNone(data) + assert data is not None + + # the Next Header field of the fragment header, as a registry enum + self.assertIsInstance(data.bufid[3], TransType) + self.assertEqual(data.bufid[3], TransType.get(frag.nxt)) + # the offset is in 8-octet units on the wire, octets in the buffer + self.assertEqual(data.fo, offset_units * 8) + self.assertFalse(data.mf) + # only the headers before the fragment header + self.assertEqual(data.ihl, ipv6.__hdr_len__) + self.assertEqual(len(data.header), ipv6.__hdr_len__) + # the payload follows the 8-octet fragment header ... + self.assertEqual(bytes(data.payload), body) + # ... and ``tl - ihl`` has to be that payload's length, not 8 more + self.assertEqual(data.tl - data.ihl, len(data.payload)) + + def test_reassembly_reconstructs_a_fragmented_datagram(self) -> None: + import dpkt + + from pcapkit.foundation.reassembly.ipv6 import IPv6 + from pcapkit.toolkit import dpkt as toolkit + + body = bytes(range(256)) * 6 # 1536 octets, a multiple of 8 + first, second = body[:1024], body[1024:] + + reasm = IPv6(strict=True) + for index, (offset_units, mf, chunk) in enumerate(( + (0, True, first), + (len(first) // 8, False, second), + ), start=1): + ipv6 = dpkt.ip6.IP6(_ipv6_fragment_bytes( + offset_units=offset_units, mf=mf, body=chunk, + )) + 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), 1) + self.assertTrue(datagrams[0].completed) + self.assertEqual(bytes(datagrams[0].payload), body) + + def test_engine_reassembles_ipv6_fragments_end_to_end(self) -> None: + """The whole ``engine='dpkt'`` path, from a capture file in.""" + import dpkt + + import pcapkit + + body = bytes(range(256)) * 6 + first, second = body[:1024], body[1024:] + + handle = tempfile.NamedTemporaryFile(suffix='.pcap', delete=False) + try: + writer = dpkt.pcap.Writer(handle) + writer.writepkt(_ipv6_fragment_frame(offset_units=0, mf=True, body=first), ts=1.0) + writer.writepkt( + _ipv6_fragment_frame(offset_units=len(first) // 8, mf=False, body=second), + ts=1.1, + ) + handle.close() + + extractor = pcapkit.extract(fin=handle.name, nofile=True, store=False, + engine='dpkt', ipv6=True, reassembly=True, + reasm_strict=True) + try: + datagrams = list(extractor.reassembly.ipv6) + finally: + close_extractor(extractor) + finally: + os.unlink(handle.name) + + self.assertEqual(len(datagrams), 1) + self.assertTrue(datagrams[0].completed) + self.assertEqual(bytes(datagrams[0].payload), body) + + +@unittest.skipUnless(HAS_RUNTIME and HAS_DPKT, 'runtime dependencies not installed') +class DPKTEngineParityTests(unittest.TestCase): + """The ``dpkt`` engine must agree with the default engine octet for octet. + + This is the check that would have caught the Data Offset defect: the split + was wrong in a way that left the datagram *lengths* untouched, because the + reassembly buffer is indexed by sequence number, so only a byte-level + comparison against a known-good engine shows it. + + """ + + def setUp(self) -> None: + purge_modules(['pcapkit']) + + def _reassemble(self, path: str, engine: str): + import pcapkit + + extractor = pcapkit.extract(fin=path, nofile=True, store=False, engine=engine, + tcp=True, reassembly=True, reasm_strict=True) + try: + return [ + ( + datagram.completed, + datagram.payload if isinstance(datagram.payload, bytes) + else b''.join(datagram.payload), + ) + for datagram in extractor.reassembly.tcp + ] + finally: + close_extractor(extractor) + + def test_tcp_reassembly_matches_the_default_engine(self) -> None: + try: + path = sample_path('test.pcap') + except FileNotFoundError as exc: + self.skipTest(str(exc)) + + default = self._reassemble(path, 'default') + dpkt_out = self._reassemble(path, 'dpkt') + + self.assertGreater(len(default), 0, 'sample capture reassembled nothing') + self.assertEqual(len(dpkt_out), len(default)) + for index, (expected, actual) in enumerate(zip(default, dpkt_out)): + with self.subTest(datagram=index): + self.assertEqual(actual[0], expected[0]) + self.assertEqual(actual[1], expected[1]) + + if __name__ == '__main__': unittest.main()