From 24814ab683e20660269bc23e4e477eb5bb6ada76 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Sat, 10 Oct 2026 20:19:43 -0400 Subject: [PATCH] fix(toolkit): pyshark traces TCP behind an IPv4 extension header as the default engine does (#1625) - tshark dissects TCP behind an AH or IPv6 extension header in IPv4 by its own rules, leaves a Mobility or HIP header's payload out of the PDML, and reads past Shim6, which the default engine leaves raw in IPv4. - For such a frame, tcp_traceflow now reads the TCP from its octets by the rules the other engines share (pypcapfile's _upper_layer and _default_accepts), from the outermost IP layer the default engine dissects: through IP-in-IP tunnels and no other, up to EXTENSION_HEADER_LIMIT headers. Over a link layer the default engine reads no IP over, the frame is read as tshark dissects it. - That layer is found by the frame's link type in the capture file, which the engine reads itself, not by frame.encap_type: tshark reads link types 12 and 14 as raw IP (101), which the default engine leaves raw. - The octets come from a second tshark process (-T ek -x -J frame), started on first use, read forward, and reaped on close. If it or the capture file cannot be read, the engine warns with PySharkWarning and traces such frames as tshark dissects them. - Drop ipv4-33/pyshark/trace from the #1609 Gap row, which this fixes. Tests: new tests/toolkit/test_pyshark_ipv4_extension_headers_1625_unit.py. --- pcapkit/foundation/engines/pyshark.py | 328 ++++- pcapkit/toolkit/pyshark.py | 508 +++++++- ...est_ext_chain_limit_agreement_1604_unit.py | 4 +- ...yshark_ipv4_extension_headers_1625_unit.py | 1079 +++++++++++++++++ 4 files changed, 1853 insertions(+), 66 deletions(-) create mode 100644 tests/toolkit/test_pyshark_ipv4_extension_headers_1625_unit.py diff --git a/pcapkit/foundation/engines/pyshark.py b/pcapkit/foundation/engines/pyshark.py index 5324bcbf85..14a62dafba 100644 --- a/pcapkit/foundation/engines/pyshark.py +++ b/pcapkit/foundation/engines/pyshark.py @@ -10,31 +10,143 @@ .. _PyShark: https://kiminewt.github.io/pyshark """ +import json +import os +import subprocess import sys +import tempfile from typing import TYPE_CHECKING, cast +from pcapkit.const.pcapng.block_type import BlockType as Enum_BlockType from pcapkit.foundation.engines.engine import EngineBase +from pcapkit.foundation.engines.pcap import PCAP as PCAP_Engine +from pcapkit.foundation.engines.pcapng import PCAPNG as PCAPNG_Engine from pcapkit.foundation.reassembly import ReassemblyManager -from pcapkit.utilities.exceptions import stacklevel +from pcapkit.protocols.misc.pcap.header import Header +from pcapkit.utilities.exceptions import FormatError, stacklevel from pcapkit.utilities.logging import get_logger -from pcapkit.utilities.warnings import AttributeWarning, warn +from pcapkit.utilities.warnings import AttributeWarning, PySharkWarning, warn __all__ = ['PyShark'] if TYPE_CHECKING: - from typing import Optional + from typing import IO, Generator, Literal, Optional, Union from pyshark.capture.file_capture import FileCapture from pyshark.packet.layers.base import BaseLayer from pyshark.packet.packet import Packet as PySharkPacket from pcapkit.foundation.extraction import Extractor + from pcapkit.toolkit.pyshark import Captured #: logging.Logger: Module-level logger, a child of the package-wide #: :data:`pcapkit.utilities.logging.logger`. logger = get_logger(__name__) +def _ek_record(line: 'bytes') -> 'Optional[Union[tuple[int, bytes], str]]': + """Parse a line of :program:`tshark`'s ``-T ek -x -J frame`` output. + + Args: + line: One line of the output. + + Returns: + :data:`None` for an index line, which each packet line follows; for a + packet line, its ``frame.number`` and its frame's captured octets, the + ``frame_raw`` that ``-x`` adds; otherwise, a phrase saying the line + cannot be parsed. + + """ + try: + record = json.loads(line) + if isinstance(record, dict) and set(record) == {'index'}: + return None + layers = record['layers'] + number, octets = layers['frame']['frame_frame_number'], layers['frame_raw'] + if isinstance(number, list): + number = number[0] + if isinstance(octets, list): + octets = octets[0] + return int(number), bytes.fromhex(octets) + except (ValueError, LookupError, TypeError): # not JSON, a field missing, or not text + return f'gave output it cannot parse: {bytes(line[:80])!r}' + + +def _linktypes(path: 'str') -> 'Generator[int, None, None]': + """Read the link type of each packet of a capture file, in order, as the default engine reads it. + + Args: + path: The PCAP or PCAP-NG file. + + Yields: + The link type the default engine dissects each packet by. A PCAP file + has one, in its global header. A PCAP-NG packet takes that of the + interface it was captured on, among the Interface Description Blocks of + its own section: an Enhanced or obsolete Packet Block names it by ID, and + a Simple Packet Block is the first's. These three are the packets the + default engine counts as frames; every other block is skipped. + + Raises: + FormatError: If a PCAP-NG block is of no byte order a Section Header + Block declares, too short for its fields, or a packet block naming + an interface its section does not describe. A block cut short by + the end of the file ends the packets instead. + + Note: + The file is opened at the first packet asked for, and closed at its end, + or when the generator is closed. + + """ + # length of the field after a block's type and length that names its link + # type (IDB) or interface (EPB, obsolete Packet Block), by block type + fields = { + int(Enum_BlockType.Interface_Description_Block): 2, + int(Enum_BlockType.Enhanced_Packet_Block): 4, + int(Enum_BlockType.Packet_Block): 2, + } # type: dict[int, int] + + with open(path, 'rb') as file: + if file.read(4) in PCAP_Engine.MAGIC_NUMBER: + file.seek(0) + linktype = int(Header(file).protocol) + while True: + yield linktype + + file.seek(0) + byteorder = None # type: Optional[Literal['big', 'little']] + interfaces = [] # type: list[int] + while True: + head = file.read(8) + if head[:4] == PCAPNG_Engine.MAGIC_NUMBER[0]: # a palindrome, the same in either byte order + head += file.read(4) + if len(head) < 12: + return + byteorder = PCAPNG_Engine.BYTE_ORDER_MAGIC.get(head[8:]) + interfaces = [] # an interface ID counts the IDBs of its own section + elif len(head) < 8: + return + if byteorder is None: + raise FormatError(f'PCAP-NG: block of no known byte order: {head!r}') + kind, length = int.from_bytes(head[:4], byteorder), int.from_bytes(head[4:8], byteorder) + size = fields.get(kind, 0) + if length < len(head) + size + 4: # the type, length, field, and trailing length + raise FormatError(f'PCAP-NG: [Block {kind}] invalid block length: {length}') + + field = file.read(size) + file.seek(length - len(head) - size - 4, os.SEEK_CUR) + if len(field) < size or len(file.read(4)) < 4: + return + + if kind == Enum_BlockType.Interface_Description_Block: + interfaces.append(int.from_bytes(field, byteorder)) + elif kind in (Enum_BlockType.Enhanced_Packet_Block, Enum_BlockType.Packet_Block, + Enum_BlockType.Simple_Packet_Block): + interface = int.from_bytes(field, byteorder) # 0 for a Simple Packet Block, which has none + if interface >= len(interfaces): + raise FormatError(f'PCAP-NG: [Block {kind}] invalid interface ID: {interface}') + yield interfaces[interface] + + def _frame_layer(packet: 'PySharkPacket') -> 'Optional[BaseLayer]': """The ``frame`` layer of a `PyShark`_ packet, or :data:`None` when the record has none. @@ -88,6 +200,21 @@ class PyShark(EngineBase['PySharkPacket']): _expkg: 'pyshark' #: Engine extraction temporary storage. _extmp: 'FileCapture' + #: The second :program:`tshark` pass :meth:`_captured` reads captured + #: octets from, once it is first asked for some. + _exraw: 'Optional[subprocess.Popen[bytes]]' + #: Where that pass writes its standard error. + _exraw_err: 'Optional[IO[bytes]]' + #: The frame number and octets :meth:`_captured` last read from it. + _exraw_held: 'Optional[tuple[int, bytes]]' + #: The link type of each packet of the capture file (see + #: :func:`_linktypes`), which :meth:`_captured` reads, once it is first + #: asked for one. + _exlink: 'Optional[Generator[int, None, None]]' + #: The count and link type of the packet :meth:`_captured` last read from it. + _exlink_held: 'Optional[tuple[int, int]]' + #: Whether reading either has failed, so :meth:`_captured` reads no more. + _exraw_failed: 'bool' ########################################################################## # Defaults. @@ -225,6 +352,12 @@ def __init__(self, extractor: 'Extractor') -> 'None': self._expkg = pyshark self._extmp = cast('FileCapture', None) + self._exraw = None + self._exraw_err = None + self._exraw_held = None + self._exlink = None + self._exlink_held = None + self._exraw_failed = False super().__init__(extractor) @@ -336,7 +469,7 @@ def read_frame(self) -> 'PySharkPacket': # trace flows if ext._flag_t: if ext._tcp: - data_tf_tcp = tcp_traceflow(packet, count=ext._frnum) + data_tf_tcp = tcp_traceflow(packet, count=ext._frnum, captured=self._captured) if data_tf_tcp is not None: ext._trace.tcp(data_tf_tcp) @@ -348,11 +481,194 @@ def read_frame(self) -> 'PySharkPacket': # return frame record return packet + def _octets_command(self) -> 'list[str]': + """The command line of the second :program:`tshark` pass :meth:`_captured` reads. + + Returns: + :program:`tshark`, found as `PyShark`_ finds it, reading the capture + with the preferences :meth:`run` gives `PyShark`_ + (:attr:`OVERRIDE_PREFS`) and no name resolution, and writing the + frames :data:`~pcapkit.toolkit.pyshark.OCTETS_FILTER` matches as + Elasticsearch JSON, one line each, with their captured octets + (``-x``) and no protocol's fields but the frame's (``-J frame``). + + .. _PyShark: https://kiminewt.github.io/pyshark + + """ + from pyshark.tshark.tshark import get_process_path # isort:skip + + from pcapkit.toolkit.pyshark import OCTETS_FILTER # isort:skip + + command = [get_process_path(), '-l', '-n', '-r', self._extractor._ifnm, '-Y', OCTETS_FILTER, + '-T', 'ek', '-x', '-J', 'frame'] + for name, value in self.OVERRIDE_PREFS.items(): + command += ['-o', f'{name}:{value}'] + return command + + def _read_octets(self, number: 'int') -> 'Union[bytes, str]': + """Read the captured octets of a frame from the second :program:`tshark` pass. + + Args: + number: The frame's ``frame.number``. + + Returns: + The octets, or a phrase saying why there are none. + + """ + if self._exraw is None: + self._exraw_err = tempfile.TemporaryFile() # pylint: disable=consider-using-with + self._exraw = subprocess.Popen(self._octets_command(), stdin=subprocess.DEVNULL, # pylint: disable=consider-using-with + stdout=subprocess.PIPE, stderr=self._exraw_err) + held = self._exraw_held + while held is None or held[0] < number: + line = cast('IO[bytes]', self._exraw.stdout).readline() + if not line: + status = self._exraw.wait() + if not status: + return f'ended before frame {number}' + error = cast('IO[bytes]', self._exraw_err) + error.seek(0) + last = error.read().decode('utf-8', 'replace').strip().splitlines()[-1:] + return f'exited with status {status}' + ''.join(f' ({text})' for text in last) + record = _ek_record(line) + if isinstance(record, str): + return record + if record is not None: + held = self._exraw_held = record + if held[0] != number: + return f'has no frame {number}' + return held[1] + + def _read_linktype(self, count: 'int') -> 'Union[int, str]': + """Read the link type the default engine dissects a frame by, from the capture file. + + Args: + count: The frame's count, as the default engine numbers frames. + + Returns: + The link type, or a phrase saying why there is none. + + """ + if self._exlink is None: + self._exlink = _linktypes(self._extractor._ifnm) + held = self._exlink_held + while held is None or held[0] < count: + linktype = next(self._exlink, None) + if linktype is None: + return f'has no packet {count}' + held = self._exlink_held = ((0 if held is None else held[0]) + 1, linktype) + if held[0] != count: + return f'was read past packet {count}' + return held[1] + + def _captured(self, packet: 'PySharkPacket') -> 'Optional[Captured]': + """Read a frame's link type and octets as the default engine reads them, which the PDML :meth:`run` reads lacks. + + :func:`~pcapkit.toolkit.pyshark.tcp_traceflow` asks for one of a frame + :data:`~pcapkit.toolkit.pyshark.OCTETS_FILTER` matches only (:issue:`1625`). + + Its link type is the one the default engine dissects it by, read from + the capture file (see :func:`_linktypes`) for the frame of this engine's + count, which numbers frames as the default engine does. tshark's + ``frame.encap_type`` cannot say which it is, as tshark reads several link + types as one (see :data:`~pcapkit.toolkit.pyshark.ENCAP_TYPE_TO_LINKTYPE`). + + Its captured octets come from a second :program:`tshark` pass over the + capture (see :meth:`_octets_command`), started on the first request. + Both passes number frames alike and deliver them in order, so the second + is read forward to the frame asked for, and the last frame it read is + kept for the next request. The capture file is read forward likewise. A + capture with no such frame reads neither. + + Args: + packet: PyShark packet of the first pass. + + Returns: + The frame's link type and captured octets; :data:`None` once either + cannot be read. + + Warns: + PySharkWarning: When the frame cannot be read: the second pass + cannot be started, exits with an error, gives output that cannot + be parsed, or lacks the frame asked for; or the capture file + cannot be read for the frame's link type. Both are then stopped, + and this and every later frame is traced as tshark dissects it, + as before :issue:`1625`, which can disagree with the default + engine. + + Note: + The second pass is a process of this engine's own rather than a + second :class:`pyshark.FileCapture`, for two reasons measured with + `PyShark`_ 0.6 and :program:`tshark` 4.6.9. Its JSON parser reads + ``frame.time_epoch``, which a Simple Packet Block lacks, and raises + :exc:`KeyError`. And each capture it creates sets a new + :class:`asyncio.SafeChildWatcher` (``Capture._setup_eventloop``), so + the process of the first is left unreaped once a second exists. + + .. _PyShark: https://kiminewt.github.io/pyshark + + """ + from pcapkit.toolkit.pyshark import Captured # isort:skip + + if self._exraw_failed: + return None + frnum = self._extractor._frnum + source = ('the capture file, read for the link type of a frame with an AH or IPv6 extension ' + 'header in IPv4,') + try: + linktype = self._read_linktype(frnum) + if isinstance(linktype, str): + reason = linktype + else: + source = ('the second tshark pass, which reads the captured octets of a frame with an AH ' + 'or IPv6 extension header in IPv4,') + octets = self._read_octets(int(packet.number)) + if isinstance(octets, bytes): + return Captured(linktype, octets) + reason = octets + except Exception as exc: # pylint: disable=broad-except + reason = f'failed: {type(exc).__name__}: {exc}' + + self._exraw_failed = True + self._stop_captured() + warn(f'Frame {frnum}: {source} {reason}; this and later such frames are traced as tshark ' + 'dissects them, which can disagree with the default engine', + PySharkWarning, stacklevel=stacklevel()) + return None + + def _stop_captured(self) -> 'None': + """Stop reading frames as :meth:`_captured` reads them. + + The second :program:`tshark` pass, if it was started, is stopped and + reaped, and the capture file, if it was opened, closed. + + """ + process, self._exraw = self._exraw, None + error, self._exraw_err = self._exraw_err, None + linktypes, self._exlink = self._exlink, None + try: + if process is not None: + if process.poll() is None: + process.kill() + process.wait() + cast('IO[bytes]', process.stdout).close() + finally: + try: + if error is not None: + error.close() + finally: + if linktypes is not None: + linktypes.close() + def close(self) -> 'None': """Close engine. This method is to be used for closing the engine instance. It is to - close the engine instance after the extraction process is finished. + close the engine instance after the extraction process is finished, + and what :meth:`_captured` reads frames from, if anything. """ - self._extmp.close() + try: + self._extmp.close() + finally: + self._stop_captured() diff --git a/pcapkit/toolkit/pyshark.py b/pcapkit/toolkit/pyshark.py index d02542554b..0850dca70b 100644 --- a/pcapkit/toolkit/pyshark.py +++ b/pcapkit/toolkit/pyshark.py @@ -18,18 +18,25 @@ """ import decimal import ipaddress -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, NamedTuple +from pcapkit.const.reg.ethertype import EtherType as Enum_EtherType from pcapkit.const.reg.linktype import LinkType as Enum_LinkType +from pcapkit.const.reg.transtype import TransType as Enum_TransType from pcapkit.foundation.traceflow.data.tcp import Packet as TF_TCP_Packet +from pcapkit.protocols.link.loopback import family_byteorder +from pcapkit.toolkit.pypcapfile import (IP_TUNNELS, IPV4_EXTENSION_HEADERS, IPV4_HEADER_LEN, + TCP_MIN_HEADER_LEN, VLAN_TAG_LEN, VLAN_TAGS, + _default_accepts, _upper_layer) from pcapkit.utilities.exceptions import MissingKeyError, ProtocolError if TYPE_CHECKING: - from typing import Any + from ipaddress import IPv4Address, IPv6Address + from typing import Any, Callable, Optional from pyshark.packet.packet import Packet -__all__ = ['packet2dict', 'tcp_traceflow', 'ENCAP_TYPE_TO_LINKTYPE', 'FILTER_NAME_TO_LINKTYPE'] +__all__ = ['packet2dict', 'tcp_traceflow', 'Captured', 'ENCAP_TYPE_TO_LINKTYPE', 'FILTER_NAME_TO_LINKTYPE'] #: ``frame.encap_type`` -- Wireshark's internal ``WTAP_ENCAP_*`` number, which :mod:`pyshark` #: exposes as ``packet.frame_info.encap_type`` -- to the matching @@ -61,6 +68,14 @@ #: 152 of them are below; ``frame.encap_type`` 32 (``editcap -T hhdlc``, DLT 121) is left out #: because :class:`LinkType` has no member for 121. An encapsulation absent from this table raises #: rather than resolving to a near-miss DLT. +#: +#: In the read direction, one key can come from several DLTs. Measured with ``tshark`` 4.6.9 over +#: every PCAP-NG link type from 0 to 65535 (bar 144 and 177, under which it rejected the test +#: frame), ``frame.encap_type`` 7 is read from DLTs 12, 14 and 101, 4 from 9 and 50, 10 from 11 +#: and 100, 11 from 16, 18, 19 and 106, 28 from 104 and 112, and 38 from 13 and 109; each other +#: key of this table from one DLT. So where :func:`tcp_traceflow` reads a frame as the default +#: engine does, it takes the link type from the capture file rather than from this table +#: (:issue:`1625`). ENCAP_TYPE_TO_LINKTYPE = { 1: Enum_LinkType.ETHERNET, # ether 2: Enum_LinkType.IEEE802_5, # tr @@ -322,6 +337,74 @@ #: on tshark 4.6.9); ``'1'``/``'0'`` is the spelling this module first parsed (#1514). _BOOLEAN_FIELD = {'1': True, '0': False, 'True': True, 'False': False} # type: dict[str, bool] +#: IPv4 protocol numbers that make :func:`tcp_traceflow` read a frame from its +#: captured octets (see :func:`_extension_header_carrier`): the AH and IPv6 +#: extension headers. The default engine reads past those of the +#: :data:`~pcapkit.toolkit.pypcapfile.IPV4_EXTENSION_HEADERS` by rules of its +#: own, and past Shim6 not at all, as its parser refuses IPv4; tshark dissects +#: the TCP behind any of them by its own rules. +OCTETS_PROTOCOLS = IPV4_EXTENSION_HEADERS | {Enum_TransType.Shim6} + +#: tshark display filter matching every frame :func:`tcp_traceflow` asks the +#: captured octets of: one with an IPv4 header whose protocol is one of the +#: :data:`OCTETS_PROTOCOLS`. It matches every ``ip.proto`` field of a frame, so +#: it matches all of those frames and maybe others. The members are joined with +#: ``||`` rather than written as a set, as tshark 4.6.9 refuses a set whose +#: members are separated by spaces only. +OCTETS_FILTER = ' || '.join(f'ip.proto == {int(protocol)}' for protocol in sorted(OCTETS_PROTOCOLS)) + +#: Length of an Ethernet header: two MAC addresses and the EtherType. +ETHERNET_HEADER_LEN = 14 + +#: Length of a BSD loopback header: the address family. +LOOPBACK_HEADER_LEN = 4 + +#: EtherTypes of the IP layers the default engine dissects over Ethernet, each +#: mapped to its IP version (see :attr:`Link.__proto__ +#: `). +ETHERTYPE_IP_VERSIONS = { + Enum_EtherType.Internet_Protocol_version_4: 4, + Enum_EtherType.Internet_Protocol_version_6: 6, +} # type: dict[int, int] + +#: Address families of the IP layers the default engine dissects over BSD +#: loopback, each mapped to its IP version (see :attr:`Loopback.__proto__ +#: `). +LOOPBACK_IP_VERSIONS = {2: 4, 24: 6, 28: 6, 30: 6} # type: dict[int, int] + + +class Captured(NamedTuple): + """A frame as the default engine reads it from the capture file (:issue:`1625`).""" + + #: Link type the default engine dissects the frame by: the PCAP file's, or + #: that of the PCAP-NG interface the frame was captured on. + linktype: 'int' + #: The frame's captured octets. + octets: 'bytes' + + +class _Segment(NamedTuple): + """The fields of a TCP segment that a flow tracing record takes from it.""" + + #: Source address of the IP layer carrying the segment. + src: 'IPv4Address | IPv6Address' + #: Destination address of that IP layer. + dst: 'IPv4Address | IPv6Address' + #: TCP source port. + srcport: 'int' + #: TCP destination port. + dstport: 'int' + #: TCP sequence number, as on the wire. + seq: 'int' + #: TCP acknowledgement number, as on the wire. + ack: 'int' + #: TCP synchronise (SYN) flag. + syn: 'bool' + #: TCP finish (FIN) flag. + fin: 'bool' + #: TCP reset (RST) flag. + rst: 'bool' + def _parse_flag(value: 'Any', name: 'str') -> 'bool': """Parse a boolean PyShark field, such as ``tcp.flags_syn``. @@ -513,7 +596,340 @@ def _has_fixed_header(tcp: 'Any') -> 'bool': return 'urgent_pointer' in tcp.field_names -def tcp_traceflow(packet: 'Packet', *, count: 'int' = -1) -> 'TF_TCP_Packet | None': +def _from_layers(packet: 'Packet') -> 'Optional[_Segment]': + """Read the TCP segment of a PyShark packet from tshark's dissection of it. + + Args: + packet: PyShark packet. + + Returns: + The packet's first TCP layer, if it has a whole fixed header (see + :func:`_has_fixed_header`), keyed by the IP layer that carries it -- in + an IP-in-IP tunnel, the innermost (see :func:`_carrier`); else + :data:`None`. + + Raises: + ProtocolError: If a TCP flag field is not a boolean tshark spelling. + + """ + ip, tcp = _carrier(packet) + if ip is None or tcp is None or not _has_fixed_header(tcp): + return None + return _Segment( + src=ipaddress.ip_address(ip.src), + dst=ipaddress.ip_address(ip.dst), + srcport=int(tcp.srcport), + dstport=int(tcp.dstport), + seq=_absolute(tcp, 'seq'), + ack=_absolute(tcp, 'ack'), + syn=_parse_flag(tcp.flags_syn, 'tcp.flags_syn'), + fin=_parse_flag(tcp.flags_fin, 'tcp.flags_fin'), + rst=_parse_flag(tcp.flags_reset, 'tcp.flags_reset'), + ) + + +def _extension_header_carrier(packet: 'Packet') -> 'Any': + """Find the IPv4 layer of a PyShark packet whose protocol names an AH or IPv6 extension header. + + Args: + packet: PyShark packet. + + Returns: + The first ``ip`` layer above the packet's first ``tcp`` layer, or of a + packet with none, whose ``ip.proto`` is one of the + :data:`OCTETS_PROTOCOLS`; else :data:`None`. + + """ + for layer in packet.layers: + name = layer.layer_name.lower() + if name == 'tcp': + break + if name == 'ip': + try: + protocol = int(str(getattr(layer, 'proto', ''))) + except ValueError: + continue + if protocol in OCTETS_PROTOCOLS: + return layer + return None + + +def _link_type(packet: 'Packet', frame: 'Any') -> 'Enum_LinkType': + """Resolve the link type of a PyShark packet's frame. + + Args: + packet: PyShark packet. + frame: Its ``frame`` layer (see :func:`_frame_layer`). + + Returns: + The link type ``frame.encap_type`` names, through + :data:`ENCAP_TYPE_TO_LINKTYPE`, or the PDML root layer's filter name, + through :data:`FILTER_NAME_TO_LINKTYPE`, where the frame layer has no + such field. + + Raises: + MissingKeyError: If the link type cannot be resolved. + + """ + # NOTE: the link type is keyed on ``frame.encap_type`` -- Wireshark's + # internal ``WTAP_ENCAP_*`` number -- through + # ``ENCAP_TYPE_TO_LINKTYPE`` (module level, above), and only on the + # PDML root layer's *filter* name when the frame layer has no such + # field. That name cannot do the job on its own: one filter name + # serves several DLTs, so keying on it would answer ``RAW`` (101) for a + # ``rawip6`` capture and ``NULL`` (0) for a ``DLT_LOOP`` one -- valid + # DLTs, wrong ones, and silent (#843). ``frame.encap_type`` + # distinguishes them; see that table's comment for the measurements. + # + # PyShark reports every field as ``LayerFieldsContainer``, a ``str`` + # subclass, so the number arrives as text and needs ``int()``; and a + # field the frame layer does not carry raises ``AttributeError`` out + # of ``pyshark.packet.layers.base.BaseLayer.__getattr__``, which is + # what the ``getattr`` default absorbs here. + # + # Neither path substitutes a DLT on a miss. A lookup raises on an + # unresolvable key rather than minting one (#775), and + # NULL and RAW are genuine DLTs -- BSD loopback and raw IP framing, + # respectively -- each meant to go with its own handler protocol + # class, so neither is an honest stand-in for "unknown link type". + # The bare ``KeyError``/``ValueError`` is re-raised as + # :exc:`~pcapkit.utilities.exceptions.MissingKeyError` -- this + # package's own house exception for a lookup miss -- rather than + # letting it escape the public :func:`tcp_traceflow`. + # + # The root layer is the first layer that is not the frame layer. + encap_type = getattr(frame, 'encap_type', None) + if encap_type is None: + name = [layer for layer in packet.layers if layer is not frame][0].layer_name + try: + return FILTER_NAME_TO_LINKTYPE[name.lower()] + except KeyError: + raise MissingKeyError(name) from None + try: + return ENCAP_TYPE_TO_LINKTYPE[int(encap_type)] + except (KeyError, ValueError): + raise MissingKeyError('frame.encap_type=%s' % encap_type) from None + + +def _network_start(linktype: 'int', octets: 'bytes') -> 'Optional[tuple[int, int]]': + """Find the outermost IP layer the default engine dissects in a frame, from its octets. + + Args: + linktype: The link type the default engine dissects the frame by, the + capture file's (see :class:`Captured`). + octets: The frame's captured octets. + + Returns: + The layer's IP version and where its header starts in ``octets``, or + :data:`None` if the default engine dissects no IP layer right after + the link layer. + + Note: + The default engine dissects a frame by its link type + (:attr:`Frame.__proto__ `). + That is the capture file's, not the one tshark's ``frame.encap_type`` + names (see :func:`_link_type`): tshark reads link types 12 and 14, raw + IP on some BSDs, as it reads ``RAW`` (101), and the default engine + leaves them raw. + Over Ethernet, it reads the EtherTypes in :data:`ETHERTYPE_IP_VERSIONS` + after the 14-octet header and every one of the + :data:`~pcapkit.toolkit.pypcapfile.VLAN_TAGS` behind it; over BSD + loopback (``NULL`` and ``LOOP``), the address families in + :data:`LOOPBACK_IP_VERSIONS` after the 4-octet family, in the byte + order :func:`~pcapkit.protocols.link.loopback.family_byteorder` gives. + ``IPV4`` and ``IPV6`` frames are IP, and a ``RAW`` one is the IP + version its first octet gives, 4 or 6. It leaves every other link type, + EtherType and family raw, and a frame cut short of its link header. + + """ + if linktype == Enum_LinkType.ETHERNET: + if len(octets) < ETHERNET_HEADER_LEN: + return None + ethertype, start = int.from_bytes(octets[12:14], 'big'), ETHERNET_HEADER_LEN + while ethertype in VLAN_TAGS: + if len(octets) < start + VLAN_TAG_LEN: + return None + ethertype, start = int.from_bytes(octets[start + 2:start + 4], 'big'), start + VLAN_TAG_LEN + version = ETHERTYPE_IP_VERSIONS.get(ethertype) + elif linktype in (Enum_LinkType.NULL, Enum_LinkType.LOOP): + if len(octets) < LOOPBACK_HEADER_LEN: + return None + family, start = octets[:LOOPBACK_HEADER_LEN], LOOPBACK_HEADER_LEN + version = LOOPBACK_IP_VERSIONS.get(int.from_bytes(family, family_byteorder(family, linktype))) + elif linktype == Enum_LinkType.RAW: + if not octets: + return None + version, start = octets[0] >> 4, 0 + elif linktype in (Enum_LinkType.IPV4, Enum_LinkType.IPV6): + version, start = (4 if linktype == Enum_LinkType.IPV4 else 6), 0 + else: + return None + if version is None or version not in (4, 6): + return None + return version, start + + +def _network(packet: 'Packet', captured: 'Captured') -> 'Optional[tuple[int, bytes]]': + """Find the outermost IP packet the default engine dissects in a PyShark packet's frame. + + Args: + packet: PyShark packet. + captured: The frame, as the default engine reads it from the capture file. + + Returns: + The packet's IP version and its octets on, to the end of the frame, as + :func:`_network_start` finds it by the frame's link type, if tshark + dissected it too: the packet's first ``ip`` or ``ipv6`` layer is of that + version, and tshark reports its version field there, with the layer's + source and destination addresses in the octets. Else :data:`None`: the + default engine dissects no IP packet right after the link layer, or + tshark dissected that layer out of other octets, e.g. another frame's. + + """ + linktype, octets = captured + found = _network_start(linktype, octets) + if found is None: + return None + version, start = found + + name = 'ip' if version == 4 else 'ipv6' + ip = next((layer for layer in packet.layers if layer.layer_name.lower() in ('ip', 'ipv6')), None) + if ip is None or ip.layer_name.lower() != name: + return None + try: + position = int(ip.version.pos) + addresses = bytes.fromhex(str(ip.src.raw_value) + str(ip.dst.raw_value)) + except (AttributeError, TypeError, ValueError): + return None + where = start + (12 if version == 4 else 8) # the source address + if (position != start or len(addresses) != (8 if version == 4 else 32) + or octets[where:where + len(addresses)] != addresses): + return None + return version, octets[start:] + + +def _from_octets(version: 'int', data: 'bytes') -> 'Optional[_Segment]': + """Read the TCP segment the default engine reads in an IP packet, from its octets. + + Args: + version: IP version of the header ``data`` starts with, ``4`` or ``6``. + data: Raw octets from the IP header on, to the end of the frame. + + Returns: + The segment, keyed by the IP layer that carries it -- in an IP-in-IP + tunnel, the innermost -- or :data:`None` if the default engine reads no + TCP segment there. Only IPv4 and IPv6 in IP (the + :data:`~pcapkit.toolkit.pypcapfile.IP_TUNNELS`) are followed: no other + tunnel, such as GRE or VXLAN, as the default engine dissects none. + + Every IPv4 header on the way must have options the default engine's + IPv4 parser accepts (see + :func:`~pcapkit.toolkit.pypcapfile._default_accepts`), and each IP + packet is followed past its headers by + :func:`~pcapkit.toolkit.pypcapfile._upper_layer`: an AH or IPv6 + extension header in IPv4 by the default engine's own parser, and the + extension headers of IPv6, up to + :data:`~pcapkit.protocols.internet.internet.EXTENSION_HEADER_LIMIT` of + them in a row; none in a later fragment. A TCP header must be at least + :data:`~pcapkit.toolkit.pypcapfile.TCP_MIN_HEADER_LEN` octets long, as + its Data Offset gives it, with that many captured. + + Behind an AH or IPv6 extension header in IPv4, the default engine reads + TCP only as its TCP parser does, so the whole of that header must be + there, with options the parser accepts. Directly over IP, it reads a + header the parser rejects out of the raw octets left behind + (:func:`pcapkit.toolkit.pcap.tcp_segment`), the rule + :func:`_has_fixed_header` applies to tshark's own dissection. + + Note: + These are the rules the pypcapfile, dpkt and scapy engines apply to + the same octets (:func:`pcapkit.toolkit.pypcapfile._transport`, + :issue:`1606`). + + """ + while True: + if version == 4 and len(data) >= IPV4_HEADER_LEN: + ihl = (data[0] & 0x0F) * 4 + if IPV4_HEADER_LEN < ihl <= len(data) and not _default_accepts('IPv4', data[:ihl]): + return None + upper = _upper_layer(version, data) + if upper is None: + return None + if upper[0] not in IP_TUNNELS: + break + version, data = IP_TUNNELS[upper[0]], upper[1] + + protocol, segment = upper + if protocol != Enum_TransType.TCP or len(segment) < TCP_MIN_HEADER_LEN: + return None + header = (segment[12] >> 4) * 4 # the Data Offset, in octets + if header < TCP_MIN_HEADER_LEN: + return None + if version == 4 and data[9] != Enum_TransType.TCP and (header > len(segment) or ( + header > TCP_MIN_HEADER_LEN and not _default_accepts('TCP', segment[:header]))): + return None + + src, dst = (data[12:16], data[16:20]) if version == 4 else (data[8:24], data[24:40]) + flags = segment[13] + return _Segment( + src=ipaddress.ip_address(src), + dst=ipaddress.ip_address(dst), + srcport=int.from_bytes(segment[0:2], 'big'), + dstport=int.from_bytes(segment[2:4], 'big'), + seq=int.from_bytes(segment[4:8], 'big'), + ack=int.from_bytes(segment[8:12], 'big'), + syn=bool(flags & 0x02), + fin=bool(flags & 0x01), + rst=bool(flags & 0x04), + ) + + +def _segment(packet: 'Packet', captured: 'Optional[Callable[[Packet], Optional[Captured]]]') -> 'Optional[_Segment]': + """Read the TCP segment of a PyShark packet that the default engine reads. + + Args: + packet: PyShark packet. + captured: Fetches a frame as the default engine reads it from the + capture file, or :data:`None` where it cannot be had (see + :func:`tcp_traceflow`). + + Returns: + If the packet has an IPv4 layer whose protocol names an AH or IPv6 + extension header (see :func:`_extension_header_carrier`), the segment + read from the octets of the frame's outermost IP packet on (see + :func:`_network` and :func:`_from_octets`), wherever that layer is; + otherwise, or where the frame cannot be had or the default engine + dissects no IP packet after its link layer, the segment of tshark's + dissection (see :func:`_from_layers`). :data:`None` if there is none. + + Raises: + ProtocolError: If a TCP flag field of tshark's dissection is not a + boolean tshark spelling. + + Note: + tshark dissects such a header and what its Next Header names by rules + of its own, and nothing past a Mobility (135) or HIP (139) header, + whose payload it leaves out of the PDML that PyShark reads. The + default engine dispatches IPv4's protocol through the registry IPv6 + uses, reads TCP behind such a header only if its TCP parser accepts + it, and none behind Shim6 (140), whose parser refuses IPv4 + (:issue:`1625`). Reading from the outermost IP packet, as the default + engine does, stops where it stops before such a header: at a tunnel it + does not dissect, such as GRE or VXLAN, which tshark does, or past + :data:`~pcapkit.protocols.internet.internet.EXTENSION_HEADER_LIMIT` + extension headers in a row of an IPv6 packet carrying it. + + """ + if captured is not None and _extension_header_carrier(packet) is not None: + frame = captured(packet) + network = None if frame is None else _network(packet, frame) + if network is not None: + return _from_octets(*network) + return _from_layers(packet) + + +def tcp_traceflow(packet: 'Packet', *, count: 'int' = -1, + captured: 'Optional[Callable[[Packet], Optional[Captured]]]' = None) -> 'TF_TCP_Packet | None': """Trace packet flow for TCP. Args: @@ -521,6 +937,14 @@ def tcp_traceflow(packet: 'Packet', *, count: 'int' = -1) -> 'TF_TCP_Packet | No count: Packet index. If not provided, default to ``-1``. The engine passes its own frame count, not tshark's ``frame.number``, which also numbers records that are not packets (#1515). + captured: Fetches a frame as the default engine reads it from the + capture file, given its packet: its link type there and its + captured octets, neither of which the PDML carries; or + :data:`None` where it cannot. It is called only for a packet + :data:`OCTETS_FILTER` matches. The engine passes + :meth:`PyShark._captured + `; without + it, such a packet is read as tshark dissects it. Returns: Data for TCP flow tracing. @@ -528,7 +952,10 @@ def tcp_traceflow(packet: 'Packet', *, count: 'int' = -1) -> 'TF_TCP_Packet | No * If the ``packet`` can be used for TCP flow tracing. A packet can be reassembled if it contains TCP layer with a whole fixed header (see :func:`_has_fixed_header`), keyed by the IP layer that carries it -- in an - IP-in-IP tunnel, the innermost (see :func:`_carrier`). + IP-in-IP tunnel, the innermost (see :func:`_carrier`). In a frame with + an AH or IPv6 extension header in IPv4, it is the TCP the default + engine reads in the frame's octets, by the frame's link type in the + capture file, instead (see :func:`_segment`). * 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`. @@ -542,53 +969,16 @@ def tcp_traceflow(packet: 'Packet', *, count: 'int' = -1) -> 'TF_TCP_Packet | No :class:`pcapkit.foundation.traceflow.tcp.TCP` """ - ip, tcp = _carrier(packet) - if ip is not None and tcp is not None and _has_fixed_header(tcp): - - # NOTE: the link type is keyed on ``frame.encap_type`` -- Wireshark's - # internal ``WTAP_ENCAP_*`` number -- through - # ``ENCAP_TYPE_TO_LINKTYPE`` (module level, above), and only on the - # PDML root layer's *filter* name when the frame layer has no such - # field. That name cannot do the job on its own: one filter name - # serves several DLTs, so keying on it would answer ``RAW`` (101) for a - # ``rawip6`` capture and ``NULL`` (0) for a ``DLT_LOOP`` one -- valid - # DLTs, wrong ones, and silent (#843). ``frame.encap_type`` - # distinguishes them; see that table's comment for the measurements. - # - # PyShark reports every field as ``LayerFieldsContainer``, a ``str`` - # subclass, so the number arrives as text and needs ``int()``; and a - # field the frame layer does not carry raises ``AttributeError`` out - # of ``pyshark.packet.layers.base.BaseLayer.__getattr__``, which is - # what the ``getattr`` default absorbs here. - # - # Neither path substitutes a DLT on a miss. A lookup raises on an - # unresolvable key rather than minting one (#775), and - # NULL and RAW are genuine DLTs -- BSD loopback and raw IP framing, - # respectively -- each meant to go with its own handler protocol - # class, so neither is an honest stand-in for "unknown link type". - # The bare ``KeyError``/``ValueError`` is re-raised as - # :exc:`~pcapkit.utilities.exceptions.MissingKeyError` -- this - # package's own house exception for a lookup miss -- rather than - # letting it escape this public function. - # - # The frame layer is found by name, not taken as ``frame_info``, which - # is ``pkt_comment`` on a commented packet; the root layer is then the - # first layer that is not the frame layer. + segment = _segment(packet, captured) + if segment is not None: + + # NOTE: the frame layer is found by name, not taken as ``frame_info``, + # which is ``pkt_comment`` on a commented packet; ``_link_type`` says + # how the link type is keyed on it. frame = _frame_layer(packet) if frame is None: raise ProtocolError(f'PyShark packet {packet.number} has no frame layer') - encap_type = getattr(frame, 'encap_type', None) - if encap_type is None: - name = [layer for layer in packet.layers if layer is not frame][0].layer_name - try: - protocol = FILTER_NAME_TO_LINKTYPE[name.lower()] - except KeyError: - raise MissingKeyError(name) from None - else: - try: - protocol = ENCAP_TYPE_TO_LINKTYPE[int(encap_type)] - except (KeyError, ValueError): - raise MissingKeyError('frame.encap_type=%s' % encap_type) from None + protocol = _link_type(packet, frame) # A Simple Packet Block carries no timestamp, so tshark reports no # ``frame.time_epoch`` for it; the PDML ``geninfo`` timestamp PyShark @@ -602,19 +992,21 @@ def tcp_traceflow(packet: 'Packet', *, count: 'int' = -1) -> 'TF_TCP_Packet | No protocol=protocol, # data link type index=count, # frame number frame=packet2dict(packet), # extracted packet - syn=_parse_flag(tcp.flags_syn, 'tcp.flags_syn'), # TCP synchronise (SYN) flag - fin=_parse_flag(tcp.flags_fin, 'tcp.flags_fin'), # TCP finish (FIN) flag - rst=_parse_flag(tcp.flags_reset, 'tcp.flags_reset'), # TCP reset (RST) flag - src=ipaddress.ip_address(ip.src), # source IP - dst=ipaddress.ip_address(ip.dst), # destination IP - srcport=int(tcp.srcport), # TCP source port - dstport=int(tcp.dstport), # TCP destination port + syn=segment.syn, # TCP synchronise (SYN) flag + fin=segment.fin, # TCP finish (FIN) flag + rst=segment.rst, # TCP reset (RST) flag + src=segment.src, # source IP + dst=segment.dst, # destination IP + srcport=segment.srcport, # TCP source port + dstport=segment.dstport, # TCP destination port timestamp=_parse_epoch(epoch), # timestamp - seq=_absolute(tcp, 'seq'), # TCP sequence number - ack=_absolute(tcp, 'ack'), # TCP acknowledgement number + seq=segment.seq, # TCP sequence number + ack=segment.ack, # TCP acknowledgement number # NOTE: PyShark reports dissected *fields*, not the octets behind # them, so there is no header or payload to hand over -- which is # the same reason this module carries no ``tcp_reassembly`` at all. + # The octets ``_segment`` reads behind an extension header are of + # those few frames alone, so are not handed over either. # ``Extractor`` refuses ``trace_analyse=True`` on this engine, so # nothing reads these two. header=b'', # unavailable 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 1664a88dec..abd1c57958 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 @@ -163,8 +163,8 @@ class TestExtensionHeaderChainAgreement(harness.RoundTripBase): 'default recursion limit, and so agrees at 1,000', 'INPUTS', 'extra, default has none', ('ipv6-33/scapy/*',)), 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')), + 'EXTENSION_HEADER_LIMIT, carried in IPv6; at 1,000 it gives up and agrees', + 'INPUTS', 'extra, default has none', ('ipv6-33/pyshark/trace',)), ) @classmethod diff --git a/tests/toolkit/test_pyshark_ipv4_extension_headers_1625_unit.py b/tests/toolkit/test_pyshark_ipv4_extension_headers_1625_unit.py new file mode 100644 index 0000000000..d98f9eedad --- /dev/null +++ b/tests/toolkit/test_pyshark_ipv4_extension_headers_1625_unit.py @@ -0,0 +1,1079 @@ +# -*- coding: utf-8 -*- +"""The pyshark engine traces the TCP the default engine reads behind an extension header in IPv4. C.f. #1625. + +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; behind such a header it reads TCP only if its TCP parser accepts the +header. tshark dissects TCP behind such a header by rules of its own, so the +pyshark engine traced TCP the default engine rejects, and none at all behind a +Mobility or HIP header, past which tshark dissects nothing and whose payload the +PDML PyShark reads leaves out. Nor does the default engine read past Shim6 in +IPv4, whose parser refuses it, where tshark does. For such a frame the engine +now reads the captured octets from a second tshark pass and applies the +default engine's rules to them +(:func:`pcapkit.toolkit.pypcapfile._past_extension_headers`), from the +outermost IP layer the default engine dissects on, so that a tunnel it does not +dissect, such as GRE or VXLAN, or the chain bound of an outer IPv6 header, +stops the read as it stops the default engine. That layer is found by the link +type in the capture file, which the engine reads for itself, as tshark reads +link types 12 and 14 as it reads raw IP (101). A frame of a link type or +EtherType the default engine dissects no IP over is read as tshark dissects +it, as before. + +That pass is a process of the engine's own. If it fails, or the capture file +cannot be read for a link type, the engine warns and traces such frames as +tshark dissects them; whether it fails or not, closing the engine reaps it, +and the first pass's tshark with it. + +Every capture is built here, so the module is in the unit tier. The tests that +run tshark skip unless the pyshark engine runs, which needs ``pyshark``, the +``tshark`` binary and Python 3.13 or older; the rest use fakes, and a Python +child process in tshark's place. + +""" +from __future__ import annotations + +import importlib.util +import json +import os +import struct +import subprocess +import sys +import tempfile +import time +import types +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) + + +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 mobility(nxt: int, length: int = 0) -> bytes: + """A Mobility header, a Binding Refresh Request, whose Header Len is ``length``.""" + return bytes([nxt, length, 0, 0, 0, 0, 0, 0]) + + +def hip(nxt: int) -> bytes: + """A HIP header of 40 octets, an I1 with no parameters.""" + return bytes([nxt, 4, 1, 0x21, 0, 0, 0, 0]) + b'\x11' * 32 + + +def shim6(nxt: int) -> bytes: + """A Shim6 payload extension header of 8 octets.""" + return bytes([nxt, 0, 0x80, 0, 0, 0, 0, 1]) + + +def tcp(number: int, *, opts: bytes = b'', words: 'int | None' = None, **flags: bool) -> 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=1000 + number, ack=1, sport=1000 + number, **flags) + 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, opts: bytes = b'', + **fields: 'Any') -> bytes: + """An IPv4 packet carrying ``payload`` as ``proto``, with options ``opts``, its Total Length ``length`` if given.""" + packet = bytearray(wire.ipv4(payload, proto=proto, **fields)) + if opts: + packet[0] = 0x40 | (5 + len(opts) // 4) + packet[2:4] = (len(packet) + len(opts)).to_bytes(2, 'big') + packet[20:20] = opts + 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[:(packet[0] & 0x0F) * 4])).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) + + +#: An MSS option of length 3, which the default engine's TCP parser rejects. +BAD_MSS = b'\x02\x03\x05\xb4' +#: An IPv4 Timestamp option of length 2, which its IPv4 parser rejects. +BAD_TIMESTAMP = b'\x44\x02\x00\x00' + +#: Frames carrying TCP behind an AH or IPv6 extension header in IPv4, each with +#: whether the default engine traces the TCP, by a name for it. Each builder +#: takes the TCP segment's number, which is its frame number. +CASES = { + 'control': (True, lambda n: eth(v4(tcp(n), 6))), + # #1625's six, the first two as controls + 'hop-by-hop': (True, lambda n: eth(v4(options(6) + tcp(n), 0))), + 'hop-by-hop, rejected tcp options': (False, lambda n: eth(v4(options(6) + tcp(n, opts=BAD_MSS), 0))), + 'hop-by-hop, tcp data offset past the capture': (False, lambda n: eth(v4(options(6) + tcp(n, words=15), 0))), + 'mobility': (True, lambda n: eth(v4(mobility(6) + tcp(n), 135))), + 'hip': (True, lambda n: eth(v4(hip(6) + tcp(n), 139))), + # the other headers + '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))), + '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))), + 'mobility cut short': (False, lambda n: eth(v4(mobility(6, 1) + tcp(n)[:4], 135))), + # Shim6, whose parser refuses IPv4, so that the default engine reads it raw + 'shim6': (False, lambda n: eth(v4(shim6(6) + tcp(n), 140))), + 'shim6 in 4in4': (False, lambda n: eth(v4(v4(shim6(6) + tcp(n), 140), 4))), + 'shim6 in 4in6': (False, lambda n: eth(v6(v4(shim6(6) + tcp(n), 140), 4))), + 'hop-by-hop then shim6': (False, lambda n: eth(v4(options(140) + shim6(6) + tcp(n), 0))), + # but not IPv6, which reads past it + 'shim6 in 6in4 behind hop-by-hop': (True, lambda n: eth(v4(options(41) + v6(shim6(6) + tcp(n), 140), 0))), + # chains + 'ah then hop-by-hop': (True, lambda n: eth(v4(ah(0) + options(6) + tcp(n), 51))), + 'hop-by-hop then ah': (True, lambda n: eth(v4(options(51) + ah(6) + tcp(n), 0))), + 'hop-by-hop then destination options': (True, lambda n: eth(v4(options(60) + options(6) + tcp(n), 0))), + 'mobility then hop-by-hop': (True, lambda n: eth(v4(mobility(0) + options(6) + tcp(n), 135))), + 'hip then destination options': (True, lambda n: eth(v4(hip(60) + options(6) + tcp(n), 139))), + '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))), + '4in4 behind mobility': (True, lambda n: eth(v4(mobility(4) + v4(tcp(n), 6), 135))), + '6in4 behind hip': (True, lambda n: eth(v4(hip(41) + v6(tcp(n), 6), 139))), + '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))), + 'mobility in 4in4': (True, lambda n: eth(v4(v4(mobility(6) + tcp(n), 135), 4))), + 'hop-by-hop in 4in4 behind hop-by-hop': (True, lambda n: eth(v4(options(4) + v4(options(6) + tcp(n), 0), 0))), + # IP 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))), + 'rejected ipv4 options behind hop-by-hop': (False, lambda n: eth(v4(options(4) + v4(tcp(n), 6, opts=BAD_TIMESTAMP), + 0))), + 'rejected ipv4 options, then hop-by-hop': (False, lambda n: eth(v4(options(6) + tcp(n), 0, opts=BAD_TIMESTAMP))), + # 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))), + '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))), + 'mobility, rejected tcp options': (False, lambda n: eth(v4(mobility(6) + tcp(n, opts=BAD_MSS), 135))), + 'hip, tcp data offset past the capture': (False, lambda n: eth(v4(hip(6) + tcp(n, words=15), 139))), + # directly over the inner IP layer, the default engine reads a rejected header raw + 'rejected tcp options in 4in4 behind hop-by-hop': (True, lambda n: eth(v4(options(4) + v4(tcp(n, opts=BAD_MSS), 6), + 0))), + 'rejected tcp options in 6in4 behind mobility': (True, lambda n: eth(v4(mobility(41) + v6(tcp(n, opts=BAD_MSS), 6), + 135))), + 'no tcp': (False, lambda n: eth(v4(options(17) + wire.udp(b'data', sport=1000 + n), 0))), + 'flags': (True, lambda n: eth(v4(mobility(6) + tcp(n, syn=True, fin=True, rst=True), 135))), + # 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(mobility(6) + tcp(n), 135)) + b'\x00' * 10), + 'later ipv4 fragment': (False, lambda n: eth(v4(mobility(6) + tcp(n), 135, offset=8))), + 'first ipv4 fragment': (True, lambda n: eth(v4(mobility(6) + tcp(n), 135, mf=True))), + # the default engine does not dissect the datagram an ICMP error quotes + 'quoted in an icmp error': (False, lambda n: eth(v4(bytes([3, 3, 0, 0, 0, 0, 0, 0]) + + v4(mobility(6) + tcp(n), 135), 1))), +} + + +def gre(packet: bytes) -> bytes: + """A GRE packet, with no options, carrying the IPv4 ``packet``.""" + return b'\x00\x00\x08\x00' + packet + + +def vxlan(frame: bytes) -> bytes: + """A UDP datagram to the VXLAN port, carrying the Ethernet ``frame`` with VNI 1.""" + return wire.udp(b'\x08\x00\x00\x00\x00\x00\x01\x00' + frame, sport=4000, dport=4789) + + +def chain(count: int, nxt: int) -> bytes: + """``count`` Destination Options headers in a row, the last naming ``nxt``.""" + return b''.join(options(60 if index < count - 1 else nxt) for index in range(count)) + + +#: Frames whose AH or IPv6 extension header in IPv4 sits inside a tunnel or +#: behind an IPv6 header, as in :data:`CASES`. The default engine reads a frame +#: from its outermost IP layer on, so it stops at a tunnel it does not dissect, +#: such as GRE or VXLAN, which tshark does, and at a chain of more than +#: :data:`~pcapkit.protocols.internet.internet.EXTENSION_HEADER_LIMIT` extension +#: headers in the IPv6 header carrying the IPv4. +OUTERMOST_CASES = { + 'mobility in gre': (False, lambda n: eth(v4(gre(v4(mobility(6) + tcp(n), 135)), 47))), + 'hip in gre': (False, lambda n: eth(v4(gre(v4(hip(6) + tcp(n), 139)), 47))), + 'hop-by-hop in gre': (False, lambda n: eth(v4(gre(v4(options(6) + tcp(n), 0)), 47))), + 'mobility in 4in4 in gre': (False, lambda n: eth(v4(gre(v4(v4(mobility(6) + tcp(n), 135), 4)), 47))), + 'hop-by-hop in 4in4 in gre': (False, lambda n: eth(v4(gre(v4(v4(options(6) + tcp(n), 0), 4)), 47))), + 'mobility in vxlan': (False, lambda n: eth(v4(vxlan(eth(v4(mobility(6) + tcp(n), 135))), 17))), + 'hip in vxlan': (False, lambda n: eth(v4(vxlan(eth(v4(hip(6) + tcp(n), 139))), 17))), + 'hop-by-hop in vxlan': (False, lambda n: eth(v4(vxlan(eth(v4(options(6) + tcp(n), 0))), 17))), + 'mobility in 4in6 past the chain limit': (False, lambda n: eth(v6(chain(33, 4) + v4(mobility(6) + tcp(n), 135), + 60))), + 'hop-by-hop in 4in6 past the chain limit': (False, lambda n: eth(v6(chain(33, 4) + v4(options(6) + tcp(n), 0), + 60))), + 'mobility in 4in6 at the chain limit': (True, lambda n: eth(v6(chain(32, 4) + v4(mobility(6) + tcp(n), 135), 60))), + 'mobility in 4in6 behind a short chain': (True, lambda n: eth(v6(chain(2, 4) + v4(mobility(6) + tcp(n), 135), 60))), + 'hip in 4in6 in 6in4 behind hop-by-hop': (True, lambda n: eth(v4(options(41) + v6(v4(hip(6) + tcp(n), 139), 4), + 0))), +} + + +def vlan(packet: bytes, *tags: int) -> bytes: + """An Ethernet frame carrying the IPv4 ``packet`` behind a VLAN tag of each TPID in ``tags``.""" + inner = b''.join(struct.pack('>HH', 1, tpid) for tpid in tags[1:]) + struct.pack('>HH', 1, 0x0800) + return wire.ethernet(inner + packet, tags[0]) + + +def loopback(packet: bytes, family: int, byteorder: str) -> bytes: + """A BSD loopback frame carrying ``packet`` as the address ``family``, written in ``byteorder``.""" + return family.to_bytes(4, byteorder) + packet + + +def sll(packet: bytes) -> bytes: + """A Linux cooked-mode (SLL) frame carrying the IPv4 ``packet``.""" + return struct.pack('>HHH8sH', 0, 1, 6, b'\x02\x00\x00\x00\x00\x01', 0x0800) + packet + + +#: Frames of each link type that carry an AH or IPv6 extension header in IPv4, +#: each with the link type, whether the default engine traces the TCP, and a +#: builder as in :data:`CASES`, by a name for it. The default engine dissects IP +#: over the link types and EtherTypes :func:`pcapkit.toolkit.pyshark._network_start` +#: names only; a frame over any other, such as MPLS, Linux SLL, or the BSD raw +#: IP of link types 12 and 14, which tshark reads as raw IP (101), is read as +#: tshark dissects it, which reads no TCP past a Mobility or HIP header. +LINK_CASES = { + 'vlan, mobility': (1, True, lambda n: vlan(v4(mobility(6) + tcp(n), 135), 0x8100)), + 'qinq, hip': (1, True, lambda n: vlan(v4(hip(6) + tcp(n), 139), 0x88A8, 0x8100)), + 'qinq, hop-by-hop, rejected tcp options': (1, False, lambda n: vlan(v4(options(6) + tcp(n, opts=BAD_MSS), 0), + 0x88A8, 0x8100)), + 'mpls, mobility': (1, False, lambda n: wire.ethernet(b'\x00\x01\x01\x40' + v4(mobility(6) + tcp(n), 135), 0x8847)), + 'null, mobility': (0, True, lambda n: loopback(v4(mobility(6) + tcp(n), 135), 2, 'little')), + 'null, mobility in 4in6': (0, True, lambda n: loopback(v6(v4(mobility(6) + tcp(n), 135), 4), 30, 'big')), + 'null, hop-by-hop, rejected tcp options': (0, False, lambda n: loopback(v4(options(6) + tcp(n, opts=BAD_MSS), 0), + 2, 'little')), + 'loop, hip': (108, True, lambda n: loopback(v4(hip(6) + tcp(n), 139), 2, 'big')), + 'loop, hop-by-hop in 4in6': (108, True, lambda n: loopback(v6(v4(options(6) + tcp(n), 0), 4), 24, 'big')), + 'raw, mobility': (101, True, lambda n: v4(mobility(6) + tcp(n), 135)), + 'raw, mobility in 4in6 past the chain limit': (101, False, lambda n: v6(chain(33, 4) + + v4(mobility(6) + tcp(n), 135), 60)), + 'ipv4, hip': (228, True, lambda n: v4(hip(6) + tcp(n), 139)), + 'ipv4, hop-by-hop, tcp data offset past the capture': (228, False, lambda n: v4(options(6) + tcp(n, words=15), 0)), + 'ipv6, mobility in 4in6': (229, True, lambda n: v6(v4(mobility(6) + tcp(n), 135), 4)), + 'sll, mobility': (113, False, lambda n: sll(v4(mobility(6) + tcp(n), 135))), + 'sll, hip': (113, False, lambda n: sll(v4(hip(6) + tcp(n), 139))), + 'bsd raw (12), mobility': (12, False, lambda n: v4(mobility(6) + tcp(n), 135)), + 'bsd raw (12), hip in 4in6': (12, False, lambda n: v6(v4(hip(6) + tcp(n), 139), 4)), + 'openbsd raw (14), hip': (14, False, lambda n: v4(hip(6) + tcp(n), 139)), + 'openbsd raw (14), mobility in 4in6': (14, False, lambda n: v6(v4(mobility(6) + tcp(n), 135), 4)), +} + + +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 traces 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 block(kind: int, body: bytes, byteorder: str = 'little') -> bytes: + """A PCAP-NG block of type ``kind`` around ``body``, zero-padded to 32 bits, in ``byteorder``.""" + order = '<' if byteorder == 'little' else '>' + body += bytes(-len(body) % 4) + return struct.pack(f'{order}II', kind, len(body) + 12) + body + struct.pack(f'{order}I', len(body) + 12) + + +def journal(entry: bytes) -> bytes: + """A systemd Journal Export Block holding ``entry``.""" + return block(9, entry) + + +def spb(octets: bytes) -> bytes: + """A Simple Packet Block of ``octets``, which carries no timestamp.""" + return block(3, struct.pack(' bytes: + """A PCAP-NG section in ``byteorder``, with an interface of each of ``linktypes``, then ``packets``. + + Each packet is ``(kind, interface, octets)``: an Enhanced Packet Block (6), + an obsolete Packet Block (2), or a Simple Packet Block (3), which names no + interface. Any other kind is a block of that type around ``octets``. + + """ + order = '<' if byteorder == 'little' else '>' + out = [block(0x0A0D0D0A, struct.pack(f'{order}IHHq', 0x1A2B3C4D, 1, 0, -1), byteorder)] + out += [block(1, struct.pack(f'{order}HHI', linktype, 0, 0xFFFF), byteorder) for linktype in linktypes] + for kind, interface, octets in packets: + head = {6: struct.pack(f'{order}IIIII', interface, 0, 1, len(octets), len(octets)), + 2: struct.pack(f'{order}HHIIII', interface, 0, 0, 1, len(octets), len(octets)), + 3: struct.pack(f'{order}I', len(octets))}.get(kind, b'') + out.append(block(kind, head + octets, byteorder)) + return b''.join(out) + + +def tshark_children() -> 'dict[int, str] | None': + """The state of each ``tshark`` process this one started, by pid, as ``ps`` reports it; :data:`None` without ``ps``.""" + try: + listing = subprocess.run(['ps', '-e', '-o', 'pid=', '-o', 'ppid=', '-o', 'stat=', '-o', 'comm='], + capture_output=True, text=True, timeout=30, check=True).stdout + except (OSError, subprocess.SubprocessError): + return None + children = {} + for line in listing.splitlines(): + fields = line.split(None, 3) + if len(fields) == 4 and fields[1] == str(os.getpid()) and 'tshark' in fields[3]: + children[int(fields[0])] = fields[2] + return children + + +def read(test: unittest.TestCase, engine: str, capture: bytes, suffix: str = 'pcap') -> 'dict[str, Any]': + """Read ``capture`` with ``engine``, with TCP reassembly and flow tracing on. + + Returns: + The input flow tracing receives for each frame number -- every field but + ``frame``, ``header`` and ``payload``, which PyShark cannot supply -- + the traced flows' labels and frames, and how many inputs TCP + reassembly received, or :data:`None` if the engine declines it. + + Skips ``test`` unless ``engine`` ran. + + """ + from pcapkit import extract + from pcapkit.foundation.reassembly.reassembly import ReassemblyBase + from pcapkit.foundation.traceflow.traceflow import TraceFlowBase + from pcapkit.utilities.exceptions import UnsupportedCall + + if engine != 'default' and importlib.util.find_spec(engine) is None: + test.skipTest(f'{engine} not installed') + reassemble, trace = ReassemblyBase.__call__, TraceFlowBase.__call__ + seen = {'tcp': 0, 'trace': {}} # type: dict[str, Any] + + def spy_tcp(this: 'Any', packet: 'Any') -> 'Any': + seen['tcp'] += 1 + return reassemble(this, packet) + + def spy_trace(this: 'Any', packet: 'Any') -> 'Any': + seen['trace'][packet.index] = {key: value for key, value in packet.items() + if key not in ('frame', 'header', 'payload')} + return trace(this, packet) + + with tempfile.TemporaryDirectory() as tmp, warnings.catch_warnings(), \ + mock.patch.object(ReassemblyBase, '__call__', spy_tcp), \ + mock.patch.object(TraceFlowBase, '__call__', spy_trace): + warnings.simplefilter('ignore') + path = os.path.join(tmp, f'capture.{suffix}') + with open(path, 'wb') as file: + file.write(capture) + 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['flows'] = [(flow.label, tuple(flow.index)) for flow in extractor.trace.tcp] + try: + declined = extractor.reassembly.tcp is None + except UnsupportedCall: + declined = True + if declined: + seen['tcp'] = None + finally: + close_extractor(extractor) + return seen + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class PySharkExtensionHeaderTraceTests(unittest.TestCase): + """The pyshark engine traces the TCP the default engine reads behind an extension header in IPv4, and no other.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def agree(self, capture: bytes, reads: 'list[int]', names: 'dict[int, str]', + suffix: str = 'pcap') -> None: + base = read(self, 'default', capture, suffix) + # what the default engine reads, as the cases say + self.assertEqual(sorted(base['trace']), reads) + seen = read(self, 'pyshark', capture, suffix) + self.assertEqual([names[number] for number in sorted(set(seen['trace']) ^ set(base['trace']))], [], + 'frames traced by one of pyshark and the default engine alone') + for number, record in base['trace'].items(): + self.assertEqual(seen['trace'][number], record, names[number]) + self.assertEqual(seen['flows'], base['flows']) + # pyshark still declines TCP reassembly + self.assertIsNone(seen['tcp']) + self.assertEqual(base['tcp'], len(reads)) + + def test_tcp_behind_an_extension_header_in_ipv4_is_traced_as_the_default_engine_reads_it(self) -> None: + frames, reads = build(CASES) + self.agree(wire.pcap([(1, number, octets) for number, octets in enumerate(frames)]), + reads, dict(enumerate(CASES, start=1))) + + def test_the_frame_is_read_from_its_outermost_ip_layer(self) -> None: + # GRE, VXLAN and an outer IPv6 header's chain bound stop it, as they + # stop the default engine, wherever the extension header in IPv4 is + frames, reads = build(OUTERMOST_CASES) + self.agree(wire.pcap([(1, number, octets) for number, octets in enumerate(frames)]), + reads, dict(enumerate(OUTERMOST_CASES, start=1))) + + def test_each_link_type_is_read_as_the_default_engine_reads_it(self) -> None: + for linktype in sorted({link for link, _, _ in LINK_CASES.values()}): + cases = {name: (reads, make) for name, (link, reads, make) in LINK_CASES.items() if link == linktype} + with self.subTest(linktype=linktype): + frames, reads = build(cases) + self.agree(wire.pcap([(1, number, octets) for number, octets in enumerate(frames)], + linktype=linktype), reads, dict(enumerate(cases, start=1))) + + def test_the_octets_are_those_of_the_frame_past_a_record_that_is_no_packet(self) -> None: + # tshark numbers a journal block as a frame, the default engine counts + # packets only, and the second tshark pass is keyed by tshark's number + cases = {name: CASES[name] for name in ('control', 'mobility', 'hip, tcp data offset past the capture', + 'hop-by-hop', 'hop-by-hop, rejected tcp options', 'flags')} + frames, reads = build(cases) + head = wire.pcapng([]) + packets = [wire.pcapng([(1, number, octets)])[len(head):] for number, octets in enumerate(frames)] + entry = journal(b'__REALTIME_TIMESTAMP=1000000\nMESSAGE=hello\n') + capture = head + entry + packets[0] + entry + b''.join(packets[1:3]) + entry + b''.join(packets[3:]) + self.agree(capture, reads, dict(enumerate(cases, start=1)), suffix='pcapng') + + def test_a_simple_packet_block_is_traced(self) -> None: + # a Simple Packet Block carries no timestamp, which PyShark's JSON + # parser could not do without + cases = {name: CASES[name] for name in ('control', 'hop-by-hop', 'mobility', + 'hop-by-hop, rejected tcp options')} + frames, reads = build(cases) + head = wire.pcapng([]) + capture = head + wire.pcapng([(1, 0, frames[0])])[len(head):] + b''.join(spb(octets) for octets in frames[1:]) + self.agree(capture, reads, dict(enumerate(cases, start=1)), suffix='pcapng') + + def test_each_interface_is_read_by_its_own_link_type(self) -> None: + # a PCAP-NG frame's link type is its interface's, among those of its + # own section; tshark reads 12 and 14 as it reads raw IP (101) + def raw(n: int) -> bytes: + return v4(mobility(6) + tcp(n), 135) + + def raw6(n: int) -> bytes: + return v6(v4(hip(6) + tcp(n), 139), 4) + + # by section: its interfaces' link types, then its blocks, each a block + # type, an interface, the octets, and whether the default engine + # traces them, or None for a block that is no packet + sections = [ + ([1, 12, 101], 'little', [(6, 0, eth(raw(1)), True), + (6, 1, raw(2), False), + (9, 0, b'__REALTIME_TIMESTAMP=1000000\nMESSAGE=hello\n', None), + (6, 2, raw(3), True), + (3, 0, eth(raw6(4)), True)]), + ([14, 101], 'big', [(3, 0, raw6(5), False), + (6, 1, raw6(6), True), + (2, 0, raw(7), False), + (2, 1, v4(options(6) + tcp(8, opts=BAD_MSS), 0), False)]), + ] + capture = b''.join(section(linktypes, [packet[:3] for packet in packets], byteorder) + for linktypes, byteorder, packets in sections) + packets = [(f'section {index}, block {kind}, interface {interface}', reads) + for index, (_, _, blocks) in enumerate(sections, start=1) + for kind, interface, _, reads in blocks if reads is not None] + self.agree(capture, [number for number, (_, reads) in enumerate(packets, start=1) if reads], + {number: name for number, (name, _) in enumerate(packets, start=1)}, suffix='pcapng') + + def test_a_second_pass_tshark_rejects_falls_back_with_a_warning(self) -> None: + # the frames are traced as tshark dissects them, as before #1625 + from pcapkit.utilities.warnings import PySharkWarning + + cases = {name: CASES[name] for name in ('control', 'hop-by-hop', 'mobility', + 'hop-by-hop, rejected tcp options')} + frames, _ = build(cases) + capture = wire.pcap([(1, number, octets) for number, octets in enumerate(frames)]) + with mock.patch('pcapkit.toolkit.pyshark.OCTETS_FILTER', 'ip.proto =='), \ + mock.patch('pcapkit.foundation.engines.pyshark.warn') as warn: + seen = read(self, 'pyshark', capture) + self.assertEqual(sorted(seen['trace']), [1, 2, 4]) + messages = [call.args[0] for call in warn.call_args_list if call.args[1] is PySharkWarning] + self.assertEqual(len(messages), 1, messages) + self.assertIn('Frame 2: the second tshark pass', messages[0]) + self.assertIn('exited with status', messages[0]) + + def test_an_early_close_leaves_no_tshark_behind(self) -> None: + from pcapkit import extract + + if tshark_children() is None: + self.skipTest('ps is not available') + frames = [(CASES['mobility'] if number % 5 == 0 else CASES['control'])[1](number) for number in range(1, 400)] + for how in ('__exit__', '_cleanup'): + with self.subTest(close=how), tempfile.TemporaryDirectory() as tmp, warnings.catch_warnings(): + warnings.simplefilter('ignore') + path = os.path.join(tmp, 'capture.pcap') + with open(path, 'wb') as file: + file.write(wire.pcap([(1, number, octets) for number, octets in enumerate(frames)])) + before = tshark_children() or {} + extractor = extract(fin=path, nofile=True, engine='pyshark', auto=False, tcp=True, trace=True, + trace_fout=os.path.join(tmp, 'trace'), trace_format='json') + try: + if extractor._exnam != 'pyshark': # pylint: disable=protected-access + self.skipTest('pyshark did not run') + for _ in range(30): + next(extractor) + self.assertIsNotNone(extractor._exeng._exraw, 'the second pass started') + self.assertEqual(len(set(tshark_children() or {}) - set(before)), 2) + finally: + if how == '__exit__': + extractor.__exit__(None, None, None) + else: + extractor._cleanup() # pylint: disable=protected-access + deadline = time.monotonic() + 3 + while True: + left = {pid: state for pid, state in (tshark_children() or {}).items() if pid not in before} + if not left or time.monotonic() > deadline: + break + time.sleep(0.1) + self.assertEqual(left, {}, 'tshark processes left behind, by pid, with their state') + + +class FakeField(str): + """A PyShark field: text, with the position and raw value tshark reports for it.""" + + def __new__(cls, show: str, pos: int = 0, raw_value: str = '') -> 'FakeField': + field = super().__new__(cls, show) + field.pos, field.raw_value = str(pos), raw_value + return field + + +class FakeLayer: + def __init__(self, layer_name: str, **fields: object) -> None: + self.layer_name = layer_name + self.field_names = tuple(fields) + for name, value in fields.items(): + setattr(self, name, value) + + +#: An Ethernet frame of IPv4 protocol 135, a Mobility header and TCP, as on the wire. +MOBILITY = eth(v4(mobility(6) + tcp(7, syn=True), 135)) + + +#: ``MOBILITY``'s IPv4 packet, as a raw IP frame. +RAW_MOBILITY = MOBILITY[14:] + + +def fake_packet(octets: bytes = MOBILITY, *, number: str = '3', start: int = 14) -> 'Any': + """A PyShark packet of a frame of the IPv4 ``octets`` from ``start`` on, as tshark dissects ``MOBILITY``. + + The frame is Ethernet, or raw IP if ``start`` is 0, as tshark names it. + + """ + ip = FakeLayer('ip', version=FakeField('4', start, '4'), proto=FakeField(str(octets[start + 9])), + src=FakeField('192.0.2.1', start + 12, octets[start + 12:start + 16].hex()), + dst=FakeField('198.51.100.2', start + 16, octets[start + 16:start + 20].hex())) + frame = FakeLayer('frame', encap_type='1' if start else '7', time_epoch='1.5') + return types.SimpleNamespace(number=number, frame_info=frame, sniff_timestamp='1.5', + layers=[frame, FakeLayer('eth' if start else 'raw'), ip, FakeLayer('mipv6')]) + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class ToolkitOctetsTests(unittest.TestCase): + """:func:`tcp_traceflow` reads such a frame from its octets, with no tshark.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_the_segment_is_read_from_the_octets(self) -> None: + import ipaddress + + from pcapkit.toolkit import pyshark as toolkit + + asked = [] # type: list[Any] + packet = fake_packet() + data = toolkit.tcp_traceflow(packet, count=2, + captured=lambda p: asked.append(p) or toolkit.Captured(1, MOBILITY)) + self.assertEqual(asked, [packet]) + self.assertEqual((data.index, data.src, data.dst, data.srcport, data.dstport, data.seq, data.ack), + (2, ipaddress.ip_address('192.0.2.1'), ipaddress.ip_address('198.51.100.2'), + 1007, 9, 1007, 1)) + self.assertEqual((data.syn, data.fin, data.rst), (True, False, False)) + + def test_the_octets_are_read_as_the_default_engine_reads_them(self) -> None: + # every case, read from its outermost IP header on + from pcapkit.toolkit import pyshark as toolkit + + fields = ('src', 'dst', 'srcport', 'dstport', 'seq', 'ack', 'syn', 'fin', 'rst') + for cases in (CASES, OUTERMOST_CASES): + frames, _ = build(cases) + base = read(self, 'default', wire.pcap([(1, number, octets) for number, octets in enumerate(frames)])) + for number, (name, octets) in enumerate(zip(cases, frames), start=1): + version = {b'\x08\x00': 4, b'\x86\xdd': 6}[octets[12:14]] + with self.subTest(case=name), warnings.catch_warnings(): + warnings.simplefilter('ignore') # what the parsers warn of rejecting a header + segment = toolkit._from_octets(version, octets[14:]) # pylint: disable=protected-access + record = base['trace'].get(number) + if record is None: + self.assertIsNone(segment) + else: + self.assertIsNotNone(segment) + self.assertEqual({field: getattr(segment, field) for field in fields}, + {field: record[field] for field in fields}) + + def test_the_outermost_ip_layer_is_found_as_the_default_engine_finds_it(self) -> None: + from pcapkit.toolkit import pyshark as toolkit + + packet4, packet6 = v4(tcp(1), 6), v6(tcp(1), 6) + for name, linktype, octets, found in ( + ('ethernet', 1, eth(packet4), (4, 14)), + ('ethernet, ipv6', 1, eth(packet6), (6, 14)), + ('vlan', 1, vlan(packet4, 0x8100), (4, 18)), + ('qinq', 1, vlan(packet4, 0x88A8, 0x8100), (4, 22)), + ('ethernet, arp', 1, wire.ethernet(packet4, 0x0806), None), + ('ethernet, mpls', 1, wire.ethernet(b'\x00\x01\x01\x40' + packet4, 0x8847), None), + ('ethernet cut short', 1, eth(packet4)[:13], None), + ('vlan tag cut short', 1, vlan(packet4, 0x8100)[:17], None), + ('null', 0, loopback(packet4, 2, 'little'), (4, 4)), + ('null, big-endian', 0, loopback(packet6, 28, 'big'), (6, 4)), + ('null, another family', 0, loopback(packet4, 7, 'little'), None), + ('null cut short', 0, b'\x02\x00\x00', None), + ('loop', 108, loopback(packet6, 24, 'big'), (6, 4)), + # a LOOP family is in network byte order, never inferred + ('loop, little-endian', 108, loopback(packet4, 2, 'little'), None), + ('raw', 101, packet4, (4, 0)), + ('raw, ipv6', 101, packet6, (6, 0)), + ('raw, another version', 101, b'\x55' + packet4[1:], None), + ('raw, empty', 101, b'', None), + ('ipv4', 228, packet4, (4, 0)), + ('ipv6', 229, packet6, (6, 0)), + ('linux sll', 113, sll(packet4), None), + ): + with self.subTest(case=name): + self.assertEqual(toolkit._network_start(linktype, octets), found) # pylint: disable=protected-access + + def test_the_octets_are_read_from_the_outermost_ip_layer_tshark_dissects(self) -> None: + # an IPv6 layer above the IPv4 one, as tshark dissects 4in6 + import ipaddress + + from pcapkit.toolkit import pyshark as toolkit + + octets = eth(v6(v4(mobility(6) + tcp(7, syn=True), 135), 4)) + ipv6 = FakeLayer('ipv6', version=FakeField('6', 14, '6'), + src=FakeField('2001:db8::1', 22, octets[22:38].hex()), + dst=FakeField('2001:db8::2', 38, octets[38:54].hex())) + ip = FakeLayer('ip', version=FakeField('4', 54, '4'), proto=FakeField('135'), + src=FakeField('192.0.2.1', 66, octets[66:70].hex()), + dst=FakeField('198.51.100.2', 70, octets[70:74].hex())) + packet = fake_packet() + packet.layers[2:3] = [ipv6, ip] + data = toolkit.tcp_traceflow(packet, captured=lambda p: toolkit.Captured(1, octets)) + self.assertEqual((data.src, data.dst, data.srcport, data.syn), + (ipaddress.ip_address('192.0.2.1'), ipaddress.ip_address('198.51.100.2'), 1007, True)) + + # nothing is read unless the outermost IP layer tshark dissects is + # the one the default engine does, in the same octets + moved = bytearray(octets) + moved[38] ^= 0xFF + for name, layers, frame in ( + ('another destination address', [ipv6, ip], bytes(moved)), + ('the ipv4 layer first', [ip, ipv6], octets), + ('the ipv4 layer alone', [ip], octets), + ('behind a vlan tag', [ipv6, ip], vlan(octets[14:], 0x8100)), + ): + with self.subTest(case=name): + packet.layers[2:-1] = layers + self.assertIsNone(toolkit.tcp_traceflow(packet, captured=lambda p, f=frame: toolkit.Captured(1, f))) + + def test_the_link_type_is_the_capture_files(self) -> None: + # not the one tshark's frame.encap_type names: it reads 12 and 14 as + # raw IP (encap 7, 101), and the default engine leaves them raw + import ipaddress + + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit import pyshark as toolkit + + packet = fake_packet(RAW_MOBILITY, start=0) + data = toolkit.tcp_traceflow(packet, captured=lambda p: toolkit.Captured(101, RAW_MOBILITY)) + self.assertEqual((data.src, data.srcport, data.syn), (ipaddress.ip_address('192.0.2.1'), 1007, True)) + # the record's link type is still tshark's, the same for each link type read + self.assertEqual(data.protocol, LinkType.RAW) + for linktype in (12, 14): + with self.subTest(linktype=linktype): + self.assertIsNone(toolkit.tcp_traceflow( + packet, captured=lambda p, l=linktype: toolkit.Captured(l, RAW_MOBILITY))) + # nor is an Ethernet frame read as raw IP, or the other way round + self.assertIsNone(toolkit.tcp_traceflow(fake_packet(), captured=lambda p: toolkit.Captured(101, MOBILITY))) + self.assertIsNone(toolkit.tcp_traceflow(packet, captured=lambda p: toolkit.Captured(1, RAW_MOBILITY))) + + def test_a_link_type_the_default_engine_reads_no_ip_in_is_left_to_tshark(self) -> None: + # tshark dissects nothing past the Mobility header, so there is no TCP; + # an unknown link type of tshark's raises only once a segment is found + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit import pyshark as toolkit + from pcapkit.utilities.exceptions import MissingKeyError + + for linktype in (113, 276, 12, 14, 999): # Linux SLL and SLL2, BSD raw IP, then none at all + with self.subTest(linktype=linktype): + self.assertIsNone(toolkit.tcp_traceflow( + fake_packet(), captured=lambda p, l=linktype: toolkit.Captured(l, MOBILITY))) + # with no frame.encap_type, the root layer names the record's link type + packet = fake_packet() + del packet.frame_info.encap_type + packet.frame_info.field_names = ('time_epoch',) + captured = lambda p: toolkit.Captured(1, MOBILITY) # noqa: E731 + self.assertEqual(toolkit.tcp_traceflow(packet, captured=captured).protocol, LinkType.ETHERNET) + packet.layers[1] = FakeLayer('sll') + with self.assertRaises(MissingKeyError): + toolkit.tcp_traceflow(packet, captured=captured) + packet = fake_packet() + packet.frame_info.encap_type = '999' + packet.layers.append(FakeLayer('tcp', urgent_pointer='0', srcport='1', dstport='2', seq_raw='3', ack_raw='4', + flags_syn='True', flags_fin='False', flags_reset='False')) + with self.assertRaises(MissingKeyError): + toolkit.tcp_traceflow(packet, captured=lambda p: toolkit.Captured(113, MOBILITY)) + + def test_no_octets_leaves_tsharks_dissection(self) -> None: + # tshark dissects nothing past the Mobility header, so there is no TCP + from pcapkit.toolkit import pyshark as toolkit + + self.assertIsNone(toolkit.tcp_traceflow(fake_packet())) + self.assertIsNone(toolkit.tcp_traceflow(fake_packet(), captured=lambda p: None)) + + def test_octets_other_than_the_layers_are_not_read(self) -> None: + from pcapkit.toolkit import pyshark as toolkit + + packet = fake_packet() + moved = bytearray(MOBILITY) + moved[26] ^= 0xFF # not the source address tshark reports + for octets in (bytes(moved), MOBILITY[:30], b'\x00' * len(MOBILITY)): + with self.subTest(octets=octets[:34].hex()): + self.assertIsNone(toolkit.tcp_traceflow(packet, captured=lambda p, o=octets: toolkit.Captured(1, o))) + packet.layers[2].version = FakeField('4', -1, '4') + self.assertIsNone(toolkit.tcp_traceflow(packet, captured=lambda p: toolkit.Captured(1, MOBILITY))) + del packet.layers[2].version + self.assertIsNone(toolkit.tcp_traceflow(packet, captured=lambda p: toolkit.Captured(1, MOBILITY))) + + def test_only_an_ipv4_extension_header_asks_for_the_octets(self) -> None: + from pcapkit.toolkit import pyshark as toolkit + + def refuse(packet: 'Any') -> 'Any': + raise AssertionError('octets asked for') + + for proto in ('6', '17', '41', '50', 'six'): + with self.subTest(proto=proto): + packet = fake_packet() + packet.layers[2].proto = FakeField(proto) + self.assertIsNone(toolkit.tcp_traceflow(packet, captured=refuse)) + packet = fake_packet() + del packet.layers[2].proto + self.assertIsNone(toolkit.tcp_traceflow(packet, captured=refuse)) + # an IP layer below the first TCP layer is not looked at + packet = fake_packet() + packet.layers.insert(2, FakeLayer('tcp')) + self.assertIsNone(toolkit.tcp_traceflow(packet, captured=refuse)) + # each of the headers, and Shim6, asks for them + for proto in ('0', '43', '44', '51', '60', '135', '139', '140'): + with self.subTest(proto=proto): + asked = [] # type: list[Any] + packet = fake_packet() + packet.layers[2].proto = FakeField(proto) + toolkit.tcp_traceflow(packet, captured=lambda p: asked.append(p)) + self.assertEqual(asked, [packet]) + + def test_the_filter_names_every_extension_header(self) -> None: + from pcapkit.toolkit import pyshark as toolkit + from pcapkit.toolkit.pypcapfile import IPV4_EXTENSION_HEADERS + + self.assertEqual(toolkit.OCTETS_FILTER, 'ip.proto == 0 || ip.proto == 43 || ip.proto == 44 || ' + 'ip.proto == 51 || ip.proto == 60 || ip.proto == 135 || ' + 'ip.proto == 139 || ip.proto == 140') + self.assertEqual(len(IPV4_EXTENSION_HEADERS), 7) + self.assertEqual(toolkit.OCTETS_PROTOCOLS - IPV4_EXTENSION_HEADERS, {140}) + + +def ek(*numbers: int) -> str: + """tshark's ``-T ek -x -J frame`` output for frames ``numbers``, each frame's octets its number.""" + lines = [] + for number in numbers: + lines.append(json.dumps({'index': {'_index': 'packets-1970-01-01'}})) + lines.append(json.dumps({'timestamp': '0', 'layers': { + 'frame_raw': bytes([number]).hex(), 'frame': {'frame_frame_number': str(number)}}})) + return ''.join(f'{line}\n' for line in lines) + + +def child(output: str = '', *, error: str = '', status: int = 0, linger: bool = False) -> 'list[str]': + """A command standing in for tshark: it writes ``output`` and ``error``, then exits with ``status``, or lingers.""" + script = (f'import sys, time\nsys.stdout.write({output!r})\nsys.stdout.flush()\n' + f'sys.stderr.write({error!r})\nsys.stderr.flush()\n' + + ('time.sleep(60)\n' if linger else '') + f'sys.exit({status})\n') + return [sys.executable, '-c', script] + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class EngineCapturedTests(unittest.TestCase): + """:meth:`PyShark._captured` reads the file and second pass forward, and falls back if either fails; no tshark.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def capture(self, data: bytes, suffix: str = 'pcap') -> str: + """Write ``data`` to a capture file removed after the test, and return its path.""" + tmp = tempfile.TemporaryDirectory() # pylint: disable=consider-using-with + self.addCleanup(tmp.cleanup) + path = os.path.join(tmp.name, f'capture.{suffix}') + with open(path, 'wb') as file: + file.write(data) + return path + + def engine(self, command: 'list[str]', capture: 'str | None' = None) -> 'Any': + from pcapkit.foundation.engines.pyshark import PyShark + + engine = PyShark.__new__(PyShark) # skip __init__, which imports pyshark + engine._extractor = types.SimpleNamespace(_ifnm=capture or self.capture(wire.pcap([])), _frnum=7) + engine._extmp = types.SimpleNamespace(close=mock.Mock()) + engine._exraw = engine._exraw_err = engine._exraw_held = None + engine._exlink = engine._exlink_held = None + engine._exraw_failed = False + engine._octets_command = mock.Mock(return_value=command) + self.addCleanup(engine._stop_captured) + return engine + + @staticmethod + def ask(engine: 'Any', number: int, count: 'int | None' = None) -> 'Any': + if count is not None: + engine._extractor._frnum = count + return engine._captured(types.SimpleNamespace(number=str(number))) + + def test_the_second_pass_starts_on_the_first_request_and_reads_forward(self) -> None: + from pcapkit.utilities.warnings import PySharkWarning + + engine = self.engine(child(ek(2, 5, 9))) + engine._octets_command.assert_not_called() + with mock.patch('pcapkit.foundation.engines.pyshark.warn') as warn: + self.assertEqual(self.ask(engine, 2), (1, b'\x02')) + process = engine._exraw + self.assertEqual(self.ask(engine, 2), (1, b'\x02')) # the frame read last is kept + self.assertEqual(self.ask(engine, 5), (1, b'\x05')) + self.assertEqual(self.ask(engine, 9), (1, b'\x09')) + warn.assert_not_called() + self.assertIsNone(self.ask(engine, 10)) # past its last frame + self.assertIsNone(self.ask(engine, 11)) + engine._octets_command.assert_called_once_with() + self.assertEqual(warn.call_count, 1) + message, category = warn.call_args.args + self.assertIs(category, PySharkWarning) + self.assertIn('Frame 7: the second tshark pass', message) + self.assertIn('ended before frame 10', message) + self.assertIsNotNone(process.returncode) # reaped + self.assertIsNone(engine._exraw) + + def test_a_failing_second_pass_falls_back_with_a_warning(self) -> None: + from pcapkit.utilities.warnings import PySharkWarning + + missing = json.dumps({'layers': {'frame': {'frame_frame_number': '4'}}}) + '\n' + for name, command, number, reason in ( + ('exits with an error', child(error='tshark: "ip.proto ==" is not a valid filter\n', status=4), + 4, 'exited with status 4 (tshark: "ip.proto ==" is not a valid filter)'), + ('exits with an error and nothing said', child(status=2), 4, 'exited with status 2'), + ('writes what is not json', child('not json\n'), 4, "gave output it cannot parse: b'not json\\n'"), + ('writes no octets', child(missing), 4, 'gave output it cannot parse'), + ('lacks the frame', child(ek(2, 5)), 4, 'has no frame 4'), + ('cannot be started', ['/nonexistent/tshark'], 4, 'failed: FileNotFoundError'), + ): + with self.subTest(case=name), mock.patch('pcapkit.foundation.engines.pyshark.warn') as warn: + engine = self.engine(command) + self.assertIsNone(self.ask(engine, number)) + self.assertEqual(warn.call_count, 1) + message, category = warn.call_args.args + self.assertIs(category, PySharkWarning) + self.assertIn(reason, message) + self.assertIn('later such frames are traced as tshark dissects them', message) + self.assertIsNone(engine._exraw) + self.assertIsNone(engine._exraw_err) + # nothing more is asked of it, and nothing more is said + self.assertIsNone(self.ask(engine, number + 1)) + self.assertEqual(warn.call_count, 1) + self.assertEqual(engine._octets_command.call_count, 1) + + with mock.patch('pcapkit.foundation.engines.pyshark.warn') as warn: + engine = self.engine([]) + engine._octets_command.side_effect = LookupError('no tshark') + self.assertIsNone(self.ask(engine, 1)) + self.assertIn('failed: LookupError: no tshark', warn.call_args.args[0]) + + def test_close_reaps_the_second_pass(self) -> None: + engine = self.engine(child(ek(1), linger=True)) + self.assertEqual(self.ask(engine, 1), (1, b'\x01')) + process = engine._exraw + self.assertIsNone(process.poll()) + engine.close() + engine._extmp.close.assert_called_once_with() + self.assertIsNotNone(process.returncode) + self.assertTrue(process.stdout.closed) + self.assertIsNone(engine._exraw) + engine.close() # and again + + # the second pass is reaped even if the first will not close + engine = self.engine(child(ek(1), linger=True)) + self.ask(engine, 1) + process = engine._exraw + engine._extmp.close.side_effect = OSError('first pass') + with self.assertRaises(OSError): + engine.close() + self.assertIsNotNone(process.returncode) + + def test_the_link_type_is_read_from_the_capture_file_by_count(self) -> None: + # by the engine's count, as the default engine numbers frames, where + # the second pass is read by tshark's frame number + capture = self.capture(section([1, 12], [(6, 1, b'a'), (9, 0, b'x'), (3, 0, b'b'), (6, 1, b'c')]) + + section([101], [(2, 0, b'd')], 'big'), 'pcapng') + engine = self.engine(child(ek(2, 4, 5, 6)), capture) + with mock.patch('pcapkit.foundation.engines.pyshark.warn') as warn: + self.assertEqual(self.ask(engine, 2, count=1), (12, b'\x02')) + self.assertEqual(self.ask(engine, 2, count=1), (12, b'\x02')) # the packet read last is kept + self.assertEqual(self.ask(engine, 4, count=2), (1, b'\x04')) + self.assertEqual(self.ask(engine, 6, count=4), (101, b'\x06')) + warn.assert_not_called() + + def test_a_capture_file_that_cannot_be_read_falls_back_with_a_warning(self) -> None: + from pcapkit.utilities.warnings import PySharkWarning + + for name, data, count, reason in ( + ('lacks the packet', section([1], [(6, 0, b'a')]), 2, 'has no packet 2'), + ('names no interface', section([1], [(6, 1, b'a')]), 1, + 'failed: FormatError: PCAP-NG: [Block 6] invalid interface ID: 1'), + ('cannot be opened', None, 1, 'failed: FileNotFoundError'), + ): + with self.subTest(case=name), mock.patch('pcapkit.foundation.engines.pyshark.warn') as warn: + capture = os.path.join(tempfile.gettempdir(), 'no-such-dir-1625', 'x.pcapng') + engine = self.engine(child(ek(1, 2)), capture if data is None else self.capture(data, 'pcapng')) + self.assertIsNone(self.ask(engine, count, count=count)) + self.assertEqual(warn.call_count, 1) + message, category = warn.call_args.args + self.assertIs(category, PySharkWarning) + self.assertIn(f'Frame {count}: the capture file, read for the link type of a frame with an AH ' + 'or IPv6 extension header in IPv4,', message) + self.assertIn(reason, message) + self.assertIn('later such frames are traced as tshark dissects them', message) + self.assertIsNone(engine._exlink) + # nothing more is asked of either, and nothing more is said + self.assertIsNone(self.ask(engine, count + 1, count=count + 1)) + self.assertEqual(warn.call_count, 1) + engine._octets_command.assert_not_called() + + # once it has read the link type, the second pass's failure stops both + engine = self.engine(child(status=3), self.capture(section([1], [(6, 0, b'a')]), 'pcapng')) + with mock.patch('pcapkit.foundation.engines.pyshark.warn') as warn: + self.assertIsNone(self.ask(engine, 1, count=1)) + self.assertIn('the second tshark pass', warn.call_args.args[0]) + self.assertIsNone(engine._exlink) + + def test_close_closes_the_capture_file(self) -> None: + engine = self.engine(child(ek(1)), self.capture(section([1], [(6, 0, b'a')]), 'pcapng')) + self.assertEqual(self.ask(engine, 1, count=1), (1, b'\x01')) + file = engine._exlink.gi_frame.f_locals['file'] + self.assertFalse(file.closed) + engine.close() + self.assertTrue(file.closed) + self.assertIsNone(engine._exlink) + + def test_linktypes_are_read_as_the_default_engine_reads_them(self) -> None: + import itertools + + from pcapkit.foundation.engines.pyshark import _linktypes + from pcapkit.utilities.exceptions import FormatError + + def read(data: bytes, limit: int = 20) -> 'list[int]': + return list(itertools.islice(_linktypes(self.capture(data)), limit)) + + # a PCAP file's, of every frame, whatever its byte order or resolution + for options in ({}, {'byteorder': 'big'}, {'nanosecond': True}): + with self.subTest(pcap=options): + self.assertEqual(read(wire.pcap([], linktype=12, **options), 3), [12, 12, 12]) + + # a PCAP-NG packet's interface's, in its own section; other blocks are not packets + first = section([1, 12], [(6, 1, b'a'), (9, 0, b'x'), (3, 0, b'b'), (4, 0, b'n'), (2, 0, b'c'), + (5, 0, b's')]) + second = section([14, 101, 229], [(2, 2, b'd'), (6, 1, b'e'), (3, 0, b'f')], 'big') + self.assertEqual(read(first + second), [12, 1, 1, 229, 101, 14]) + self.assertEqual(read(first + second[:-3]), [12, 1, 1, 229, 101]) # a block cut short ends them + self.assertEqual(read(first + second[:10]), [12, 1, 1]) + self.assertEqual(read(section([1], [])), []) + + for name, data, message in ( + ('an interface the section does not describe', first + section([1], [(6, 1, b'a')]), + 'invalid interface ID: 1'), + ('the interface of a previous section', first + section([], [(3, 0, b'a')]), 'invalid interface ID: 0'), + ('an unknown byte order', section([1], [(6, 0, b'a')])[:8] + b'\x00\x00\x00\x00' + + section([1], [(6, 0, b'a')])[12:], 'block of no known byte order'), + ('a block shorter than 12 octets', section([1], []) + struct.pack(' None: + if importlib.util.find_spec('pyshark') is None: + self.skipTest('pyshark not installed') + from pcapkit.foundation.engines.pyshark import PyShark + from pcapkit.toolkit.pyshark import OCTETS_FILTER + + engine = PyShark.__new__(PyShark) # skip __init__, which imports pyshark + engine._extractor = types.SimpleNamespace(_ifnm='in.pcap') + with mock.patch('pyshark.tshark.tshark.get_process_path', return_value='/bin/tshark'): + self.assertEqual(engine._octets_command(), [ + '/bin/tshark', '-l', '-n', '-r', 'in.pcap', '-Y', OCTETS_FILTER, '-T', 'ek', '-x', '-J', 'frame', + '-o', 'ip.defragment:FALSE', '-o', 'ipv6.defragment:FALSE']) + + def test_ek_records_are_parsed(self) -> None: + from pcapkit.foundation.engines.pyshark import _ek_record + + index, packet = ek(2).splitlines() + self.assertIsNone(_ek_record(index.encode())) + self.assertEqual(_ek_record(packet.encode()), (2, b'\x02')) + listed = json.dumps({'layers': {'frame_raw': ['0a0b', 0, 2], 'frame': {'frame_frame_number': ['3']}}}) + self.assertEqual(_ek_record(listed.encode()), (3, b'\x0a\x0b')) + for line in (b'[]', b'{}', b'{"index": 1, "x": 2}', b'\xff', + b'{"layers": {"frame_raw": "zz", "frame": {"frame_frame_number": "1"}}}', + b'{"layers": {"frame_raw": 1, "frame": {"frame_frame_number": "1"}}}', + b'{"layers": {"frame_raw": "00", "frame": {"frame_frame_number": "one"}}}'): + with self.subTest(line=line): + self.assertIn('gave output it cannot parse', _ek_record(line)) + + def test_read_frame_hands_tcp_traceflow_the_captured_frame(self) -> None: + from pcapkit.toolkit import pyshark as toolkit + + engine = self.engine([]) + engine._extmp = types.SimpleNamespace(next=lambda: fake_packet(number='1')) + engine._extractor = types.SimpleNamespace(_frnum=0, _vfunc=lambda e, f: None, _flag_q=True, + _flag_d=False, _flag_t=True, _tcp=True, + _trace=types.SimpleNamespace(tcp=mock.Mock())) + with mock.patch.object(toolkit, 'tcp_traceflow', return_value=None) as traceflow: + engine.read_frame() + self.assertEqual(traceflow.call_args.kwargs, {'count': 1, 'captured': engine._captured}) + + +if __name__ == '__main__': + unittest.main()