From b86351c52da222954892559f200f4f47abd9df43 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Fri, 9 Oct 2026 23:54:44 -0400 Subject: [PATCH] fix(protocols): a later IPv4 fragment's data is raw, not an upper-layer header (#1553) A datagram with a non-zero Fragment Offset now passes no next-layer code to _decode_next_layer, so its data is a Raw leaf with no protocol name, as Wireshark shows it, instead of a bogus UDP/TCP header read from mid-datagram octets. info.protocol, the lengths and packet.payload are unchanged, so reassembly and every round trip are too. First fragments and unfragmented datagrams are still dissected. Mirrors #1552's IPv6 fix for #1545. --- pcapkit/protocols/internet/ipv4.py | 12 +- .../test_ipv4_later_fragment_raw_unit.py | 162 ++++++++++++++++++ tests/protocols/internet/test_ipv4_unit.py | 8 +- 3 files changed, 179 insertions(+), 3 deletions(-) create mode 100644 tests/protocols/internet/test_ipv4_later_fragment_raw_unit.py diff --git a/pcapkit/protocols/internet/ipv4.py b/pcapkit/protocols/internet/ipv4.py index c1d3ab3888..23f419d50a 100644 --- a/pcapkit/protocols/internet/ipv4.py +++ b/pcapkit/protocols/internet/ipv4.py @@ -376,7 +376,17 @@ def read(self, length: 'Optional[int]' = None, *, # pylint: disable=unused-argu else: _plen = 0 - return self._decode_next_layer(ipv4, ipv4.protocol, _plen, packet=__packet__) + # NOTE: A later fragment, one with a non-zero Fragment Offset, carries a + # slice of the datagram's data from that offset on, and that slice + # starts with no upper-layer header (:rfc:`791#section-3.2`), though + # ``protocol`` above still names the one the first fragment carries. No + # code is dispatched on, so the registry's fallback leaves the octets as + # :class:`~pcapkit.protocols.misc.raw.Raw` with no protocol to alias it + # as, and ``'UDP' in frame`` does not hold of a slice of UDP payload + # (:issue:`1553`). Reassembly takes ``packet.payload``, the octets as + # captured, which this leaves as they were. + return self._decode_next_layer(ipv4, None if ipv4.offset else ipv4.protocol, _plen, + packet=__packet__) def make(self, tos_pre: 'Enum_ToSPrecedence | StdlibEnum | AenumEnum | int | str' = Enum_ToSPrecedence.Routine, diff --git a/tests/protocols/internet/test_ipv4_later_fragment_raw_unit.py b/tests/protocols/internet/test_ipv4_later_fragment_raw_unit.py new file mode 100644 index 0000000000..9f3d7699a9 --- /dev/null +++ b/tests/protocols/internet/test_ipv4_later_fragment_raw_unit.py @@ -0,0 +1,162 @@ +# -*- coding: utf-8 -*- +"""A later IPv4 fragment's data is raw, not an upper-layer header. + +GitHub issue #1553: :class:`~pcapkit.protocols.internet.ipv4.IPv4` handed the +payload of a datagram with a non-zero Fragment Offset to the upper-layer +dispatch on its Protocol field, so a slice from the middle of a UDP or TCP +datagram read as a UDP or TCP header of its own -- ports and lengths taken from +payload octets. It is the IPv4 counterpart of #1545. + +:rfc:`791#section-3.2` puts the upper-layer header in the first fragment only. +A later one carries a slice of the datagram's data from its offset on, and that +slice starts with no header, so it is now a +:class:`~pcapkit.protocols.misc.raw.Raw` leaf, as Wireshark shows it (``data``, +under ``-o ip.defragment:FALSE``). A first fragment (offset zero, MF set) and +an unfragmented datagram are still dissected as before. + +Every case builds its own octets in memory and reads no capture. Classes are +imported inside each test, after :func:`~tests._support.reimport_once_per_class`. + +""" + +import unittest + +from tests._support import reimport_once_per_class + +#: Source and destination addresses. +ADDRS = bytes((1, 1, 1, 1, 2, 2, 2, 2)) + +#: Upper layers by name: the Protocol code, the protocol's alias, and a header +#: with payload that dissects as that protocol when it is one. +UPPER = { + 'UDP': (17, 'UDP', bytes.fromhex('c822138900280000') + bytes(range(0x40, 0x60))), + 'TCP': (6, 'TCP', bytes.fromhex('9c40000900000001000000005000ffff0000000078')), +} + +#: Fragment Offsets of a later fragment, in 8-octet units: the smallest, the +#: issue's 185 (1480 octets, one Ethernet MTU's worth), and the largest. +LATER_OFFSETS = (1, 185, 0x1fff) + +#: IPv4 options, padded to a 4-octet boundary: none, and NOP NOP NOP EOOL. +OPTIONS = (b'', bytes((1, 1, 1, 0))) + + +def datagram(code: int, payload: bytes, *, offset: int = 0, more: bool = False, + options: bytes = b'') -> bytes: + """An IPv4 datagram with Protocol ``code``, at ``offset`` 8-octet units.""" + ihl = 5 + len(options) // 4 + return (bytes((0x40 | ihl, 0)) + (ihl * 4 + len(payload)).to_bytes(2, 'big') + + (0x1234).to_bytes(2, 'big') + ((more << 13) | offset).to_bytes(2, 'big') + + bytes((64, code)) + b'\x00\x00' + ADDRS + options + payload) + + +class TestIPv4LaterFragmentRaw(unittest.TestCase): + """Pin a later fragment's data as raw, and a first one's as dissected.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _parse(self, data: bytes) -> 'object': + from pcapkit.protocols.internet.ipv4 import IPv4 + + ipv4 = IPv4(data) + self.assertEqual(ipv4.data, data) + return ipv4 + + def assert_rebuilds(self, ipv4: 'object') -> None: + """``ipv4`` rebuilds byte for byte from its info and its info as a dict.""" + from pcapkit.protocols.internet.ipv4 import IPv4 + + self.assertEqual(IPv4.from_data(ipv4.info).data, ipv4.data) + self.assertEqual(IPv4.from_data(ipv4.info.to_dict()).data, ipv4.data) + + def test_later_fragment_data_is_raw(self) -> None: + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.misc.raw import Raw + + for name, (code, alias, upper) in UPPER.items(): + for options in OPTIONS: + for offset in LATER_OFFSETS: + for more in (True, False): + with self.subTest(upper=name, options=options.hex(), offset=offset, more=more): + ipv4 = self._parse(datagram(code, upper, offset=offset, more=more, + options=options)) + self.assertNotIn(alias, ipv4) + self.assertEqual(str(ipv4.protochain), 'IPv4:Raw') + self.assertIs(type(ipv4.payload), Raw) + self.assertIsNone(ipv4.payload.info.protocol) + self.assertEqual(ipv4.payload.data, upper) + + # the Protocol field still names the first fragment's + # upper layer, and what reassembly reads is untouched + self.assertEqual(ipv4.info.protocol, TransType(code)) + self.assertEqual(ipv4.info.offset, offset * 8) + self.assertEqual(ipv4.info.flags.mf, more) + self.assertEqual(ipv4.info.hdr_len, 20 + len(options)) + self.assertEqual(ipv4.info.len, 20 + len(options) + len(upper)) + self.assertEqual(bytes(ipv4.packet.payload), upper) + self.assert_rebuilds(ipv4) + + def test_first_and_unfragmented_datagrams_are_dissected(self) -> None: + # offset zero with MF set is a first fragment; with MF clear, the whole + # datagram + for name, (code, alias, upper) in UPPER.items(): + for options in OPTIONS: + for more in (True, False): + with self.subTest(upper=name, options=options.hex(), more=more): + ipv4 = self._parse(datagram(code, upper, more=more, options=options)) + self.assertIn(alias, ipv4) + self.assertEqual(str(ipv4.protochain), f'IPv4:{alias}:Raw') + self.assertEqual(ipv4.payload.data, upper) + self.assertEqual(ipv4.info.protocol, code) + self.assert_rebuilds(ipv4) + + def test_issue_repro_later_fragment_in_a_frame(self) -> None: + # the issue's reproduction, carried in an Ethernet frame as a capture + # would carry it: ``'UDP' in frame`` no longer holds + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.link.ethernet import Ethernet + from pcapkit.protocols.misc.raw import Raw + + made = IPv4(protocol='UDP', offset=185, payload=Raw(bytes(range(40))), + src='1.1.1.1', dst='2.2.2.2') + ipv4 = self._parse(made.data) + self.assertEqual(ipv4.info.offset, 1480) + self.assertEqual(str(ipv4.protochain), 'IPv4:Raw') + self.assertEqual(ipv4.payload.data, bytes(range(40))) + self.assert_rebuilds(ipv4) + + wire = bytes.fromhex('020000000002' '020000000001' '0800') + made.data + frame = Ethernet(wire) + self.assertEqual(frame.data, wire) + self.assertNotIn('UDP', frame) + self.assertEqual(str(frame.protochain), 'Ethernet:IPv4:Raw') + self.assertEqual(Ethernet.from_data(frame.info.to_dict()).data, wire) + + def test_made_later_fragment_parses_as_raw(self) -> None: + # make -> parse: a later fragment built with a UDP object as its payload + # writes the UDP octets, reads back as raw (as IPv6 does since #1545), + # and rebuilds to the octets made + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.misc.raw import Raw + from pcapkit.protocols.transport.udp import UDP + + upper = UDP(srcport=51234, dstport=5001, payload=bytes(range(0x40, 0x60))) + for options in (None, [bytes((1, 1, 1, 0))]): + with self.subTest(options=options): + kwargs = {} if options is None else {'options': options} + made = IPv4(protocol='UDP', offset=185, mf=True, payload=upper, + src='1.1.1.1', dst='2.2.2.2', **kwargs) + self.assertEqual(made.data[made.info.hdr_len:], upper.data) + self.assertNotIn('UDP', made) + self.assertIs(type(made.payload), Raw) + + ipv4 = self._parse(made.data) + self.assertNotIn('UDP', ipv4) + self.assertIs(type(ipv4.payload), Raw) + self.assertEqual(ipv4.payload.data, upper.data) + self.assert_rebuilds(ipv4) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/internet/test_ipv4_unit.py b/tests/protocols/internet/test_ipv4_unit.py index 73d9de850e..283c8dbca7 100644 --- a/tests/protocols/internet/test_ipv4_unit.py +++ b/tests/protocols/internet/test_ipv4_unit.py @@ -1033,7 +1033,10 @@ def test_ipv4_properties_read_and_make_cover_packet_paths(self) -> None: self.assertEqual(decoded_ip.hdr_len, 28) self.assertEqual(decoded_ip.options[OptionNumber.SID].sid, 55) self.assertEqual(decoded_ip.offset, 8) - self.assertEqual(next_type, TransType.TCP) + # a later fragment's data starts with no TCP header, so no upper layer + # is dispatched on, though ``protocol`` still names it (#1553) + self.assertIsNone(next_type) + self.assertEqual(decoded_ip.protocol, TransType.TCP) self.assertEqual(payload_length, 4) proto.__header__ = ipv4_schema.IPv4( @@ -1052,8 +1055,9 @@ def test_ipv4_properties_read_and_make_cover_packet_paths(self) -> None: ) proto._decode_next_layer.reset_mock() self.assertEqual(proto.read(length=24), 'decoded') - decoded_no_options = proto._decode_next_layer.call_args.args[0] + decoded_no_options, next_type, _ = proto._decode_next_layer.call_args.args self.assertFalse(hasattr(decoded_no_options, 'options')) + self.assertEqual(next_type, TransType.UDP) # offset 0: dispatched proto.__header__ = ipv4_schema.IPv4( vihl={'version': 6, 'ihl': 5},