From e9d316ad886141f7f75f3e71ae6afe9924ae054f 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, and leaves the payload of a Mobility or HIP header out of the PDML, so the pyshark engine traced TCP the default engine rejects there and missed the TCP behind protocols 135 and 139. - tcp_traceflow now reads such a frame's TCP from its octets, by the rules the other engines share (pypcapfile's _upper_layer and _default_accepts), up to EXTENSION_HEADER_LIMIT headers and through IP-in-IP tunnels. - The engine reads those octets from a second tshark process of its own (-T ek -x -J frame, filtered to such frames), started on first use, read forward by frame number, and reaped on close. If it fails, 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 | 184 ++++- pcapkit/toolkit/pyshark.py | 269 ++++++- ...est_ext_chain_limit_agreement_1604_unit.py | 4 +- ...yshark_ipv4_extension_headers_1625_unit.py | 668 ++++++++++++++++++ 4 files changed, 1103 insertions(+), 22 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..d5080eb319 100644 --- a/pcapkit/foundation/engines/pyshark.py +++ b/pcapkit/foundation/engines/pyshark.py @@ -10,19 +10,22 @@ .. _PyShark: https://kiminewt.github.io/pyshark """ +import json +import subprocess import sys +import tempfile from typing import TYPE_CHECKING, cast from pcapkit.foundation.engines.engine import EngineBase from pcapkit.foundation.reassembly import ReassemblyManager from pcapkit.utilities.exceptions import 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, Optional, Union from pyshark.capture.file_capture import FileCapture from pyshark.packet.layers.base import BaseLayer @@ -35,6 +38,34 @@ 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 _frame_layer(packet: 'PySharkPacket') -> 'Optional[BaseLayer]': """The ``frame`` layer of a `PyShark`_ packet, or :data:`None` when the record has none. @@ -88,6 +119,15 @@ class PyShark(EngineBase['PySharkPacket']): _expkg: 'pyshark' #: Engine extraction temporary storage. _extmp: 'FileCapture' + #: The second :program:`tshark` pass :meth:`_octets` 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:`_octets` last read from it. + _exraw_held: 'Optional[tuple[int, bytes]]' + #: Whether that pass has failed, so :meth:`_octets` reads no more. + _exraw_failed: 'bool' ########################################################################## # Defaults. @@ -225,6 +265,10 @@ 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._exraw_failed = False super().__init__(extractor) @@ -336,7 +380,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, octets=self._octets) if data_tf_tcp is not None: ext._trace.tcp(data_tf_tcp) @@ -348,11 +392,141 @@ 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:`_octets` 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 _octets(self, packet: 'PySharkPacket') -> 'Optional[bytes]': + """Read the captured octets of a frame, which the PDML :meth:`run` reads does not carry. + + :func:`~pcapkit.toolkit.pyshark.tcp_traceflow` asks for them of a frame + :data:`~pcapkit.toolkit.pyshark.OCTETS_FILTER` matches only (:issue:`1625`). + They 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. A capture with no such frame never starts it. + + Args: + packet: PyShark packet of the first pass. + + Returns: + The frame's captured octets; :data:`None` once the second pass has + failed. + + Warns: + PySharkWarning: When the second pass fails: it cannot be started, + exits with an error, gives output that cannot be parsed, or + lacks the frame asked for. It is 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 + + """ + if self._exraw_failed: + return None + try: + found = self._read_octets(int(packet.number)) + except Exception as exc: # pylint: disable=broad-except + found = f'failed: {type(exc).__name__}: {exc}' + if isinstance(found, bytes): + return found + + self._exraw_failed = True + self._stop_octets() + warn(f'Frame {self._extractor._frnum}: the second tshark pass, which reads the captured octets ' + f'of a frame with an AH or IPv6 extension header in IPv4, {found}; 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_octets(self) -> 'None': + """Stop the second :program:`tshark` pass, if :meth:`_octets` started it, and reap it.""" + process, self._exraw = self._exraw, None + error, self._exraw_err = self._exraw_err, None + try: + if process is not None: + if process.poll() is None: + process.kill() + process.wait() + cast('IO[bytes]', process.stdout).close() + finally: + if error is not None: + error.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 the second pass :meth:`_octets` started, if any. """ - self._extmp.close() + try: + self._extmp.close() + finally: + self._stop_octets() diff --git a/pcapkit/toolkit/pyshark.py b/pcapkit/toolkit/pyshark.py index d02542554b..3cbcdd0667 100644 --- a/pcapkit/toolkit/pyshark.py +++ b/pcapkit/toolkit/pyshark.py @@ -18,14 +18,18 @@ """ import decimal import ipaddress -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, NamedTuple 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.toolkit.pypcapfile import (IP_TUNNELS, IPV4_EXTENSION_HEADERS, IPV4_HEADER_LEN, + TCP_MIN_HEADER_LEN, _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 @@ -322,6 +326,38 @@ #: 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] +#: 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:`~pcapkit.toolkit.pypcapfile.IPV4_EXTENSION_HEADERS` (see +#: :func:`_extension_header_carrier`). 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(IPV4_EXTENSION_HEADERS)) + + +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 +549,200 @@ 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:`~pcapkit.toolkit.pypcapfile.IPV4_EXTENSION_HEADERS`; 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 IPV4_EXTENSION_HEADERS: + return layer + return None + + +def _header_offset(ip: 'Any', octets: 'bytes') -> 'Optional[int]': + """Find where an IPv4 layer of a PyShark packet starts in the octets of its frame. + + Args: + ip: PyShark IPv4 layer. + octets: The frame's captured octets. + + Returns: + The position tshark reports for the layer's ``ip.version`` field, if + the octets there are a version 4 header with the layer's source and + destination addresses; else :data:`None`, as when tshark dissected the + layer out of octets other than the frame's own, e.g. reassembled ones. + + """ + try: + start = int(ip.version.pos) + addresses = bytes.fromhex(str(ip.src.raw_value) + str(ip.dst.raw_value)) + except (AttributeError, TypeError, ValueError): + return None + if (start < 0 or len(octets) < start + IPV4_HEADER_LEN or octets[start] >> 4 != 4 + or octets[start + 12:start + 20] != addresses): + return None + return start + + +def _from_octets(data: 'bytes') -> 'Optional[_Segment]': + """Read the TCP segment the default engine reads in an IPv4 packet, from its octets. + + Args: + data: Raw octets from the IPv4 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. + + 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, 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`). + + """ + version = 4 + 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', octets: 'Optional[Callable[[Packet], Optional[bytes]]]') -> 'Optional[_Segment]': + """Read the TCP segment of a PyShark packet that the default engine reads. + + Args: + packet: PyShark packet. + octets: Fetches the captured octets of a frame, or :data:`None` where + they 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 that layer's octets on (see :func:`_from_octets`); otherwise, + or where its octets cannot be had, 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, and reads TCP behind such a header only if its TCP parser + accepts it (:issue:`1625`). + + """ + if octets is not None and (ip := _extension_header_carrier(packet)) is not None: + frame = octets(packet) + if frame is not None and (start := _header_offset(ip, frame)) is not None: + return _from_octets(frame[start:]) + return _from_layers(packet) + + +def tcp_traceflow(packet: 'Packet', *, count: 'int' = -1, + octets: 'Optional[Callable[[Packet], Optional[bytes]]]' = None) -> 'TF_TCP_Packet | None': """Trace packet flow for TCP. Args: @@ -521,6 +750,12 @@ 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). + octets: Fetches the captured octets of a frame, given its packet, or + :data:`None` where it cannot. It is called only for a packet + :data:`OCTETS_FILTER` matches. The engine passes + :meth:`PyShark._octets + `; without it, + such a packet is read as tshark dissects it. Returns: Data for TCP flow tracing. @@ -528,7 +763,9 @@ 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`). Behind an AH or + IPv6 extension header in IPv4, it is the TCP the default engine reads + in the frame's octets 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,8 +779,8 @@ 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): + segment = _segment(packet, octets) + if segment is not None: # NOTE: the link type is keyed on ``frame.encap_type`` -- Wireshark's # internal ``WTAP_ENCAP_*`` number -- through @@ -602,19 +839,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 613ba92488..3aa37c2187 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',)), Gap(1606, 'dpkt and scapy dissect no IPv6 extension header carried in IPv4 ' '(dpkt keeps the payload of IPv4 protocol 60 as bytes, scapy as Raw), so find no ' 'TCP behind one, where the default engine reads past it at any length up to ' 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..98808b1f67 --- /dev/null +++ b/tests/toolkit/test_pyshark_ipv4_extension_headers_1625_unit.py @@ -0,0 +1,668 @@ +# -*- 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. 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`). + +That pass is a process of the engine's own. If it fails, 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 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))), + # 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 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) -> bytes: + """A little-endian PCAP-NG block of type ``kind`` around ``body``, zero-padded to 32 bits.""" + body += bytes(-len(body) % 4) + return struct.pack(' 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(' '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_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_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)) + + +def fake_packet(octets: bytes = MOBILITY, *, number: str = '3') -> 'Any': + """A PyShark packet of an Ethernet frame of IPv4 ``octets``, as tshark dissects ``MOBILITY``.""" + ip = FakeLayer('ip', version=FakeField('4', 14, '4'), proto=FakeField(str(octets[23])), + src=FakeField('192.0.2.1', 26, octets[26:30].hex()), + dst=FakeField('198.51.100.2', 30, octets[30:34].hex())) + frame = FakeLayer('frame', encap_type='1', time_epoch='1.5') + return types.SimpleNamespace(number=number, frame_info=frame, sniff_timestamp='1.5', + layers=[frame, FakeLayer('eth'), 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, octets=lambda p: asked.append(p) or 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 carried in IPv4, read from its IPv4 header on + from pcapkit.toolkit import pyshark as toolkit + + frames, _ = build(CASES) + base = read(self, 'default', wire.pcap([(1, number, octets) for number, octets in enumerate(frames)])) + fields = ('src', 'dst', 'srcport', 'dstport', 'seq', 'ack', 'syn', 'fin', 'rst') + for number, (name, octets) in enumerate(zip(CASES, frames), start=1): + if octets[12:14] != b'\x08\x00': + continue + with self.subTest(case=name), warnings.catch_warnings(): + warnings.simplefilter('ignore') # what the parsers warn of rejecting a header + segment = toolkit._from_octets(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_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(), octets=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, octets=lambda p, o=octets: o)) + packet.layers[2].version = FakeField('4', -1, '4') + self.assertIsNone(toolkit.tcp_traceflow(packet, octets=lambda p: MOBILITY)) + del packet.layers[2].version + self.assertIsNone(toolkit.tcp_traceflow(packet, octets=lambda p: 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') -> bytes: + raise AssertionError('octets asked for') + + for proto in ('6', '17', 'six'): + with self.subTest(proto=proto): + packet = fake_packet() + packet.layers[2].proto = FakeField(proto) + self.assertIsNone(toolkit.tcp_traceflow(packet, octets=refuse)) + packet = fake_packet() + del packet.layers[2].proto + self.assertIsNone(toolkit.tcp_traceflow(packet, octets=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, octets=refuse)) + + 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') + self.assertEqual(len(IPV4_EXTENSION_HEADERS), 7) + + +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 EngineOctetsTests(unittest.TestCase): + """:meth:`PyShark._octets` reads its second pass forward, falls back when it fails, and reaps it; no tshark.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def engine(self, command: 'list[str]') -> 'Any': + from pcapkit.foundation.engines.pyshark import PyShark + + engine = PyShark.__new__(PyShark) # skip __init__, which imports pyshark + engine._extractor = types.SimpleNamespace(_ifnm='in.pcap', _frnum=7) + engine._extmp = types.SimpleNamespace(close=mock.Mock()) + engine._exraw = engine._exraw_err = engine._exraw_held = None + engine._exraw_failed = False + engine._octets_command = mock.Mock(return_value=command) + return engine + + @staticmethod + def ask(engine: 'Any', number: int) -> 'Any': + return engine._octets(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), b'\x02') + process = engine._exraw + self.assertEqual(self.ask(engine, 2), b'\x02') # the frame read last is kept + self.assertEqual(self.ask(engine, 5), b'\x05') + self.assertEqual(self.ask(engine, 9), 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), 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_command_reads_the_capture_as_the_first_pass_does(self) -> 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_octets(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, 'octets': engine._octets}) + + +if __name__ == '__main__': + unittest.main()