diff --git a/pcapkit/toolkit/scapy.py b/pcapkit/toolkit/scapy.py index 1257552a37..746b4e65db 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 a82a2d35d5..8e95a8ef00 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()