diff --git a/tests/foundation/_roundtrip.py b/tests/foundation/_roundtrip.py new file mode 100644 index 0000000000..8f68a2cf5d --- /dev/null +++ b/tests/foundation/_roundtrip.py @@ -0,0 +1,307 @@ +# -*- coding: utf-8 -*- +"""Shared harness for the foundation round-trip modules. C.f. #1202. + +The owner's rule is that every round trip the library promises passes and +matches exactly. Four foundation paths make such a promise, and each has a +module beside this one that enumerates its cases: + +* reassembly recovers the payload its fragments or segments were cut from + (:file:`reassembly/test_reassembly_payload_roundtrip_unit.py`); +* flow tracing writes each flow to a file that reads back as exactly that + flow's frames (:file:`traceflow/test_traceflow_reread_roundtrip_unit.py`, + and the ``_runtime`` twin over :file:`examples/captures/`); +* every offline engine reports what the ``default`` engine reports, field for + field (:file:`engines/test_engine_agreement_runtime.py`); +* a registrar handed back the entry it shipped with leaves every registry it + touches as it was (:file:`registry/test_registry_symmetry_unit.py`). + +A case that does not close goes in its module's ``KNOWN_FAILURES``, one +:class:`Gap` per root cause, in the style of +:mod:`tests.protocols._edge_roundtrip`. The table is asserted in both +directions: a case it does not list must pass, and a case it lists must still +fail as recorded, so a fixed defect turns the module red until its entry goes. + +Nothing here imports :mod:`pcapkit`. + +""" + +from __future__ import annotations + +import fnmatch +import struct +import unittest +from typing import TYPE_CHECKING, NamedTuple + +from tests._support import time_limit + +if TYPE_CHECKING: + from typing import Optional + +#: Whole seconds one case may take. +CASE_TIMEOUT = 60 + + +class Outcome(NamedTuple): + """What one check found.""" + + #: ``'OK'``, ``'SKIP'`` (the case cannot run here, e.g. an engine's + #: dependency is missing), or one of the module's failure statuses. + status: 'str' + #: What differed, observed against expected. + detail: 'str' = '' + + +class Gap(NamedTuple): + """One root cause, and every case it stops.""" + + #: GitHub issue tracking the defect. + issue: 'Optional[int]' + #: The defect, with the ``file:line`` that causes it. + defect: 'str' + #: Status every case below reports, or the statuses of which each must + #: report one. + status: 'str | tuple[str, ...]' + #: Substring every failure detail contains, or a tuple of which each + #: detail contains at least one. + fragment: 'str | tuple[str, ...]' + #: Labels of the cases this defect stops, as :mod:`fnmatch` patterns. Each + #: pattern matches at least one case, and every case it matches fails. + cases: 'tuple[str, ...]' + + +def _as_tuple(value: 'str | tuple[str, ...]') -> 'tuple[str, ...]': + return (value,) if isinstance(value, str) else value + + +############################################################################### +# Wire builders, so the unit-tier modules read no capture. +############################################################################### + +#: Ethernet addresses every built frame uses. +MAC_A, MAC_B = bytes.fromhex('020000000001'), bytes.fromhex('020000000002') +#: IPv4 and IPv6 endpoints every built packet uses, as octets. +IPV4_A, IPV4_B = bytes([192, 0, 2, 1]), bytes([198, 51, 100, 2]) +IPV6_A = bytes.fromhex('20010db8000000000000000000000001') +IPV6_B = bytes.fromhex('20010db8000000000000000000000002') + + +def checksum(data: 'bytes') -> 'int': + """RFC 1071 Internet checksum.""" + if len(data) % 2: + data += b'\x00' + total = sum(int.from_bytes(data[i:i + 2], 'big') for i in range(0, len(data), 2)) + while total >> 16: + total = (total & 0xFFFF) + (total >> 16) + return ~total & 0xFFFF + + +def ethernet(payload: 'bytes', ethertype: 'int', *, reverse: 'bool' = False) -> 'bytes': + """An Ethernet II frame.""" + src, dst = (MAC_B, MAC_A) if reverse else (MAC_A, MAC_B) + return dst + src + ethertype.to_bytes(2, 'big') + payload + + +def ipv4(payload: 'bytes', *, ident: 'int' = 0x1234, offset: 'int' = 0, mf: 'bool' = False, + df: 'bool' = False, proto: 'int' = 17, reverse: 'bool' = False) -> 'bytes': + """An IPv4 packet; ``offset`` is in octets and must be a multiple of 8.""" + src, dst = (IPV4_B, IPV4_A) if reverse else (IPV4_A, IPV4_B) + flags = (0x4000 if df else 0) | (0x2000 if mf else 0) | (offset // 8) + header = bytearray(b'\x45\x00' + (20 + len(payload)).to_bytes(2, 'big') + + ident.to_bytes(2, 'big') + flags.to_bytes(2, 'big') + + bytes([64, proto]) + b'\x00\x00' + src + dst) + header[10:12] = checksum(bytes(header)).to_bytes(2, 'big') + return bytes(header) + payload + + +def ipv6(payload: 'bytes', *, nxt: 'int', reverse: 'bool' = False) -> 'bytes': + """An IPv6 packet whose first extension header or upper layer is ``nxt``.""" + src, dst = (IPV6_B, IPV6_A) if reverse else (IPV6_A, IPV6_B) + return (b'\x60\x00\x00\x00' + len(payload).to_bytes(2, 'big') + bytes([nxt, 64]) + + src + dst + payload) + + +def ipv6_fragment(payload: 'bytes', *, ident: 'int' = 0x12345678, offset: 'int' = 0, + mf: 'bool' = False, nxt: 'int' = 17) -> 'bytes': + """An IPv6 packet carrying one Fragment header (RFC 8200 section 4.5).""" + word = offset | (1 if mf else 0) + header = bytes([nxt, 0]) + word.to_bytes(2, 'big') + ident.to_bytes(4, 'big') + return ipv6(header + payload, nxt=44) + + +def udp(data: 'bytes', *, sport: 'int' = 40000, dport: 'int' = 9) -> 'bytes': + """A UDP datagram; port 9 (discard) dispatches to no application layer.""" + return (sport.to_bytes(2, 'big') + dport.to_bytes(2, 'big') + + (8 + len(data)).to_bytes(2, 'big') + b'\x00\x00' + data) + + +def tcp(payload: 'bytes', *, seq: 'int', ack: 'int' = 0, syn: 'bool' = False, + fin: 'bool' = False, rst: 'bool' = False, sport: 'int' = 40000, + dport: 'int' = 9) -> 'bytes': + """A TCP segment with no options; the ACK flag is set whenever ``ack`` is.""" + flags = (0x01 if fin else 0) | (0x02 if syn else 0) | (0x04 if rst else 0) | (0x10 if ack else 0) + return (sport.to_bytes(2, 'big') + dport.to_bytes(2, 'big') + (seq % (1 << 32)).to_bytes(4, 'big') + + ack.to_bytes(4, 'big') + bytes([0x50, flags]) + b'\xff\xff\x00\x00\x00\x00' + payload) + + +def pcap(frames: 'list[tuple[int, int, bytes]]', *, linktype: 'int' = 1, + nanosecond: 'bool' = False, byteorder: 'str' = 'little') -> 'bytes': + """A PCAP savefile of ``(seconds, fraction, frame)`` records. + + ``fraction`` is microseconds, or nanoseconds when ``nanosecond`` is set. + + """ + order = '<' if byteorder == 'little' else '>' + magic = 0xA1B23C4D if nanosecond else 0xA1B2C3D4 + out = [struct.pack(f'{order}IHHiIII', magic, 2, 4, 0, 0, 0xFFFF, linktype)] + for seconds, fraction, frame in frames: + out.append(struct.pack(f'{order}IIII', seconds, fraction, len(frame), len(frame)) + frame) + return b''.join(out) + + +def pcapng(frames: 'list[tuple[int, int, bytes]]', *, linktype: 'int' = 1, + nanosecond: 'bool' = False, byteorder: 'str' = 'little') -> 'bytes': + """A PCAP-NG file of one section and one interface, one EPB per record. + + ``fraction`` is microseconds, or nanoseconds when ``nanosecond`` is set, in + which case the interface carries ``if_tsresol`` 9. + + """ + order = '<' if byteorder == 'little' else '>' + + def block(kind: 'int', body: 'bytes') -> 'bytes': + body += b'\x00' * (-len(body) % 4) + size = 12 + len(body) + return struct.pack(f'{order}II', kind, size) + body + struct.pack(f'{order}I', size) + + shb = block(0x0A0D0D0A, struct.pack(f'{order}IHHq', 0x1A2B3C4D, 1, 0, -1)) + options = b'' + if nanosecond: # if_tsresol = 9, then opt_endofopt + options = struct.pack(f'{order}HH', 9, 1) + b'\x09\x00\x00\x00' + struct.pack(f'{order}HH', 0, 0) + idb = block(0x00000001, struct.pack(f'{order}HHI', linktype, 0, 0xFFFF) + options) + out = [shb, idb] + for seconds, fraction, frame in frames: + ticks = seconds * (10 ** 9 if nanosecond else 10 ** 6) + fraction + out.append(block(0x00000006, struct.pack(f'{order}IIIII', 0, ticks >> 32, ticks & 0xFFFFFFFF, + len(frame), len(frame)) + frame)) + return b''.join(out) + + +class Savefile(NamedTuple): + """A PCAP savefile, read without :mod:`pcapkit`.""" + + #: ``'little'`` or ``'big'``. + byteorder: 'str' + nanosecond: 'bool' + linktype: 'int' + #: ``(seconds, fraction, captured length, original length, octets)``. + records: 'tuple[tuple[int, int, int, int, bytes], ...]' + + +def read_pcap(data: 'bytes') -> 'Savefile': + """Read a PCAP savefile with :mod:`struct` alone, as an independent reference.""" + magic = data[:4] + table = {b'\xd4\xc3\xb2\xa1': ('little', False), b'\xa1\xb2\xc3\xd4': ('big', False), + b'\x4d\x3c\xb2\xa1': ('little', True), b'\xa1\xb2\x3c\x4d': ('big', True)} + byteorder, nanosecond = table[magic] + order = '<' if byteorder == 'little' else '>' + linktype, = struct.unpack_from(f'{order}I', data, 20) + records, offset = [], 24 + while offset < len(data): + sec, frac, incl, orig = struct.unpack_from(f'{order}IIII', data, offset) + records.append((sec, frac, incl, orig, data[offset + 16:offset + 16 + incl])) + offset += 16 + incl + return Savefile(byteorder, nanosecond, linktype, tuple(records)) + + +def payload_pattern(size: 'int', seed: 'int' = 1) -> 'bytes': + """``size`` octets, none of them zero, so a zero-filled hole shows.""" + return bytes((seed + 7 * i) % 255 + 1 for i in range(size)) + + +class RoundTripBase(unittest.TestCase): + """The table checks, shared by the foundation round-trip modules. + + A subclass sets :attr:`STATUSES` and :attr:`KNOWN_FAILURES`, and + implements :meth:`labels` and :meth:`outcome`. This class is not collected + itself, because :meth:`labels` is empty here. + + """ + + #: Failure statuses this module reports. + STATUSES = () # type: tuple[str, ...] + #: Known failures, one entry per root cause. + KNOWN_FAILURES = () # type: tuple[Gap, ...] + + def labels(self) -> 'list[str]': + """Every case label, in the order the cases run.""" + return [] + + def outcome(self, label: 'str') -> 'Outcome': + """Run one case.""" + raise NotImplementedError + + def setUp(self) -> None: + if not self.STATUSES: + self.skipTest('base class') + + # -- helpers ---------------------------------------------------------- + + def _gaps(self, labels: 'list[str]') -> 'dict[str, Gap]': + table = {} # type: dict[str, Gap] + for gap in self.KNOWN_FAILURES: + for pattern in gap.cases: + for label in fnmatch.filter(labels, pattern): + self.assertIs(table.setdefault(label, gap), gap, + f'{label} is listed under two root causes') + return table + + def _run(self, label: 'str') -> 'Outcome': + try: + with time_limit(CASE_TIMEOUT): + return self.outcome(label) + except TimeoutError as exc: + return Outcome('TIMEOUT', str(exc)) + + # -- the tables ------------------------------------------------------- + + def test_labels_are_unique(self) -> None: + labels = self.labels() + self.assertTrue(labels, 'no cases') + self.assertEqual(len(labels), len(set(labels)), 'duplicate case labels') + + def test_known_failures_name_real_cases(self) -> None: + labels = self.labels() + stale = [pattern for gap in self.KNOWN_FAILURES for pattern in gap.cases + if not fnmatch.filter(labels, pattern)] + self.assertEqual(stale, [], 'KNOWN_FAILURES patterns that match no case') + for gap in self.KNOWN_FAILURES: + self.assertTrue(gap.cases, f'empty KNOWN_FAILURES entry: {gap.defect}') + self.assertTrue(gap.fragment, f'no failure fragment: {gap.defect}') + for status in _as_tuple(gap.status): + self.assertIn(status, self.STATUSES + ('TIMEOUT',), gap.defect) + self._gaps(labels) + + def test_round_trip_is_exact_or_a_recorded_gap(self) -> None: + labels = self.labels() + gaps = self._gaps(labels) + for label in labels: + with self.subTest(case=label): + outcome = self._run(label) + if outcome.status == 'SKIP': + self.skipTest(outcome.detail) + gap = gaps.get(label) + if gap is None: + self.assertEqual( + outcome.status, 'OK', + f'{label} does not round-trip: {outcome.status}: {outcome.detail}. ' + 'If this is a new defect, add it to KNOWN_FAILURES under its root cause.') + continue + self.assertIn( + outcome.status, _as_tuple(gap.status), + f'{label} was recorded as failing (#{gap.issue}: {gap.defect}) but came ' + f'back {outcome.status}: {outcome.detail}. If the defect is fixed, ' + 'delete the label from its entry.') + self.assertTrue( + any(fragment in outcome.detail for fragment in _as_tuple(gap.fragment)), + f'{label} fails, but not in the recorded way ({gap.defect}); ' + f'detail was {outcome.detail!r}') diff --git a/tests/foundation/engines/test_engine_agreement_runtime.py b/tests/foundation/engines/test_engine_agreement_runtime.py new file mode 100644 index 0000000000..2acb389412 --- /dev/null +++ b/tests/foundation/engines/test_engine_agreement_runtime.py @@ -0,0 +1,371 @@ +# -*- coding: utf-8 -*- +"""Every offline engine reports what the ``default`` engine reports. C.f. #1202. + +Each engine in :attr:`Extractor.__engine__ +` reads every capture under +:file:`examples/captures/` with IPv4, IPv6 and TCP reassembly and TCP flow +tracing on, and is compared, field for field, with the ``default`` engine over +the same capture: + +``frames`` + the number of frames read; +``timestamps``, ``octets`` + each frame's timestamp, and its octets where the engine's frame object + keeps them -- ``pyshark`` reports dissected fields only, and ``pypcapfile`` + replaces the octets with its decoded layers, so their ``octets`` skip; +``ipv4``, ``ipv6``, ``tcp`` + every :class:`~pcapkit.foundation.reassembly.data.ip.Packet` or + :class:`~pcapkit.foundation.reassembly.data.tcp.Packet` the engine's toolkit + hands the reassembler, keyed by frame number -- every field; +``trace`` + every :class:`~pcapkit.foundation.traceflow.data.tcp.Packet` handed to the + flow tracer, keyed by frame number -- every field but ``frame``, which is + the engine's own frame object; +``datagrams`` + every reassembled datagram, every field but the lazily parsed ``packet``; +``flows`` + every traced flow's label and index, and the octets of its PCAP file. + +The inputs are captured by wrapping :meth:`ReassemblyBase.__call__ +` and +:meth:`TraceFlowBase.__call__ +`, the one +entry point every engine feeds. + +An engine whose dependency is missing, or which declines this interpreter, makes +:class:`~pcapkit.foundation.extraction.Extractor` fall back to ``default`` with +an :class:`~pcapkit.utilities.warnings.EngineWarning`; its cases are skipped with +that warning as the reason, never compared, since a fallback agrees with +``default`` trivially. Refusals the engines document are skipped too, with the +refusal as the reason: ``pypcap``, ``pcap_ct`` and ``pypcapfile`` reject PCAP-NG +with a :exc:`~pcapkit.utilities.exceptions.FormatError`, and an engine that +declines reassembly or flow tracing, or one protocol of it, warns and leaves it +unset (``pypcap`` and ``pcap_ct`` decline both, ``pyshark`` reassembly, +``pypcapfile`` IPv6). Any other exception is an ``ERROR``. + +This module reads generated captures, so it is in the fixture-dependent tier +(see :mod:`tests._tiers`) and runs after ``make samples``. + +""" + +from __future__ import annotations + +import importlib.util +import os +import tempfile +import unittest +import warnings +from typing import TYPE_CHECKING +from unittest import mock + +from tests._support import close_extractor, reimport_once_per_class, sample_path +from tests._tiers import SAMPLE_ROOT +from tests.foundation import _roundtrip as harness +from tests.foundation._roundtrip import Gap, Outcome + +if TYPE_CHECKING: + from typing import Any, Optional + +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None + for name in ('tbtrim', 'aenum', 'chardet', 'dictdumper')) + +#: Every capture on disk, enumerated rather than hand-picked. +CAPTURES = sorted(name for name in os.listdir(SAMPLE_ROOT) + if name.endswith(('.pcap', '.cap', '.pcapng'))) +#: Every engine but ``default``, as :attr:`Extractor.__engine__` names them. +ENGINES = ('dpkt', 'scapy', 'pyshark', 'pypcap', 'pcap_ct', 'pypcapfile') +ASPECTS = ('frames', 'timestamps', 'octets', 'ipv4', 'ipv6', 'tcp', 'trace', 'datagrams', 'flows') +#: Engines whose frame objects keep no octets to compare, and why. +NO_OCTETS = { + 'pyshark': 'pyshark reports dissected fields, not the frame octets', + 'pypcapfile': 'the pypcapfile engine replaces the frame octets with decoded layers', +} +#: Captures with at least one TCP flow, measured on ``00d232890``. +TRACED = ('http.pcap', 'http6.cap', 'in.pcap', 'many_interfaces.pcapng', 'options-tcp.pcap', + 'options-transport.pcap', 'stream.pcap', 'tcp.pcap', 'test.pcap') +#: The PCAP captures among them whose flows are all over IPv4. +TRACED_IPV4_ONLY_PCAP = ('http.pcap', 'in.pcap', 'options-tcp.pcap', 'options-transport.pcap') + + +def _plain(value: 'Any') -> 'Any': + return bytes(value) if isinstance(value, bytearray) else value + + +def _fields(packet: 'Any', skip: 'tuple[str, ...]' = ()) -> 'dict[str, Any]': + return {key: _plain(value) for key, value in packet.items() if key not in skip} + + +def _record(engine: 'str', frame: 'Any') -> 'tuple[float, Optional[bytes]]': + """``(timestamp, octets)`` of one frame; octets are :data:`None` per :data:`NO_OCTETS`.""" + if engine == 'default': + info = frame.info + stamp = info.time_epoch if 'time_epoch' in info else info.timestamp_epoch + return float(stamp), bytes(info.packet) + if engine == 'dpkt': + from pcapkit.toolkit.dpkt import packet2bytes, packet2timestamp + return float(packet2timestamp(frame)), packet2bytes(frame) + if engine == 'scapy': + return float(frame.time), bytes(getattr(frame, 'original', None) or bytes(frame)) + if engine in ('pypcap', 'pcap_ct'): + stamp, octets = frame + return float(stamp), bytes(octets) + if engine == 'pypcapfile': + from pcapkit.toolkit.pypcapfile import packet2timestamp + return float(packet2timestamp(frame)), None + if engine == 'pyshark': + return float(frame.sniff_timestamp), None + raise AssertionError(engine) + + +class _Run: + """One engine's reading of one capture.""" + + def __init__(self, engine: 'str', capture: 'str') -> None: + from pcapkit import extract + from pcapkit.foundation.reassembly.reassembly import ReassemblyBase + from pcapkit.foundation.traceflow.traceflow import TraceFlowBase + from pcapkit.utilities.warnings import EngineWarning + + self.fallback = None # type: Optional[str] + self.inputs = {'ipv4': {}, 'ipv6': {}, 'tcp': {}, 'trace': {}} # type: dict[str, dict[int, Any]] + inputs = self.inputs + reassemble, trace = ReassemblyBase.__call__, TraceFlowBase.__call__ + + def spy_reassembly(this: 'Any', packet: 'Any') -> 'None': + inputs[type(this).__name__.lower()].setdefault(packet.num, []).append(_fields(packet)) + return reassemble(this, packet) + + def spy_trace(this: 'Any', packet: 'Any') -> 'None': + inputs['trace'].setdefault(packet.index, []).append(_fields(packet, ('frame',))) + return trace(this, packet) + + with warnings.catch_warnings(), tempfile.TemporaryDirectory() as tracedir, \ + mock.patch.object(ReassemblyBase, '__call__', spy_reassembly), \ + mock.patch.object(TraceFlowBase, '__call__', spy_trace), \ + mock.patch('pcapkit.foundation.extraction.warn') as warn: + warnings.simplefilter('ignore') + fallen = [] # type: list[str] + try: + extractor = extract(fin=sample_path(capture), nofile=True, store=True, + engine=engine, reassembly=True, ipv4=True, ipv6=True, + tcp=True, trace=True, trace_fout=tracedir, trace_format='pcap') + except Exception as exc: # pylint: disable=broad-except + self.error = f'{type(exc).__name__}: {exc}' # type: Optional[str] + fallen = self._fallen(warn, EngineWarning) + self.fallback = '; '.join(fallen) or None + return + self.error = None + close_extractor(extractor) + fallen = self._fallen(warn, EngineWarning) + if fallen or extractor._exnam != engine: # pylint: disable=protected-access + self.fallback = '; '.join(fallen) or f'ran as {extractor._exnam}' # pylint: disable=protected-access + return + self.length = extractor.length + self.records = [_record(engine, frame) for frame in extractor.frame] + # An engine that declines a protocol says so with a warning and leaves + # it None; one that declines reassembly or tracing outright makes the + # property raise. Either way the aspect is None and is skipped, not + # compared (``pypcap``, ``pyshark``; ``pypcapfile`` for IPv6). + self.datagrams = {} # type: dict[str, Optional[list[dict[str, Any]]]] + reasm = self._property(extractor, 'reassembly') + for kind in ('ipv4', 'ipv6', 'tcp'): + datagrams = getattr(reasm, kind, None) if reasm is not None else None + self.datagrams[kind] = (None if datagrams is None + else [_fields(d, ('packet',)) for d in datagrams]) + self.flows = None # type: Optional[list[tuple[str, tuple[int, ...], bytes]]] + trace = self._property(extractor, 'trace') + if trace is not None and trace.tcp is not None: + self.flows = [] + for flow in trace.tcp: + with open(flow.fpout, 'rb') as file: + self.flows.append((flow.label, tuple(flow.index), file.read())) + + @staticmethod + def _fallen(warn: 'Any', category: 'type') -> 'list[str]': + return [str(call.args[0]) for call in warn.call_args_list + if len(call.args) > 1 and call.args[1] is category] + + @staticmethod + def _property(extractor: 'Any', name: 'str') -> 'Any': + from pcapkit.utilities.exceptions import UnsupportedCall + + try: + return getattr(extractor, name) + except UnsupportedCall: + return None + + +_RUNS = {} # type: dict[tuple[str, str], _Run] + + +def _run(engine: 'str', capture: 'str') -> '_Run': + key = (engine, capture) + if key not in _RUNS: + _RUNS[key] = _Run(engine, capture) + return _RUNS[key] + + +def _first_difference(got: 'dict[int, Any]', want: 'dict[int, Any]') -> 'Optional[str]': + for number in sorted(set(got) | set(want)): + if number not in got: + return f'frame {number}: missing, default has {len(want[number])} input(s)' + if number not in want: + return f'frame {number}: extra, default has none' + if got[number] != want[number]: + if len(got[number]) != len(want[number]): + return f'frame {number}: {len(got[number])} inputs != {len(want[number])}' + for mine, theirs in zip(got[number], want[number]): + diff = [f'{key} {mine.get(key)!r:.70} != {theirs.get(key)!r:.70}' + for key in sorted(set(mine) | set(theirs)) if mine.get(key) != theirs.get(key)] + if diff: + return f'frame {number}: ' + '; '.join(diff) + return None + + +def compare(engine: 'str', capture: 'str', aspect: 'str') -> 'Outcome': + run = _run(engine, capture) + if run.fallback is not None: + return Outcome('SKIP', f'{engine} did not run: {run.fallback}') + if run.error is not None: + if 'reads PCAP savefiles only' in run.error: # a documented refusal, not a reading + return Outcome('SKIP', f'{engine} refuses {capture}: {run.error}') + return Outcome('ERROR', run.error) + base = _run('default', capture) + if aspect in ('ipv4', 'ipv6', 'tcp') and run.datagrams[aspect] is None: + return Outcome('SKIP', f'{engine} declines {aspect} reassembly') + if aspect == 'trace' and run.flows is None: + return Outcome('SKIP', f'{engine} declines flow tracing') + if aspect == 'frames': + if run.length != base.length: + return Outcome('FRAMES', f'{run.length} frames != {base.length}') + return Outcome('OK') + if aspect in ('timestamps', 'octets'): + if aspect == 'octets' and engine in NO_OCTETS: + return Outcome('SKIP', NO_OCTETS[engine]) + field = 0 if aspect == 'timestamps' else 1 + mine, theirs = [r[field] for r in run.records], [r[field] for r in base.records] + for number, (one, two) in enumerate(zip(mine, theirs), 1): + if one != two: + return Outcome('RECORDS', f'frame {number}: {aspect[:-1]} {one!r:.70} != {two!r:.70}') + if len(mine) != len(theirs): + return Outcome('RECORDS', f'{len(mine)} {aspect} != {len(theirs)}') + return Outcome('OK') + if aspect in run.inputs: + diff = _first_difference(run.inputs[aspect], base.inputs[aspect]) + return Outcome('INPUTS', diff) if diff else Outcome('OK') + if aspect == 'datagrams': + kinds = [kind for kind in ('ipv4', 'ipv6', 'tcp') if run.datagrams[kind] is not None] + if not kinds: + return Outcome('SKIP', f'{engine} declines reassembly') + for kind in kinds: + mine, theirs = run.datagrams[kind], base.datagrams[kind] + if mine != theirs: + if len(mine) != len(theirs): + return Outcome('DATAGRAMS', f'{kind}: {len(mine)} datagrams != {len(theirs)}') + for position, (one, two) in enumerate(zip(mine, theirs)): + diff = [key for key in one if one[key] != two.get(key)] + if diff: + return Outcome('DATAGRAMS', f'{kind} datagram {position}: {", ".join(diff)} differ') + return Outcome('OK') + if aspect == 'flows': + if run.flows is None: + return Outcome('SKIP', f'{engine} declines flow tracing') + if [f[:2] for f in run.flows] != [f[:2] for f in base.flows]: + return Outcome('FLOWS', f'flows {[f[:2] for f in run.flows]!r:.200} != ' + f'{[f[:2] for f in base.flows]!r:.200}') + for (label, _, mine), (_, _, theirs) in zip(run.flows, base.flows): + if mine != theirs: + return Outcome('FLOWS', f'{label}: file of {len(mine)} octets != {len(theirs)} ' + f'(head {mine[:24].hex()} != {theirs[:24].hex()})') + return Outcome('OK') + raise AssertionError(aspect) + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class TestEngineAgreement(harness.RoundTripBase): + """Every engine that runs agrees with ``default`` on every capture.""" + + STATUSES = ('ERROR', 'FRAMES', 'RECORDS', 'INPUTS', 'DATAGRAMS', 'FLOWS') + KNOWN_FAILURES = ( + Gap(1507, "the dpkt, scapy and pypcapfile engines write every flow as JSON when asked " + "for trace_format='pcap': their tcp_traceflow adapters build the frame with " + 'packet2dict (pcapkit/toolkit/dpkt.py:600, pcapkit/toolkit/scapy.py:363, ' + 'pcapkit/toolkit/pypcapfile.py:620), which PCAPIO cannot write, so Extractor ' + 'swaps the format with a FormatWarning (pcapkit/foundation/extraction.py:1440-1444)', + 'FLOWS', 'file of ', tuple(f'{capture}/{engine}/flows' for engine in ('dpkt', 'scapy') + for capture in TRACED) + + tuple(f'{capture}/pypcapfile/flows' for capture in TRACED_IPV4_ONLY_PCAP)), + Gap(1512, 'the pypcapfile engine reads no frame from a big-endian PCAP, without a warning: ' + 'pcapfile.savefile.load_savefile (pypcapfile 0.12.0) yields no packets for one, ' + 'and the engine iterates that empty list (pcapkit/foundation/engines/' + 'pypcapfile.py:235, :249)', + ('FRAMES', 'RECORDS', 'INPUTS', 'DATAGRAMS'), + ('0 frames != 3', '0 timestamps != 3', 'frame 1: missing', '0 datagrams != 3'), + tuple(f'{capture}/pypcapfile/{aspect}' + for capture in ('big_endian.pcap', 'big_endian_nanosecond.pcap') + for aspect in ('frames', 'timestamps', 'ipv4', 'datagrams'))), + Gap(1513, 'the pypcapfile engine leaves TCP over IPv6 out of TCP reassembly and flow ' + 'tracing, without a warning: _network returns None for any frame that is not ' + 'IPv4 (pcapkit/toolkit/pypcapfile.py:262), so tcp_reassembly (:540) and ' + 'tcp_traceflow (:603) skip it; only IPv6 reassembly is declined with a warning ' + '(pcapkit/foundation/engines/pypcapfile.py:229)', + ('INPUTS', 'DATAGRAMS', 'FLOWS'), ('missing, default has', 'tcp: ', 'flows ['), + tuple(f'{capture}/pypcapfile/{aspect}' + for capture in ('http6.cap', 'stream.pcap', 'tcp.pcap', 'test.pcap') + for aspect in ('tcp', 'trace', 'datagrams', 'flows'))), + Gap(1514, "the pyshark engine raises ValueError on the first TCP frame it traces: " + 'tcp_traceflow reads the flags with bool(int(tcp.flags_syn)) (pcapkit/toolkit/' + "pyshark.py:415-417), and tshark 4.x reports them as 'True' and 'False'", + 'ERROR', "ValueError: invalid literal for int() with base 10: ", + tuple(f'{capture}/pyshark/*' for capture in TRACED)), + Gap(1515, 'the pyshark engine counts a Systemd Journal Export Block as a frame: tshark ' + 'numbers it as one, and read_frame takes every tshark frame ' + '(pcapkit/foundation/engines/pyshark.py:245-248)', + ('FRAMES', 'RECORDS'), ('6 frames != 5', 'frame 5: timestamp '), + ('test.pcapng/pyshark/frames', 'test.pcapng/pyshark/timestamps')), + Gap(1501, 'the dpkt engine hands reassembly the Decimal timestamp dpkt reads from a ' + 'nanosecond PCAP (pcapkit/foundation/engines/dpkt.py:309), where every other ' + 'engine hands it a float (pcapkit/toolkit/pcap.py:73)', + 'INPUTS', "timestamp Decimal('1500000000.123456789') != 1500000000.1234567", + ('big_endian_nanosecond.pcap/dpkt/ipv4',)), + Gap(1520, "pcapkit's own PCAPNGReader for the dpkt engine yields only Enhanced and " + 'obsolete Packet Blocks, so a Simple Packet Block is dropped and every later ' + 'frame shifts by one (pcapkit/foundation/engines/dpkt.py:124)', + ('FRAMES', 'RECORDS', 'INPUTS', 'DATAGRAMS'), + ('4 frames != 5', 'frame 4: ', '2 datagrams != 3'), + ('test.pcapng/dpkt/frames', 'test.pcapng/dpkt/timestamps', 'test.pcapng/dpkt/octets', + 'test.pcapng/dpkt/ipv4', + 'test.pcapng/dpkt/datagrams')), + Gap(1502, 'the default engine does not dissect LINKTYPE_RAW (101) frames, which scapy reads ' + 'as IPv4: neither Frame.__proto__ nor PCAPNG.__proto__ registers LinkType.RAW ' + '(pcapkit/protocols/misc/pcap/frame.py:90, pcapkit/protocols/misc/pcapng.py:699), ' + 'so those frames are Raw and never reach reassembly', + ('INPUTS', 'DATAGRAMS'), ('extra, default has none', 'datagrams != '), + ('test.pcapng/scapy/ipv4', 'test.pcapng/scapy/datagrams', + 'many_interfaces.pcapng/scapy/ipv4', 'many_interfaces.pcapng/scapy/datagrams')), + Gap(1503, 'the scapy engine dates a Simple Packet Block, which carries no timestamp, with ' + 'the wall-clock time it was read (scapy leaves Packet.time at construction; ' + 'pcapkit/toolkit/scapy.py:186 forwards it), where default reports 0', + 'RECORDS', 'frame 4: timestamp ', ('test.pcapng/scapy/timestamps',)), + ) + + def setUp(self) -> None: + super().setUp() + reimport_once_per_class(self) + + @classmethod + def tearDownClass(cls) -> None: + _RUNS.clear() + super().tearDownClass() + + def labels(self) -> 'list[str]': + return [f'{capture}/{engine}/{aspect}' for capture in CAPTURES for engine in ENGINES + for aspect in ASPECTS] + + def outcome(self, label: 'str') -> 'Outcome': + capture, engine, aspect = label.split('/') + return compare(engine, capture, aspect) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/foundation/reassembly/test_reassembly_payload_roundtrip_unit.py b/tests/foundation/reassembly/test_reassembly_payload_roundtrip_unit.py new file mode 100644 index 0000000000..17ef0a84dd --- /dev/null +++ b/tests/foundation/reassembly/test_reassembly_payload_roundtrip_unit.py @@ -0,0 +1,309 @@ +# -*- coding: utf-8 -*- +"""Reassembly recovers the payload its fragments or segments were cut from. C.f. #1202. + +Each case cuts a known payload -- a UDP datagram for IPv4 and IPv6, a byte +stream for TCP -- into fragments or segments, puts them on the wire in some +order, writes them to an in-memory PCAP savefile and runs that through +:func:`~pcapkit.interface.extract` with reassembly on. The round trip closes +when exactly one datagram comes back per payload, it is +:attr:`~pcapkit.foundation.reassembly.data.data.Completion.COMPLETE`, its octets +are the payload's, and its ``index`` names the frames that carried it. + +The orders are the ones a capture produces: in order, reversed, the last piece +first, interleaved, every piece twice (a retransmission), and with one extra +piece overlapping two others with the same octets. Each runs in strict and in +non-strict mode, which differ only in how a datagram with holes is reported -- +so for a complete datagram they must agree. + +When every IP fragment arrives twice, the second copy of the last one comes +after its datagram completed. Per the owner's ruling on #1507 (item 1), that +copy opens a new buffer and surfaces as a second, ``PARTIAL`` datagram, as RFC +791, Linux, FreeBSD, Suricata, Zeek and Wireshark all do, so those cases expect +both datagrams. + +Every case builds its own octets and reads no capture. Classes are imported +inside each test, after :func:`~tests._support.reimport_once_per_class`. + +""" + +from __future__ import annotations + +import importlib.util +import io +import unittest +import warnings +from typing import TYPE_CHECKING, NamedTuple + +from tests._support import close_extractor, reimport_once_per_class +from tests.foundation import _roundtrip as harness +from tests.foundation._roundtrip import (Gap, Outcome, ethernet, ipv4, ipv6, + ipv6_fragment, payload_pattern, pcap, tcp, udp) + +if TYPE_CHECKING: + from typing import Any + +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None + for name in ('tbtrim', 'aenum', 'chardet', 'dictdumper')) + + +class Stream(NamedTuple): + """One payload, and the frames that carry it.""" + + #: Payload the frames were cut from, as the datagram reports it: a tuple of + #: runs for a strict ``PARTIAL`` datagram. + payload: 'bytes | tuple[bytes, ...]' + #: 1-based frame numbers carrying any of it. + frames: 'tuple[int, ...]' + #: Expected datagram header, or :data:`None` not to check it. + header: 'bytes | None' = None + #: Expected completion. + completed: 'str' = 'COMPLETE' + + +class Case(NamedTuple): + """One capture to reassemble.""" + + label: 'str' + #: ``'ipv4'``, ``'ipv6'`` or ``'tcp'``. + kind: 'str' + frames: 'tuple[bytes, ...]' + streams: 'tuple[Stream, ...]' + strict: 'bool' + + +#: How each payload is cut, as piece lengths; an IP piece but the last is a +#: multiple of 8 octets. +IP_CUTS = { + 'halves': (600, 597), + 'thirds': (400, 400, 397), + 'uneven': (504, 8, 496, 189), + 'eights': (8,) * 149 + (5,), +} +TCP_CUTS = { + 'halves': (1500, 1500), + 'thirds': (1000, 1000, 1000), + 'uneven': (1, 999, 1460, 540), + 'small': (30,) * 100, +} +#: Orders the pieces are sent in, as functions of the piece count. +ORDERS = { + 'forward': lambda n: list(range(n)), + 'reverse': lambda n: list(reversed(range(n))), + 'last-first': lambda n: [n - 1] + list(range(n - 1)), + 'interleave': lambda n: list(range(0, n, 2)) + list(range(1, n, 2)), + 'duplicate': lambda n: [i for i in range(n) for _ in (0, 1)], +} + +#: The IP payload: a UDP datagram of 1197 octets, so the last piece is short. +IP_PAYLOAD = udp(payload_pattern(1189)) +#: The second IP payload, for two datagrams in flight at once. +IP_PAYLOAD_2 = udp(payload_pattern(709, seed=3), sport=40001) +#: The TCP byte stream. +TCP_STREAM = payload_pattern(3000, seed=5) + + +def _octets(payload: 'bytes | tuple[bytes, ...]') -> 'int': + """Octets in a payload, summing the runs of a strict ``PARTIAL`` one.""" + return len(payload) if isinstance(payload, bytes) else sum(len(run) for run in payload) + + +def _pieces(total: 'int', cuts: 'tuple[int, ...]') -> 'list[tuple[int, int]]': + assert sum(cuts) == total, (sum(cuts), total) + out, offset = [], 0 + for size in cuts: + out.append((offset, size)) + offset += size + return out + + +def _ip_frame(kind: 'str', payload: 'bytes', offset: 'int', size: 'int', ident: 'int') -> 'bytes': + more = offset + size < len(payload) + chunk = payload[offset:offset + size] + if kind == 'ipv4': + return ethernet(ipv4(chunk, ident=ident, offset=offset, mf=more), 0x0800) + return ethernet(ipv6_fragment(chunk, ident=ident, offset=offset, mf=more), 0x86DD) + + +def _ip_header(kind: 'str', payload: 'bytes', size: 'int', ident: 'int') -> 'bytes | None': + """The header the offset-0 fragment carried, for IPv4.""" + if kind != 'ipv4': + return None + return ipv4(payload[:size], ident=ident, offset=0, mf=size < len(payload))[:20] + + +def _ip_cases() -> 'list[Case]': + cases = [] + for kind in ('ipv4', 'ipv6'): + for cut, sizes in IP_CUTS.items(): + pieces = _pieces(len(IP_PAYLOAD), sizes) + header = _ip_header(kind, IP_PAYLOAD, sizes[0], 0x1234) + for order, permute in ORDERS.items(): + frames = tuple(_ip_frame(kind, IP_PAYLOAD, *pieces[i], 0x1234) + for i in permute(len(pieces))) + for strict in (True, False): + if order == 'duplicate': + # the last frame repeats the last fragment after completion (#1507) + offset, size = pieces[-1] + late = IP_PAYLOAD[offset:offset + size] + streams = ( + Stream(IP_PAYLOAD, tuple(range(1, len(frames))), header), + Stream((late,) if strict else bytes(offset) + late, (len(frames),), + b'' if kind == 'ipv4' else None, 'PARTIAL'), + ) + else: + streams = (Stream(IP_PAYLOAD, tuple(range(1, len(frames) + 1)), header),) + cases.append(Case(f'{kind}/{cut}/{order}/{"strict" if strict else "loose"}', + kind, frames, streams, strict)) + # one extra piece straddling a boundary, with the same octets + frames = [_ip_frame(kind, IP_PAYLOAD, *piece, 0x1234) for piece in pieces] + end = next(offset + size for offset, size in pieces if offset + size >= 24) - 8 + frames.insert(1, _ip_frame(kind, IP_PAYLOAD, 8, (end - 8) - (end - 8) % 8, 0x1234)) + stream = Stream(IP_PAYLOAD, tuple(range(1, len(frames) + 1)), header) + for strict in (True, False): + cases.append(Case(f'{kind}/{cut}/overlap/{"strict" if strict else "loose"}', + kind, tuple(frames), (stream,), strict)) + + # unfragmented: one packet, offset 0 and MF clear + if kind == 'ipv4': + frame = ethernet(ipv4(IP_PAYLOAD, ident=0x0101), 0x0800) + else: + frame = ethernet(ipv6_fragment(IP_PAYLOAD, ident=0x0101), 0x86DD) + header = _ip_header(kind, IP_PAYLOAD, len(IP_PAYLOAD), 0x0101) + for strict in (True, False): + cases.append(Case(f'{kind}/atomic/{"strict" if strict else "loose"}', kind, (frame,), + (Stream(IP_PAYLOAD, (1,), header),), strict)) + + # two datagrams in flight, their fragments interleaved + first = [_ip_frame(kind, IP_PAYLOAD, *piece, 0x2001) + for piece in _pieces(len(IP_PAYLOAD), IP_CUTS['thirds'])] + second = [_ip_frame(kind, IP_PAYLOAD_2, *piece, 0x2002) + for piece in _pieces(len(IP_PAYLOAD_2), (352, 365))] + frames = (first[0], second[1], first[2], second[0], first[1]) + streams = (Stream(IP_PAYLOAD, (1, 3, 5), _ip_header(kind, IP_PAYLOAD, 400, 0x2001)), + Stream(IP_PAYLOAD_2, (2, 4), _ip_header(kind, IP_PAYLOAD_2, 352, 0x2002))) + for strict in (True, False): + cases.append(Case(f'{kind}/two-datagrams/{"strict" if strict else "loose"}', kind, + frames, streams, strict)) + return cases + + +def _tcp_cases() -> 'list[Case]': + cases = [] + for isn_name, isn in (('isn-low', 1000), ('isn-wrap', (1 << 32) - 1700)): + syn = ethernet(ipv4(tcp(b'', seq=isn, syn=True), proto=6), 0x0800) + for cut, sizes in TCP_CUTS.items(): + pieces = _pieces(len(TCP_STREAM), sizes) + + def segment(offset: 'int', size: 'int') -> 'bytes': + return ethernet(ipv4(tcp(TCP_STREAM[offset:offset + size], + seq=isn + 1 + offset, ack=1), proto=6), 0x0800) + + fin = ethernet(ipv4(tcp(b'', seq=isn + 1 + len(TCP_STREAM), ack=1, fin=True), + proto=6), 0x0800) + for order, permute in ORDERS.items(): + data = [segment(*pieces[i]) for i in permute(len(pieces))] + frames = (syn, *data, fin) + stream = Stream(TCP_STREAM, tuple(range(2, len(frames) + 1))) + for strict in (True, False): + cases.append(Case(f'tcp/{isn_name}/{cut}/{order}/{"strict" if strict else "loose"}', + 'tcp', frames, (stream,), strict)) + # the FIN reordered ahead of the data + data = [segment(*piece) for piece in reversed(pieces)] + frames = (syn, fin, *data) + stream = Stream(TCP_STREAM, tuple(range(2, len(frames) + 1))) + for strict in (True, False): + cases.append(Case(f'tcp/{isn_name}/{cut}/fin-first/{"strict" if strict else "loose"}', + 'tcp', frames, (stream,), strict)) + # one extra segment overlapping the first two with the same octets + data = [segment(*piece) for piece in pieces] + data.insert(1, segment(1, pieces[0][1] + pieces[1][1] - 2)) + frames = (syn, *data, fin) + stream = Stream(TCP_STREAM, tuple(range(2, len(frames) + 1))) + for strict in (True, False): + cases.append(Case(f'tcp/{isn_name}/{cut}/overlap/{"strict" if strict else "loose"}', + 'tcp', frames, (stream,), strict)) + return cases + + +CASES = {case.label: case for case in _ip_cases() + _tcp_cases()} + + +def run_case(case: 'Case') -> 'Outcome': + """Reassemble ``case`` and compare each datagram with the payload it came from.""" + from pcapkit import extract + + # a millisecond apart, so no case outlasts a reassembly timeout + data = pcap([(1_500_000_000, 1000 * i, frame) for i, frame in enumerate(case.frames)]) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + try: + extractor = extract(fin=io.BytesIO(data), nofile=True, store=False, reassembly=True, + reasm_strict=case.strict, **{case.kind: True}) + except Exception as exc: # pylint: disable=broad-except + return Outcome('ERROR', f'{type(exc).__name__}: {exc}') + close_extractor(extractor) + datagrams = list(getattr(extractor.reassembly, case.kind)) + if len(datagrams) != len(case.streams): + return Outcome('COUNT', f'{len(datagrams)} datagrams != {len(case.streams)}: ' + + ', '.join(f'{d.completed.name} {d.index}' for d in datagrams)) + by_payload = {} # type: dict[Any, Any] + for datagram in datagrams: + payload = datagram.payload + key = bytes(payload) if isinstance(payload, (bytes, bytearray)) else tuple(payload) + by_payload[key] = datagram + for number, stream in enumerate(case.streams): + datagram = by_payload.get(stream.payload) + if datagram is None: + sizes = [(d.completed.name, len(d.payload) if isinstance(d.payload, (bytes, bytearray)) + else tuple(len(run) for run in d.payload)) for d in datagrams] + return Outcome('PAYLOAD', f'stream {number}: no datagram carries the ' + f'{_octets(stream.payload)}-octet payload; got {sizes}') + if datagram.completed.name != stream.completed: + return Outcome('INCOMPLETE', f'stream {number}: {datagram.completed.name} != ' + f'{stream.completed}') + if set(datagram.index) != set(stream.frames): + return Outcome('INDEX', f'stream {number}: index {datagram.index} != frames {stream.frames}') + if stream.header is not None and bytes(datagram.header) != stream.header: + return Outcome('HEADER', f'stream {number}: header {bytes(datagram.header).hex()} != ' + f'{stream.header.hex()}') + return Outcome('OK') + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class TestReassemblyPayloadRoundTrip(harness.RoundTripBase): + """Every cut, order and mode recovers the payload exactly.""" + + STATUSES = ('ERROR', 'COUNT', 'PAYLOAD', 'INCOMPLETE', 'INDEX', 'HEADER') + KNOWN_FAILURES = () # type: tuple[Gap, ...] + + def setUp(self) -> None: + super().setUp() + reimport_once_per_class(self) + + def labels(self) -> 'list[str]': + return list(CASES) + + def outcome(self, label: 'str') -> 'Outcome': + return run_case(CASES[label]) + + def test_builders_carry_the_payload(self) -> None: + """The wire builders put the payload where the reassembler looks for it.""" + from pcapkit.protocols.link.ethernet import Ethernet + + frame = ethernet(ipv6(udp(b'abc'), nxt=17), 0x86DD) + self.assertEqual(len(frame), 14 + 40 + 11) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + parsed = Ethernet(frame, len(frame)) + self.assertEqual(parsed.info.type.value, 0x86DD) + self.assertEqual(bytes(parsed['IPv6'].info.src.packed), bytes.fromhex( + '20010db8000000000000000000000001')) + # every case's frames carry the whole payload at least once + for case in CASES.values(): + with self.subTest(case=case.label): + self.assertGreaterEqual(sum(len(f) for f in case.frames), + sum(_octets(s.payload) for s in case.streams)) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/foundation/registry/test_registry_symmetry_unit.py b/tests/foundation/registry/test_registry_symmetry_unit.py new file mode 100644 index 0000000000..d4b885daf2 --- /dev/null +++ b/tests/foundation/registry/test_registry_symmetry_unit.py @@ -0,0 +1,466 @@ +# -*- coding: utf-8 -*- +"""A registrar handed back what it shipped with leaves every registry as it was. C.f. #1202. + +:mod:`pcapkit.foundation.registry` has no ``unregister``: the way to undo a +registration is to register the entry it displaced, which the owner ruled a +registrar must accept for the entry a code shipped with (#1363). So the round +trip here is *override, then restore*: snapshot every registry any public +registrar writes, register a custom entry over a shipped code, register the +shipped entry back, and compare every registry with the snapshot -- the same +keys, and under each key the very object that was there. + +Each registrar runs in these variants, where they apply: + +``noop`` + hand back the shipped entry without overriding it first; +``override`` + override with a custom class (or method pair and schema) whose name is new + to the package, then restore the shipped entry object; +``override-same-name`` + the same, with a custom class named like the one it displaces; +``restore-by-name`` + override, then restore by ``(module, class)`` strings, which build a + descriptor equal to the shipped one. + +and the comparison is split in two, so that two root causes cannot hide behind +one another: ``dispatch`` covers the tables the registrar names -- the +``__proto__`` it targets, the option and schema registries, the extractor and +flow-tracing tables -- and ``names`` covers :data:`pcapkit.protocols.__proto__`, +the by-name registry the protocol registrars all funnel into. + +The scope is the 36 public registrars in +:data:`pcapkit.foundation.registry.__all__`. Of those, the callback registrars +(``register_reassembly_*_callback`` and ``register_traceflow_tcp_callback``) +only append, and offer nothing to undo an append with, so they have no case +here; nor does ``register_protocol_code``, which funnels into the same +:meth:`ProtocolBase.register ` +the ``register_tcp`` cases reach. Class-level registrars with no foundation wrapper are out of scope: +``SCTP.register_chunk``, ``register_parameter`` and ``register_cause`` +(covered by :file:`tests/protocols/transport/test_sctp_unit.py`), +``Schema.register``, the PCAP-NG ``Option.register`` and ``ESP.register``. + +Every case restores the registries it touched, whatever its outcome. Classes are +imported inside each test, after :func:`~tests._support.reimport_once_per_class`. + +""" + +from __future__ import annotations + +import importlib +import importlib.util +import unittest +import warnings +from typing import TYPE_CHECKING, NamedTuple + +from tests._support import reimport_once_per_class +from tests.foundation import _roundtrip as harness +from tests.foundation._roundtrip import Gap, Outcome + +if TYPE_CHECKING: + from typing import Any, Callable + +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None + for name in ('tbtrim', 'aenum', 'chardet', 'dictdumper')) + + +def resolve(path: 'str') -> 'Any': + module, _, name = path.partition(':') + return getattr(importlib.import_module(module), name) + + +#: Protocol classes whose ``__proto__`` a module registrar writes. +DISPATCH = { + 'Frame': 'pcapkit.protocols.misc.pcap.frame:Frame', + 'PCAPNG': 'pcapkit.protocols.misc.pcapng:PCAPNG', + 'Link': 'pcapkit.protocols.link.link:Link', + 'Internet': 'pcapkit.protocols.internet.internet:Internet', + 'TCP': 'pcapkit.protocols.transport.tcp:TCP', + 'UDP': 'pcapkit.protocols.transport.udp:UDP', + 'SCTP': 'pcapkit.protocols.transport.sctp:SCTP', +} + + +class MethodRegistrar(NamedTuple): + """A registrar keyed by an option-like code, with its two tables.""" + + name: 'str' + #: Protocol class and the attribute holding its method table. + proto: 'str' + table: 'str' + #: Schema class whose ``__enum__`` (or ``registry``) holds the schemas. + schema: 'str' + #: Codes to override, as ``module:Enum.member`` paths. + codes: 'tuple[str, ...]' + + +METHOD_REGISTRARS = ( + MethodRegistrar('register_ipv4_option', 'pcapkit.protocols.internet.ipv4:IPv4', '__option__', + 'pcapkit.protocols.schema.internet.ipv4:Option', + ('pcapkit.const.ipv4.option_number:OptionNumber.NOP',)), + MethodRegistrar('register_hip_parameter', 'pcapkit.protocols.internet.hip:HIP', '__parameter__', + 'pcapkit.protocols.schema.internet.hip:Parameter', + ('pcapkit.const.hip.parameter:Parameter.R1_Counter',)), + MethodRegistrar('register_hopopt_option', 'pcapkit.protocols.internet.hopopt:HOPOPT', '__option__', + 'pcapkit.protocols.schema.internet.hopopt:Option', + ('pcapkit.const.ipv6.option:Option.PadN',)), + MethodRegistrar('register_ipv6_opts_option', 'pcapkit.protocols.internet.ipv6_opts:IPv6_Opts', + '__option__', 'pcapkit.protocols.schema.internet.ipv6_opts:Option', + ('pcapkit.const.ipv6.option:Option.PadN',)), + MethodRegistrar('register_ipv6_route_routing', 'pcapkit.protocols.internet.ipv6_route:IPv6_Route', + '__routing__', 'pcapkit.protocols.schema.internet.ipv6_route:RoutingType', + ('pcapkit.const.ipv6.routing:Routing.Type_2_Routing_Header',)), + MethodRegistrar('register_mh_message', 'pcapkit.protocols.internet.mh:MH', '__message__', + 'pcapkit.protocols.schema.internet.mh:Packet', + ('pcapkit.const.mh.packet:Packet.Home_Test_Init',)), + MethodRegistrar('register_mh_option', 'pcapkit.protocols.internet.mh:MH', '__option__', + 'pcapkit.protocols.schema.internet.mh:Option', + ('pcapkit.const.mh.option:Option.PadN',)), + MethodRegistrar('register_mh_extension', 'pcapkit.protocols.internet.mh:MH', '__extension__', + 'pcapkit.protocols.schema.internet.mh:CGAExtension', + ('pcapkit.const.mh.cga_extension:CGAExtension.Exp_FFFD',)), + MethodRegistrar('register_tcp_option', 'pcapkit.protocols.transport.tcp:TCP', '__option__', + 'pcapkit.protocols.schema.transport.tcp:Option', + ('pcapkit.const.tcp.option:Option.No_Operation',)), + MethodRegistrar('register_tcp_mp_option', 'pcapkit.protocols.transport.tcp:TCP', '__mp_option__', + 'pcapkit.protocols.schema.transport.tcp:MPTCP', + ('pcapkit.const.tcp.mp_tcp_option:MPTCPOption.MP_JOIN',)), + MethodRegistrar('register_http_frame', 'pcapkit.protocols.application.httpv2:HTTP', '__frame__', + 'pcapkit.protocols.schema.application.httpv2:FrameType', + ('pcapkit.const.http.frame:Frame.HEADERS',)), + MethodRegistrar('register_pcapng_block', 'pcapkit.protocols.misc.pcapng:PCAPNG', '__block__', + 'pcapkit.protocols.schema.misc.pcapng:BlockType', + ('pcapkit.const.pcapng.block_type:BlockType.Interface_Description_Block',)), + MethodRegistrar('register_pcapng_option', 'pcapkit.protocols.misc.pcapng:PCAPNG', '__option__', + 'pcapkit.protocols.schema.misc.pcapng:Option', + ('pcapkit.const.pcapng.option_type:OptionType.opt_comment', + 'pcapkit.const.pcapng.option_type:OptionType.if_name', + 'pcapkit.const.pcapng.option_type:OptionType.epb_flags')), + MethodRegistrar('register_pcapng_record', 'pcapkit.protocols.misc.pcapng:PCAPNG', '__record__', + 'pcapkit.protocols.schema.misc.pcapng:NameResolutionRecord', + ('pcapkit.const.pcapng.record_type:RecordType.nrb_record_ipv4',)), + MethodRegistrar('register_pcapng_secrets', 'pcapkit.protocols.misc.pcapng:PCAPNG', '__secrets__', + 'pcapkit.protocols.schema.misc.pcapng:DSBSecrets', + ('pcapkit.const.pcapng.secrets_type:SecretsType.WireGuard_Key_Log',)), +) + + +class ModuleRegistrar(NamedTuple): + """A registrar keyed by a dispatch code or a name, taking a class or descriptor.""" + + name: 'str' + #: The code, as a ``module:Enum.member`` path, or a literal ``int`` or ``str``. + code: 'str | int' + #: Where the shipped entry is read from: a ``DISPATCH`` name, or + #: ``'engine'``/``'reassembly'``/``'traceflow'``/``'output'``/``'trace-output'``. + source: 'str' + #: The base a custom class must subclass, as a ``module:Class`` path. + base: 'str' + #: Extra positional arguments after the module, e.g. ``'tcp'`` for apptype. + extra: 'tuple[str, ...]' = () + + +MODULE_REGISTRARS = ( + ModuleRegistrar('register_linktype', 'pcapkit.const.reg.linktype:LinkType.ETHERNET', 'Frame', + 'pcapkit.protocols.link.link:Link'), + ModuleRegistrar('register_pcap', 'pcapkit.const.reg.linktype:LinkType.ETHERNET', 'Frame', + 'pcapkit.protocols.link.link:Link'), + ModuleRegistrar('register_pcapng', 'pcapkit.const.reg.linktype:LinkType.ETHERNET', 'PCAPNG', + 'pcapkit.protocols.link.link:Link'), + ModuleRegistrar('register_ethertype', + 'pcapkit.const.reg.ethertype:EtherType.Address_Resolution_Protocol', 'Link', + 'pcapkit.protocols.link.link:Link'), + ModuleRegistrar('register_transtype', 'pcapkit.const.reg.transtype:TransType.TCP', 'Internet', + 'pcapkit.protocols.transport.transport:Transport'), + ModuleRegistrar('register_tcp', 80, 'TCP', 'pcapkit.protocols.application.application:Application'), + ModuleRegistrar('register_udp', 1701, 'UDP', 'pcapkit.protocols.link.link:Link'), + ModuleRegistrar('register_apptype', 80, 'TCP', + 'pcapkit.protocols.application.application:Application', ('tcp',)), + ModuleRegistrar('register_sctp', + 'pcapkit.const.sctp.payload_protocol_identifier:PayloadProtocolIdentifier.' + 'PayloadProtocolIdentifier_3GPP_NG_Application_Protocol', 'SCTP', + 'pcapkit.protocols.application.application:Application'), + ModuleRegistrar('register_extractor_engine', 'dpkt', 'engine', + 'pcapkit.foundation.engines.engine:Engine'), + ModuleRegistrar('register_extractor_reassembly', 'ipv4', 'reassembly', + 'pcapkit.foundation.reassembly.reassembly:Reassembly'), + ModuleRegistrar('register_extractor_traceflow', 'tcp', 'traceflow', + 'pcapkit.foundation.traceflow.traceflow:TraceFlow'), + ModuleRegistrar('register_dumper', 'json', 'output', 'dictdumper:JSON'), + ModuleRegistrar('register_extractor_dumper', 'json', 'output', 'dictdumper:JSON'), + ModuleRegistrar('register_traceflow_dumper', 'json', 'trace-output', 'dictdumper:JSON'), +) + +MODULE_VARIANTS = ('noop', 'override', 'override-same-name', 'restore-by-name') +METHOD_VARIANTS = ('noop', 'override') +ASPECTS = ('dispatch', 'names') + + +def _code(spec: 'str | int') -> 'Any': + if isinstance(spec, str) and ':' in spec: + module, _, rest = spec.partition(':') + enum, _, member = rest.partition('.') + return getattr(getattr(importlib.import_module(module), enum), member) + return spec + + +############################################################################### +# Snapshots. +############################################################################### + +def _tables() -> 'dict[str, dict[Any, Any]]': + """Every registry a public registrar writes, by name, as live objects.""" + from pcapkit.foundation.extraction import Extractor + from pcapkit.foundation.traceflow import TraceFlow + from pcapkit.protocols import __proto__ as names + + tables = {'names': names} # type: dict[str, dict[Any, Any]] + for label, path in DISPATCH.items(): + tables[f'{label}.__proto__'] = resolve(path).__proto__ + for registrar in METHOD_REGISTRARS: + proto = resolve(registrar.proto) + tables[f'{proto.__name__}.{registrar.table}'] = getattr(proto, registrar.table) + schema = resolve(registrar.schema) + registry = getattr(schema, 'registry', None) + if registry is not None and all(isinstance(value, dict) for value in registry.values()): + tables[f'{schema.__name__}.registry'] = registry + for namespace, table in registry.items(): + tables[f'{schema.__name__}.registry[{namespace}]'] = table + else: + tables[f'{schema.__module__}.{schema.__name__}.__enum__'] = schema.__enum__ + tables['Extractor.__engine__'] = Extractor.__engine__ + tables['Extractor.__reassembly__'] = Extractor.__reassembly__ + tables['Extractor.__traceflow__'] = Extractor.__traceflow__ + tables['Extractor.__output__'] = Extractor.__output__ + tables['TraceFlow.__output__'] = TraceFlow.__output__ + return tables + + +def _snapshot() -> 'dict[str, dict[Any, Any]]': + return {name: dict(table) for name, table in _tables().items()} + + +def _restore(snapshot: 'dict[str, dict[Any, Any]]') -> None: + tables = _tables() + for name, copy in snapshot.items(): + tables[name].clear() + tables[name].update(copy) + + +def _same(one: 'Any', two: 'Any') -> 'bool': + if one is two: + return True + if isinstance(one, tuple) and isinstance(two, tuple): + return len(one) == len(two) and all(_same(a, b) for a, b in zip(one, two)) + return isinstance(one, (str, int)) and type(one) is type(two) and one == two + + +def _diff(before: 'dict[str, dict[Any, Any]]', aspect: 'str') -> 'list[str]': + out = [] + for name, table in _tables().items(): + if (name == 'names') != (aspect == 'names'): + continue + old = before[name] + for key in sorted(set(old) | set(table), key=repr): + if key not in table: + out.append(f'{name}[{key!r}] removed') + elif key not in old: + out.append(f'{name}[{key!r}] added {table[key]!r}') + elif not _same(table[key], old[key]): + out.append(f'{name}[{key!r}] {old[key]!r} -> {table[key]!r}') + return out + + +############################################################################### +# Cases. +############################################################################### + +def _shipped(registrar: 'ModuleRegistrar', code: 'Any') -> 'Any': + from pcapkit.foundation.extraction import Extractor + from pcapkit.foundation.traceflow import TraceFlow + + if registrar.source in DISPATCH: + return resolve(DISPATCH[registrar.source]).__proto__[code] + return { + 'engine': Extractor.__engine__, 'reassembly': Extractor.__reassembly__, + 'traceflow': Extractor.__traceflow__, 'output': Extractor.__output__, + 'trace-output': TraceFlow.__output__, + }[registrar.source][code] + + +def _module_steps(registrar: 'ModuleRegistrar', variant: 'str') -> 'list[Callable[[], None]]': + """The calls one module-registrar case makes, in order.""" + from pcapkit.corekit.module import ModuleDescriptor + from pcapkit.foundation import registry + + register = getattr(registry, registrar.name) + code = _code(registrar.code) + shipped = _shipped(registrar, code) + ext = () # type: tuple[Any, ...] + kwargs = {} # type: dict[str, Any] + if registrar.source in ('output', 'trace-output'): + shipped, extension = shipped + kwargs = {'ext': extension} + descriptor = shipped if isinstance(shipped, ModuleDescriptor) else None + target = descriptor.klass if descriptor is not None else shipped + base = resolve(registrar.base) + name = target.__name__ if variant == 'override-same-name' else 'RoundTripCustom' + custom = type(name, (base,), {'__module__': __name__}) + ext = registrar.extra + + def call(*args: 'Any') -> 'Callable[[], None]': + return lambda: register(code, *args, *ext, **kwargs) + + if variant == 'noop': + return [call(shipped)] + if variant == 'restore-by-name': + module, klass = ((descriptor.module, descriptor.name) if descriptor is not None + else (target.__module__, target.__name__)) + return [call(custom), call(module, klass)] + return [call(custom), call(shipped)] + + +def _method_steps(registrar: 'MethodRegistrar', code_path: 'str', variant: 'str') -> 'list[Callable[[], None]]': + from pcapkit.foundation import registry + + register = getattr(registry, registrar.name) + proto, schema = resolve(registrar.proto), resolve(registrar.schema) + code = _code(code_path) + key = code + if registrar.name == 'register_pcapng_option': + from pcapkit.protocols.misc.pcapng import _option_key + key = _option_key(code) + namespace = code.name.split('_')[0] + shipped_schema = schema.registry[namespace][code] + else: + shipped_schema = schema.__enum__[code] + shipped = getattr(proto, registrar.table)[key] + if variant == 'noop': + return [lambda: register(code, shipped, schema=shipped_schema)] + + def parser(*args: 'Any', **kwargs: 'Any') -> 'Any': + raise NotImplementedError + + custom_schema = type('RoundTripCustom', (schema,), {'__module__': __name__}) + return [lambda: register(code, (parser, parser), schema=custom_schema), + lambda: register(code, shipped, schema=shipped_schema)] + + +def _labels() -> 'list[str]': + labels = [] + for module in MODULE_REGISTRARS: + labels.extend(f'{module.name}/{variant}/{aspect}' for variant in MODULE_VARIANTS + for aspect in ASPECTS) + for method in METHOD_REGISTRARS: + for code in method.codes: + member = code.rpartition('.')[2] + labels.extend(f'{method.name}:{member}/{variant}/{aspect}' for variant in METHOD_VARIANTS + for aspect in ASPECTS) + labels.extend(f'register_protocol/HTTP/{aspect}' for aspect in ASPECTS) + return labels + + +def _steps(label: 'str') -> 'list[Callable[[], None]]': + head, variant, _ = label.split('/') + if head == 'register_protocol': + from pcapkit.foundation.registry import register_protocol + from pcapkit.protocols import __proto__ as names + from pcapkit.protocols.application import httpv2 + + shipped = names['HTTP'] + return [lambda: register_protocol(httpv2.HTTP), lambda: register_protocol(shipped)] + name, _, member = head.partition(':') + for module in MODULE_REGISTRARS: + if module.name == name: + return _module_steps(module, variant) + for method in METHOD_REGISTRARS: + if method.name == name: + code, = (code for code in method.codes if code.rpartition('.')[2] == member) + return _method_steps(method, code, variant) + raise AssertionError(label) + + +def run_case(label: 'str') -> 'Outcome': + """Snapshot, run the case's calls, compare one aspect, and put everything back.""" + aspect = label.rpartition('/')[2] + before = _snapshot() + try: + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + try: + for step in _steps(label): + step() + except Exception as exc: # pylint: disable=broad-except + return Outcome('RAISED', f'{type(exc).__name__}: {exc}') + diff = _diff(before, aspect) + finally: + _restore(before) + if diff: + return Outcome('CHANGED', '; '.join(diff)) + return Outcome('OK') + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class TestRegistrySymmetry(harness.RoundTripBase): + """Override then restore leaves every registry exactly as shipped.""" + + STATUSES = ('RAISED', 'CHANGED') + KNOWN_FAILURES = ( + Gap(1504, 'restoring a shipped dispatch entry over an override stores the class it names, ' + 'not the shipped ModuleDescriptor: each register resolves a descriptor unequal ' + 'to the incumbent (pcapkit/protocols/protocol.py:924, link/link.py:148, ' + 'internet/internet.py:168, transport/transport.py:122, misc/pcap/frame.py:155, ' + 'misc/pcapng.py:1027, transport/sctp.py:748), where the foundation registrars ' + 'keep the shipped object (#1363, #1364)', + 'CHANGED', ("ModuleDescriptor(module='pcapkit.protocols.",), + tuple(f'{name}/{variant}/dispatch' + for name in ('register_linktype', 'register_pcap', 'register_pcapng', + 'register_ethertype', 'register_transtype', 'register_tcp', + 'register_udp', 'register_apptype', 'register_sctp') + for variant in ('override', 'override-same-name', 'restore-by-name'))), + Gap(1505, 'a protocol registrar adds the class it registers to pcapkit.protocols.__proto__ ' + 'by name (pcapkit/foundation/registry/protocols.py:235, reached from e.g. :416), ' + 'and restoring the code leaves that name behind: nothing removes it', + 'CHANGED', "names['ROUNDTRIPCUSTOM'] added", + tuple(f'{name}/{variant}/names' + for name in ('register_linktype', 'register_pcap', 'register_pcapng', + 'register_ethertype', 'register_transtype', 'register_tcp', + 'register_udp', 'register_apptype', 'register_sctp') + for variant in ('override', 'restore-by-name'))), + ) + + def setUp(self) -> None: + super().setUp() + reimport_once_per_class(self) + + def labels(self) -> 'list[str]': + return _labels() + + def outcome(self, label: 'str') -> 'Outcome': + return run_case(label) + + def test_every_public_registrar_is_covered_or_excused(self) -> None: + """A registrar added tomorrow needs a case, or a reason it has none.""" + from pcapkit.foundation import registry + + covered = {r.name for r in MODULE_REGISTRARS} | {r.name for r in METHOD_REGISTRARS} + excused = { + 'register_protocol': 'covered by the register_protocol/HTTP case', + 'register_protocol_code': 'funnels into ProtocolBase.register, as register_tcp does', + 'register_reassembly_ipv4_callback': 'append-only, nothing to undo with', + 'register_reassembly_ipv6_callback': 'append-only, nothing to undo with', + 'register_reassembly_tcp_callback': 'append-only, nothing to undo with', + 'register_traceflow_tcp_callback': 'append-only, nothing to undo with', + } + self.assertEqual(sorted(set(registry.__all__) - covered - set(excused)), []) + self.assertEqual(sorted((covered | set(excused)) - set(registry.__all__)), []) + + def test_case_leaves_registries_as_found(self) -> None: + """The harness's own restore puts back what each case changed.""" + before = _snapshot() + for label in self.labels()[:8]: + run_case(label) + self.assertEqual(_diff(before, 'dispatch') + _diff(before, 'names'), []) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/foundation/traceflow/test_traceflow_reread_roundtrip_runtime.py b/tests/foundation/traceflow/test_traceflow_reread_roundtrip_runtime.py new file mode 100644 index 0000000000..26a918b507 --- /dev/null +++ b/tests/foundation/traceflow/test_traceflow_reread_roundtrip_runtime.py @@ -0,0 +1,124 @@ +# -*- coding: utf-8 -*- +"""Flow tracing over every sample capture reads back as exactly its frames. C.f. #1202. + +The capture-driven twin of :mod:`tests.foundation.traceflow.\ +test_traceflow_reread_roundtrip_unit`: every capture under +:file:`examples/captures/` is traced to PCAP, and each flow's file is read back +with :mod:`struct` alone and compared, record by record, with the frames the +flow's ``index`` names, as :mod:`pcapkit` read them from the input -- the +timestamp exactly (as a :class:`~decimal.Decimal`), the captured and original +lengths, and the octets. + +A PCAP input is traced in its own byte order and resolution, so its records +must come back byte for byte. A PCAP-NG input is traced in nanoseconds, which +holds every timestamp the sample interfaces record. + +This module reads generated captures, so it is in the fixture-dependent tier +(see :mod:`tests._tiers`) and runs after ``make samples``. + +""" + +from __future__ import annotations + +import decimal +import importlib.util +import os +import tempfile +import unittest +import warnings + +from tests._support import close_extractor, reimport_once_per_class, sample_path +from tests._tiers import SAMPLE_ROOT +from tests.foundation import _roundtrip as harness +from tests.foundation._roundtrip import Gap, Outcome, read_pcap + +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None + for name in ('tbtrim', 'aenum', 'chardet', 'dictdumper')) + +#: Every capture on disk, enumerated rather than hand-picked. +CAPTURES = sorted(name for name in os.listdir(SAMPLE_ROOT) + if name.endswith(('.pcap', '.cap', '.pcapng'))) + + +def _record(frame: 'object') -> 'tuple[decimal.Decimal, int, int, bytes]': + """``(timestamp, captured length, original length, octets)`` of an input frame.""" + info = frame.info # type: ignore[attr-defined] + if 'time_epoch' in info: + return (decimal.Decimal(info.time_epoch), info.cap_len, info.len, bytes(info.packet)) + return (decimal.Decimal(info.timestamp_epoch), info.captured_len, info.original_len, + bytes(info.packet)) + + +def run_capture(name: 'str') -> 'Outcome': + """Trace one capture to PCAP and read every flow back.""" + from pcapkit import extract + + path = sample_path(name) + with open(path, 'rb') as file: + head = file.read(24) + pcapng = head[:4] == b'\x0a\x0d\x0d\x0a' + if pcapng: + order, nsec, linktypes = 'little', True, None + else: + reference = read_pcap(head) + order, nsec, linktypes = reference.byteorder, reference.nanosecond, reference.linktype + + with warnings.catch_warnings(), tempfile.TemporaryDirectory() as tracedir: + warnings.simplefilter('ignore') + try: + extractor = extract(fin=path, nofile=True, store=True, trace=True, tcp=True, + trace_fout=tracedir, trace_format='pcap', + trace_byteorder=order, trace_nanosecond=nsec) + except Exception as exc: # pylint: disable=broad-except + return Outcome('ERROR', f'{type(exc).__name__}: {exc}') + close_extractor(extractor) + frames = {frame.info.number: frame for frame in extractor.frame} + scale = decimal.Decimal(10) ** (9 if nsec else 6) + for flow in extractor.trace.tcp: + with open(flow.fpout, 'rb') as file: + written = read_pcap(file.read()) + if (written.byteorder, written.nanosecond) != (order, nsec): + return Outcome('HEADER', f'{flow.label}: {(written.byteorder, written.nanosecond)} ' + f'!= {(order, nsec)}') + if linktypes is not None and written.linktype != linktypes: + return Outcome('HEADER', f'{flow.label}: linktype {written.linktype} != {linktypes}') + if len(written.records) != len(flow.index): + return Outcome('COUNT', f'{flow.label}: {len(written.records)} records != ' + f'{len(flow.index)} indexed') + for (sec, frac, incl, orig, octets), number in zip(written.records, flow.index): + got = (decimal.Decimal(sec) + decimal.Decimal(frac) / scale, incl, orig, octets) + want = _record(frames[number]) + if got != want: + fields = ('timestamp', 'incl_len', 'orig_len', 'octets') + diff = '; '.join(f'{field} {g!r:.60} != {w!r:.60}' for field, g, w + in zip(fields, got, want) if g != w) + return Outcome('RECORD', f'{flow.label} frame {number}: {diff}') + return Outcome('OK') + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class TestTraceFlowRereadRuntime(harness.RoundTripBase): + """Every sample capture's flows read back as exactly their frames.""" + + STATUSES = ('ERROR', 'HEADER', 'COUNT', 'RECORD') + KNOWN_FAILURES = ( + Gap(1500, 'a microsecond PCAP-NG interface traced to a nanosecond PCAP keeps its ' + 'microsecond fraction under the nanosecond magic: PCAPIO._append_value writes ' + 'frame_info.ts_usec unscaled (pcapkit/dumpkit/pcap.py:206-207) and tcp_traceflow ' + "converts at the block's own resolution (pcapkit/foundation/engines/pcapng.py:304)", + 'RECORD', 'timestamp ', ('many_interfaces.pcapng',)), + ) + + def setUp(self) -> None: + super().setUp() + reimport_once_per_class(self) + + def labels(self) -> 'list[str]': + return list(CAPTURES) + + def outcome(self, label: 'str') -> 'Outcome': + return run_capture(label) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/foundation/traceflow/test_traceflow_reread_roundtrip_unit.py b/tests/foundation/traceflow/test_traceflow_reread_roundtrip_unit.py new file mode 100644 index 0000000000..8b838595d2 --- /dev/null +++ b/tests/foundation/traceflow/test_traceflow_reread_roundtrip_unit.py @@ -0,0 +1,253 @@ +# -*- coding: utf-8 -*- +"""Flow tracing writes each flow to a file that reads back as exactly its frames. C.f. #1202. + +Each case builds a capture of TCP conversations as an in-memory savefile, +traces it through :func:`~pcapkit.interface.extract` with +``trace_format='pcap'``, and reads every flow's output file back twice: with +:mod:`struct` alone, as an independent reference, and with :mod:`pcapkit`. The +round trip closes when + +* each output file's global header carries the byte order, timestamp + resolution and link type asked for; +* its records are, in order, the input records the flow's ``index`` names -- + the same seconds, fraction, captured and original length, and octets; +* :mod:`pcapkit` reads it back as that many frames; and +* every TCP frame of the input lands in exactly one flow, and no other frame + lands in any. + +The conversations cover the shapes :mod:`pcapkit.foundation.traceflow.tcp` +distinguishes: a whole connection, two interleaved, one mixed with UDP, one over +IPv6, a port pair reused after a close, a reset, one left half-open, and the +per-direction ``trace_bidirectional=False`` mode. Each runs in every byte order +and resolution the PCAP writer offers. + +Every case builds its own octets and reads no capture. Classes are imported +inside each test, after :func:`~tests._support.reimport_once_per_class`. + +""" + +from __future__ import annotations + +import importlib.util +import io +import os +import tempfile +import unittest +import warnings +from typing import TYPE_CHECKING, NamedTuple + +from tests._support import close_extractor, reimport_once_per_class +from tests.foundation import _roundtrip as harness +from tests.foundation._roundtrip import (Gap, Outcome, ethernet, ipv4, + ipv6, pcap, pcapng, read_pcap, tcp, udp) + +if TYPE_CHECKING: + from typing import Callable + +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None + for name in ('tbtrim', 'aenum', 'chardet', 'dictdumper')) + + +def _segment(payload: 'bytes', *, reply: 'bool', v6: 'bool', port: 'int', **flags: 'int | bool') -> 'bytes': + """One TCP segment between the two built hosts, from the client unless ``reply``.""" + sport, dport = (9, port) if reply else (port, 9) + seg = tcp(payload, sport=sport, dport=dport, **flags) # type: ignore[arg-type] + if v6: + return ethernet(ipv6(seg, nxt=6, reverse=reply), 0x86DD, reverse=reply) + return ethernet(ipv4(seg, proto=6, reverse=reply), 0x0800, reverse=reply) + + +def connection(port: 'int' = 40000, *, v6: 'bool' = False, close: 'str' = 'fin', + isn: 'int' = 1000) -> 'list[bytes]': + """A client-server exchange: handshake, a request, a response, and a close. + + ``close`` is ``'fin'`` for the four-way close, ``'rst'`` for a reset, or + ``'none'`` to leave the connection open. + + """ + srv = isn + 50000 + req, rsp = b'GET / HTTP/1.0\r\n\r\n', b'HTTP/1.0 200 OK\r\n\r\nhello' + out = [ + _segment(b'', reply=False, v6=v6, port=port, seq=isn, syn=True), + _segment(b'', reply=True, v6=v6, port=port, seq=srv, ack=isn + 1, syn=True), + _segment(req, reply=False, v6=v6, port=port, seq=isn + 1, ack=srv + 1), + _segment(rsp, reply=True, v6=v6, port=port, seq=srv + 1, ack=isn + 1 + len(req)), + ] + cseq, sseq = isn + 1 + len(req), srv + 1 + len(rsp) + if close == 'fin': + out += [ + _segment(b'', reply=False, v6=v6, port=port, seq=cseq, ack=sseq, fin=True), + _segment(b'', reply=True, v6=v6, port=port, seq=sseq, ack=cseq + 1, fin=True), + _segment(b'', reply=False, v6=v6, port=port, seq=cseq + 1, ack=sseq + 1), + ] + elif close == 'rst': + out.append(_segment(b'', reply=False, v6=v6, port=port, seq=cseq, ack=sseq, rst=True)) + return out + + +def _interleave(*streams: 'list[bytes]') -> 'list[bytes]': + out = [] # type: list[bytes] + for index in range(max(len(stream) for stream in streams)): + out.extend(stream[index] for stream in streams if index < len(stream)) + return out + + +def _mixed() -> 'list[bytes]': + frames = connection() + noise = [ethernet(ipv4(udp(b'noise %d' % i)), 0x0800) for i in range(3)] + return [frames[0], noise[0], *frames[1:4], noise[1], *frames[4:], noise[2]] + + +#: Scenario name -> (frames, bidirectional). +SCENARIOS = { + 'one-connection': (lambda: connection(), True), + 'two-interleaved': (lambda: _interleave(connection(40000), connection(40001, isn=7000)), True), + 'mixed-udp': (_mixed, True), + 'ipv6': (lambda: connection(v6=True), True), + 'reused-port': (lambda: connection() + connection(isn=90000), True), + 'reset': (lambda: connection(close='rst'), True), + 'half-open': (lambda: connection(close='none'), True), + 'unidirectional': (lambda: connection(), False), +} # type: dict[str, tuple[Callable[[], list[bytes]], bool]] + +#: Variant name -> (container, input byte order, input nanosecond, output byte +#: order, output nanosecond). A coarser output keeps the whole microseconds. +VARIANTS = { + 'pcap/little-usec': ('pcap', 'little', False, 'little', False), + 'pcap/big-usec': ('pcap', 'big', False, 'big', False), + 'pcap/little-nsec': ('pcap', 'little', True, 'little', True), + 'pcap/big-nsec': ('pcap', 'big', True, 'big', True), + 'pcap/big-to-little': ('pcap', 'big', False, 'little', False), + 'pcap/usec-to-nsec': ('pcap', 'little', False, 'little', True), + 'pcap/nsec-to-usec': ('pcap', 'little', True, 'little', False), + 'pcapng/usec': ('pcapng', 'little', False, 'little', False), + 'pcapng/nsec': ('pcapng', 'little', True, 'little', True), + 'pcapng/big-usec': ('pcapng', 'big', False, 'big', False), + 'pcapng/usec-to-nsec': ('pcapng', 'little', False, 'little', True), + 'pcapng/nsec-to-usec': ('pcapng', 'little', True, 'little', False), +} + + +class Case(NamedTuple): + label: 'str' + frames: 'tuple[bytes, ...]' + bidirectional: 'bool' + variant: 'tuple[str, str, bool, str, bool]' + + +def _rescale(fraction: 'int', in_nsec: 'bool', out_nsec: 'bool') -> 'int': + """A timestamp fraction in the output's resolution.""" + if in_nsec == out_nsec: + return fraction + return fraction * 1000 if out_nsec else fraction // 1000 + + +CASES = {f'{scenario}/{variant}': Case(f'{scenario}/{variant}', tuple(build()), bidirectional, + VARIANTS[variant]) + for scenario, (build, bidirectional) in SCENARIOS.items() for variant in VARIANTS} + + +def _is_tcp(frame: 'bytes') -> 'bool': + ethertype = int.from_bytes(frame[12:14], 'big') + return (ethertype == 0x0800 and frame[14 + 9] == 6) or (ethertype == 0x86DD and frame[14 + 6] == 6) + + +def run_case(case: 'Case') -> 'Outcome': + """Trace ``case`` to PCAP, then read every flow's file back.""" + from pcapkit import extract + + container, in_order, in_nsec, out_order, out_nsec = case.variant + # the fraction is distinct per frame and, in nanoseconds, finer than a microsecond + step = 1001 if in_nsec else 1 + records = [(1_500_000_000 + i, (i * 7919 * step) % (10 ** 9 if in_nsec else 10 ** 6), frame) + for i, frame in enumerate(case.frames)] + build = pcap if container == 'pcap' else pcapng + data = build(records, nanosecond=in_nsec, byteorder=in_order) + source = [(sec, _rescale(frac, in_nsec, out_nsec), len(frame), len(frame), frame) + for sec, frac, frame in records] + + with warnings.catch_warnings(), tempfile.TemporaryDirectory() as tracedir: + warnings.simplefilter('ignore') + try: + extractor = extract(fin=io.BytesIO(data), nofile=True, store=False, trace=True, tcp=True, + trace_fout=tracedir, trace_format='pcap', + trace_byteorder=out_order, trace_nanosecond=out_nsec, + trace_bidirectional=case.bidirectional) + except Exception as exc: # pylint: disable=broad-except + return Outcome('ERROR', f'{type(exc).__name__}: {exc}') + close_extractor(extractor) + flows = list(extractor.trace.tcp) + + seen = [] # type: list[int] + for flow in flows: + seen.extend(flow.index) + if not os.path.isfile(flow.fpout): + return Outcome('FILE', f'{flow.label}: no file at {flow.fpout}') + with open(flow.fpout, 'rb') as file: + written = read_pcap(file.read()) + header = (written.byteorder, written.nanosecond, written.linktype) + if header != (out_order, out_nsec, 1): + return Outcome('HEADER', f'{flow.label}: header {header} != {(out_order, out_nsec, 1)}') + want = tuple(source[number - 1] for number in flow.index) + if len(written.records) != len(want): + return Outcome('COUNT', f'{flow.label}: {len(written.records)} records != ' + f'{len(want)} indexed {flow.index}') + for position, (got, expected) in enumerate(zip(written.records, want)): + if got != expected: + fields = ('seconds', 'fraction', 'incl_len', 'orig_len', 'octets') + diff = '; '.join(f'{name} {g!r:.60} != {e!r:.60}' for name, g, e + in zip(fields, got, expected) if g != e) + return Outcome('RECORD', f'{flow.label} record {position + 1} ' + f'(frame {flow.index[position]}): {diff}') + try: + reread = extract(fin=flow.fpout, nofile=True, store=False) + except Exception as exc: # pylint: disable=broad-except + return Outcome('REREAD', f'{flow.label}: {type(exc).__name__}: {exc}') + close_extractor(reread) + if reread.length != len(want): + return Outcome('REREAD', f'{flow.label}: pcapkit reads {reread.length} frames != {len(want)}') + + tcp_frames = [number for number, frame in enumerate(case.frames, 1) if _is_tcp(frame)] + if sorted(seen) != tcp_frames: + return Outcome('COVER', f'flows index {sorted(seen)} != TCP frames {tcp_frames}') + return Outcome('OK') + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class TestTraceFlowRereadRoundTrip(harness.RoundTripBase): + """Every scenario and variant reads back as exactly its flows' frames.""" + + STATUSES = ('ERROR', 'FILE', 'HEADER', 'COUNT', 'RECORD', 'REREAD', 'COVER') + KNOWN_FAILURES = ( + Gap(1500, 'a flow traced at another timestamp resolution than its input keeps the ' + "input's fraction under the output's magic: PCAPIO._append_value writes " + 'frame_info.ts_usec unscaled (pcapkit/dumpkit/pcap.py:206-207), and for PCAP-NG ' + 'input tcp_traceflow converts at the block\'s own resolution ' + '(pcapkit/foundation/engines/pcapng.py:304)', + 'RECORD', 'fraction ', ('*/usec-to-nsec', '*/nsec-to-usec')), + ) + + def setUp(self) -> None: + super().setUp() + reimport_once_per_class(self) + + def labels(self) -> 'list[str]': + return list(CASES) + + def outcome(self, label: 'str') -> 'Outcome': + return run_case(CASES[label]) + + def test_reference_reader_reads_the_builder(self) -> None: + """:func:`read_pcap` and :func:`pcap` agree, so the reference is sound.""" + frames = connection() + for order in ('little', 'big'): + for nsec in (False, True): + with self.subTest(order=order, nsec=nsec): + records = [(1, 999_999_999 if nsec else 999_999, frame) for frame in frames] + saved = read_pcap(pcap(records, byteorder=order, nanosecond=nsec)) + self.assertEqual((saved.byteorder, saved.nanosecond, saved.linktype), (order, nsec, 1)) + self.assertEqual(saved.records, tuple((s, f, len(b), len(b), b) for s, f, b in records)) + + +if __name__ == '__main__': + unittest.main()