From 6c7b4d0667673a5d166f2f9d10b240860bb7dcc1 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Sat, 10 Oct 2026 19:37:00 -0400 Subject: [PATCH] fix(toolkit): dpkt and scapy find TCP in IPv4 where the default engine does (#1606, #1617) - dpkt and scapy step over an AH or IPv6 extension header carried directly in IPv4 (protocols 0, 43, 44, 51, 60, 135, 139) by the default engine's own rules, via pypcapfile's _past_extension_headers, bounded at EXTENSION_HEADER_LIMIT; the TCP segment or IP-in-IP tunnel behind them is decoded from the captured octets. Behind such a header, TCP is read only as the default TCP parser accepts it. - dpkt keeps a tunnel it raises anything decoding undecoded, as its engine keeps such a frame raw, rather than ending the extraction. - dpkt's packet2dict and packet2chain treat only a packet-like payload as a layer; a str payload, such as VRRP's '', is leaf data and ends the chain. - scapy's carrier walk now runs top-down and stops at a later IPv4 fragment, which scapy still dissects IPv6 and AH in. - Drop the #1606 gap row from the #1604 chain-limit agreement test. --- pcapkit/toolkit/dpkt.py | 150 +++++++- pcapkit/toolkit/scapy.py | 128 +++++-- ...est_ext_chain_limit_agreement_1604_unit.py | 6 - .../test_tcp_carrier_walk_1606_1617_unit.py | 348 ++++++++++++++++++ 4 files changed, 589 insertions(+), 43 deletions(-) create mode 100644 tests/toolkit/test_tcp_carrier_walk_1606_1617_unit.py diff --git a/pcapkit/toolkit/dpkt.py b/pcapkit/toolkit/dpkt.py index a0e9a8d61e..0c1cbd9189 100644 --- a/pcapkit/toolkit/dpkt.py +++ b/pcapkit/toolkit/dpkt.py @@ -23,7 +23,8 @@ 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.toolkit.pypcapfile import _default_accepts +from pcapkit.toolkit.pypcapfile import (IPV4_EXTENSION_HEADERS, TCP_MIN_HEADER_LEN, + _default_accepts, _past_extension_headers) from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall if TYPE_CHECKING: @@ -462,6 +463,112 @@ def _ipv4_accepted(ipv4: 'IP') -> 'bool': return _default_accepts('IPv4', ipv4.pack_hdr() + bytes(ipv4.opts)) +def _past_ipv4_extension_headers(packet: 'Packet', ipv4: 'IP') -> 'Optional[tuple[int, bytes]]': + """Step over the headers at the head of a DPKT IPv4 packet's payload that the default engine reads past. + + Args: + packet: DPKT packet ``ipv4`` is a layer of, whose octets + :func:`_layer2bytes` slices. + ipv4: DPKT IPv4 packet, whose protocol is one of the + :data:`~pcapkit.toolkit.pypcapfile.IPV4_EXTENSION_HEADERS`. + + Returns: + The protocol number after the last of those headers and the octets from + its header on, as far as the Total Length goes -- a Total Length of + ``0``, which TCP segmentation offload leaves, runs to the end of the + frame -- or :data:`None` if the default engine reads no upper-layer + header there: ``ipv4`` is a later fragment, whose data is a slice of the + datagram's payload (:rfc:`791#section-3.2`), or it leaves one of the + headers undissected, which + :func:`~pcapkit.toolkit.pypcapfile._past_extension_headers` decides by + the default engine's own rules, up to + :data:`~pcapkit.protocols.internet.internet.EXTENSION_HEADER_LIMIT` + headers in a row (:issue:`1604`). + + Note: + The default engine dispatches an IPv4 packet's protocol through the + registry IPv6 uses, so it reads an AH or IPv6 extension header there + and what its Next Header names. `DPKT`_ decodes protocol 0 as IPv4 and + an AH as :class:`dpkt.ah.AH`, and keeps the rest as octets, so the + octets are read instead (:issue:`1606`). + + .. _DPKT: https://dpkt.readthedocs.io + + """ + if ipv4.offset: + return None + wire, _ = _layer2bytes(packet, ipv4) + ihl = ipv4.hl * 4 + return _past_extension_headers(ipv4.p, wire[ihl:max(ipv4.len, ihl)] if ipv4.len else wire[ihl:]) + + +def _decode_tcp(segment: 'bytes') -> 'Optional[TCP]': + """Decode a TCP segment the default engine reads behind an AH or IPv6 extension header in IPv4. + + Args: + segment: Raw octets from the TCP header on, as far as the IPv4 packet + carrying it goes (see :func:`_past_ipv4_extension_headers`). + + Returns: + The segment, with ``segment`` attached by :func:`attach_buffer` so that + its header is the captured one; or :data:`None` if the default engine + reads no TCP there: fewer than :data:`TCP_MIN_HEADER_LEN + ` octets, a Data Offset + counting fewer or more than were captured, or options its TCP parser + rejects (see :func:`~pcapkit.toolkit.pypcapfile._default_accepts`). + + Note: + Directly over IP, the default engine reads a TCP header its parser + rejects out of the raw octets left behind + (:func:`pcapkit.toolkit.pcap.tcp_segment`), but behind another header + it does not, so the parser's answer stands here, as for the pypcapfile + engine. + + """ + import dpkt # isort:skip + + hdr_len = (segment[12] >> 4) * 4 if len(segment) >= TCP_MIN_HEADER_LEN else 0 + if hdr_len < TCP_MIN_HEADER_LEN or hdr_len > len(segment): + return None + if hdr_len > TCP_MIN_HEADER_LEN and not _default_accepts('TCP', segment[:hdr_len]): + return None + tcp = dpkt.tcp.TCP(segment) + attach_buffer(tcp, segment) + return tcp + + +def _decode_ip(protocol: 'int', data: 'bytes') -> 'Optional[IP | IP6]': + """Decode an IP packet the default engine reads behind an AH or IPv6 extension header in IPv4. + + Args: + protocol: The IP-in-IP protocol number, one of the :data:`IP_TUNNELS`. + data: Raw octets from the IP header on, as far as the IPv4 packet + carrying it goes (see :func:`_past_ipv4_extension_headers`). + + Returns: + The packet as `DPKT`_ decodes one in a tunnel, with ``data`` attached by + :func:`attach_buffer`, so that :func:`_layer2bytes` slices its layers + out of the captured octets; or :data:`None` if decoding it raises + anything at all, as :class:`~pcapkit.foundation.engines.dpkt.DPKT` then + keeps a frame raw. `DPKT`_ raises more than its own + :exc:`~dpkt.dpkt.UnpackError`: dpkt 1.9.8 reads ``frag_off`` from + whatever header follows an IPv6 Fragment header, so one followed by + another extension header raises :exc:`AttributeError` (:issue:`1351`). + + .. _DPKT: https://dpkt.readthedocs.io + + """ + import dpkt # isort:skip + + klass = dpkt.ip.IP if protocol == Enum_TransType.IPv4 else dpkt.ip6.IP6 + try: + ip = klass(data) + except Exception: # as the DPKT engine does, for a whole frame + return None + attach_buffer(ip, data) + return ip + + def _tcp_carrier(packet: 'Packet') -> 'tuple[Optional[IP | IP6], Optional[TCP]]': """Fetch the IP layer of a DPKT packet that carries its TCP segment, and the segment. @@ -476,14 +583,20 @@ def _tcp_carrier(packet: 'Packet') -> 'tuple[Optional[IP | IP6], Optional[TCP]]' rejects the header of an IPv4 packet on the way (see :func:`_ipv4_accepted`). A tunnel is a packet whose protocol, or for IPv6 the next header after its extension headers, is one of the - :data:`IP_TUNNELS`. + :data:`IP_TUNNELS`. An IPv4 packet whose protocol names an AH or IPv6 + extension header carries what the default engine reads past them (see + :func:`_past_ipv4_extension_headers`), a tunnel or a TCP segment, which + are decoded from the captured octets (see :func:`_decode_ip` and + :func:`_decode_tcp`). Note: `DPKT`_ decodes an IPv4 or IPv6 packet inside another (:rfc:`2003`, :rfc:`2473`, :rfc:`4213`) as its ``data``. The TCP endpoints are the innermost packet's addresses, as for the default engine (:func:`pcapkit.toolkit.pcap.tcp_segment`); the outer ones are the - tunnel's (:issue:`1581`). + tunnel's (:issue:`1581`). A segment found behind an extension header is + no layer of ``packet``, so it carries its own octets, which + :func:`_layer2bytes` then hands back (:issue:`1606`). .. _DPKT: https://dpkt.readthedocs.io @@ -493,9 +606,19 @@ def _tcp_carrier(packet: 'Packet') -> 'tuple[Optional[IP | IP6], Optional[TCP]]' ip = _network(packet) if ip is None: return None, None + root = packet # the packet ``ip`` is a layer of while True: - if isinstance(ip, dpkt.ip.IP) and not _ipv4_accepted(ip): - return None, None + if isinstance(ip, dpkt.ip.IP): + if not _ipv4_accepted(ip): + return None, None + if ip.p in IPV4_EXTENSION_HEADERS: + upper = _past_ipv4_extension_headers(root, ip) + if upper is not None and upper[0] == Enum_TransType.TCP: + return ip, _decode_tcp(upper[1]) + if upper is None or upper[0] not in IP_TUNNELS or (inner := _decode_ip(*upper)) is None: + return ip, None + root = ip = inner + continue if getattr(ip, 'p', None) not in IP_TUNNELS or not isinstance(ip.data, (dpkt.ip.IP, dpkt.ip6.IP6)): break ip = ip.data @@ -503,6 +626,8 @@ def _tcp_carrier(packet: 'Packet') -> 'tuple[Optional[IP | IP6], Optional[TCP]]' tcp = getattr(ip, 'tcp', None) # type: Optional[TCP] if tcp is None and type(getattr(ip, 'data', None)).__name__ == 'TCP': tcp = cast('TCP', ip.data) + if tcp is not None and root is not packet: + attach_buffer(tcp, _layer2bytes(root, tcp)[0]) return ip, tcp @@ -542,12 +667,13 @@ def packet2chain(packet: 'Packet') -> 'str': packet: DPKT packet. Returns: - Colon (``:``) separated list of protocol chain. + Colon (``:``) separated list of protocol chain. It ends at the first + payload that is no layer, as in :func:`packet2dict`, e.g. VRRP's ``''``. """ chain = [type(packet).__name__] payload = packet.data - while not isinstance(payload, bytes): + while hasattr(payload, '__hdr_fields__'): chain.append(type(payload).__name__) payload = payload.data return ':'.join(chain) @@ -563,7 +689,13 @@ def packet2dict(packet: 'Packet', timestamp: 'float', *, data_link: Data link type. Returns: - Dict[str, Any]: A :obj:`dict` mapping of packet data. + Dict[str, Any]: A :obj:`dict` mapping of packet data. A layer's payload + is mapped as a layer of its own only where it is one, i.e. has the + header fields a :class:`dpkt.dpkt.Packet` has; any other is leaf data. + `DPKT`_ leaves some undecoded payloads as a :obj:`str` rather than as + :obj:`bytes`, e.g. ``''`` for VRRP's. + + .. _DPKT: https://dpkt.readthedocs.io """ def wrapper(packet: 'Packet') -> 'dict[str, Any]': @@ -571,7 +703,7 @@ def wrapper(packet: 'Packet') -> 'dict[str, Any]': for field in packet.__hdr_fields__: dict_[field] = getattr(packet, field, None) payload = packet.data - if not isinstance(payload, bytes): + if hasattr(payload, '__hdr_fields__'): dict_[type(payload).__name__] = wrapper(payload) return dict_ diff --git a/pcapkit/toolkit/scapy.py b/pcapkit/toolkit/scapy.py index df091e0bb7..61e4bc700d 100644 --- a/pcapkit/toolkit/scapy.py +++ b/pcapkit/toolkit/scapy.py @@ -47,7 +47,9 @@ 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.toolkit.pypcapfile import _default_accepts +from pcapkit.toolkit.pypcapfile import (IP_TUNNELS, IPV4_EXTENSION_HEADERS, IPV4_HEADER_LEN, + IPV6_HEADER_LEN, TCP_MIN_HEADER_LEN, _default_accepts, + _past_extension_headers) 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 @@ -332,17 +334,62 @@ def _ipv4_accepted(ipv4: 'IP') -> 'bool': return _default_accepts('IPv4', octets[:max((octets[0] & 0x0F) * 4, 20)]) -def _carrier(layer: 'Packet') -> 'IP | IPv6 | None': - """Fetch the IP layer that carries ``layer``. +def _tcp_behind_extension_headers(segment: 'bytes') -> 'tuple[bytes, bytes] | None': + """Split a TCP segment the default engine reads behind an AH or IPv6 extension header in IPv4. Args: - layer: A layer of a dissected Scapy packet. + segment: Raw octets from the TCP header on, as far as the IPv4 layer + carrying it goes (see :func:`_tcp_carrier`). + + Returns: + The header, as many octets as the Data Offset counts, and the payload + after it; or :data:`None` if the default engine reads no TCP there: + fewer than :data:`TCP_MIN_HEADER_LEN + ` octets, a Data Offset + counting fewer or more than were captured, or options its TCP parser + rejects (see :func:`~pcapkit.toolkit.pypcapfile._default_accepts`). + + Note: + Directly over IP, the default engine reads a TCP header its parser + rejects out of the raw octets left behind + (:func:`pcapkit.toolkit.pcap.tcp_segment`), but behind another header + it does not, so the parser's answer stands here, as for the pypcapfile + engine. + + """ + hdr_len = (segment[12] >> 4) * 4 if len(segment) >= TCP_MIN_HEADER_LEN else 0 + if hdr_len < TCP_MIN_HEADER_LEN or hdr_len > len(segment): + return None + if hdr_len > TCP_MIN_HEADER_LEN and not _default_accepts('TCP', segment[:hdr_len]): + return None + return segment[:hdr_len], segment[hdr_len:] + + +def _tcp_carrier(packet: 'Packet') -> 'tuple[IP | IPv6, TCP, bytes, bytes] | None': + """Fetch the TCP segment of a Scapy packet, and the IP layer that carries it. + + Args: + packet: Scapy packet. Returns: The nearest :class:`~scapy.layers.inet.IP` or - :class:`~scapy.layers.inet6.IPv6` layer above ``layer``, or - :data:`None` if there is none, or the default engine rejects the - header of an IPv4 layer above it (see :func:`_ipv4_accepted`). + :class:`~scapy.layers.inet6.IPv6` layer above the packet's TCP layer, + that layer, and the segment's header and payload (see + :func:`_tcp_segment`); or :data:`None` if there is no such IP layer, or + the default engine reads no TCP there: an IPv4 layer above it has a + header that engine rejects (see :func:`_ipv4_accepted`), or is a later + fragment, whose data is a slice of the datagram's payload with no + upper-layer header in it (:rfc:`791#section-3.2`, :issue:`1617`). + + An IPv4 layer whose protocol names an AH or IPv6 extension header + carries what the default engine reads past those headers, by its own + rules and up to + :data:`~pcapkit.protocols.internet.internet.EXTENSION_HEADER_LIMIT` of + them in a row (see + :func:`~pcapkit.toolkit.pypcapfile._past_extension_headers`): a tunnel, + walked on as above, or a TCP segment (see + :func:`_tcp_behind_extension_headers`), each dissected from the + octets (:issue:`1606`). Note: The nearest is the innermost IP layer of an IP-in-IP tunnel @@ -351,22 +398,51 @@ def _carrier(layer: 'Packet') -> 'IP | IPv6 | None': (:func:`pcapkit.toolkit.pcap.tcp_segment`). ``packet['IP']`` finds the outermost IPv4 layer instead, which is the tunnel's, or with IPv4 carried in IPv6 the inner one (:issue:`1581`). The extension header - layers `Scapy`_ dissects between an IP layer and its payload are not + layers `Scapy`_ dissects between an IPv6 layer and its payload are not IP layers, so are stepped over. + Inside IPv4, `Scapy`_ dissects an AH alone of those headers, and behind + one only IPv6 and, for a Next Header of ``0``, IPv4, so the default + engine's rules are applied to the octets instead. It also dissects IPv6 + and AH inside a later fragment, binding neither to a zero fragment + offset as it binds TCP. + """ - from scapy.layers.inet import IP + from scapy.layers.inet import IP, TCP, IPerror from scapy.layers.inet6 import IPv6 + from scapy.packet import NoPayload + tcp = cast('TCP | None', packet.getlayer('TCP')) carrier = None # type: IP | IPv6 | None - above = layer.underlayer - while above is not None: - if isinstance(above, IP) and not _ipv4_accepted(above): + layer = packet # type: Packet + while layer is not tcp: + if isinstance(layer, NoPayload): return None - if carrier is None and isinstance(above, (IP, IPv6)): - carrier = above - above = above.underlayer - return carrier + if isinstance(layer, IP): + if not _ipv4_accepted(layer) or layer.frag: + return None + carrier = layer + # not the datagram an ICMP error quotes, which the default engine does not dissect + if layer.proto in IPV4_EXTENSION_HEADERS and not isinstance(layer, IPerror): + upper = _past_extension_headers(layer.proto, _within_datagrams(layer.payload)) + if upper is not None and upper[0] == Enum_TransType.TCP: + segment = _tcp_behind_extension_headers(upper[1]) + return None if segment is None else (carrier, TCP(upper[1]), *segment) + if upper is None or upper[0] not in IP_TUNNELS: + return None + version, data = IP_TUNNELS[upper[0]], upper[1] + if len(data) < (IPV4_HEADER_LEN if version == 4 else IPV6_HEADER_LEN): + return None + layer = IP(data) if version == 4 else IPv6(data) + tcp = cast('TCP | None', layer.getlayer('TCP')) + continue + elif isinstance(layer, IPv6): + carrier = layer + layer = layer.payload + + if carrier is None or (segment := _tcp_segment(tcp)) is None: + return None + return carrier, tcp, *segment def ipv4_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv4Address] | None': @@ -512,8 +588,10 @@ def tcp_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'TCP_Packet | None * If the ``packet`` can be used for TCP reassembly. A packet can be reassembled 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, - behind no IPv4 header the default engine rejects. It is keyed by the IP layer - that carries it -- in an IP-in-IP tunnel, the innermost (see :func:`_carrier`). + behind no IPv4 header the default engine rejects and in no later IPv4 fragment, + or a TCP segment the default engine reads behind an AH or IPv6 extension header + in IPv4. It is keyed by the IP layer that carries it -- in an IP-in-IP tunnel, + the innermost (see :func:`_tcp_carrier`). * 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`. @@ -521,11 +599,8 @@ def tcp_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'TCP_Packet | None :class:`pcapkit.foundation.reassembly.tcp.TCP` """ - if 'TCP' in packet: - tcp = cast('TCP', packet['TCP']) - if (ip := _carrier(tcp)) is None or (segment := _tcp_segment(tcp)) is None: - return None - header, payload = segment + if (found := _tcp_carrier(packet)) is not None: + ip, tcp, header, payload = found raw_len = len(payload) # payload length, header excludes data = TCP_Packet( @@ -572,11 +647,8 @@ def tcp_traceflow(packet: 'Packet', *, count: 'int' = -1) -> 'TF_TCP_Packet | No :class:`pcapkit.foundation.traceflow.tcp.TCP` """ - if 'TCP' in packet: - tcp = cast('TCP', packet['TCP']) - if (ip := _carrier(tcp)) is None or (segment := _tcp_segment(tcp)) is None: - return None - header, payload = segment + if (found := _tcp_carrier(packet)) is not None: + ip, tcp, header, payload = found # NOTE: the link type of the interface the packet was read from, which # :class:`~pcapkit.foundation.engines.scapy.Scapy` attaches with diff --git a/tests/foundation/engines/test_ext_chain_limit_agreement_1604_unit.py b/tests/foundation/engines/test_ext_chain_limit_agreement_1604_unit.py index 613ba92488..1664a88dec 100644 --- a/tests/foundation/engines/test_ext_chain_limit_agreement_1604_unit.py +++ b/tests/foundation/engines/test_ext_chain_limit_agreement_1604_unit.py @@ -165,12 +165,6 @@ class TestExtensionHeaderChainAgreement(harness.RoundTripBase): Gap(1609, 'tshark dissects up to about 496 extension headers, so finds the TCP past ' 'EXTENSION_HEADER_LIMIT, carried in IPv4 or IPv6; at 1,000 it gives up and agrees', 'INPUTS', 'extra, default has none', ('ipv4-33/pyshark/trace', 'ipv6-33/pyshark/trace')), - Gap(1606, 'dpkt and scapy dissect no IPv6 extension header carried in IPv4 ' - '(dpkt keeps the payload of IPv4 protocol 60 as bytes, scapy as Raw), so find no ' - 'TCP behind one, where the default engine reads past it at any length up to ' - 'EXTENSION_HEADER_LIMIT', - 'INPUTS', 'missing, default has', ('ipv4-30/dpkt/*', 'ipv4-32/dpkt/*', - 'ipv4-30/scapy/*', 'ipv4-32/scapy/*')), ) @classmethod diff --git a/tests/toolkit/test_tcp_carrier_walk_1606_1617_unit.py b/tests/toolkit/test_tcp_carrier_walk_1606_1617_unit.py new file mode 100644 index 0000000000..f89b8a3890 --- /dev/null +++ b/tests/toolkit/test_tcp_carrier_walk_1606_1617_unit.py @@ -0,0 +1,348 @@ +# -*- coding: utf-8 -*- +"""The dpkt and scapy engines find TCP in IPv4 where the default engine does. C.f. #1606, #1617. + +The default engine dispatches an IPv4 packet's protocol through the registry +IPv6 uses, so it reads an AH or IPv6 extension header carried directly in IPv4, +and the TCP segment or IP-in-IP tunnel behind it, up to +:data:`~pcapkit.protocols.internet.internet.EXTENSION_HEADER_LIMIT` headers in a +row. `dpkt`_ decodes protocol 0 as IPv4 and keeps protocols 43, 44, 60, 135 and +139 as octets, and `Scapy`_ keeps all of them as ``Raw`` but AH, behind which it +dissects no TCP; so neither engine handed that TCP on (#1606). Both now step over +the headers by the default engine's own rules, as the pypcapfile engine does +(:func:`pcapkit.toolkit.pypcapfile._past_extension_headers`). + +A later IPv4 fragment carries no upper-layer header (:rfc:`791#section-3.2`), so +the default engine reads no TCP in one, but `Scapy`_ binds IPv6 to IPv4 protocol +41, and AH to protocol 51, whatever the fragment offset, and its engine handed on +the TCP it then dissected (#1617). It now stops at such a fragment. + +Every capture is built here, so the module is in the unit tier. + +.. _dpkt: https://dpkt.readthedocs.io +.. _Scapy: https://scapy.net + +""" +from __future__ import annotations + +import importlib.util +import os +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 +from tests.foundation import _roundtrip as wire + +if TYPE_CHECKING: + from typing import Any, Callable + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + +#: The engines compared with the default engine. +ENGINES = ('dpkt', 'scapy') + + +def has(engine: str) -> bool: + """Whether ``engine``'s third-party package is installed.""" + return importlib.util.find_spec(engine) is not None + + +def options(nxt: int) -> bytes: + """A Hop-by-Hop or Destination Options header of 8 octets, one PadN option in it.""" + return bytes([nxt, 0, 1, 4, 0, 0, 0, 0]) + + +def ah(nxt: int, length: int = 4) -> bytes: + """An Authentication Header whose Payload Length is ``length``, its ICV filled in.""" + fixed = bytes([nxt, length, 0, 0]) + (0x100).to_bytes(4, 'big') + (7).to_bytes(4, 'big') + return fixed + b'\xaa' * max((length + 2) * 4 - len(fixed), 0) + + +def fragment(nxt: int, offset: int = 0) -> bytes: + """An IPv6 Fragment header; ``offset`` is in octets.""" + return bytes([nxt, 0]) + offset.to_bytes(2, 'big') + (0xDEADBEEF).to_bytes(4, 'big') + + +def tcp(number: int, *, opts: bytes = b'', words: 'int | None' = None) -> bytes: + """A TCP segment carrying ``data`` on a port pair of its own, with ``opts`` and a Data Offset of ``words``.""" + segment = wire.tcp(b'data', seq=1, ack=1, sport=1000 + number) + words = 5 + len(opts) // 4 if words is None else words + return segment[:12] + bytes([words << 4]) + segment[13:20] + opts + segment[20:] + + +def v4(payload: bytes, proto: int, *, length: 'int | None' = None, **fields: 'Any') -> bytes: + """An IPv4 packet carrying ``payload`` as ``proto``, its Total Length ``length`` if given.""" + packet = bytearray(wire.ipv4(payload, proto=proto, **fields)) + if length is not None: + packet[2:4] = length.to_bytes(2, 'big') + packet[10:12] = b'\x00\x00' + packet[10:12] = wire.checksum(bytes(packet[:20])).to_bytes(2, 'big') + return bytes(packet) + + +def v6(payload: bytes, nxt: int) -> bytes: + """An IPv6 packet carrying ``payload`` as ``nxt``.""" + return wire.ipv6(payload, nxt=nxt) + + +def eth(packet: bytes) -> bytes: + """An Ethernet frame carrying the IPv4 or IPv6 ``packet``.""" + return wire.ethernet(packet, 0x86DD if packet[0] >> 4 == 6 else 0x0800) + + +#: Frames carrying TCP behind an AH or IPv6 extension header in IPv4, each with +#: whether the default engine reads the TCP, by a name for it. Each builder takes +#: the TCP segment it is to carry, given by its frame number. +EXTENSION_HEADERS = { + 'hop-by-hop': (True, lambda n: eth(v4(options(6) + tcp(n), 0))), + 'destination options': (True, lambda n: eth(v4(options(6) + tcp(n), 60))), + 'routing': (True, lambda n: eth(v4(bytes([6, 0, 0, 0, 0, 0, 0, 0]) + tcp(n), 43))), + 'fragment': (True, lambda n: eth(v4(fragment(6) + tcp(n), 44))), + # the default engine reads past an IPv6 Fragment header inside IPv4 whatever its offset + 'later fragment header': (True, lambda n: eth(v4(fragment(6, 40) + tcp(n), 44))), + 'mobility': (True, lambda n: eth(v4(bytes([6, 0, 0, 0, 0, 0, 0, 0]) + tcp(n), 135))), + 'hip': (True, lambda n: eth(v4(bytes([6, 4, 1, 0x21, 0, 0, 0, 0]) + b'\x11' * 32 + tcp(n), 139))), + 'ah': (True, lambda n: eth(v4(ah(6) + tcp(n), 51))), + 'ah without icv': (True, lambda n: eth(v4(ah(6, 1) + tcp(n), 51))), + # a Payload Length of 0 declares 8 octets, fewer than the fixed header + 'ah too short': (False, lambda n: eth(v4(ah(6, 0) + tcp(n), 51))), + # Scapy reads a Next Header of 0 as IPv4 + 'ah then hop-by-hop': (True, lambda n: eth(v4(ah(0) + options(6) + tcp(n), 51))), + 'hop-by-hop then destination options': (True, lambda n: eth(v4(options(60) + options(6) + tcp(n), 0))), + 'hop-by-hop then ah': (True, lambda n: eth(v4(options(51) + ah(6) + tcp(n), 0))), + 'chain at the limit': (True, lambda n: eth(v4(b''.join(options(60 if i < 31 else 6) for i in range(32)) + + tcp(n), 60))), + 'chain past the limit': (False, lambda n: eth(v4(b''.join(options(60 if i < 32 else 6) for i in range(33)) + + tcp(n), 60))), + # tunnels behind the headers, and the headers inside tunnels + '4in4 behind hop-by-hop': (True, lambda n: eth(v4(options(4) + v4(tcp(n), 6), 0))), + '6in4 behind hop-by-hop': (True, lambda n: eth(v4(options(41) + v6(tcp(n), 6), 0))), + '6in4 behind ah': (True, lambda n: eth(v4(ah(41) + v6(tcp(n), 6), 51))), + 'hop-by-hop in 4in4': (True, lambda n: eth(v4(v4(options(6) + tcp(n), 0), 4))), + 'hop-by-hop in 4in6': (True, lambda n: eth(v6(v4(options(6) + tcp(n), 0), 4))), + 'hop-by-hop in 4in4 behind hop-by-hop': (True, lambda n: eth(v4(options(4) + v4(options(6) + tcp(n), 0), 0))), + # IPv4 behind the header that the default engine does not read + 'short ipv4 behind hop-by-hop': (False, lambda n: eth(v4(options(4) + v4(tcp(n), 6)[:12], 0))), + 'short ipv6 behind hop-by-hop': (False, lambda n: eth(v4(options(41) + v6(tcp(n), 6)[:30], 0))), + 'bad ipv4 behind hop-by-hop': (False, lambda n: eth(v4(options(4) + b'\x55' + v4(tcp(n), 6)[1:], 0))), + # TCP behind the header, read as the TCP parser reads it + 'tcp options': (True, lambda n: eth(v4(options(6) + tcp(n, opts=b'\x02\x04\x05\xb4'), 0))), + 'rejected tcp options': (False, lambda n: eth(v4(options(6) + tcp(n, opts=b'\x02\x03\x05\xb4'), 0))), + 'tcp data offset past the capture': (False, lambda n: eth(v4(options(6) + tcp(n, words=15), 0))), + 'tcp data offset below 5': (False, lambda n: eth(v4(options(6) + tcp(n, words=4), 0))), + 'tcp cut short': (False, lambda n: eth(v4(options(6) + tcp(n)[:16], 0))), + 'no tcp': (False, lambda n: eth(v4(options(17) + wire.udp(b'data', sport=1000 + n), 0))), + # the Total Length bounds what is read, and a zero one does not + 'total length short of the tcp': (False, lambda n: eth(v4(options(6) + tcp(n), 0, length=40))), + 'total length under the header': (False, lambda n: eth(v4(options(6) + tcp(n), 0, length=12))), + 'zero total length': (True, lambda n: eth(v4(options(6) + tcp(n), 0, length=0))), + 'trailer': (True, lambda n: eth(v4(options(6) + tcp(n), 0)) + b'\x00' * 10), + 'later ipv4 fragment': (False, lambda n: eth(v4(options(6) + tcp(n), 0, offset=8))), + # Scapy dissects the datagram an ICMP error quotes, the default engine does not + 'quoted in an icmp error': (False, lambda n: eth(v4(bytes([3, 3, 0, 0, 0, 0, 0, 0]) + + v4(options(6) + tcp(n), 0), 1))), + 'control': (True, lambda n: eth(v4(tcp(n), 6))), +} + +#: Frames carrying TCP in a later IPv4 fragment or in its first, each with whether +#: the default engine reads the TCP, by a name for it. +FRAGMENTS = { + '6in4 later fragment': (False, lambda n: eth(v4(v6(tcp(n), 6), 41, offset=40))), + '4in4 later fragment': (False, lambda n: eth(v4(v4(tcp(n), 6), 4, offset=40))), + 'ah later fragment': (False, lambda n: eth(v4(ah(41) + v6(tcp(n), 6), 51, offset=40))), + 'tcp later fragment': (False, lambda n: eth(v4(tcp(n), 6, offset=40))), + # in a tunnel, the inner header's offset counts too + '6in4 later fragment in 4in4': (False, lambda n: eth(v4(v4(v6(tcp(n), 6), 41, offset=40), 4))), + '6in4 later fragment in 4in6': (False, lambda n: eth(v6(v4(v6(tcp(n), 6), 41, offset=40), 4))), + '6in4 first fragment': (True, lambda n: eth(v4(v6(tcp(n), 6), 41, mf=True))), + '6in4': (True, lambda n: eth(v4(v6(tcp(n), 6), 41))), +} + + +def build(cases: 'dict[str, tuple[bool, Callable[[int], bytes]]]') -> 'tuple[list[bytes], list[int]]': + """The frames of ``cases``, and the numbers of those the default engine reads TCP from.""" + numbered = list(enumerate(cases.values(), start=1)) + return ([make(number) for number, (_, make) in numbered], + [number for number, (reads, _) in numbered if reads]) + + +def read(test: unittest.TestCase, engine: str, frames: 'list[bytes]') -> 'dict[str, Any]': + """Read ``frames`` with ``engine``, with TCP reassembly and flow tracing. + + Returns: + Every input the engine hands TCP reassembly and flow tracing, keyed by + frame number -- every field, but a trace input's ``frame`` -- the + reassembled datagrams' ``id``, and the traced flows' labels and frames. + + """ + from pcapkit import extract + from pcapkit.foundation.reassembly.tcp import TCP + from pcapkit.foundation.traceflow.traceflow import TraceFlowBase + + seen = {'tcp': {}, 'trace': {}} # type: dict[str, Any] + reassemble, trace = TCP.__call__, TraceFlowBase.__call__ + + def fields(packet: 'Any', skip: 'tuple[str, ...]' = ()) -> 'dict[str, Any]': + return {key: bytes(value) if isinstance(value, bytearray) else value + for key, value in packet.items() if key not in skip} + + def spy_tcp(this: 'Any', packet: 'Any') -> 'Any': + seen['tcp'][packet.num] = fields(packet) + return reassemble(this, packet) + + def spy_trace(this: 'Any', packet: 'Any') -> 'Any': + seen['trace'][packet.index] = fields(packet, ('frame',)) + return trace(this, packet) + + records = [(1, number, octets) for number, octets in enumerate(frames)] + with tempfile.TemporaryDirectory() as tmp, warnings.catch_warnings(), \ + mock.patch.object(TCP, '__call__', spy_tcp), \ + mock.patch.object(TraceFlowBase, '__call__', spy_trace): + path = os.path.join(tmp, 'carrier.pcap') + with open(path, 'wb') as file: + file.write(wire.pcap(records)) + warnings.simplefilter('ignore') + extractor = extract(fin=path, nofile=True, engine=engine, reassembly=True, tcp=True, + trace=True, trace_fout=os.path.join(tmp, 'trace'), trace_format='json') + try: + if extractor._exnam != engine: # pylint: disable=protected-access + test.skipTest(f'{engine} did not run') + seen['ids'] = [datagram.id for datagram in extractor.reassembly.tcp] + seen['flows'] = [(flow.label, tuple(flow.index)) for flow in extractor.trace.tcp] + finally: + close_extractor(extractor) + return seen + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class TCPCarrierWalkTests(unittest.TestCase): + """dpkt and scapy hand on the TCP the default engine reads in IPv4, and no other.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def agree(self, cases: 'dict[str, tuple[bool, Callable[[int], bytes]]]') -> None: + frames, reads = build(cases) + base = read(self, 'default', frames) + # what the default engine reads, as the cases say + self.assertEqual(sorted(base['tcp']), reads) + self.assertEqual(sorted(base['trace']), reads) + self.assertEqual(len(base['ids']), len(reads)) + names = dict(enumerate(cases, start=1)) + for engine in ENGINES: + with self.subTest(engine=engine): + if not has(engine): + self.skipTest(f'{engine} not installed') + seen = read(self, engine, frames) + for aspect in ('tcp', 'trace'): + self.assertEqual( + [names[number] for number in sorted(set(seen[aspect]) ^ set(base[aspect]))], [], + f'{aspect}: frames read by one of {engine} and the default engine alone') + self.assertEqual(seen[aspect], base[aspect], aspect) + for aspect in ('ids', 'flows'): + self.assertEqual(seen[aspect], base[aspect], aspect) + + def test_tcp_behind_an_extension_header_in_ipv4_is_read_as_the_default_engine_reads_it(self) -> None: + # #1606 + self.agree(EXTENSION_HEADERS) + + def test_no_tcp_is_read_in_a_later_ipv4_fragment(self) -> None: + # #1617 + self.agree(FRAGMENTS) + + def test_a_dpkt_packet_built_by_hand_is_walked_too(self) -> None: + # With no record octets attached, the IPv4 payload is DPKT's own + # re-serialisation of it, which here is the octets it was parsed from. + if not has('dpkt'): + self.skipTest('dpkt not installed') + import dpkt + + from pcapkit.toolkit import dpkt as toolkit + + segment = tcp(1, opts=b'\x02\x04\x05\xb4') + for name, packet in (('hop-by-hop', v4(options(6) + segment, 0)), + ('ah', v4(ah(6) + segment, 51)), + ('4in4 behind hop-by-hop', v4(options(4) + v4(segment, 6), 0))): + with self.subTest(case=name): + data = toolkit.tcp_reassembly(dpkt.ethernet.Ethernet(eth(packet)), 1.0) + self.assertIsNotNone(data) + self.assertEqual((data.header, bytes(data.payload)), (segment[:24], segment[24:])) + + def test_a_tunnel_dpkt_cannot_decode_ends_no_extraction(self) -> None: + # dpkt 1.9.8 reads ``frag_off`` from whatever header follows an IPv6 + # Fragment header (#1351), so the IPv6 packet behind this Hop-by-Hop + # header raises AttributeError. The tunnel is left undecoded, as the + # engine leaves a frame it cannot decode, and the next frame is read. + # Whether TCP is read in it, as the default engine does, is #1626. + if not has('dpkt'): + self.skipTest('dpkt not installed') + from pcapkit.toolkit import dpkt as toolkit + + inner = v6(fragment(0) + options(6) + tcp(1), 44) + self.assertIsNone(toolkit._decode_ip(41, inner)) # pylint: disable=protected-access + seen = read(self, 'dpkt', [eth(v4(options(41) + inner, 0)), eth(v4(tcp(2), 6))]) + self.assertIn(2, seen['tcp']) + self.assertIn(2, seen['trace']) + + def test_a_dpkt_payload_that_is_no_packet_is_leaf_data(self) -> None: + # dpkt decodes IPv4 protocol 0 as IPv4, so this Hop-by-Hop header and the + # start of the TCP segment behind it, from source port 112, read as IPv4 + # carrying VRRP, whose data dpkt leaves as ``''``. Tracing the segment + # converts the frame with ``packet2dict``, which takes that for a layer. + if not has('dpkt'): + self.skipTest('dpkt not installed') + import dpkt + + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit import dpkt as toolkit + + octets = eth(v4(options(6) + wire.tcp(b'data', seq=1, ack=1, sport=112), 0)) + packet = dpkt.ethernet.Ethernet(octets) + self.assertEqual(packet.ip.data.vrrp.data, '') + info = toolkit.packet2dict(packet, 1.0, data_link=LinkType.ETHERNET) + self.assertEqual(sorted(info['ETHERNET']['IP']['IP']['VRRP']), sorted(dpkt.vrrp.VRRP.__hdr_fields__)) + + frames = [octets, eth(v4(tcp(2), 6))] + base = read(self, 'default', frames) + self.assertEqual((sorted(base['tcp']), sorted(base['trace'])), ([1, 2], [1, 2])) + seen = read(self, 'dpkt', frames) + for aspect in ('tcp', 'trace', 'ids', 'flows'): + self.assertEqual(seen[aspect], base[aspect], aspect) + + def test_a_dpkt_payload_that_is_no_packet_ends_the_verbose_chain(self) -> None: + # The same frame as above, whose chain ``packet2chain`` prints for each + # frame in verbose mode: VRRP's ``''`` ends it. + if not has('dpkt'): + self.skipTest('dpkt not installed') + import contextlib + import io + + import dpkt + + from pcapkit import extract + from pcapkit.toolkit import dpkt as toolkit + + octets = eth(v4(options(6) + wire.tcp(b'data', seq=1, ack=1, sport=112), 0)) + out = io.StringIO() + with tempfile.TemporaryDirectory() as tmp, warnings.catch_warnings(), contextlib.redirect_stdout(out): + warnings.simplefilter('ignore') + path = os.path.join(tmp, 'vrrp.pcap') + with open(path, 'wb') as file: + file.write(wire.pcap([(1, 0, octets), (1, 1, eth(v4(tcp(2), 6)))])) + extractor = extract(fin=path, nofile=True, engine='dpkt', verbose=True) + try: + ran = extractor._exnam # pylint: disable=protected-access + finally: + close_extractor(extractor) + if ran != 'dpkt': + self.skipTest('dpkt did not run') + self.assertEqual([line for line in out.getvalue().splitlines() if line.startswith('Frame')], + ['Frame 1: Ethernet:IP:IP:VRRP', 'Frame 2: Ethernet:IP:TCP']) + self.assertEqual(toolkit.packet2chain(dpkt.ethernet.Ethernet(octets)), 'Ethernet:IP:IP:VRRP') + + +if __name__ == '__main__': + unittest.main()