fix(reassembly): a TSO fragment's length is its captured length, and RCVBT covers the last block (#1555, #1565)

pcapkit/foundation/reassembly/ip.py

@@ -126,11 +126,14 @@ def reassembly(self, info: 'Packet[_AT]') -> 'None':
         # the header of the fragment at offset zero is the reassembled datagram's
         header = b'' if FO else self._rectify_header(info.header, BUFID[3])
 
-        # initialise buffer with BUFID
+        # initialise buffer with BUFID; ``RCVBT`` holds one entry per 8-octet
+        # block of the data buffer, its partial last block included -- 8191
+        # covered octets 0 to 65527 only, so a fragment reaching octet 65528,
+        # e.g. at the largest Fragment Offset, indexed past it (:issue:`1565`)
         if BUFID not in self._buffer:
             self._buffer[BUFID] = Buffer(
                 TDL=-1,                              # Total Data Length
-                RCVBT=bytearray(8191),              # Fragment Received Bit Table
+                RCVBT=bytearray(8192),              # Fragment Received Bit Table
                 index=[],                           # index record
                 header=header,                      # header buffer
                 datagram=bytearray(65535),          # data buffer
@@ -151,8 +154,34 @@ def reassembly(self, info: 'Packet[_AT]') -> 'None':
         # octets the capture actually holds can be written: a snaplen-truncated
         # fragment carries fewer, and anything past the declared length is not
         # fragment data. The data buffer is preallocated, so the write is also
-        # clipped to it -- its length never changes.
-        length = TL - IHL
+        # clipped to it.
+        #
+        # A Total Length of 0 declares nothing. It is what TCP segmentation
+        # offload leaves in a capture taken on the sending host, and the
+        # datagram then runs to the end of the captured frame -- which is the
+        # payload every adapter hands over (:issue:`1547`). Wireshark ("presumed
+        # TSO") substitutes that length for the field before it reassembles
+        # anything, and so does this, whatever **MF** and the offset say: an
+        # unfragmented one is then whole rather than ``-IHL`` octets long
+        # (:issue:`1555`). Linux's BIG TCP zeroes the field precisely because
+        # the datagram outgrows the 65535 octets it can declare, so the buffer
+        # grows to hold such a datagram rather than clipping it, ``RCVBT`` in
+        # step with it.
+        #
+        # With nothing declared, a snaplen cut cannot be seen in the length.
+        # A fragment with **MF** set must carry a multiple of 8 octets, so its
+        # partial last block is a cut and is left unmarked below. Any other cut
+        # -- an unfragmented frame's, or a final fragment's -- cannot be told
+        # from a whole one, since the frame's original length is not part of
+        # the input, and the datagram comes out ``COMPLETE`` but short.
+        if TL == 0:
+            length = len(info.payload)
+            extent = FO + length
+            if extent > len(buf.datagram):
+                buf.datagram.extend(bytes(extent - len(buf.datagram)))
+                buf.RCVBT.extend(bytes((extent + 7) // 8 - len(buf.RCVBT)))
+        else:
+            length = TL - IHL
         held = max(min(len(info.payload), length, len(buf.datagram) - FO), 0)
         payload = info.payload[:held]
 
@@ -179,9 +208,11 @@ def reassembly(self, info: 'Packet[_AT]') -> 'None':
         # Set RCVBT bits (in 8 octets) for the blocks this fragment actually
         # filled. A fragment held in full marks its last block even when it is
         # short, as the final fragment's is; a truncated one leaves its partial
-        # last block clear, since the octets after the cut are a real hole.
+        # last block clear, since the octets after the cut are a real hole --
+        # as does a non-final fragment with Total Length 0, whose partial last
+        # block can only be a cut.
         start = FO // 8
-        if held == length:
+        if held == length and not (TL == 0 and MF):
             stop = (FO + held + 7) // 8
         else:
             stop = (FO + held) // 8
@@ -192,7 +223,7 @@ def reassembly(self, info: 'Packet[_AT]') -> 'None':
         # get total data length (header excludes); it stays ``-1`` until the
         # fragment with MF clear has arrived, and ``0`` is a valid length
         if not MF:
-            buf.__update__(TDL=TL - IHL + FO)
+            buf.__update__(TDL=length + FO)
         TDL = buf.TDL
 
         # when datagram is reassembled in whole

tests/foundation/reassembly/test_ip_last_fragment_offset_1565_unit.py

@@ -0,0 +1,136 @@
+# -*- coding: utf-8 -*-
+"""IP reassembly buffers a fragment that reaches octet 65,528.
+
+GitHub issue #1565. The data buffer holds 65,535 octets, but the received-bit
+table ``RCVBT`` had only 8,191 entries, one per 8-octet block, which cover
+octets 0 to 65,527. A fragment whose data reached octet 65,528 -- the last
+fragment at the largest Fragment Offset, 8191, among them -- made
+:meth:`IP._detect_conflicts <pcapkit.foundation.reassembly.ip.IP._detect_conflicts>`
+index past the table, and the whole extraction raised :exc:`IndexError`. The
+table now has 8,192 entries, the partial last block included.
+
+Every case builds its fragments in memory and reads no capture. Classes are
+imported inside each test, after :func:`~tests._support.reimport_once_per_class`.
+
+"""
+
+from ipaddress import ip_address
+import unittest
+
+from tests._support import reimport_once_per_class
+
+#: The largest Fragment Offset, 8191 eight-octet units, in octets.
+LAST = 65528
+
+#: The octets before :data:`LAST`, with no zero octet in them, so a hole shows.
+BODY = (bytes(range(1, 256)) * 257)[:LAST]
+
+
+class TestIPReassemblyLastFragmentOffset(unittest.TestCase):
+    """Pin what the IP reassemblers report for a fragment at the largest offset."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+
+    def _packet(self, *, num: int, fo: int, mf: bool, payload: bytes, tl: 'int | None' = None):
+        from pcapkit.const.reg.transtype import TransType
+        from pcapkit.foundation.reassembly.data.ip import Packet
+
+        bufid = (ip_address('192.0.2.1'), ip_address('198.51.100.2'), 7, TransType.UDP)
+        return Packet(bufid, num, fo, 20, mf, 20 + len(payload) if tl is None else tl,
+                      b'H' * 20, bytearray(payload), 1000.0 + num)
+
+    def _classes(self):
+        from pcapkit.foundation.reassembly.ipv4 import IPv4
+        from pcapkit.foundation.reassembly.ipv6 import IPv6
+
+        return IPv4, IPv6
+
+    def _body(self):
+        """Fragments carrying :data:`BODY`, each at most 65,515 octets as a Total Length allows."""
+        return (self._packet(num=1, fo=0, mf=True, payload=BODY[:32768]),
+                self._packet(num=2, fo=32768, mf=True, payload=BODY[32768:]))
+
+    def test_a_lone_fragment_at_the_largest_offset_is_buffered(self) -> None:
+        from pcapkit.foundation.reassembly.data.data import Completion
+
+        for cls in self._classes():
+            for size in (1, 7):
+                for mf in (False, True):
+                    for strict in (True, False):
+                        with self.subTest(cls=cls.__name__, size=size, mf=mf, strict=strict):
+                            tail = b'A' * size
+                            reasm = cls(strict=strict)
+                            reasm(self._packet(num=1, fo=LAST, mf=mf, payload=tail))
+                            datagram, = reasm.datagram
+                            self.assertIs(datagram.completed, Completion.PARTIAL)
+                            self.assertEqual(datagram.index, (1,))
+                            if strict:
+                                run, = datagram.payload
+                                self.assertEqual(run[:size], tail)
+                            elif mf:
+                                # the length is unknown, and the prefix before
+                                # the first hole is empty
+                                self.assertEqual(datagram.payload, b'')
+                            else:
+                                self.assertEqual(datagram.payload, bytes(LAST) + tail)
+
+    def test_the_last_fragment_at_the_largest_offset_completes_its_datagram(self) -> None:
+        from pcapkit.foundation.reassembly.data.data import Completion
+
+        for cls in self._classes():
+            for size in (1, 7):
+                for strict in (True, False):
+                    with self.subTest(cls=cls.__name__, size=size, strict=strict):
+                        tail = b'A' * size
+                        reasm = cls(strict=strict)
+                        for packet in self._body():
+                            reasm(packet)
+                        reasm(self._packet(num=3, fo=LAST, mf=False, payload=tail))
+                        datagram, = reasm.datagram
+                        self.assertIs(datagram.completed, Completion.COMPLETE)
+                        self.assertEqual(datagram.index, (1, 2, 3))
+                        self.assertEqual(datagram.payload, BODY + tail)
+                        self.assertEqual(datagram.conflict, ())
+
+    def test_a_non_final_fragment_at_the_largest_offset_leaves_the_datagram_open(self) -> None:
+        from pcapkit.foundation.reassembly.data.data import Completion
+
+        for cls in self._classes():
+            for size in (1, 7):
+                for strict in (True, False):
+                    with self.subTest(cls=cls.__name__, size=size, strict=strict):
+                        tail = b'A' * size
+                        reasm = cls(strict=strict)
+                        for packet in self._body():
+                            reasm(packet)
+                        reasm(self._packet(num=3, fo=LAST, mf=True, payload=tail))
+                        datagram, = reasm.datagram
+                        self.assertIs(datagram.completed, Completion.PARTIAL)
+                        self.assertEqual(datagram.index, (1, 2, 3))
+                        if strict:
+                            run, = datagram.payload
+                        else:
+                            run = datagram.payload
+                        self.assertEqual(run[:LAST + size], BODY + tail)
+
+    def test_a_tso_fragment_at_the_largest_offset_composes_with_the_buffer_growth(self) -> None:
+        # Total Length 0 (#1555): 7 octets end exactly at the preallocated
+        # buffer's end, 9 run past it and grow it.
+        from pcapkit.foundation.reassembly.data.data import Completion
+
+        for size in (7, 9):
+            for strict in (True, False):
+                with self.subTest(size=size, strict=strict):
+                    tail = b'A' * size
+                    reasm = self._classes()[0](strict=strict)
+                    for packet in self._body():
+                        reasm(packet)
+                    reasm(self._packet(num=3, fo=LAST, mf=False, payload=tail, tl=0))
+                    datagram, = reasm.datagram
+                    self.assertIs(datagram.completed, Completion.COMPLETE)
+                    self.assertEqual(datagram.payload, BODY + tail)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/foundation/reassembly/test_ip_tso_total_length_1555_runtime.py

@@ -0,0 +1,174 @@
+# -*- coding: utf-8 -*-
+"""A TSO capture reassembles to the same IPv4 datagrams as the unmodified one.
+
+GitHub issue #1555. With the four IPv4 frames of :file:`tcp.pcap` given Total
+Length 0, as TCP segmentation offload leaves them, ``extract(ipv4=True,
+reassembly=True)`` reported four ``PARTIAL`` datagrams with no payload under
+every engine, against four ``COMPLETE`` ones of 44, 32, 244 and 32 octets on
+the unmodified capture: the reassembler took the data length from ``tl - ihl``.
+Wireshark does not reassemble such a frame at all -- it is one segment -- so the
+expected result is the unmodified capture's, the zeroed header aside.
+
+A two-fragment UDP datagram with the Total Length of one fragment zeroed is
+read the way Wireshark 4.6.9 reads it, as that fragment's captured length.
+
+The captures are built in memory, the TSO one from a generated sample, so the
+module 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 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 = ('default',) + 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)
+
+#: A libpcap global header: little-endian, version 2.4, snaplen 65535, Ethernet.
+PCAP_HEADER = struct.pack('<IHHiIII', 0xA1B2C3D4, 2, 4, 0, 0, 65535, 1)
+
+
+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 _with_total_length(frame: 'bytes', total_length: 'int', *, fix_checksum: 'bool') -> 'bytes':
+    """``frame``, an Ethernet frame carrying an option-less IPv4 header, 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_capture(*, fix_checksum: 'bool') -> 'bytes':
+    """``tcp.pcap`` with every IPv4 frame's Total Length zeroed."""
+    with open(sample_path('tcp.pcap'), 'rb') as file:
+        source = file.read()
+    output, rest, number = bytearray(source[:24]), source[24:], 0
+    while rest:
+        number += 1
+        record, incl_len = rest[:16], struct.unpack('<I', rest[8:12])[0]
+        frame = rest[16:16 + incl_len]
+        if number in IPV4_FRAMES:
+            frame = _with_total_length(frame, 0, fix_checksum=fix_checksum)
+        output += record + frame
+        rest = rest[16 + incl_len:]
+    return bytes(output)
+
+
+def _fragment_capture(zeroed: 'int') -> 'tuple[bytes, bytes]':
+    """A UDP datagram in two fragments, fragment ``zeroed`` with Total Length 0.
+
+    Returns:
+        The capture, and the datagram's payload (UDP header and data).
+
+    """
+    data = (bytes(range(1, 256)) * 2)[:96]
+    payload = struct.pack('!HHHH', 1111, 2222, 8 + len(data), 0) + data
+    ether = bytes.fromhex('020000000002 020000000001 0800')  # destination, source, IPv4
+    output = bytearray(PCAP_HEADER)
+    for index, (offset, more, chunk) in enumerate(((0, 1, payload[:56]), (7, 0, payload[56:]))):
+        total_length = 0 if index == zeroed else 20 + len(chunk)
+        header = bytearray(struct.pack('!BBHHHBBH4s4s', 0x45, 0, total_length, 0x1234,
+                                       (more << 13) | offset, 64, 17, 0,
+                                       bytes((192, 0, 2, 1)), bytes((198, 51, 100, 2))))
+        header[10:12] = _checksum(bytes(header))
+        frame = ether + bytes(header) + chunk
+        output += struct.pack('<IIII', 1000 + index, 0, len(frame), len(frame)) + frame
+    return bytes(output), payload
+
+
+@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed')
+class TSOReassemblyTests(unittest.TestCase):
+    """A Total Length of 0 reassembles to the datagram the frame carries, under every engine."""
+
+    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', octets: 'bytes') -> 'str':
+        path = os.path.join(self.tmp, name)
+        with open(path, 'wb') as file:
+            file.write(octets)
+        return path
+
+    def _extract(self, fin: 'str', engine: 'str', strict: 'bool') -> 'Any':
+        from pcapkit import extract
+
+        with warnings.catch_warnings():
+            warnings.simplefilter('ignore')
+            extractor = extract(fin=fin, nofile=True, engine=engine, ipv4=True,
+                                reassembly=True, reasm_strict=strict)
+        self.addCleanup(close_extractor, extractor)
+        return extractor
+
+    def test_tso_capture_reassembles_like_the_unmodified_one(self) -> None:
+        from pcapkit.foundation.reassembly.data.data import Completion
+
+        for fix_checksum in (True, False):
+            octets = _tso_capture(fix_checksum=fix_checksum)
+            fin = self._write(f'tso-{fix_checksum}.pcap', octets)
+            for engine in ENGINES:
+                for strict in (True, False):
+                    with self.subTest(engine=engine, strict=strict,
+                                      checksum='fixed' if fix_checksum else 'stale'):
+                        expected = self._extract(sample_path('tcp.pcap'), engine, strict).reassembly.ipv4
+                        actual = self._extract(fin, engine, strict).reassembly.ipv4
+
+                        self.assertEqual([datagram.completed for datagram in actual],
+                                         [Completion.COMPLETE] * 4)
+                        self.assertEqual([len(datagram.payload) for datagram in actual], [44, 32, 244, 32])
+                        self.assertEqual([(datagram.completed, datagram.index, datagram.payload,
+                                           datagram.conflict) for datagram in actual],
+                                         [(datagram.completed, datagram.index, datagram.payload,
+                                           datagram.conflict) for datagram in expected])
+                        self.assertEqual([datagram.packet.info.to_dict() for datagram in actual],
+                                         [datagram.packet.info.to_dict() for datagram in expected])
+
+                        # the header is the frame's own, Total Length 0 and all
+                        self.assertEqual([datagram.header[2:4] for datagram in actual], [b'\x00\x00'] * 4)
+                        self.assertEqual([datagram.header[:2] + datagram.header[4:10]
+                                          for datagram in actual],
+                                         [datagram.header[:2] + datagram.header[4:10]
+                                          for datagram in expected])
+
+    def test_a_zero_total_length_fragment_is_reassembled(self) -> None:
+        from pcapkit.foundation.reassembly.data.data import Completion
+
+        for zeroed in (0, 1):
+            octets, payload = _fragment_capture(zeroed)
+            fin = self._write(f'fragment-{zeroed}.pcap', octets)
+            for engine in ENGINES:
+                for strict in (True, False):
+                    with self.subTest(zeroed=zeroed, engine=engine, strict=strict):
+                        datagram, = self._extract(fin, engine, strict).reassembly.ipv4
+                        self.assertIs(datagram.completed, Completion.COMPLETE)
+                        self.assertEqual(datagram.index, (1, 2))
+                        self.assertEqual(datagram.payload, payload)
+                        self.assertEqual(datagram.header[2:4], b'\x00\x00' if zeroed == 0 else b'\x00\x4c')
+
+
+if __name__ == '__main__':
+    unittest.main()