From 821aa1003d515ca09c0e2ed1d0442588360b56d3 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Fri, 9 Oct 2026 22:30:00 -0400 Subject: [PATCH] fix(protocols): an IPv4 Total Length of 0 is TSO, and one inside the header is bogus (#1547) - A Total Length of 0, as TCP segmentation offload leaves it on the sending host, handed the next layer 0 - hdr_len octets, so TCP fell back to Raw and the flow was never traced. The payload is now the rest of the captured frame with no trailer, as Wireshark and dpkt read it. - Any other Total Length shorter than the header is Wireshark's "Bogus IP length": no payload layer and no negative length reaching one; the octets after the header are kept as the trailer. - Both write the Total Length back as it was, so parse -> rebuild, make -> parse and from_data(info.to_dict()) stay byte-exact. A Total Length past the capture is unchanged. tests/protocols/internet, schema, foundation/reassembly and the capture round trip pass. --- pcapkit/protocols/internet/ipv4.py | 14 +- pcapkit/protocols/schema/internet/ipv4.py | 8 +- ...test_ipv4_tso_total_length_1547_runtime.py | 280 ++++++++++++++++++ 3 files changed, 299 insertions(+), 3 deletions(-) create mode 100644 tests/protocols/internet/test_ipv4_tso_total_length_1547_runtime.py diff --git a/pcapkit/protocols/internet/ipv4.py b/pcapkit/protocols/internet/ipv4.py index 3464a3cab5..c1d3ab3888 100644 --- a/pcapkit/protocols/internet/ipv4.py +++ b/pcapkit/protocols/internet/ipv4.py @@ -359,8 +359,20 @@ def read(self, length: 'Optional[int]' = None, *, # pylint: disable=unused-argu # octets, and parsing the next layer from none would default-fill a # header that was never captured, which ``from_data`` then rebuilds # (:issue:`1167`). A zero length makes it a ``NoPayload`` instead. + # + # A Total Length of 0 is what TCP segmentation offload leaves in a + # capture taken on the sending host: the payload is then the rest of + # the captured frame, with no trailer, as Wireshark ("presumed TSO") and + # dpkt read it. Any other Total Length shorter than the header is + # Wireshark's "Bogus IP length", which it dissects no further: no + # payload, and the octets after the header are the trailer, as they are + # for a Total Length equal to the header's. Either way the rebuild + # writes the Total Length back as it was (:issue:`1547`). if len(self) > ipv4.hdr_len: - _plen = ipv4.len - ipv4.hdr_len + if ipv4.len == 0: + _plen = len(self) - ipv4.hdr_len + else: + _plen = max(ipv4.len - ipv4.hdr_len, 0) else: _plen = 0 diff --git a/pcapkit/protocols/schema/internet/ipv4.py b/pcapkit/protocols/schema/internet/ipv4.py index 5c1a4b3df3..36aee6b79d 100644 --- a/pcapkit/protocols/schema/internet/ipv4.py +++ b/pcapkit/protocols/schema/internet/ipv4.py @@ -732,8 +732,12 @@ class IPv4(Schema): ) #: Padding. padding: 'bytes' = PaddingField(length=lambda pkt: pkt.get('__option_padding__', 0)) - #: Payload. - payload: 'bytes' = PayloadField(length=lambda pkt: pkt['length'] - pkt['vihl']['ihl'] * 4) + #: Payload. A Total Length of 0, as TCP segmentation offload leaves it in a + #: capture taken on the sending host, takes the rest of the data and leaves + #: no trailer; any other Total Length shorter than the header leaves the + #: payload empty and the rest to the trailer (:issue:`1547`). + payload: 'bytes' = PayloadField(length=lambda pkt: ( + -1 if pkt['length'] == 0 else max(pkt['length'] - pkt['vihl']['ihl'] * 4, 0))) #: Octets captured past the Total Length, such as Ethernet minimum-frame #: padding. A :class:`~pcapkit.corekit.fields.misc.PayloadField` with no #: length takes whatever the payload leaves on unpack and writes its value diff --git a/tests/protocols/internet/test_ipv4_tso_total_length_1547_runtime.py b/tests/protocols/internet/test_ipv4_tso_total_length_1547_runtime.py new file mode 100644 index 0000000000..895a759a86 --- /dev/null +++ b/tests/protocols/internet/test_ipv4_tso_total_length_1547_runtime.py @@ -0,0 +1,280 @@ +# -*- coding: utf-8 -*- +"""An IPv4 Total Length of 0 is TCP segmentation offload, not a payload of -20. + +GitHub issue #1547. A capture taken on a host that offloads TCP segmentation +(TSO/GSO) records its outgoing IPv4 headers with ``total_length`` 0, since the +NIC fills it in later. :meth:`IPv4.read ` +handed the next layer ``0 - hdr_len`` octets, so TCP fell back to +:class:`~pcapkit.protocols.misc.raw.Raw` and the flow was never traced. A Total +Length of 0 now means the rest of the captured frame, with no trailer, as dpkt +(``ip.py``) and Wireshark ("presumed TSO") read it, and the rebuild writes the 0 +back. Following the owner's ruling to match Wireshark, any other Total Length +shorter than the header is a "Bogus IP length": no payload layer, the octets +after the header kept as the trailer, and the length written back as it was. + +The cases are built in memory from :file:`tcp.pcap` with its four IPv4 frames' +Total Length zeroed, once with the header checksum fixed and once left stale, +since a TSO capture may show either. The module reads a generated sample, so it +belongs to the fixture-dependent tier. + +""" +from __future__ import annotations + +import importlib.util +import os +import struct +import tempfile +import unittest +import warnings +from typing import TYPE_CHECKING +from unittest import mock + +from tests._support import close_extractor, reimport_once_per_class, sample_path + +if TYPE_CHECKING: + from typing import Any + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) +ENGINES = tuple(name for name in ('dpkt', 'scapy') if importlib.util.find_spec(name) is not None) + +#: One-based numbers of the IPv4 frames of ``tcp.pcap``; the rest are IPv6. +IPV4_FRAMES = (1, 2, 6, 7) + +#: The flows every engine traces in ``tcp.pcap``, TSO or not. +FLOWS = [(1, 2, 6, 7), (3, 4, 5)] + + +def _checksum(header: 'bytes') -> 'bytes': + """RFC 1071 Internet checksum of ``header``, its checksum field zeroed.""" + total = sum(struct.unpack(f'!{len(header) // 2}H', header)) + while total >> 16: + total = (total & 0xFFFF) + (total >> 16) + return struct.pack('!H', ~total & 0xFFFF) + + +def _records() -> 'tuple[bytes, list[list[Any]]]': + """``tcp.pcap``'s global header and ``[ts_sec, ts_usec, orig_len, octets]`` records.""" + with open(sample_path('tcp.pcap'), 'rb') as file: + source = file.read() + header, rest, records = source[:24], source[24:], [] + while rest: + ts_sec, ts_usec, incl_len, orig_len = struct.unpack(' 'bytes': + """``frame``, an Ethernet frame carrying IPv4, with the given Total Length.""" + header = bytearray(frame[14:34]) + header[2:4] = struct.pack('!H', total_length) + if fix_checksum: + header[10:12] = b'\x00\x00' + header[10:12] = _checksum(bytes(header)) + return frame[:14] + bytes(header) + frame[34:] + + +def _tso_records(*, fix_checksum: 'bool') -> 'tuple[bytes, list[list[Any]]]': + """``tcp.pcap`` with every IPv4 frame's Total Length zeroed.""" + header, records = _records() + for number in IPV4_FRAMES: + records[number - 1][3] = _with_total_length(records[number - 1][3], 0, fix_checksum=fix_checksum) + return header, records + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class TSOTotalLengthTests(unittest.TestCase): + """A zero Total Length reads the rest of the frame and is written back as 0.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _parse(self, octets: 'bytes') -> 'Any': + from pcapkit.protocols.link.ethernet import Ethernet + from pcapkit.utilities.warnings import SchemaWarning + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + parsed = Ethernet(octets, len(octets)) + self.assertEqual([str(w.message) for w in caught if issubclass(w.category, SchemaWarning)], []) + return parsed + + def test_tcp_is_parsed_and_every_rebuild_is_byte_exact(self) -> None: + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.link.ethernet import Ethernet + from pcapkit.protocols.transport.tcp import TCP + + _, original = _records() + for fix_checksum in (True, False): + _, records = _tso_records(fix_checksum=fix_checksum) + for number in IPV4_FRAMES: + with self.subTest(frame=number, checksum='fixed' if fix_checksum else 'stale'): + octets = records[number - 1][3] + parsed = self._parse(octets) + ipv4 = parsed.payload + + self.assertIsInstance(ipv4, IPv4) + self.assertIsInstance(ipv4.payload, TCP) + self.assertEqual(ipv4.info.len, 0) + self.assertNotIn('trailer', ipv4.info) + self.assertEqual(ipv4.info.checksum, octets[24:26]) + self.assertEqual(ipv4.payload.data, octets[34:]) + + # The same TCP header the frame carries with its real length. + expected = self._parse(original[number - 1][3]).payload.payload + self.assertEqual(ipv4.payload.info.to_dict(), expected.info.to_dict()) + + # parse -> rebuild, from the info object and from its dict + self.assertEqual(Ethernet.from_data(parsed.info).data, octets) + self.assertEqual(Ethernet.from_data(parsed.info.to_dict()).data, octets) + self.assertEqual(IPv4.from_data(ipv4.info).data, octets[14:]) + self.assertEqual(IPv4.from_data(ipv4.info.to_dict()).data, octets[14:]) + + first = self._parse(_tso_records(fix_checksum=True)[1][0][3]).payload.payload.info + self.assertEqual((first.srcport.port, first.dstport.port), (22, 53406)) + + def test_make_with_a_zero_total_length_round_trips(self) -> None: + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.transport.tcp import TCP + + segment = _records()[1][0][3][34:] + made = IPv4(total_length=0, protocol='TCP', ttl=64, src='10.20.30.130', dst='10.20.30.131', + payload=segment) + self.assertEqual(made.data[2:4], b'\x00\x00') + self.assertEqual(made.data[20:], segment) + + parsed = IPv4(made.data, len(made.data)) + self.assertIsInstance(parsed.payload, TCP) + self.assertEqual(parsed.info.len, 0) + self.assertEqual(IPv4.from_data(parsed.info).data, made.data) + self.assertEqual(IPv4.from_data(parsed.info.to_dict()).data, made.data) + + # Without an explicit Total Length it is computed, as before. + computed = IPv4(protocol='TCP', payload=segment) + self.assertEqual(struct.unpack('!H', computed.data[2:4])[0], 20 + len(segment)) + + def test_schema_splits_payload_and_trailer(self) -> None: + from pcapkit.protocols.schema.internet.ipv4 import IPv4 as Schema_IPv4 + + frame = _records()[1][0][3] + # 0 is TSO: the rest is payload. 1 to 19 are bogus: the rest is trailer. + for total_length, payload in ((0, frame[34:]), (1, b''), (10, b''), (19, b'')): + with self.subTest(total_length=total_length): + octets = _with_total_length(frame, total_length, fix_checksum=True)[14:] + schema = Schema_IPv4.unpack(octets) + self.assertEqual(schema.length, total_length) + self.assertEqual(schema.payload, payload) + self.assertEqual(schema.payload + schema.trailer, octets[20:]) + self.assertEqual(schema.pack(), octets) + + def test_a_bogus_total_length_is_not_dissected_further(self) -> None: + # A non-zero Total Length shorter than the header is not TSO: Wireshark + # 4.6.9 reports "Total Length: 10 bytes (bogus, less than header length + # 20)" with the expert note "Bogus IP length", and its protocol list + # stops at ``eth:ethertype:ip``. So no payload layer is parsed from it, + # no negative length reaches one, the octets after the header are kept + # as the trailer, and the Total Length is written back as it was. + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.link.ethernet import Ethernet + from pcapkit.protocols.misc.null import NoPayload + + frame = _records()[1][0][3] + lengths: 'list[int]' = [] + decode = IPv4._decode_next_layer + + def spy(self: 'Any', dict_: 'Any', proto: 'Any', length: 'Any' = None, **kwargs: 'Any') -> 'Any': + lengths.append(length) + return decode(self, dict_, proto, length, **kwargs) + + for total_length in (1, 10, 19): + for fix_checksum in (True, False): + with self.subTest(total_length=total_length, checksum='fixed' if fix_checksum else 'stale'): + octets = _with_total_length(frame, total_length, fix_checksum=fix_checksum) + lengths.clear() + with mock.patch.object(IPv4, '_decode_next_layer', spy): + parsed = self._parse(octets) + ipv4 = parsed.payload + + self.assertEqual(lengths, [0]) + self.assertIsInstance(ipv4.payload, NoPayload) + self.assertEqual(str(parsed.protochain), 'Ethernet:IPv4') + self.assertEqual(ipv4.info.len, total_length) + self.assertEqual(ipv4.info.trailer, octets[34:]) + + self.assertEqual(Ethernet.from_data(parsed.info).data, octets) + self.assertEqual(Ethernet.from_data(parsed.info.to_dict()).data, octets) + self.assertEqual(IPv4.from_data(ipv4.info).data, octets[14:]) + self.assertEqual(IPv4.from_data(ipv4.info.to_dict()).data, octets[14:]) + + # make -> parse -> rebuild + made = IPv4(total_length=10, protocol='TCP', trailer=frame[34:]) + self.assertEqual(made.data[2:4], b'\x00\x0a') + parsed_made = IPv4(made.data, len(made.data)) + self.assertIsInstance(parsed_made.payload, NoPayload) + self.assertEqual(IPv4.from_data(parsed_made.info.to_dict()).data, made.data) + + def test_a_total_length_past_the_capture_is_unchanged(self) -> None: + # The other direction, which Wireshark reports as "IPv4 total length + # exceeds packet length (64 bytes)" while still dissecting TCP: the + # declared length is kept for the rebuild (#1155), and #1547 leaves it be. + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.link.ethernet import Ethernet + from pcapkit.protocols.transport.tcp import TCP + + frame = _records()[1][0][3] + octets = _with_total_length(frame, 100, fix_checksum=True) + parsed = self._parse(octets) + ipv4 = parsed.payload + self.assertIsInstance(ipv4.payload, TCP) + self.assertEqual(ipv4.payload.data, octets[34:]) + self.assertEqual(ipv4.info.len, 100) + self.assertNotIn('trailer', ipv4.info) + self.assertEqual(Ethernet.from_data(parsed.info).data, octets) + self.assertEqual(IPv4.from_data(ipv4.info.to_dict()).data, octets[14:]) + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class TSOTraceTests(unittest.TestCase): + """The default engine traces a TSO capture's IPv4 flow, as dpkt and scapy do.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + tmp = tempfile.TemporaryDirectory() # pylint: disable=consider-using-with + self.addCleanup(tmp.cleanup) + self.tmp = tmp.name + + def _write(self, name: 'str', header: 'bytes', records: 'list[list[Any]]') -> 'str': + path = os.path.join(self.tmp, name) + with open(path, 'wb') as file: + file.write(header) + for ts_sec, ts_usec, orig_len, octets in records: + file.write(struct.pack(' 'Any': + from pcapkit import extract + + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + extractor = extract(fin=fin, nofile=True, engine=engine, tcp=True, trace=True, + reassembly=True, trace_fout=tempfile.mkdtemp(dir=self.tmp), + trace_format='json') + self.addCleanup(close_extractor, extractor) + return extractor + + def test_flows_are_traced_and_reassembled(self) -> None: + baseline = self._extract(sample_path('tcp.pcap'), 'default') + expected_tcp = [tuple(datagram.index) for datagram in baseline.reassembly.tcp] + for fix_checksum in (True, False): + fin = self._write(f'tso-{fix_checksum}.pcap', *_tso_records(fix_checksum=fix_checksum)) + for engine in ('default',) + ENGINES: + with self.subTest(engine=engine, checksum='fixed' if fix_checksum else 'stale'): + extractor = self._extract(fin, engine) + self.assertEqual([tuple(flow.index) for flow in extractor.trace.tcp], FLOWS) + self.assertEqual([tuple(datagram.index) for datagram in extractor.reassembly.tcp], + expected_tcp) + + +if __name__ == '__main__': + unittest.main()