From 73e36ea0b74e045171d6249f2a449e75ae72c5de Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Mon, 14 Sep 2026 00:47:08 -0400 Subject: [PATCH 1/2] ipv6: advance past the last extension header before the next layer The extension-header loop broke out of the walk before advancing the payload, so a fragmented datagram handed the fragment header to the next layer as if it were the transport header: a UDP fragment reported srcport 4352, dstport 1 and length 1, which are the fragment header's next-header and reserved octets, its offset-and-flags hextet, and the top half of the identification. The advance now runs before every exit from the loop. tests/protocols/internet was pinning the wrong values and now pins the real header (51234 -> 5001, length 4778), plus a synthetic chain per shape - none, fragment only, hop-by-hop with destination options, and hop-by-hop with a 16-octet routing header then fragment - since no committed capture carries more than one extension header. Closes #348. --- examples/generators/pcap.py | 14 +- pcapkit/protocols/internet/ipv6.py | 9 +- .../internet/test_ipv6_extension_runtime.py | 123 +++++++++++++++++- 3 files changed, 134 insertions(+), 12 deletions(-) diff --git a/examples/generators/pcap.py b/examples/generators/pcap.py index 25f719f929..d79a9b482a 100644 --- a/examples/generators/pcap.py +++ b/examples/generators/pcap.py @@ -76,12 +76,14 @@ appears nowhere else) is solved for. The checksums are still computed by :mod:`scapy` from the finished packet, and the captures stay well-formed. 2. ``tests/protocols/internet/test_ipv6_extension_runtime.py`` reads UDP - ports 4352 and 1 and a UDP length of 1 out of the first fragment. Those - are not the real UDP header: :mod:`pcapkit` hands the next layer the bytes - starting at the *fragment header* rather than after it, so ``4352`` is its - next-header and reserved octets (``0x1100``), ``1`` is its offset-and-flags - hextet, and the length is the top half of the identification. The fixture - is an ordinary fragmented datagram; the test pins pcapkit's behaviour. + ports 51234 and 5001 and a UDP length of 4778 out of the *first* fragment + of ``ipv6.pcap``'s fragmented datagram. Only that fragment carries the UDP + header, and the length it declares is that of the whole reassembled + datagram, not of the 1448 octets the fragment holds -- so the numbers look + inconsistent with the fragment they are read from, and are not. The three + later fragments carry no transport header at all; pcapkit decodes one out + of their continuation bytes regardless, which is why those frames are not + asserted on beyond their fragment header and payload length. """ diff --git a/pcapkit/protocols/internet/ipv6.py b/pcapkit/protocols/internet/ipv6.py index a52f6fad31..cbc2881db2 100644 --- a/pcapkit/protocols/internet/ipv6.py +++ b/pcapkit/protocols/internet/ipv6.py @@ -344,6 +344,12 @@ def _decode_next_layer(self, ipv6: 'Data_IPv6', proto: 'Optional[int]' = None, # keep record of extension headers self._exthdr.add(ex_proto, next_) + # update payload for the next header -- either the next extension + # header, or the upper layer protocol should this be the last one; + # this must happen before any exit from the loop, since the payload + # is what gets handed to ``super()._decode_next_layer`` below + payload = payload[next_.length:] + # keep original data after fragment header if ex_proto == Enum_ExtensionHeader.IPv6_Frag: ipv6.__update__({ @@ -351,9 +357,6 @@ def _decode_next_layer(self, ipv6: 'Data_IPv6', proto: 'Optional[int]' = None, }) break - # update payload for next extension header - payload = payload[next_.length:] - # record real header & payload length (headers exclude) ipv6.__update__({ 'hdr_len': hdr_len, diff --git a/tests/protocols/internet/test_ipv6_extension_runtime.py b/tests/protocols/internet/test_ipv6_extension_runtime.py index 5daacb314a..bb9c513ae4 100644 --- a/tests/protocols/internet/test_ipv6_extension_runtime.py +++ b/tests/protocols/internet/test_ipv6_extension_runtime.py @@ -8,6 +8,56 @@ RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) +#: Next-header values, spelled out so the datagram builders below read as the +#: header chains they describe. +NH_HOPOPT = 0 +NH_UDP = 17 +NH_ROUTE = 43 +NH_FRAG = 44 +NH_OPTS = 60 + +#: Link-local endpoints of the hand-built datagrams. +SRC_ADDR = bytes.fromhex('fe80' + '0000' * 6 + '0001') +DST_ADDR = bytes.fromhex('fe80' + '0000' * 6 + '0002') + +#: Ports and body of the UDP datagram carried by every hand-built chain. +UDP_SRCPORT = 51234 +UDP_DSTPORT = 5001 +UDP_BODY = bytes(range(0x40, 0x60)) +#: The eight octets the UDP layer is expected to be handed, whatever precedes +#: them -- ``c8 22 13 89 00 28 00 00``. +UDP_HEADER = (UDP_SRCPORT.to_bytes(2, 'big') + UDP_DSTPORT.to_bytes(2, 'big') + + (8 + len(UDP_BODY)).to_bytes(2, 'big') + b'\x00\x00') + +#: Fragment identification, matching the one in ``examples/captures/ipv6.pcap``. +FRAG_ID = 110308 + + +def build_ipv6(next_header: int, payload: bytes) -> bytes: + """Build an IPv6 datagram carrying ``payload`` after the fixed header.""" + return (b'\x60\x00\x00\x00' + len(payload).to_bytes(2, 'big') + + bytes([next_header, 64]) + SRC_ADDR + DST_ADDR + payload) + + +def build_option_header(next_header: int) -> bytes: + """Build an 8-octet hop-by-hop or destination options header (one PadN).""" + return bytes([next_header, 0]) + b'\x01\x04' + b'\x00' * 4 + + +def build_routing_header(next_header: int) -> bytes: + """Build a 16-octet routing header of an unassigned routing type. + + Deliberately *not* eight octets long, so that a chain walk advancing by a + fixed stride rather than by each header's own length would be caught. + + """ + return bytes([next_header, 1, 253, 0]) + b'\x00' * 12 + + +def build_fragment_header(next_header: int) -> bytes: + """Build an 8-octet fragment header for the first of several fragments.""" + return bytes([next_header, 0]) + b'\x00\x01' + FRAG_ID.to_bytes(4, 'big') + @unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') class IPv6ExtensionRuntimeTests(unittest.TestCase): @@ -41,14 +91,18 @@ def test_ipv6_fragment_chain_records_extension_header_metadata(self) -> None: self.assertTrue(frag.info.mf) self.assertEqual(frag.info.id, 110308) - self.assertEqual(int(udp.info.srcport), 4352) - self.assertEqual(int(udp.info.dstport), 1) - self.assertEqual(udp.info.len, 1) + self.assertEqual(int(udp.info.srcport), 51234) + self.assertEqual(int(udp.info.dstport), 5001) + self.assertEqual(udp.info.len, 4778) self.assertEqual(type(udp.payload).__name__, 'Raw') self.assertEqual(len(ipv6.info.fragment.header), 48) self.assertEqual(len(ipv6.info.fragment.payload), 1448) + # the transport header has to be read from *after* the fragment header, + # i.e. from the first octets of the data the fragment header carries + self.assertEqual(bytes(udp.packet.header), bytes(ipv6.info.fragment.payload)[:8]) + def test_ipv6_fragment_extension_forbids_direct_payload_accessors(self) -> None: from pcapkit.utilities.exceptions import UnsupportedCall @@ -72,5 +126,68 @@ def test_ipv6_fragment_extension_forbids_direct_payload_accessors(self) -> None: _ = frag.protochain +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class IPv6ExtensionPayloadOffsetTests(unittest.TestCase): + """The next layer is read from after the *last* extension header. + + The datagrams here are built from bytes rather than read out of + ``examples/captures/``: no committed capture carries more than one IPv6 + extension header, and the captures are pinned byte-for-byte by the rest of + the suite, so a chain of several headers cannot be obtained from them. + Parsing goes through :class:`~pcapkit.protocols.internet.ipv6.IPv6` + directly, which is the layer that walks the chain -- the link layer and the + capture container play no part in it. + + """ + + def setUp(self) -> None: + purge_modules(['pcapkit']) + + def _parse(self, next_header: int, extensions: bytes): + from pcapkit.protocols.internet.ipv6 import IPv6 + + data = build_ipv6(next_header, extensions + UDP_HEADER + UDP_BODY) + return IPv6(data, len(data)) + + def test_transport_header_is_read_after_the_extension_header_chain(self) -> None: + cases = [ + # chain, first next-header, extension header bytes, + # expected extension headers, expected hdr_len + ('none', NH_UDP, b'', [], 40), + ('fragment', NH_FRAG, build_fragment_header(NH_UDP), ['IPv6_Frag'], 48), + ('hop-by-hop, destination options', NH_HOPOPT, + build_option_header(NH_OPTS) + build_option_header(NH_UDP), + ['HOPOPT', 'IPv6_Opts'], 56), + ('hop-by-hop, routing, fragment', NH_HOPOPT, + build_option_header(NH_ROUTE) + build_routing_header(NH_FRAG) + + build_fragment_header(NH_UDP), + ['HOPOPT', 'IPv6_Route', 'IPv6_Frag'], 72), + ] + + for chain, next_header, extensions, exthdrs, hdr_len in cases: + with self.subTest(chain=chain): + ipv6 = self._parse(next_header, extensions) + udp = ipv6.payload + + self.assertEqual([key.name for key in ipv6.extension_headers.keys()], exthdrs) + self.assertEqual(ipv6.info.hdr_len, hdr_len) + self.assertEqual(ipv6.info.raw_len, len(UDP_HEADER) + len(UDP_BODY)) + self.assertEqual(ipv6.info.protocol.name, 'UDP') + + self.assertEqual(type(udp).__name__, 'UDP') + self.assertEqual(bytes(udp.packet.header), UDP_HEADER) + self.assertEqual(int(udp.info.srcport), UDP_SRCPORT) + self.assertEqual(int(udp.info.dstport), UDP_DSTPORT) + self.assertEqual(udp.info.len, len(UDP_HEADER) + len(UDP_BODY)) + self.assertEqual(bytes(udp.payload.info.packet), UDP_BODY) + + if 'IPv6_Frag' in exthdrs: + # the fragment record and the next layer describe the same + # octets, the ones following the whole chain + self.assertEqual(len(ipv6.info.fragment.header), hdr_len) + self.assertEqual(bytes(ipv6.info.fragment.payload), + UDP_HEADER + UDP_BODY) + + if __name__ == '__main__': unittest.main() From 0ce947dfa4f854a625412043068959f5fe61edb0 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Mon, 14 Sep 2026 15:03:42 +0000 Subject: [PATCH 2/2] tests: correct ipv6 fixture path references Co-authored-by: JarryShaw <15666417+JarryShaw@users.noreply.github.com> --- tests/protocols/internet/test_ipv6_extension_runtime.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/protocols/internet/test_ipv6_extension_runtime.py b/tests/protocols/internet/test_ipv6_extension_runtime.py index bb9c513ae4..75fc79329f 100644 --- a/tests/protocols/internet/test_ipv6_extension_runtime.py +++ b/tests/protocols/internet/test_ipv6_extension_runtime.py @@ -29,7 +29,7 @@ UDP_HEADER = (UDP_SRCPORT.to_bytes(2, 'big') + UDP_DSTPORT.to_bytes(2, 'big') + (8 + len(UDP_BODY)).to_bytes(2, 'big') + b'\x00\x00') -#: Fragment identification, matching the one in ``examples/captures/ipv6.pcap``. +#: Fragment identification, matching the one in ``examples/sample/ipv6.pcap``. FRAG_ID = 110308 @@ -131,7 +131,7 @@ class IPv6ExtensionPayloadOffsetTests(unittest.TestCase): """The next layer is read from after the *last* extension header. The datagrams here are built from bytes rather than read out of - ``examples/captures/``: no committed capture carries more than one IPv6 + ``examples/sample/``: no committed capture carries more than one IPv6 extension header, and the captures are pinned byte-for-byte by the rest of the suite, so a chain of several headers cannot be obtained from them. Parsing goes through :class:`~pcapkit.protocols.internet.ipv6.IPv6`