From 402d851c8c457405267c94035801a6a49ed8e079 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Tue, 6 Oct 2026 23:03:29 -0400 Subject: [PATCH] fix(protocols): a payload layer parsed from a short stream rebuilds default-filled octets Every _import_next_layer picked NoPayload only for a declared length of zero, so a payload captured short of the next layer's header was parsed into zero-filled fields that from_data then rebuilt. One shared helper, ProtocolBase._parse_next_layer, now backs the ProtocolBase, Internet and IPv6 implementations: no captured octets gives NoPayload, and fewer than the next layer's __length_hint__ gives Raw, unless the header it parsed fits anyway (the L2TPv2 hint is its longest form). test.pcapng frame 3, cut inside the IPv4 header, is now Ethernet:Raw, so the two tests that pinned its zero-filled IPv4 rebuild are updated. Closes #1170 --- pcapkit/protocols/internet/internet.py | 9 +- pcapkit/protocols/internet/ipv6.py | 8 +- pcapkit/protocols/protocol.py | 51 +++++++- ...t_ipv4_header_truncated_payload_runtime.py | 35 +++--- .../test_declared_length_roundtrip_runtime.py | 22 ++-- .../test_next_layer_short_header_unit.py | 116 ++++++++++++++++++ 6 files changed, 199 insertions(+), 42 deletions(-) create mode 100644 tests/protocols/test_next_layer_short_header_unit.py diff --git a/pcapkit/protocols/internet/internet.py b/pcapkit/protocols/internet/internet.py index e311a22aa1..f6514b0ed1 100644 --- a/pcapkit/protocols/internet/internet.py +++ b/pcapkit/protocols/internet/internet.py @@ -265,14 +265,13 @@ def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *, # if length is None: length = len(file_) - if length == 0: + if min(length, len(file_)) == 0: from pcapkit.protocols.misc.null import NoPayload as protocol # isort: skip # pylint: disable=import-outside-toplevel elif self._sigterm: from pcapkit.protocols.misc.raw import Raw as protocol # isort: skip # pylint: disable=import-outside-toplevel else: protocol = self._lookup_next_layer(self.__proto__, proto) - next_ = protocol(file_, length, version=version, extension=extension, # type: ignore[abstract] - alias=proto, packet=packet, layer=self._exlayer, protocol=self._exproto, - __context__=self._exctx) - return next_ + return self._parse_next_layer(protocol, file_, length, version=version, extension=extension, + alias=proto, packet=packet, layer=self._exlayer, + protocol=self._exproto, __context__=self._exctx) diff --git a/pcapkit/protocols/internet/ipv6.py b/pcapkit/protocols/internet/ipv6.py index 1cb9c82f2d..8e73d3e8df 100644 --- a/pcapkit/protocols/internet/ipv6.py +++ b/pcapkit/protocols/internet/ipv6.py @@ -520,7 +520,7 @@ def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *, # if length is None: length = len(file_) - if length == 0: + if min(length, len(file_)) == 0: from pcapkit.protocols.misc.null import \ NoPayload as protocol # isort: skip # pylint: disable=import-outside-toplevel elif self._sigterm: @@ -530,9 +530,9 @@ def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *, # protocol = self._lookup_next_layer(self.__proto__, proto) try: - next_ = protocol(file_, length, version=version, extension=extension, # type: ignore[abstract] - alias=proto, packet=packet, layer=self._exlayer, protocol=self._exproto, - __context__=self._exctx) + next_ = self._parse_next_layer(protocol, file_, length, version=version, extension=extension, + alias=proto, packet=packet, layer=self._exlayer, + protocol=self._exproto, __context__=self._exctx) except Exception as exc: from pcapkit.protocols.internet.ipv6_ext import \ IPv6_Ext # isort: skip # pylint: disable=import-outside-toplevel diff --git a/pcapkit/protocols/protocol.py b/pcapkit/protocols/protocol.py index 9958c859eb..5acb10c9cd 100644 --- a/pcapkit/protocols/protocol.py +++ b/pcapkit/protocols/protocol.py @@ -1820,16 +1820,59 @@ def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *, if length is None: length = len(file_) - if length == 0: + if min(length, len(file_)) == 0: from pcapkit.protocols.misc.null import NoPayload as protocol # isort: skip # pylint: disable=import-outside-toplevel elif self._sigterm: from pcapkit.protocols.misc.raw import Raw as protocol # isort: skip # pylint: disable=import-outside-toplevel else: protocol = self._lookup_next_layer(self.__proto__, proto) - next_ = protocol(file_, length, alias=proto, packet=packet, - layer=self._exlayer, protocol=self._exproto, - __context__=self._exctx) # type: ignore[abstract] + return self._parse_next_layer(protocol, file_, length, alias=proto, packet=packet, + layer=self._exlayer, protocol=self._exproto, + __context__=self._exctx) + + @staticmethod + def _parse_next_layer(klass: 'Type[ProtocolBase]', file_: 'bytes', length: 'int', + **kwargs: 'Any') -> 'ProtocolBase': + r"""Parse the next layer, or keep it as raw octets if its header was not captured. + + Arguments: + klass: next layer protocol class + file\_: payload of the current layer + length: valid (*non-padding*) length + **kwargs: keyword arguments for the next layer's constructor + + Returns: + Instance of ``klass``, or of :class:`~pcapkit.protocols.misc.raw.Raw` + when fewer octets were captured than its header needs. + + Note: + :meth:`Schema.unpack ` + zero-fills a field the stream ends inside, so a header parsed from + too few octets is filled with values that were never on the wire, and + :meth:`from_data ` rebuilds them (:issue:`1170`). + Such a payload is kept as :class:`~pcapkit.protocols.misc.raw.Raw` + instead, as an unparseable one is. + + The captured octets are what ``file_`` holds, which may be fewer than + the ``length`` the current header declares. They are short of the + header when there are fewer than the next layer's + :meth:`__length_hint__ `, unless the + header it parsed, :attr:`length ` octets long, fits + in them anyway: for a variable-length header such as + :class:`~pcapkit.protocols.link.l2tpv2.L2TPv2` the hint is the longest + form rather than the shortest. A next layer that reports no hint is + kept as parsed. + + """ + next_ = klass(file_, length, **kwargs) # type: ignore[abstract] + + captured = min(length, len(file_)) + length_hint = getattr(next_, '__length_hint__', None) + hint = None if length_hint is None else length_hint() + if hint is not None and captured < hint and captured < next_.length: + from pcapkit.protocols.misc.raw import Raw # isort: skip # pylint: disable=import-outside-toplevel + next_ = Raw(file_, length, **kwargs) return next_ def _check_term_threshold(self) -> bool: diff --git a/tests/protocols/internet/test_ipv4_header_truncated_payload_runtime.py b/tests/protocols/internet/test_ipv4_header_truncated_payload_runtime.py index 10d0a79096..eaebb18af8 100644 --- a/tests/protocols/internet/test_ipv4_header_truncated_payload_runtime.py +++ b/tests/protocols/internet/test_ipv4_header_truncated_payload_runtime.py @@ -3,11 +3,12 @@ GitHub issue #1167. :meth:`IPv4.read ` handed the next layer the length the header declared even when the capture held -none of its octets. ``test.pcapng`` frame 3 is cut inside the IPv4 header (18 of -20 octets), so its UDP layer was parsed from an empty stream into an all-zero -header, and :meth:`~pcapkit.protocols.protocol.ProtocolBase.from_data` rebuilt -those 8 octets as if they had been on the wire. Such a payload is now a -:class:`~pcapkit.protocols.misc.null.NoPayload`. +none of its octets, so its UDP layer was parsed from an empty stream into an +all-zero header, and :meth:`~pcapkit.protocols.protocol.ProtocolBase.from_data` +rebuilt those 8 octets as if they had been on the wire. Such a payload is now a +:class:`~pcapkit.protocols.misc.null.NoPayload`. ``test.pcapng`` frame 3, cut +inside the IPv4 header itself (18 of 20 octets), is kept as raw octets +(:issue:`1170`). The capture case reads a generated sample, so this module belongs to the fixture-dependent tier. The in-memory cases need no capture. @@ -36,24 +37,20 @@ def setUp(self) -> None: def test_capture_cut_inside_the_header(self) -> None: from pcapkit.interface import extract - from pcapkit.protocols.internet.ipv4 import IPv4 - from pcapkit.protocols.misc.null import NoPayload + from pcapkit.protocols.link.ethernet import Ethernet + from pcapkit.protocols.misc.raw import Raw extractor = extract(fin=sample_path('test.pcapng'), fout='/tmp/out', format='tree', store=True, nofile=True) self.addCleanup(close_extractor, extractor) - parsed = list(extractor.frame)[2][IPv4] - - self.assertEqual(parsed.data, bytes.fromhex('4500012ca8370000fa11178a00000000ffff')) - self.assertIsInstance(parsed.payload, NoPayload) - self.assertEqual(str(parsed.protochain), 'IPv4') - self.assertNotIn('udp', parsed.info) - - # The rebuild is the header alone: the 18 captured octets, then the two - # the parse already filled into ``dst``, and no UDP header after them. - rebuilt = IPv4.from_data(parsed.info) - self.assertEqual(rebuilt.data, parsed.data + b'\x00\x00') - self.assertEqual(rebuilt.info.len, 300) + parsed = list(extractor.frame)[2][Ethernet] + + # 18 octets are short of the 20-octet IPv4 header, so they are kept as + # raw octets (:issue:`1170`) rather than parsed into a ``dst`` whose low + # half is zero-filled, and the rebuild is exactly what was captured. + self.assertIsInstance(parsed.payload, Raw) + self.assertEqual(parsed.payload.data, bytes.fromhex('4500012ca8370000fa11178a00000000ffff')) + self.assertEqual(Ethernet.from_data(parsed.info).data, parsed.data) def test_header_without_payload_octets(self) -> None: from pcapkit.protocols.internet.ipv4 import IPv4 diff --git a/tests/protocols/test_declared_length_roundtrip_runtime.py b/tests/protocols/test_declared_length_roundtrip_runtime.py index 2d868af986..24eef94440 100644 --- a/tests/protocols/test_declared_length_roundtrip_runtime.py +++ b/tests/protocols/test_declared_length_roundtrip_runtime.py @@ -98,20 +98,22 @@ def test_complete_frames_still_round_trip(self) -> None: def test_header_truncated_frame_keeps_its_total_length(self) -> None: """``test.pcapng`` frame 3 is cut inside the IPv4 header itself. - Only 18 of the 20 header octets were recorded, so the rebuild cannot be - byte for byte -- it writes a whole header. What it must not do is replace - the declared 300 with the 18 that were kept. + Only 18 of the 20 header octets were recorded, so they are kept as raw + octets rather than parsed into a header (:issue:`1170`). The rebuild is + then byte for byte, and keeps the declared 300 rather than the 18 that + were kept. """ - from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.link.ethernet import Ethernet + from pcapkit.protocols.misc.raw import Raw - parsed = self._frames('test.pcapng')[2][IPv4] - self.assertEqual(parsed.info.len, 300) - self.assertEqual(len(parsed.data), 18) + parsed = self._frames('test.pcapng')[2][Ethernet] + self.assertIsInstance(parsed.payload, Raw) + self.assertEqual(len(parsed.payload.data), 18) + self.assertEqual(parsed.payload.data[2:4], (300).to_bytes(2, 'big')) - rebuilt = IPv4.from_data(parsed.info) - self.assertEqual(rebuilt.info.len, 300) - self.assertEqual(rebuilt.data[:18], parsed.data) + rebuilt = Ethernet.from_data(parsed.info) + self.assertEqual(rebuilt.data, parsed.data) @unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') diff --git a/tests/protocols/test_next_layer_short_header_unit.py b/tests/protocols/test_next_layer_short_header_unit.py new file mode 100644 index 0000000000..e2f0f3b20c --- /dev/null +++ b/tests/protocols/test_next_layer_short_header_unit.py @@ -0,0 +1,116 @@ +# -*- coding: utf-8 -*- +"""A payload captured short of the next layer's header is kept as raw octets. + +GitHub issue #1170. Every ``_import_next_layer`` parsed the next layer from +whatever octets were captured, however few, and only picked +:class:`~pcapkit.protocols.misc.null.NoPayload` for a declared length of zero. +:meth:`Schema.unpack ` zero-fills +the fields the stream ends inside, so +:meth:`~pcapkit.protocols.protocol.ProtocolBase.from_data` rebuilt a header that +was never captured. A payload with no captured octets is now a ``NoPayload``, +and one short of the next layer's header a +:class:`~pcapkit.protocols.misc.raw.Raw`. + +""" +from __future__ import annotations + +import importlib.util +import unittest + +from tests._support import reimport_once_per_class + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + +#: A complete 20-octet IPv4 header declaring a 300-octet UDP datagram. +IPV4 = bytes.fromhex('4500012ca8370000fa11178a00000000ffffffff') +#: A complete 40-octet IPv6 header declaring a 168-octet UDP payload. +IPV6 = bytes.fromhex('6000000000a81140') + bytes(15) + b'\x01' + bytes(15) + b'\x02' +#: A UDP header (port 53 to 53, length 32) and the first of its payload octets. +UDP = bytes.fromhex('0035003500200000ab') +#: An Ethernet header carrying IPv4. +ETHERNET = bytes.fromhex('ffffffffffff0000000000010800') + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class ShortNextLayerHeaderTests(unittest.TestCase): + """The rebuild of a short payload is the octets that were captured.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_ipv6_header_without_payload_octets(self) -> None: + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.misc.null import NoPayload + + parsed = IPv6(IPV6, len(IPV6)) + self.assertIsInstance(parsed.payload, NoPayload) + self.assertEqual(IPv6.from_data(parsed.info).data, IPV6) + + def test_ipv6_partial_udp_header(self) -> None: + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.misc.raw import Raw + + packet = IPV6 + UDP[:4] + parsed = IPv6(packet, len(packet)) + self.assertIsInstance(parsed.payload, Raw) + self.assertEqual(IPv6.from_data(parsed.info).data, packet) + + def test_ipv4_partial_udp_header(self) -> None: + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.misc.raw import Raw + + for size in (1, 4, 7): + with self.subTest(size=size): + packet = IPV4 + UDP[:size] + parsed = IPv4(packet, len(packet)) + self.assertIsInstance(parsed.payload, Raw) + self.assertEqual(parsed.payload.data, UDP[:size]) + self.assertEqual(IPv4.from_data(parsed.info).data, packet) + + def test_udp_header_without_payload_octets(self) -> None: + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.misc.null import NoPayload + from pcapkit.protocols.transport.udp import UDP as UDP_ + + packet = IPV4 + UDP[:8] + parsed = IPv4(packet, len(packet)) + self.assertIsInstance(parsed.payload, UDP_) + self.assertIsInstance(parsed.payload.payload, NoPayload) + self.assertEqual(IPv4.from_data(parsed.info).data, packet) + + def test_ipv4_header_cut_short(self) -> None: + from pcapkit.protocols.link.ethernet import Ethernet + from pcapkit.protocols.misc.raw import Raw + + frame = ETHERNET + IPV4[:18] + parsed = Ethernet(frame, len(frame)) + self.assertIsInstance(parsed.payload, Raw) + self.assertEqual(Ethernet.from_data(parsed.info).data, frame) + + def test_header_shorter_than_its_length_hint_still_parses(self) -> None: + # A 12-octet L2TPv2 ZLB acknowledgement (flags T, L and S) is a whole + # header, though ``L2TPv2.__length_hint__`` reports its 16-octet form. + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.link.l2tpv2 import L2TPv2 + + zlb = bytes.fromhex('c802000c000100000001' '0002') + udp = bytes.fromhex('06a506a5') + (8 + len(zlb)).to_bytes(2, 'big') + b'\x00\x00' + zlb + ipv4 = IPV4[:2] + (20 + len(udp)).to_bytes(2, 'big') + IPV4[4:] + packet = ipv4 + udp + parsed = IPv4(packet, len(packet)) + self.assertIsInstance(parsed.payload.payload, L2TPv2) + self.assertEqual(parsed.payload.payload.info.length, 12) + + def test_captured_header_is_still_parsed(self) -> None: + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.transport.udp import UDP as UDP_ + + packet = IPV4 + UDP + parsed = IPv4(packet, len(packet)) + self.assertIsInstance(parsed.payload, UDP_) + self.assertEqual(parsed.payload.info.srcport.port, 53) + + +if __name__ == '__main__': + unittest.main()