From 4ce6ee3806e786c1d8590cf82ddcbf727f374f64 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Sat, 10 Oct 2026 02:41:21 -0400 Subject: [PATCH] fix(toolkit): the scapy adapters stop at the IP datagram's declared length (#1584) Scapy files the octets past an IP layer's Total Length or Payload Length as a Padding layer, and bytes() of every layer above it ends with them. - ipv6_reassembly took the fragment's data, and so its tl, with the frame's trailer on the end: an 80-octet datagram reassembled to 84 octets. - tcp_reassembly and tcp_traceflow did the same to the TCP payload, len and last. A segment now ends at the shortest end any enclosing IP layer declares, and is dropped, as the default engine drops it, when that holds under 20 octets or the Data Offset counts under 5. - Jumbograms use pcapkit's own Jumbo Payload Length, since Scapy misreads it and splits nothing off. - ipv4_reassembly is unchanged: it hands the trailer on as the default engine's adapter does, and the reassembler cuts at tl - ihl. Tests: the adapters match the default engine on every frame tested, and each new test fails without its fix. --- pcapkit/toolkit/scapy.py | 105 ++++++++++- tests/toolkit/test_scapy_unit.py | 296 ++++++++++++++++++++++++++++++- 2 files changed, 392 insertions(+), 9 deletions(-) diff --git a/pcapkit/toolkit/scapy.py b/pcapkit/toolkit/scapy.py index 1257552a3..746b4e65d 100644 --- a/pcapkit/toolkit/scapy.py +++ b/pcapkit/toolkit/scapy.py @@ -46,6 +46,7 @@ from pcapkit.foundation.traceflow.data.data import FrameRecord from pcapkit.foundation.traceflow.data.tcp import Packet as TF_TCP_Packet from pcapkit.protocols.data.misc.pcap.frame import FrameInfo as Data_FrameInfo +from pcapkit.protocols.schema.internet.ipv6 import jumbo_payload_length from pcapkit.utilities.compat import ModuleNotFoundError # pylint: disable=redefined-builtin from pcapkit.utilities.exceptions import MissingKeyError, ModuleNotFound, stacklevel from pcapkit.utilities.warnings import ScapyWarning, warn @@ -226,6 +227,80 @@ def packet2frame(packet: 'Packet') -> 'FrameRecord': ), time_epoch=decimal.Decimal(epoch)) +def _within_datagrams(layer: 'Packet') -> 'bytes': + """Return the octets from ``layer`` on that lie inside every IP datagram carrying it. + + Args: + layer: A layer of a dissected Scapy packet. + + Returns: + The octets of ``layer`` and everything after it, up to the nearest end + that the Total Length or Payload Length of an IP layer above it declares. + + Note: + When `Scapy`_ dissects an IP layer, it files whatever follows the + datagram's declared length as a :class:`~scapy.packet.Padding` layer -- + an Ethernet frame's padding up to its 60-octet minimum, or a trailer + such as an FCS -- and ``bytes()`` of *any* layer above that one ends + with it. Those octets are no part of the datagram, and the default + engine never hands them over (:issue:`1584`), so they are cut off here + by the default engine's rules: an IPv4 Total Length of zero, as TCP + segmentation offload leaves, runs the datagram to the end of the frame + (:issue:`1547`), and an IPv6 Payload Length of zero gives way to a + Jumbo Payload option (:issue:`1434`). The latter is not left to Scapy, + which reads that option's length in host byte order and adds the + Hop-by-Hop Options header to it a second time. + + Every IP layer above ``layer`` bounds it, not only the nearest, as each + bounds what the default engine hands the next: a datagram tunnelled in + IP may declare more than the one carrying it holds, or nothing at all. + + """ + from scapy.layers.inet import IP + from scapy.layers.inet6 import IPv6 + + data = bytes(layer) + end = len(data) + carrier = layer.underlayer if data else None + while carrier is not None: + length = None + if isinstance(carrier, IPv6): + if carrier.plen is not None: # it is until a packet built by hand is built + # a zero is a jumbogram's, or else it declares nothing + length = carrier.plen or jumbo_payload_length(carrier.nh, bytes(carrier.payload)) or 0 + elif isinstance(carrier, IP) and carrier.len: # a zero is TSO's + length = max(carrier.len - carrier.ihl * 4, 0) + if length is not None: + # ``bytes()`` of a layer spans it and every layer after it, padding + # included, so the difference is the octets between the end of the + # IP header and ``layer`` + end = min(end, max(length - (len(carrier.payload) - len(data)), 0)) + carrier = carrier.underlayer + return data[:end] + + +def _tcp_segment(tcp: 'TCP') -> 'tuple[bytes, bytes] | None': + """Split the TCP segment ``tcp`` heads into its header and its payload. + + Args: + tcp: TCP layer of a dissected Scapy packet. + + Returns: + The header, as many octets as the Data Offset counts, and the payload + after it, up to the end of the datagrams that carry the segment, c.f. + ``_within_datagrams()``; or :data:`None` if the datagrams hold fewer + than the 20 octets of the fixed header, or the Data Offset counts fewer + (:rfc:`9293#section-3.1`), since the default engine then reads no + segment at all. + + """ + segment = _within_datagrams(tcp) + hdr_len = tcp.dataofs * 4 + if len(segment) < 20 or hdr_len < 20: + return None + return segment[:hdr_len], segment[hdr_len:] + + def ipv4_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv4Address] | None': """Make data for IPv4 reassembly. @@ -271,6 +346,10 @@ def ipv4_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv4Ad mf=bool(ipv4.flags.MF), # more fragment flag tl=ipv4.len, # total length, header includes header=bytes(ipv4)[:ipv4.ihl * 4], # raw bytes type header + # NOTE: this keeps the octets past the Total Length, as the default + # engine's adapter does, since the reassembly machinery cuts the + # payload at ``tl - ihl`` itself. :func:`ipv6_reassembly` derives + # its ``tl`` from the payload, so it has to cut them (:issue:`1584`). payload=bytearray(bytes(ipv4.payload)), # raw bytearray type payload timestamp=float(packet.time), # capture timestamp ) @@ -316,7 +395,9 @@ def ipv6_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv6Ad # the Fragment header, which is the only part of the header the # reassembled packet keeps (:rfc:`8200#section-4.5`). hdr_len = len(ipv6) - len(ipv6_frag) - payload = bytearray(bytes(ipv6_frag.payload)) + # the fragment's data stops at the Payload Length, so neither it nor + # ``tl`` below counts the frame's trailer (:issue:`1584`) + payload = bytearray(_within_datagrams(ipv6_frag.payload)) data = IP_Packet( bufid=( @@ -358,7 +439,8 @@ def tcp_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'TCP_Packet | None Data for TCP reassembly. * If the ``packet`` can be used for TCP reassembly. A packet can be reassembled - if it contains TCP layer (:class:`scapy.layers.inet.TCP`). + if it contains TCP layer (:class:`scapy.layers.inet.TCP`) whose fixed header + lies inside its datagram, and whose Data Offset counts at least that header. * If the ``packet`` can be reassembled, then the :obj:`dict` mapping of data for TCP reassembly (:term:`reasm.tcp.packet`) will be returned; otherwise, returns :data:`None`. @@ -375,8 +457,11 @@ def tcp_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'TCP_Packet | None if 'TCP' in packet: tcp = cast('TCP', packet['TCP']) + if (segment := _tcp_segment(tcp)) is None: + return None + header, payload = segment - raw_len = len(tcp.payload) # payload length, header excludes + raw_len = len(payload) # payload length, header excludes data = TCP_Packet( bufid=( ipaddress.ip_address(ip.src), # source IP address @@ -390,8 +475,8 @@ def tcp_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'TCP_Packet | None syn=bool(tcp.flags.S), # synchronise flag fin=bool(tcp.flags.F), # finish flag rst=bool(tcp.flags.R), # reset connection flag - header=bytes(tcp)[:tcp.dataofs * 4], # raw bytes type header - payload=bytearray(bytes(tcp.payload)), # raw bytearray type payload + header=header, # raw bytes type header + payload=bytearray(payload), # raw bytearray type payload first=tcp.seq, # first sequence number of payload last=tcp.seq + raw_len - 1, # last sequence number of payload len=raw_len, # payload length, header excludes @@ -412,7 +497,8 @@ def tcp_traceflow(packet: 'Packet', *, count: 'int' = -1) -> 'TF_TCP_Packet | No Data for TCP reassembly. * If the ``packet`` can be used for TCP flow tracing. A packet can be reassembled - if it contains TCP layer (:class:`scapy.layers.inet.TCP`). + if it contains TCP layer (:class:`scapy.layers.inet.TCP`), on the terms + :func:`tcp_reassembly` sets. * If the ``packet`` can be reassembled, then the :obj:`dict` mapping of data for TCP flow tracing (:term:`trace.tcp.packet`) will be returned; otherwise, returns :data:`None`. @@ -423,6 +509,9 @@ def tcp_traceflow(packet: 'Packet', *, count: 'int' = -1) -> 'TF_TCP_Packet | No if 'TCP' in packet: ip = cast('IP', packet['IP']) if 'IP' in packet else cast('IPv6', packet['IPv6']) tcp = cast('TCP', packet['TCP']) + if (segment := _tcp_segment(tcp)) is None: + return None + header, payload = segment # NOTE: the link type of the interface the packet was read from, which # :class:`~pcapkit.foundation.engines.scapy.Scapy` attaches with @@ -475,8 +564,8 @@ def tcp_traceflow(packet: 'Packet', *, count: 'int' = -1) -> 'TF_TCP_Packet | No timestamp=float(packet.time), # capture timestamp seq=tcp.seq, # TCP sequence number ack=tcp.ack, # TCP acknowledgement number - header=bytes(tcp)[:tcp.dataofs * 4], # raw bytes type header - payload=bytearray(bytes(tcp.payload)), # raw bytearray type payload + header=header, # raw bytes type header + payload=bytearray(payload), # raw bytearray type payload ) return data return None diff --git a/tests/toolkit/test_scapy_unit.py b/tests/toolkit/test_scapy_unit.py index a82a2d35d..8e95a8ef0 100644 --- a/tests/toolkit/test_scapy_unit.py +++ b/tests/toolkit/test_scapy_unit.py @@ -4,15 +4,35 @@ from ipaddress import ip_address import builtins import importlib +import os +import struct +import tempfile import unittest +import warnings from unittest import mock -from tests._support import purge_modules, reimport_once_per_class +from tests._support import close_extractor, purge_modules, reimport_once_per_class HAS_SCAPY = importlib.util.find_spec('scapy') is not None RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) +#: Octets an Ethernet frame carries past its IP datagram -- an FCS, say. +TRAILER = bytes.fromhex('deadbeef') +#: The datagram #1584 sends as two 40-octet fragments. +DATAGRAM = bytes(range(80)) + + +def write_pcap(frames: 'list[bytes]') -> 'str': + """Write ``frames`` to a temporary Ethernet PCAP file and return its path.""" + fd, path = tempfile.mkstemp(suffix='.pcap') + with os.fdopen(fd, 'wb') as file: + file.write(struct.pack(' None: self.assertEqual(before, after) +@unittest.skipUnless(HAS_RUNTIME and HAS_SCAPY, 'runtime dependencies not installed') +class ScapyTrailerTests(unittest.TestCase): + """The adapters stop at the datagram's declared length (GitHub issue #1584). + + Scapy files the octets past an IP layer's Total Length or Payload Length as a + ``Padding`` layer, and ``bytes()`` of every layer above it ends with them, so + an IPv6 fragment's data and a TCP segment's payload both carried the frame's + trailer. Every frame here is written to a capture and read back through the + engine, and the default engine reads the same capture for comparison. + + """ + + #: The TCP payload each of :meth:`_tcp_frames` carries. + TCP_PAYLOADS = { + 'padded': b'X', + 'trailer': b'YZ', + 'pure-ack-padded': b'', + 'ipv6-ext-header-trailer': b'six', + 'ip-in-ip-trailers': b'in', + 'jumbogram-trailer': b'jmb', + 'zero-total-length': b'tso-data', + } + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _ether(self): + from scapy.layers.l2 import Ether + + return Ether(src='aa:aa:aa:aa:aa:aa', dst='bb:bb:bb:bb:bb:bb') + + def _ip_frames(self) -> 'list[bytes]': + """``DATAGRAM`` as IPv4, then as IPv6, fragments; each last one has a trailer.""" + from scapy.layers.inet import IP + from scapy.layers.inet6 import IPv6, IPv6ExtHdrFragment + from scapy.packet import Raw + + ipv4 = {'src': '192.0.2.1', 'dst': '198.51.100.1', 'id': 7, 'proto': 17} + ipv6 = {'src': '2001:db8::1', 'dst': '2001:db8::2'} + return [ + bytes(self._ether() / IP(**ipv4, flags='MF', frag=0) / Raw(DATAGRAM[:40])), + bytes(self._ether() / IP(**ipv4, frag=5) / Raw(DATAGRAM[40:])) + TRAILER, + bytes(self._ether() / IPv6(**ipv6) / + IPv6ExtHdrFragment(nh=17, offset=0, m=1, id=99) / Raw(DATAGRAM[:40])), + bytes(self._ether() / IPv6(**ipv6) / + IPv6ExtHdrFragment(nh=17, offset=5, m=0, id=99) / Raw(DATAGRAM[40:])) + TRAILER, + ] + + def _tcp_frames(self) -> 'list[bytes]': + """One frame per :attr:`TCP_PAYLOADS` entry, in its order.""" + from scapy.layers.inet import IP, TCP + from scapy.layers.inet6 import IPv6, IPv6ExtHdrDestOpt, IPv6ExtHdrHopByHop, Jumbo + from scapy.packet import Raw + + ipv4 = {'src': '192.0.2.1', 'dst': '198.51.100.1'} + ipv6 = {'src': '2001:db8::1', 'dst': '2001:db8::2'} + + def padded(frame: 'bytes') -> 'bytes': + return frame + bytes(60 - len(frame)) # Ethernet's minimum frame + + inner = bytes(IP(src='10.0.0.1', dst='10.0.0.2') / TCP(sport=5, dport=6, flags='PA') / + Raw(b'in')) + b'\xaa\xbb' + tso = bytes(self._ether() / IP(**ipv4) / TCP(sport=9, dport=10, flags='PA') / Raw(b'tso-data')) + return [ + padded(bytes(self._ether() / IP(**ipv4) / TCP(sport=1, dport=2, seq=100, flags='PA') / + Raw(b'X'))), + bytes(self._ether() / IP(**ipv4) / TCP(sport=1, dport=2, seq=101, flags='PA') / + Raw(b'YZ')) + TRAILER, + padded(bytes(self._ether() / IP(**ipv4) / TCP(sport=3, dport=4, seq=7, flags='A'))), + bytes(self._ether() / IPv6(**ipv6) / IPv6ExtHdrDestOpt() / + TCP(sport=1, dport=2, flags='PA') / Raw(b'six')) + TRAILER, + bytes(self._ether() / IP(**ipv4, proto=4) / Raw(inner)) + TRAILER, + # the Jumbo Payload Length counts the Hop-by-Hop Options header too + bytes(self._ether() / IPv6(**ipv6, plen=0, nh=0) / + IPv6ExtHdrHopByHop(nh=6, options=[Jumbo(jumboplen=8 + 20 + 3)]) / + TCP(sport=7, dport=8, flags='PA') / Raw(b'jmb')) + TRAILER, + # a Total Length of 0 runs the datagram to the end of the frame + tso[:16] + b'\x00\x00' + tso[18:], + ] + + def _extract(self, frames: 'list[bytes]', engine: 'str' = 'scapy', **kwargs: 'object'): + import pcapkit + + path = write_pcap(frames) + self.addCleanup(os.remove, path) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + extractor = pcapkit.extract(fin=path, nofile=True, engine=engine, **kwargs) + self.addCleanup(close_extractor, extractor) + return extractor + + def test_ipv6_fragment_data_stops_at_the_payload_length(self) -> None: + from pcapkit.toolkit import scapy as toolkit + + packets = self._extract(self._ip_frames(), store=True).frame + first, last = (toolkit.ipv6_reassembly(packet) for packet in packets[2:]) + assert first is not None and last is not None + self.assertEqual(bytes(first.payload), DATAGRAM[:40]) + # 44 octets and a ``tl`` of 84 before the fix + self.assertEqual(bytes(last.payload), DATAGRAM[40:]) + self.assertEqual((last.ihl, last.tl), (40, 80)) + + def test_reassembled_datagrams_are_the_datagram_sent(self) -> None: + extractor = self._extract(self._ip_frames(), store=False, reassembly=True, ip=True) + for name, datagrams in (('ipv4', extractor.reassembly.ipv4), + ('ipv6', extractor.reassembly.ipv6)): + with self.subTest(version=name): + datagram, = datagrams + # only ``Completion.COMPLETE`` is truthy + self.assertTrue(datagram.completed) + # IPv6 came back 84 octets long, ending in the trailer + self.assertEqual(datagram.payload, DATAGRAM) + + def test_tcp_payload_stops_at_the_datagram_end(self) -> None: + from scapy.layers.inet import TCP + + from pcapkit.toolkit import scapy as toolkit + + packets = self._extract(self._tcp_frames(), store=True).frame + self.assertEqual(len(packets), len(self.TCP_PAYLOADS)) + for (name, expected), packet in zip(self.TCP_PAYLOADS.items(), packets): + with self.subTest(frame=name): + header = bytes(packet[TCP])[:20] + segment = toolkit.tcp_reassembly(packet) + assert segment is not None + self.assertEqual(bytes(segment.payload), expected) + self.assertEqual(segment.len, len(expected)) + self.assertEqual(segment.last, segment.first + len(expected) - 1) + self.assertEqual(segment.header, header) + + flow = toolkit.tcp_traceflow(packet) + assert flow is not None + self.assertEqual(bytes(flow.payload), expected) + self.assertEqual(flow.header, header) + + def test_tcp_reassembly_is_the_stream_sent(self) -> None: + def streams(engine: 'str') -> 'dict[tuple[str, int], bytes]': + extractor = self._extract(self._tcp_frames(), engine=engine, store=False, + reassembly=True, tcp=True) + return {(str(datagram.id.src[0]), datagram.id.src[1]): datagram.payload + for datagram in extractor.reassembly.tcp} + + scapy = streams('scapy') + # ``b'X\x00\x00\x00\x00\x00\xef'`` before the fix: the first segment's + # padding took the place of the second's data, and its trailer followed + self.assertEqual(scapy.get(('192.0.2.1', 1)), b'XYZ') + self.assertEqual(scapy.get(('2001:db8::1', 1)), b'six') + self.assertEqual(scapy, streams('default')) + + def test_every_adapter_agrees_with_the_default_engine(self) -> None: + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit import pcap as toolkit_pcap + from pcapkit.toolkit import scapy as toolkit + + frames = self._ip_frames() + self._tcp_frames() + packets = self._extract(frames, store=True).frame + defaults = self._extract(frames, engine='default', store=True).frame + self.assertEqual(len(packets), len(defaults)) + + def fields(data: 'object', names: 'tuple[str, ...]') -> 'object': + if data is None: + return None + return {name: bytes(value) if isinstance(value, (bytes, bytearray)) else value + for name, value in ((name, getattr(data, name)) for name in names)} + + ip_fields = ('fo', 'ihl', 'mf', 'tl', 'header', 'payload') + tcp_fields = ('dsn', 'ack', 'header', 'payload', 'first', 'last', 'len') + trace_fields = ('seq', 'ack', 'header', 'payload') + compared = 0 + for index, (packet, default) in enumerate(zip(packets, defaults)): + for adapter, expected, actual in ( + ('ipv4', fields(toolkit_pcap.ipv4_reassembly(default), ip_fields), + fields(toolkit.ipv4_reassembly(packet), ip_fields)), + ('ipv6', fields(toolkit_pcap.ipv6_reassembly(default), ip_fields), + fields(toolkit.ipv6_reassembly(packet), ip_fields)), + ('tcp', fields(toolkit_pcap.tcp_reassembly(default), tcp_fields), + fields(toolkit.tcp_reassembly(packet), tcp_fields)), + ('trace', fields(toolkit_pcap.tcp_traceflow(default, data_link=LinkType.ETHERNET), + trace_fields), + fields(toolkit.tcp_traceflow(packet), trace_fields)), + ): + compared += expected is not None + with self.subTest(frame=index, adapter=adapter): + self.assertEqual(actual, expected) + # every IPv4 frame here leaves DF clear, so each of the 7 makes an IPv4 + # record; then the 2 IPv6 fragments, and a TCP and a trace record for + # each TCP frame + self.assertEqual(compared, 7 + 2 + 2 * len(self.TCP_PAYLOADS)) + + def _assert_tcp_records(self, frames: 'list[bytes]', expected: 'list[bytes | None]') -> None: + """Each frame gives ``expected`` TCP payload, or no record, as on the default engine.""" + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit import pcap as toolkit_pcap + from pcapkit.toolkit import scapy as toolkit + + def octets(record: 'object') -> 'object': + return None if record is None else (bytes(record.header), bytes(record.payload)) + + packets = self._extract(frames, store=True).frame + defaults = self._extract(frames, engine='default', store=True).frame + self.assertEqual(len(packets), len(expected)) + for index, (packet, default, payload) in enumerate(zip(packets, defaults, expected)): + with self.subTest(frame=index): + segment = toolkit.tcp_reassembly(packet) + flow = toolkit.tcp_traceflow(packet) + self.assertEqual(None if segment is None else bytes(segment.payload), payload) + self.assertEqual(None if flow is None else bytes(flow.payload), payload) + self.assertEqual(octets(segment), octets(toolkit_pcap.tcp_reassembly(default))) + self.assertEqual(octets(flow), octets(toolkit_pcap.tcp_traceflow( + default, data_link=LinkType.ETHERNET))) + + def test_a_data_offset_below_the_fixed_header_gives_no_segment(self) -> None: + from scapy.layers.inet import IP, TCP + from scapy.packet import Raw + + # a header shorter than 20 octets is malformed (:rfc:`9293#section-3.1`), + # and the default engine reads no segment; the payload here had started + # with the last of the header's octets + self._assert_tcp_records([ + bytes(self._ether() / IP(src='192.0.2.1', dst='198.51.100.1') / + TCP(sport=1, dport=2, flags='PA', dataofs=dataofs) / Raw(b'abcdef')) + TRAILER + for dataofs in (0, 4) + ], [None, None]) + + def test_a_total_length_below_the_header_gives_no_segment(self) -> None: + from scapy.layers.inet import IP, TCP + from scapy.packet import Raw + + frame = bytes(self._ether() / IP(src='192.0.2.1', dst='198.51.100.1') / + TCP(sport=1, dport=2, flags='PA') / Raw(b'abcdef')) + # 10 octets is less than the IPv4 header, and 30 holds only half the TCP + # header; the former had come back as an empty record + self._assert_tcp_records([frame[:16] + struct.pack('!H', length) + frame[18:] + for length in (10, 30)], [None, None]) + + def test_a_tunnelled_segment_ends_with_the_shortest_datagram(self) -> None: + from scapy.layers.inet import IP, TCP + from scapy.packet import Raw + + def inner(data: 'bytes') -> 'bytearray': + return bytearray(bytes(IP(src='10.0.0.1', dst='10.0.0.2') / + TCP(sport=5, dport=6, flags='PA') / Raw(data))) + + outer = {'src': '192.0.2.1', 'dst': '198.51.100.1', 'proto': 4} + tso = inner(b'tsoin') + tso[2:4] = b'\x00\x00' # the inner datagram declares no length at all + # the first fragment of the outer datagram ends with the TCP header, + # though the inner datagram declares 40 octets of data after it + overrun = inner(b'x' * 40)[:40] + self._assert_tcp_records([ + bytes(self._ether() / IP(**outer) / Raw(bytes(tso))) + TRAILER, + bytes(self._ether() / IP(**outer, id=1, flags='MF') / Raw(bytes(overrun))) + TRAILER, + ], [b'tsoin', b'']) + + def test_data_with_no_length_to_stop_at_is_taken_whole(self) -> None: + from scapy.layers.inet import TCP + from scapy.layers.inet6 import IPv6, IPv6ExtHdrFragment + from scapy.packet import Raw + + from pcapkit.toolkit import scapy as toolkit + + # a segment with no IP layer above it at all + tcp = TCP(bytes(TCP(sport=1, dport=2, flags='PA') / Raw(b'bare'))) + self.assertEqual(toolkit._tcp_segment(tcp), (bytes(tcp)[:20], b'bare')) + + # a fragment built by hand, whose Payload Length is only filled in when + # the packet is built + packet = self._ether() / IPv6() / IPv6ExtHdrFragment(id=1) / Raw(b'hand') + self.assertIsNone(packet[IPv6].plen) + fragment = toolkit.ipv6_reassembly(packet) + assert fragment is not None + self.assertEqual(bytes(fragment.payload), b'hand') + + if __name__ == '__main__': unittest.main()