diff --git a/tests/foundation/_roundtrip.py b/tests/foundation/_roundtrip.py index 8f68a2cf5d..a7ceadffa1 100644 --- a/tests/foundation/_roundtrip.py +++ b/tests/foundation/_roundtrip.py @@ -2,7 +2,7 @@ """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 +matches exactly. Three 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 @@ -10,8 +10,6 @@ * 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`). diff --git a/tests/foundation/engines/test_engine_agreement_retry_unit.py b/tests/foundation/engines/test_engine_agreement_retry_unit.py deleted file mode 100644 index d42b0502ea..0000000000 --- a/tests/foundation/engines/test_engine_agreement_retry_unit.py +++ /dev/null @@ -1,91 +0,0 @@ -# -*- coding: utf-8 -*- -"""The engine agreement module retries a pyshark run once on a tshark crash. - -GitHub issue #1534: tshark now and then exits 255 on a capture it reads cleanly -on a rerun, and pyshark raises ``TSharkCrashException``. The agreement module -(:file:`test_engine_agreement_runtime.py`) caches one run per engine and -capture, so that one crash failed all nine pyshark aspects of the capture. - -Each case replaces :func:`pcapkit.extract` with a stand-in, so nothing here -reads a capture or needs pyshark; the exception is matched by class name, so a -class of the same name stands in for pyshark's. - -""" - -import types -import unittest -import warnings -from unittest import mock - -from tests._support import reimport_once_per_class -from tests.foundation._roundtrip import Outcome -from tests.foundation.engines import test_engine_agreement_runtime as agreement - -#: A capture git tracks, so the unit-tier guard lets its name through. -CAPTURE = 'in.pcap' - - -class TSharkCrashException(Exception): - """Stands in for ``pyshark.capture.capture.TSharkCrashException``.""" - - -@unittest.skipUnless(agreement.HAS_RUNTIME, 'runtime dependencies not installed') -class TestTSharkCrashRetry(unittest.TestCase): - """Pin how the agreement module's runner handles a pyshark run that raises.""" - - def setUp(self) -> None: - reimport_once_per_class(self) - - def _compare(self, *errors: 'Exception') -> 'tuple[Outcome, int, list[str]]': - """``compare`` the pyshark ``frames`` aspect, its runs raising ``errors`` in turn. - - Returns the outcome, the number of pyshark runs, and every warning raised. - - """ - import pcapkit - - pending = list(errors) - calls = [] # type: list[str] - - def extract(**kwargs: 'object') -> 'object': - engine = kwargs['engine'] - calls.append(engine) - if engine == 'pyshark' and pending: - raise pending.pop(0) - return types.SimpleNamespace(_exnam=engine, length=2, frame=(), - reassembly=None, trace=None) - - with mock.patch.dict(agreement._RUNS, clear=True), \ - mock.patch.object(pcapkit, 'extract', extract), \ - warnings.catch_warnings(record=True) as caught: - warnings.simplefilter('always') - outcome = agreement.compare('pyshark', CAPTURE, 'frames') - return outcome, calls.count('pyshark'), [str(item.message) for item in caught] - - def test_one_crash_is_retried_and_logged(self) -> None: - crash = TSharkCrashException('TShark (pid 1) seems to have crashed (retcode: 255).\n' - 'Last error line: None') - outcome, runs, logged = self._compare(crash) - self.assertEqual(outcome, Outcome('OK')) - self.assertEqual(runs, 2) - self.assertEqual(logged, [f'pyshark on {CAPTURE}: TSharkCrashException: TShark (pid 1) ' - 'seems to have crashed (retcode: 255). Retrying once (#1534).']) - - def test_a_crash_that_persists_is_an_error(self) -> None: - crashes = [TSharkCrashException(f'TShark (pid {pid}) seems to have crashed (retcode: 255).') - for pid in (1, 2)] - outcome, runs, logged = self._compare(*crashes) - self.assertEqual(outcome, Outcome('ERROR', 'TSharkCrashException: TShark (pid 2) seems ' - 'to have crashed (retcode: 255).')) - self.assertEqual(runs, 2) - self.assertEqual(len(logged), 1) - - def test_any_other_error_is_not_retried(self) -> None: - outcome, runs, logged = self._compare(ValueError('invalid literal')) - self.assertEqual(outcome, Outcome('ERROR', 'ValueError: invalid literal')) - self.assertEqual(runs, 1) - self.assertEqual(logged, []) - - -if __name__ == '__main__': - unittest.main() diff --git a/tests/foundation/engines/test_engine_agreement_runtime.py b/tests/foundation/engines/test_engine_agreement_runtime.py deleted file mode 100644 index 74586500fd..0000000000 --- a/tests/foundation/engines/test_engine_agreement_runtime.py +++ /dev/null @@ -1,389 +0,0 @@ -# -*- 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 except ``frame``, which - is the engine's own frame object, and except ``header`` and ``payload`` for - an engine in :data:`NO_TCP_OCTETS`; -``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 unless - the engine is in :data:`JSON_TRACE`. - -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). ``pypcapfile`` also warns when a capture carries TCP over -IPv6, which it leaves out of TCP reassembly and flow tracing: on such a capture -its ``tcp`` and ``trace`` inputs and TCP ``datagrams`` are compared with the -``default`` engine's over IPv4 only, and its ``flows`` skip. 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 ipaddress -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', -} -#: Engines whose flow-tracing input carries no TCP octets, and why. Their ``trace`` -#: comparison leaves out ``header`` and ``payload`` and compares every other field. -NO_TCP_OCTETS = { - 'pyshark': 'pyshark reports dissected fields, not the octets behind them, so tcp_traceflow ' - "hands the tracer empty ones and Extractor refuses trace_analyse for it " - '(pcapkit/foundation/extraction.py:1446-1455; ruling on #1507)', -} -#: Engines whose flow files are JSON whatever ``trace_format`` asks for, and why. Their -#: ``flows`` comparison checks each flow's label and frame indices, not the file's octets. -JSON_TRACE = { - 'pyshark': 'pyshark cannot supply the frame bytes a PCAP trace needs, so Extractor writes ' - "JSON with a FormatWarning (pcapkit/foundation/extraction.py:1440-1444; ruling " - 'on #1507)', -} -#: 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 AttributeWarning, EngineWarning - - self.fallback = None # type: Optional[str] - #: The ``pypcapfile`` toolkit's warning that it leaves TCP over IPv6 out. - self.tcp_declined = 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, \ - mock.patch('pcapkit.toolkit.pypcapfile.warn') as toolkit_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.tcp_declined = '; '.join(self._fallen(toolkit_warn, AttributeWarning)) or None - 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] -#: What pyshark raises when tshark exits non-zero, by class name, so this module -#: imports without pyshark. -TSHARK_CRASH = 'TSharkCrashException' - - -def _run(engine: 'str', capture: 'str') -> '_Run': - key = (engine, capture) - if key not in _RUNS: - run = _Run(engine, capture) - # tshark sometimes exits 255 on a capture it reads cleanly on a rerun - # (#1534), and the run is cached, so one crash fails every aspect. Retry - # once, visibly; a crash that persists stays an ERROR. - if run.error is not None and run.error.split(':', 1)[0] == TSHARK_CRASH: - warnings.warn(f'{engine} on {capture}: {run.error.splitlines()[0]} ' - 'Retrying once (#1534).', RuntimeWarning) - run = _Run(engine, capture) - _RUNS[key] = run - return _RUNS[key] - - -def _without_tcp_octets(inputs: 'dict[int, Any]') -> 'dict[int, Any]': - """``trace`` inputs without ``header`` and ``payload``, per :data:`NO_TCP_OCTETS`.""" - return {number: [{key: value for key, value in packet.items() if key not in ('header', 'payload')} - for packet in packets] for number, packets in inputs.items()} - - -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 _over_ipv4(record: 'dict[str, Any]') -> 'bool': - """Whether a TCP trace or reassembly input, or a TCP datagram, is over IPv4.""" - if 'src' in record: - address = record['src'] - elif 'bufid' in record: - address = record['bufid'][0] - else: - address = record['id'].src[0] - return isinstance(address, ipaddress.IPv4Address) - - -def _ipv4_only(base: '_Run', aspect: 'str') -> 'Any': - """``base``'s ``tcp`` or ``trace`` inputs, or its TCP datagrams, over IPv4 only.""" - if aspect == 'datagrams': - return [datagram for datagram in base.datagrams['tcp'] if _over_ipv4(datagram)] - return {number: records for number, records in base.inputs[aspect].items() - if _over_ipv4(records[0])} - - -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) - if run.tcp_declined is not None and aspect == 'flows': # the rest narrow to IPv4 below - return Outcome('SKIP', f'{engine} warns: {run.tcp_declined}') - 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: - mine, theirs = run.inputs[aspect], base.inputs[aspect] - if run.tcp_declined is not None and aspect in ('tcp', 'trace'): - theirs = _ipv4_only(base, aspect) - if aspect == 'trace' and engine in NO_TCP_OCTETS: - mine, theirs = _without_tcp_octets(mine), _without_tcp_octets(theirs) - diff = _first_difference(mine, theirs) - 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 kind == 'tcp' and run.tcp_declined is not None: - theirs = _ipv4_only(base, 'datagrams') - 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}') - if engine in JSON_TRACE: - return Outcome('OK') - 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 pypcapfile engine writes every flow as JSON when asked for " - "trace_format='pcap': its tcp_traceflow adapter builds the frame with " - 'packet2dict (pcapkit/toolkit/pypcapfile.py:620), which PCAPIO cannot write, so ' - 'Extractor swaps the format with a FormatWarning ' - '(pcapkit/foundation/extraction.py:1439-1443)', - 'FLOWS', 'file of ', tuple(f'{capture}/pypcapfile/flows' for capture in TRACED_IPV4_ONLY_PCAP)), - ) - - 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/engines/test_ext_chain_limit_agreement_1604_unit.py b/tests/foundation/engines/test_ext_chain_limit_agreement_1604_unit.py deleted file mode 100644 index c88f75058b..0000000000 --- a/tests/foundation/engines/test_ext_chain_limit_agreement_1604_unit.py +++ /dev/null @@ -1,230 +0,0 @@ -# -*- coding: utf-8 -*- -"""Every engine stops a chain of IPv6 extension headers where ``default`` does. C.f. #1604. - -The ``default`` engine dissects at most -:data:`~pcapkit.protocols.internet.internet.EXTENSION_HEADER_LIMIT` extension -headers in a row, and leaves the rest raw, so the TCP behind a longer chain is -not found. Each engine here reads one capture per chain length -- 30, 32, 33 and -1,000 Destination Options headers, then a TCP segment -- carried in IPv4 and in -IPv6, with TCP reassembly and flow tracing on, and its ``tcp`` and ``trace`` -inputs are compared with ``default``'s, as in -:mod:`tests.foundation.engines.test_engine_agreement_runtime`. - -``pypcapfile`` walks the chain itself and stops at the same depth. It leaves TCP -over IPv6 out, so it is compared over IPv4 only, and ``declined`` checks that -it warns of leaving out TCP over IPv6 exactly where ``default`` finds some. It -imports on Python 3.11 only, and is skipped elsewhere. ``dpkt``'s adapters walk -the chain off the octets, and stop at the same depth too. ``scapy`` walks the -chain over the octets by the default engine's rules, rather than through Scapy's -own layers, so stops there as well (#1609). - -``pyshark`` reads a frame tshark finds TCP in from its octets, by the default -engine's rules, so stops there too (#1639). Measured on tshark 4.6.9. - -Every capture is built in a temporary directory, so no fixture is read. - -""" - -from __future__ import annotations - -import importlib.util -import ipaddress -import os -import tempfile -import unittest -import warnings -from typing import TYPE_CHECKING -from unittest import mock - -from tests._support import close_extractor, reimport_once_per_class -from tests.foundation import _roundtrip as 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')) - -#: Chain lengths: under the limit, at it, one past it, and far past it. -CHAINS = (30, 32, 33, 1000) -#: What carries the chain. -CARRIERS = ('ipv4', 'ipv6') -#: Every engine compared with ``default``, as :attr:`Extractor.__engine__ -#: ` names them. ``pypcap`` -#: and ``pcap_ct`` decline both reassembly and flow tracing, so are left out. -ENGINES = ('dpkt', 'scapy', 'pyshark', 'pypcapfile') -#: Engines whose flow-tracing input carries no TCP octets, c.f. -#: :data:`tests.foundation.engines.test_engine_agreement_runtime.NO_TCP_OCTETS`. -NO_TCP_OCTETS = ('pyshark',) - - -def capture(carrier: 'str', count: 'int') -> 'bytes': - """A PCAP savefile of one frame: ``count`` Destination Options headers, then TCP.""" - headers = b''.join(bytes([60 if index < count - 1 else 6, 0, 1, 4, 0, 0, 0, 0]) - for index in range(count)) - payload = headers + harness.tcp(b'payload', seq=1) - if carrier == 'ipv4': - frame = harness.ethernet(harness.ipv4(payload, proto=60), 0x0800) - else: - frame = harness.ethernet(harness.ipv6(payload, nxt=60), 0x86DD) - return harness.pcap([(1, 0, frame)]) - - -def _fields(packet: 'Any', skip: 'tuple[str, ...]' = ()) -> 'dict[str, Any]': - return {key: bytes(value) if isinstance(value, bytearray) else value - for key, value in packet.items() if key not in skip} - - -class _Run: - """One engine's reading of one capture.""" - - def __init__(self, engine: 'str', path: 'str') -> None: - from pcapkit import extract - from pcapkit.foundation.reassembly.reassembly import ReassemblyBase - from pcapkit.foundation.traceflow.traceflow import TraceFlowBase - from pcapkit.utilities.exceptions import UnsupportedCall - from pcapkit.utilities.warnings import AttributeWarning, EngineWarning - - self.error = self.fallback = None # type: Optional[str] - #: What the ``pypcapfile`` toolkit warned it leaves out, if anything. - self.declined = None # type: Optional[str] - #: Whether the engine declines TCP reassembly, or flow tracing. - self.no_tcp = self.no_trace = False - self.inputs = {'tcp': {}, 'trace': {}} # type: dict[str, dict[int, list[dict[str, Any]]]] - inputs = self.inputs - reassemble, trace = ReassemblyBase.__call__, TraceFlowBase.__call__ - - def spy_reassembly(this: 'Any', packet: 'Any') -> 'None': - inputs['tcp'].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, \ - mock.patch('pcapkit.toolkit.pypcapfile.warn') as toolkit_warn: - warnings.simplefilter('ignore') - try: - extractor = extract(fin=path, nofile=True, engine=engine, reassembly=True, - tcp=True, trace=True, trace_fout=tracedir, trace_format='json') - except Exception as exc: # pylint: disable=broad-except - self.error = f'{type(exc).__name__}: {exc}' - return - close_extractor(extractor) - fallen = [str(call.args[0]) for call in warn.call_args_list - if len(call.args) > 1 and call.args[1] is 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.declined = '; '.join(str(call.args[0]) for call in toolkit_warn.call_args_list - if len(call.args) > 1 and call.args[1] is AttributeWarning - and 'is left out of TCP reassembly' in str(call.args[0])) or None - for name in ('reassembly', 'trace'): - try: - found = getattr(getattr(extractor, name), 'tcp', None) - except UnsupportedCall: - found = None - setattr(self, 'no_tcp' if name == 'reassembly' else 'no_trace', found is None) - - -_RUNS = {} # type: dict[tuple[str, str], _Run] - - -def _over_ipv4(record: 'dict[str, Any]') -> 'bool': - address = record['bufid'][0] if 'bufid' in record else record['src'] - return isinstance(address, ipaddress.IPv4Address) - - -def _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]: - return f'frame {number}: {got[number]!r:.120} != {want[number]!r:.120}' - return None - - -@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') -class TestExtensionHeaderChainAgreement(harness.RoundTripBase): - """Every engine that runs stops the chain where ``default`` does, or is a recorded gap.""" - - STATUSES = ('ERROR', 'INPUTS', 'DECLINED') - KNOWN_FAILURES = () # type: tuple[Gap, ...] - - @classmethod - def setUpClass(cls) -> None: - super().setUpClass() - cls.tmp = tempfile.TemporaryDirectory() # pylint: disable=consider-using-with - for carrier in CARRIERS: - for count in CHAINS: - with open(cls.path(carrier, count), 'wb') as file: - file.write(capture(carrier, count)) - - @classmethod - def tearDownClass(cls) -> None: - _RUNS.clear() - cls.tmp.cleanup() - super().tearDownClass() - - @classmethod - def path(cls, carrier: 'str', count: 'int') -> 'str': - return os.path.join(cls.tmp.name, f'{carrier}-{count}.pcap') - - def setUp(self) -> None: - super().setUp() - reimport_once_per_class(self) - - def labels(self) -> 'list[str]': - return [f'{carrier}-{count}/{engine}/{aspect}' - for carrier in CARRIERS for count in CHAINS for engine in ENGINES - for aspect in ('tcp', 'trace', 'declined') - if aspect != 'declined' or engine == 'pypcapfile'] - - def run_of(self, engine: 'str', name: 'str') -> '_Run': - if (engine, name) not in _RUNS: - carrier, count = name.split('-') - _RUNS[engine, name] = _Run(engine, self.path(carrier, int(count))) - return _RUNS[engine, name] - - def outcome(self, label: 'str') -> 'Outcome': - name, engine, aspect = label.split('/') - run = self.run_of(engine, name) - if run.fallback is not None: - return Outcome('SKIP', f'{engine} did not run: {run.fallback}') - if run.error is not None: - return Outcome('ERROR', run.error) - base = self.run_of('default', name) - if base.error is not None or base.fallback is not None: - return Outcome('ERROR', f'default: {base.error or base.fallback}') - - over_ipv6 = any(not _over_ipv4(records[0]) for records in base.inputs['tcp'].values()) - if aspect == 'declined': - if (run.declined is not None) != over_ipv6: - return Outcome('DECLINED', f'warned {run.declined!r}, but default finds TCP over IPv6: ' - f'{over_ipv6}') - return Outcome('OK') - if (aspect == 'tcp' and run.no_tcp) or (aspect == 'trace' and run.no_trace): - return Outcome('SKIP', f'{engine} declines {aspect}') - - mine, theirs = run.inputs[aspect], base.inputs[aspect] - if run.declined is not None: - theirs = {number: records for number, records in theirs.items() if _over_ipv4(records[0])} - if aspect == 'trace' and engine in NO_TCP_OCTETS: - def strip(inputs: 'dict[int, Any]') -> 'dict[int, Any]': - return {number: [{key: value for key, value in record.items() - if key not in ('header', 'payload')} for record in records] - for number, records in inputs.items()} - mine, theirs = strip(mine), strip(theirs) - diff = _difference(mine, theirs) - return Outcome('INPUTS', diff) if diff else Outcome('OK') - - -if __name__ == '__main__': - unittest.main() diff --git a/tests/foundation/engines/test_new_engine_parity_runtime.py b/tests/foundation/engines/test_new_engine_parity_runtime.py index 6a5272e8b7..e692700df9 100644 --- a/tests/foundation/engines/test_new_engine_parity_runtime.py +++ b/tests/foundation/engines/test_new_engine_parity_runtime.py @@ -1,30 +1,22 @@ -"""End-to-end agreement between the new engines and the ``default`` engine. +"""End-to-end runs of the ``pypcap`` and ``pypcapfile`` engines. :class:`~pcapkit.foundation.extraction.Extractor` falls back to its own parser with an :class:`~pcapkit.utilities.warnings.EngineWarning` when the engine module -is missing, so a naive "does it agree with the default engine?" test passes most -loudly when the engine under test never ran at all. Every test here therefore -proves the engine ran first -- by the recorded engine name, the engine class, and -the absence of any :class:`~pcapkit.utilities.warnings.EngineWarning` -- and only -then compares. +is missing, so a test of an engine passes most loudly when the engine under test +never ran at all. Every test here therefore proves the engine ran first -- by the +recorded engine name, the engine class, and the absence of any +:class:`~pcapkit.utilities.warnings.EngineWarning` -- and only then checks what +it reports. This module is named ``*_runtime.py`` deliberately, which puts it in the -fixture-dependent tier (see :mod:`tests._tiers`). It reads ``arp.pcap``, -``tcp.pcap``, ``ipv4.pcap`` and ``test.pcapng``, none of which git tracks -- +fixture-dependent tier (see :mod:`tests._tiers`). It reads ``tcp.pcap``, +``ipv6.pcap`` and ``test.pcapng``, none of which git tracks -- :file:`examples/generators/make_samples.py` writes them -- so it cannot be unit-tier, and it was one only because the guard added in GitHub pull request #393 could not see it: the reads go through ``sample_path(capture)`` with a *variable*, which only the runtime half of the guard catches, and that half was never reached on a machine where the engine packages were absent and every test skipped. -Installing an engine made all four reads fail at once. - -Note also that the frame objects a stored extraction hands back are not a usable -source of raw bytes: :attr:`Frame.packet -` re-reads from the (by then -exhausted) input stream, and ``Frame.info.packet`` holds the *undecoded remainder* -rather than the frame. The comparison below therefore uses the fields that are -recorded verbatim off the wire -- the per-record capture length and timestamp from -the PCAP record header, and the Ethernet header of each frame. +Installing an engine made its reads fail at once. """ from __future__ import annotations @@ -69,20 +61,6 @@ def _importable(*modules: str) -> bool: HAS_PYPCAP = _importable('pcap') and not _importable('pcap._pcap') HAS_PYPCAPFILE = _importable('pcapfile.savefile', 'pcapfile.linklayer') -#: Captures the parity comparison runs over. All are Ethernet PCAP savefiles. -CAPTURES = ('in.pcap', 'arp.pcap', 'tcp.pcap', 'ipv4.pcap') - - -def ethernet_of(frame) -> tuple[str, str, int]: - """Fetch ``(src, dst, ethertype)`` from a ``default`` engine frame.""" - info = frame['Ethernet'].info - return str(info.src), str(info.dst), int(info.type) - - -def mac(raw) -> str: - """Render six raw bytes as a colon-separated MAC address.""" - return ':'.join(f'{octet:02x}' for octet in bytes(raw)) - @unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') class NewEngineParityTests(unittest.TestCase): @@ -143,89 +121,6 @@ def test_pypcapfile_engine_really_ran(self) -> None: for frame in extractor.frame: self.assertIsInstance(frame, pcap_packet) - ########################################################################## - # Agreement with the default engine. - ########################################################################## - - @unittest.skipUnless(HAS_PYPCAP, 'pypcap not installed') - def test_pypcap_agrees_with_the_default_engine(self) -> None: - from pcapkit.const.reg.linktype import LinkType - - for capture in CAPTURES: - with self.subTest(capture=capture): - base = self.extract('default', capture) - engine = self.extract('pypcap', capture) - - self.assertEqual(engine.length, base.length) - self.assertEqual(len(engine.frame), base.length) - self.assertEqual(engine.engine.dlink, LinkType.ETHERNET) - - for index, expected in enumerate(base.frame): - timestamp, packet = engine.frame[index] - self.assertEqual(len(packet), expected.info.frame_info.incl_len, - f'{capture} frame {index + 1}: capture length') - self.assertEqual(timestamp, float(expected.info.time_epoch), - f'{capture} frame {index + 1}: timestamp') - - src, dst, ethertype = ethernet_of(expected) - self.assertEqual(mac(packet[0:6]), dst, - f'{capture} frame {index + 1}: ethernet dst') - self.assertEqual(mac(packet[6:12]), src, - f'{capture} frame {index + 1}: ethernet src') - self.assertEqual(int.from_bytes(packet[12:14], 'big'), ethertype, - f'{capture} frame {index + 1}: ethertype') - - @unittest.skipUnless(HAS_PYPCAPFILE, 'pypcapfile not installed') - def test_pypcapfile_agrees_with_the_default_engine(self) -> None: - from pcapkit.const.reg.linktype import LinkType - from pcapkit.toolkit.pypcapfile import packet2timestamp - - for capture in CAPTURES: - with self.subTest(capture=capture): - base = self.extract('default', capture) - engine = self.extract('pypcapfile', capture) - - self.assertEqual(engine.length, base.length) - self.assertEqual(len(engine.frame), base.length) - self.assertEqual(engine.engine.dlink, LinkType.ETHERNET) - - for index, expected in enumerate(base.frame): - packet = engine.frame[index] - self.assertEqual(packet.capture_len, expected.info.frame_info.incl_len, - f'{capture} frame {index + 1}: capture length') - self.assertEqual(packet.packet_len, expected.info.frame_info.orig_len, - f'{capture} frame {index + 1}: original length') - self.assertEqual(packet2timestamp(packet), float(expected.info.time_epoch), - f'{capture} frame {index + 1}: timestamp') - - src, dst, ethertype = ethernet_of(expected) - ethernet = packet.packet - self.assertEqual(type(ethernet).__name__, 'Ethernet') - # ``pypcapfile``'s ``Ethernet.dst``/``.src`` are already - # colon-separated ASCII (see ``Ethernet.__init__`` in - # ``pcapfile/protocols/linklayer/ethernet.py``), unlike the - # raw 6-byte slices ``mac()`` exists to format above -- so - # they only need decoding, not another hex pass. - self.assertEqual(ethernet.dst.decode('ascii'), dst, - f'{capture} frame {index + 1}: ethernet dst') - self.assertEqual(ethernet.src.decode('ascii'), src, - f'{capture} frame {index + 1}: ethernet src') - self.assertEqual(ethernet.type, ethertype, - f'{capture} frame {index + 1}: ethertype') - - @unittest.skipUnless(HAS_PYPCAPFILE, 'pypcapfile not installed') - def test_pypcapfile_ipv4_reassembly_agrees_with_the_default_engine(self) -> None: - base = self.extract('default', 'ipv4.pcap', reassembly=True, ipv4=True) - engine = self.extract('pypcapfile', 'ipv4.pcap', reassembly=True, ipv4=True) - - def payloads(datagrams): - return sorted(bytes(datagram.payload) if isinstance(datagram.payload, (bytes, bytearray)) - else tuple(bytes(part) for part in datagram.payload) - for datagram in datagrams) - - self.assertTrue(base.reassembly.ipv4) - self.assertEqual(payloads(engine.reassembly.ipv4), payloads(base.reassembly.ipv4)) - ########################################################################## # Capability gaps, asserted rather than described. ########################################################################## @@ -238,27 +133,6 @@ def test_pypcapfile_reports_no_ipv6_reassembly(self) -> None: base = self.extract('default', 'ipv6.pcap', reassembly=True, ip=True) self.assertTrue(base.reassembly.ipv6) - @unittest.skipUnless(HAS_PYPCAPFILE, 'pypcapfile not installed') - def test_pypcapfile_traces_only_the_ipv4_half_of_a_mixed_capture(self) -> None: - # ``tcp.pcap`` carries four IPv4 and three IPv6 TCP frames; ``pypcapfile`` - # can only see the former, so it finds strictly fewer flows. - base = self.extract('default', 'tcp.pcap', trace=True, tcp=True, - trace_fout='/tmp/parity-trace-default') - engine = self.extract('pypcapfile', 'tcp.pcap', trace=True, tcp=True, - trace_fout='/tmp/parity-trace-pypcapfile') - - def traced(extractor) -> set[int]: - return {index for stream in extractor.trace.tcp for index in stream.index} - - ipv4_frames = {number for number, frame in enumerate(base.frame, start=1) - if 'IPv4' in frame} - self.assertTrue(ipv4_frames) - self.assertTrue(traced(engine)) - # every frame it traced is one it could decode ... - self.assertLessEqual(traced(engine), ipv4_frames) - # ... and that is a strict subset of what the default engine traced - self.assertLess(traced(engine), traced(base)) - @unittest.skipUnless(HAS_PYPCAP, 'pypcap not installed') def test_pypcap_reports_neither_reassembly_nor_flow_tracing(self) -> None: from pcapkit.utilities.exceptions import UnsupportedCall diff --git a/tests/foundation/engines/test_pypcapfile_linktype_and_layer_bound_agreement_1609_1667_unit.py b/tests/foundation/engines/test_pypcapfile_linktype_and_layer_bound_agreement_1609_1667_unit.py deleted file mode 100644 index 25474acf3b..0000000000 --- a/tests/foundation/engines/test_pypcapfile_linktype_and_layer_bound_agreement_1609_1667_unit.py +++ /dev/null @@ -1,309 +0,0 @@ -# -*- coding: utf-8 -*- -"""``pypcapfile`` hands over what ``default`` does, by link type and to the frame layer bound. C.f. #1609, #1667. - -The ``default`` engine dissects the IP packet of the ``NULL``, ``LOOP``, -``RAW``, ``IPV4`` and ``IPV6`` link types, which `pypcapfile`_ 0.12.0 has no -decoder for (#1667), and at most -:data:`~pcapkit.protocols.protocol.FRAME_LAYER_LIMIT` layers of a frame, each -VLAN tag, IP header, AH and IPv6 extension header one of them (#1609). Both -engines read one capture per case, built in a temporary directory, with IPv4 -and TCP reassembly and flow tracing on, and ``pypcapfile``'s ``tcp``, ``trace`` -and ``ipv4`` inputs are compared with ``default``'s field for field, as in -:mod:`tests.foundation.engines.test_ext_chain_limit_agreement_1604_unit`. - -``pypcapfile`` leaves TCP over IPv6 out, so its ``tcp`` and ``trace`` inputs are -compared over IPv4 only, and ``declined`` checks that it warns of leaving some -out exactly where ``default`` finds any. ``unread`` checks that the engine warns -of a link type left unread exactly where ``default`` dissects nothing of it. It -imports on Python 3.11 only, and is skipped elsewhere. - -.. _pypcapfile: https://github.com/kisom/pypcapfile - -""" - -from __future__ import annotations - -import importlib.util -import ipaddress -import os -import tempfile -import unittest -import warnings -from typing import TYPE_CHECKING -from unittest import mock - -from tests._support import close_extractor, reimport_once_per_class -from tests.foundation import _roundtrip as harness -from tests.foundation._roundtrip import 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')) - -#: ``LINKTYPE_*`` numbers, by name. -LINKTYPES = {'NULL': 0, 'ETHERNET': 1, 'LOOP': 108, 'RAW': 101, 'IPV4': 228, 'IPV6': 229, - 'DLT12': 12, 'LINUX_SLL': 113} -#: Link types the ``default`` engine dissects nothing of. -UNREAD = ('DLT12', 'LINUX_SLL') - -TCP = harness.tcp(b'payload', seq=1, syn=True) -TCP4 = harness.ipv4(TCP, proto=6) -TCP6 = harness.ipv6(TCP, nxt=6) -UDP = harness.udp(b'u' * 24) -FRAG4 = (harness.ipv4(UDP[:16], proto=17, mf=True), harness.ipv4(UDP[16:], proto=17, offset=16)) -FRAG6 = (harness.ipv6_fragment(UDP[:16], mf=True), harness.ipv6_fragment(UDP[16:], offset=16)) -#: Payloads, each the IP version it starts with and the packets of its frames. -PAYLOADS = { - 'ipv4-tcp': (4, [TCP4]), - 'ipv6-tcp': (6, [TCP6]), - 'ipv4-frags': (4, list(FRAG4)), - 'ipv6-frags': (6, list(FRAG6)), - '4in6-tcp': (6, [harness.ipv6(TCP4, nxt=4)]), - '4in6-frags': (6, [harness.ipv6(fragment, nxt=4) for fragment in FRAG4]), -} - - -def link(name: 'str', packet: 'bytes', version: 'int') -> 'bytes': - """``packet`` as a frame of the link type ``name``.""" - if name == 'ETHERNET': - return harness.ethernet(packet, 0x0800 if version == 4 else 0x86DD) - if name in ('NULL', 'LOOP'): - family = 2 if version == 4 else 24 - return family.to_bytes(4, 'little' if name == 'NULL' else 'big') + packet - if name == 'LINUX_SLL': - return (b'\x00\x00\x00\x01\x00\x06' + b'\x02' * 8 - + (0x0800 if version == 4 else 0x86DD).to_bytes(2, 'big') + packet) - return packet - - -def nest4(inner: 'bytes', count: 'int', proto: 'int') -> 'bytes': - """``count`` IPv4 headers, the innermost carrying ``inner`` as ``proto``.""" - packet = harness.ipv4(inner, proto=proto) - for _ in range(count - 1): - packet = harness.ipv4(packet, proto=4) - return packet - - -def nest6(inner: 'bytes', count: 'int', nxt: 'int') -> 'bytes': - """``count`` IPv6 headers, the innermost carrying ``inner`` as ``nxt``.""" - packet = harness.ipv6(inner, nxt=nxt) - for _ in range(count - 1): - packet = harness.ipv6(packet, nxt=41) - return packet - - -def pairs(count: 'int', version: 'int') -> 'bytes': - """``count`` pairs of an IPv4 header and an AH, or of an IPv6 header and a Hop-by-Hop Options header, then TCP.""" - packet = TCP - for index in range(count): - nxt = 6 if index == 0 else (4 if version == 4 else 41) - if version == 4: - ah = bytes([nxt, 4, 0, 0]) + (0x100).to_bytes(4, 'big') + (1).to_bytes(4, 'big') + b'\xaa' * 12 - packet = harness.ipv4(ah + packet, proto=51) - else: - packet = harness.ipv6(bytes([nxt, 0, 1, 4, 0, 0, 0, 0]) + packet, nxt=0) - return packet - - -def chain(count: 'int', last: 'int') -> 'bytes': - """``count`` Destination Options headers, the last naming ``last``.""" - return b''.join(bytes([60 if index < count - 1 else last, 0, 1, 4, 0, 0, 0, 0]) for index in range(count)) - - -def tagged(packet: 'bytes', ethertype: 'int', count: 'int', tpid: 'int') -> 'bytes': - """An Ethernet frame carrying ``packet`` behind ``count`` VLAN tags of ``tpid``.""" - tags = b''.join((100 + index).to_bytes(2, 'big') + (tpid if index < count - 1 else ethertype).to_bytes(2, 'big') - for index in range(count)) - return harness.MAC_B + harness.MAC_A + tpid.to_bytes(2, 'big') + tags + packet - - -def cases() -> 'dict[str, tuple[int, list[bytes]]]': - """Every case, as the link type of its capture and its frames.""" - out = {} # type: dict[str, tuple[int, list[bytes]]] - for name, linktype in LINKTYPES.items(): - for kind, (version, packets) in PAYLOADS.items(): - if (name, version) not in (('IPV4', 6), ('IPV6', 4)): - out[f'{name}-{kind}'] = (linktype, [link(name, packet, version) for packet in packets]) - out['NULL-big-endian-family'] = (0, [(2).to_bytes(4, 'big') + TCP4]) - out['LOOP-little-endian-family'] = (108, [(2).to_bytes(4, 'little') + TCP4]) - out['NULL-af28-ipv6-tcp'] = (0, [(28).to_bytes(4, 'little') + TCP6]) - out['RAW-version-5'] = (101, [b'\x55' + TCP4[1:]]) - out['RAW-6in4-tcp'] = (101, [harness.ipv4(TCP6, proto=41)]) - out['RAW-tcp-behind-ah'] = (101, [pairs(1, 4)]) - for count in (62, 63): - out[f'ETHERNET-tcp-in-{count}-ipv4'] = (1, [harness.ethernet(nest4(TCP, count, 6), 0x0800)]) - out[f'ETHERNET-ipv4-frags-in-{count}-ipv6'] = ( - 1, [harness.ethernet(nest6(fragment, count, 4), 0x86DD) for fragment in FRAG4]) - out[f'NULL-tcp-in-{count}-ipv4'] = (0, [(2).to_bytes(4, 'little') + nest4(TCP, count, 6)]) - out[f'IPV6-4in6-tcp-in-{count}-ipv6'] = (229, [nest6(TCP4, count, 4)]) - for count in (63, 64): - out[f'RAW-tcp-in-{count}-ipv4'] = (101, [nest4(TCP, count, 6)]) - out[f'RAW-ipv4-frags-in-{count}-ipv6'] = (101, [nest6(fragment, count, 4) for fragment in FRAG4]) - for count in (61, 62): - out[f'ETHERNET-4in6-tcp-in-{count}-ipv6'] = (1, [harness.ethernet(nest6(TCP4, count, 4), 0x86DD)]) - out[f'ETHERNET-ipv6-tcp-behind-{count}-tags'] = (1, [tagged(TCP6, 0x86DD, count, 0x8100)]) - for count in (31, 32): - out[f'ETHERNET-tcp-behind-{count}-ipv4-ah'] = (1, [harness.ethernet(pairs(count, 4), 0x0800)]) - out[f'ETHERNET-tcp-behind-{count}-ipv6-hbh'] = (1, [harness.ethernet(pairs(count, 6), 0x86DD)]) - out[f'ETHERNET-tcp-in-31-ipv4-behind-{count}-headers'] = ( - 1, [harness.ethernet(nest4(chain(count, 6) + TCP, 31, 60), 0x0800)]) - for count in (61, 62, 63): - for tpid in (0x8100, 0x88A8): - out[f'ETHERNET-tcp-behind-{count}-tags-{tpid:04x}'] = (1, [tagged(TCP4, 0x0800, count, tpid)]) - out[f'ETHERNET-ipv4-frags-behind-{count}-tags-{tpid:04x}'] = ( - 1, [tagged(fragment, 0x0800, count, tpid) for fragment in FRAG4]) - return out - - -#: Every case, built once. -CASES = cases() - - -def _fields(packet: 'Any', skip: 'tuple[str, ...]' = ()) -> 'dict[str, Any]': - return {key: bytes(value) if isinstance(value, bytearray) else value - for key, value in packet.items() if key not in skip} - - -class _Run: - """One engine's reading of one capture.""" - - def __init__(self, engine: 'str', path: '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 AttributeWarning, EngineWarning - - self.error = self.fallback = None # type: Optional[str] - #: What the ``pypcapfile`` toolkit warned it leaves out, if anything. - self.declined = None # type: Optional[str] - #: Whether the ``pypcapfile`` engine warned it leaves the link type unread. - self.unread = False - self.inputs = {'ipv4': {}, 'tcp': {}, 'trace': {}} # type: dict[str, dict[int, list[dict[str, 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, \ - mock.patch('pcapkit.foundation.engines.pypcapfile.warn') as engine_warn, \ - mock.patch('pcapkit.toolkit.pypcapfile.warn') as toolkit_warn: - warnings.simplefilter('ignore') - try: - extractor = extract(fin=path, nofile=True, engine=engine, reassembly=True, ipv4=True, - tcp=True, trace=True, trace_fout=tracedir, trace_format='json') - except Exception as exc: # pylint: disable=broad-except - self.error = f'{type(exc).__name__}: {exc}' - return - close_extractor(extractor) - fallen = [str(call.args[0]) for call in warn.call_args_list - if len(call.args) > 1 and call.args[1] is 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.declined = '; '.join(str(call.args[0]) for call in toolkit_warn.call_args_list - if len(call.args) > 1 and call.args[1] is AttributeWarning - and 'is left out of TCP reassembly' in str(call.args[0])) or None - self.unread = any('unrecognised link layer protocol' in str(call.args[0]) - for call in engine_warn.call_args_list) - - -_RUNS = {} # type: dict[tuple[str, str], _Run] - - -def _over_ipv4(record: 'dict[str, Any]') -> 'bool': - address = record['bufid'][0] if 'bufid' in record else record['src'] - return isinstance(address, ipaddress.IPv4Address) - - -def _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]: - return f'frame {number}: {got[number]!r:.120} != {want[number]!r:.120}' - return None - - -@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') -class TestPyPCAPFileLinkTypeAndLayerBoundAgreement(harness.RoundTripBase): - """``pypcapfile`` hands over what ``default`` does over every case, or is a recorded gap.""" - - STATUSES = ('ERROR', 'INPUTS', 'DECLINED', 'UNREAD') - KNOWN_FAILURES = () - - @classmethod - def setUpClass(cls) -> None: - super().setUpClass() - cls.tmp = tempfile.TemporaryDirectory() # pylint: disable=consider-using-with - for name, (linktype, frames) in CASES.items(): - with open(cls.path(name), 'wb') as file: - file.write(harness.pcap([(1, index, frame) for index, frame in enumerate(frames)], - linktype=linktype)) - - @classmethod - def tearDownClass(cls) -> None: - _RUNS.clear() - cls.tmp.cleanup() - super().tearDownClass() - - @classmethod - def path(cls, name: 'str') -> 'str': - return os.path.join(cls.tmp.name, f'{name}.pcap') - - def setUp(self) -> None: - super().setUp() - reimport_once_per_class(self) - - def labels(self) -> 'list[str]': - return [f'{name}/{aspect}' for name in CASES - for aspect in ('tcp', 'trace', 'ipv4', 'declined', 'unread')] - - def run_of(self, engine: 'str', name: 'str') -> '_Run': - if (engine, name) not in _RUNS: - _RUNS[engine, name] = _Run(engine, self.path(name)) - return _RUNS[engine, name] - - def outcome(self, label: 'str') -> 'Outcome': - name, aspect = label.split('/') - run = self.run_of('pypcapfile', name) - if run.fallback is not None: - return Outcome('SKIP', f'pypcapfile did not run: {run.fallback}') - if run.error is not None: - return Outcome('ERROR', run.error) - base = self.run_of('default', name) - if base.error is not None or base.fallback is not None: - return Outcome('ERROR', f'default: {base.error or base.fallback}') - - if aspect == 'unread': - if run.unread != name.startswith(UNREAD): - return Outcome('UNREAD', f'warned of the link type left unread: {run.unread}') - return Outcome('OK') - over_ipv6 = any(not _over_ipv4(records[0]) for records in base.inputs['tcp'].values()) - if aspect == 'declined': - if (run.declined is not None) != over_ipv6: - return Outcome('DECLINED', f'warned {run.declined!r}, but default finds TCP over IPv6: ' - f'{over_ipv6}') - return Outcome('OK') - - mine, theirs = run.inputs[aspect], base.inputs[aspect] - if aspect != 'ipv4': - theirs = {number: records for number, records in theirs.items() if _over_ipv4(records[0])} - diff = _difference(mine, theirs) - return Outcome('INPUTS', diff) if diff else Outcome('OK') - - -if __name__ == '__main__': - unittest.main() diff --git a/tests/foundation/traceflow/test_traceflow_snapped_capture_runtime.py b/tests/foundation/traceflow/test_traceflow_snapped_capture_runtime.py index c82c2e9c99..aff57b947d 100644 --- a/tests/foundation/traceflow/test_traceflow_snapped_capture_runtime.py +++ b/tests/foundation/traceflow/test_traceflow_snapped_capture_runtime.py @@ -44,6 +44,8 @@ UNIDIRECTIONAL = [((1, 7), (1, 7), ()), ((2, 6), (2, 6), ()), ((3,), (3,), ()), ((4, 5), (4, 5), ())] #: Frame numbers of each reassembled TCP datagram, as :mod:`dpkt` gives them. DATAGRAMS = [(2, 6), (3,), (4, 5)] +#: The octets of each of those datagrams' payloads. +DATAGRAM_LENGTHS = [208, 17, 17] def snapped_records(cut: 'int' = 4) -> 'tuple[bytes, list[tuple[int, int, int, bytes]]]': @@ -91,7 +93,7 @@ def pcap_records(data: 'bytes') -> 'list[tuple[int, int, int, bytes]]': @unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') class SnappedCaptureTraceTests(unittest.TestCase): - """The default engine traces the snapped capture as the other engines do.""" + """Every engine traces and reassembles every frame of the snapped capture.""" def setUp(self) -> None: reimport_once_per_class(self) @@ -130,13 +132,13 @@ def test_the_default_engine_traces_every_frame(self) -> None: self.assertEqual([flow[:3] for flow in flows], expected) @unittest.skipUnless(ENGINES, 'neither dpkt nor scapy installed') - def test_the_other_engines_agree(self) -> None: + def test_the_other_engines_trace_every_frame(self) -> None: for container in CONTAINERS: - for bidirectional in (True, False): - base = self._flows(container, trace_bidirectional=bidirectional) + for bidirectional, expected in ((True, FLOWS), (False, UNIDIRECTIONAL)): for engine in ENGINES: with self.subTest(container=container, bidirectional=bidirectional, engine=engine): - self.assertEqual(self._flows(container, engine, trace_bidirectional=bidirectional), base) + flows = self._flows(container, engine, trace_bidirectional=bidirectional) + self.assertEqual([flow[:3] for flow in flows], expected) def _trace_files(self, container: 'str', engine: 'str' = 'default') -> 'dict[str, bytes]': fout = tempfile.mkdtemp(dir=self.tmp) @@ -147,22 +149,23 @@ def _trace_files(self, container: 'str', engine: 'str' = 'default') -> 'dict[str files[flow.label] = file.read() return files - def test_each_traced_pcap_holds_the_captured_records(self) -> None: + def _assert_traced_records(self, engine: 'str') -> None: # Byte for byte: each flow's file holds its frames' records as captured, # snapped octets and original length alike (#1543). _, records = snapped_records() for container in CONTAINERS: with self.subTest(container=container): - files = self._trace_files(container) + files = self._trace_files(container, engine) self.assertEqual(len(files), len(FLOWS)) for (index, _, _), data in zip(FLOWS, files.values()): self.assertEqual(pcap_records(data), [records[number - 1] for number in index]) + def test_each_traced_pcap_holds_the_captured_records(self) -> None: + self._assert_traced_records('default') + @unittest.skipUnless('dpkt' in ENGINES, 'dpkt not installed') - def test_each_traced_pcap_matches_dpkt(self) -> None: - for container in CONTAINERS: - with self.subTest(container=container): - self.assertEqual(self._trace_files(container), self._trace_files(container, 'dpkt')) + def test_each_dpkt_traced_pcap_holds_the_captured_records(self) -> None: + self._assert_traced_records('dpkt') def _datagrams(self, container: 'str', engine: 'str' = 'default') -> 'list[tuple]': extractor = self._extract(container, engine, reassembly=True, reasm_store=True) @@ -174,10 +177,11 @@ def test_reassembly_takes_every_frame(self) -> None: self.assertEqual([index for index, _ in self._datagrams(container)], DATAGRAMS) @unittest.skipUnless('dpkt' in ENGINES, 'dpkt not installed') - def test_reassembly_matches_dpkt(self) -> None: + def test_dpkt_reassembly_takes_every_frame(self) -> None: for container in CONTAINERS: with self.subTest(container=container): - self.assertEqual(self._datagrams(container), self._datagrams(container, 'dpkt')) + self.assertEqual([(index, len(payload)) for index, payload in self._datagrams(container, 'dpkt')], + list(zip(DATAGRAMS, DATAGRAM_LENGTHS))) if __name__ == '__main__': diff --git a/tests/integration/test_engine_parity.py b/tests/integration/test_engine_parity.py index bb7c728220..4aa488c05d 100644 --- a/tests/integration/test_engine_parity.py +++ b/tests/integration/test_engine_parity.py @@ -1,15 +1,15 @@ # -*- coding: utf-8 -*- -"""End-to-end parity between the extraction engines. +"""End-to-end runs of each extraction engine over the same captures. Translates :file:`examples/legacy_smoke/test_engine.py`, which ran the same -capture through each engine and wrote four reports nobody compared, into -assertions that the engines agree. +capture through each engine and wrote four reports nobody checked, into +assertions on what each engine reports, against constants. ``pyshark`` is left out on purpose -- not for want of :program:`tshark`, which the ``pypcap-parity`` job (the one whose selection collects this file) now installs, but because on Python 3.14 it fails inside its own -``get_event_loop()`` before it reads a byte, so it cannot take part in a parity -comparison that runs on every version; +``get_event_loop()`` before it reads a byte, so it cannot take part in a check +that runs on every version; :file:`tests/integration/test_engine_runtime.py` already pins that. The ``pipeline`` and ``server`` engines are commented out in the original script and are not covered here either. @@ -23,7 +23,7 @@ from tests.integration._helpers import HAS_DPKT, HAS_RUNTIME, HAS_SCAPY, EndToEndTestCase #: Captures every engine is run over, with the frame count each must report. -#: All three are small: the point is agreement, not throughput. +#: All are small: the point is what each engine reports, not throughput. PARITY_CAPTURES = { 'in.pcap': 6, 'arp.pcap': 2, @@ -31,7 +31,11 @@ 'http6.cap': 26, } -#: Engines to compare, with the gate that decides whether each is available. +#: The Ethernet header of each frame of ``arp.pcap``, as captured. +ARP_ETHERNET_HEADERS = (bytes.fromhex('000c297d1db4000c2919dc610806'), + bytes.fromhex('000c2919dc61000c297d1db40806')) + +#: Engines to run, with the gate that decides whether each is available. ENGINES = ( ('default', HAS_RUNTIME), ('dpkt', HAS_DPKT), @@ -46,19 +50,14 @@ class EngineParityTests(EndToEndTestCase): def available(self) -> 'list[str]': return [name for name, present in ENGINES if present] - def test_every_engine_reports_the_same_frame_count(self) -> None: + def test_every_engine_reports_the_frame_count(self) -> None: for capture, expected in PARITY_CAPTURES.items(): - counts = {} for engine in self.available(): with self.subTest(capture=capture, engine=engine): extractor = self.extract(fin=sample_path(capture), nofile=True, store=False, engine=engine) - counts[engine] = extractor.length self.assertEqual(extractor.length, expected) - with self.subTest(capture=capture): - self.assertEqual(set(counts.values()), {expected}) - def test_every_engine_writes_a_report(self) -> None: for engine in self.available(): with self.subTest(engine=engine): @@ -88,48 +87,40 @@ def test_engine_macro_selects_the_same_engine_as_its_name(self) -> None: [str(frame.protochain) for frame in by_name.frame]) @unittest.skipUnless(HAS_DPKT, 'dpkt not installed') - def test_dpkt_engine_agrees_on_the_protocol_chain(self) -> None: - # The engines hand back their own packet objects, so the chains are - # spelled differently; the toolkit is what maps one onto the other. + def test_dpkt_engine_spells_the_protocol_chain_in_its_own_names(self) -> None: + # The engine hands back dpkt's own packet objects, and the toolkit spells + # their chain in dpkt's names. from pcapkit.toolkit.dpkt import packet2chain - native = self.extract(fin=sample_path('in.pcap'), nofile=True, store=True, - engine='default') foreign = self.extract(fin=sample_path('in.pcap'), nofile=True, store=True, engine='dpkt') - self.assertEqual(str(native.frame[0].protochain), 'Ethernet:IPv6:IPv6_ICMP') self.assertEqual(packet2chain(foreign.frame[0]), 'Ethernet:IP6:ICMP6') - self.assertEqual(str(native.frame[2].protochain), 'Ethernet:IPv4:TCP') self.assertEqual(packet2chain(foreign.frame[2]), 'Ethernet:IP:TCP') @unittest.skipUnless(HAS_SCAPY, 'scapy not installed') - def test_scapy_engine_returns_the_same_link_layer_bytes(self) -> None: - native = self.extract(fin=sample_path('arp.pcap'), nofile=True, store=True, - engine='default') + def test_scapy_engine_returns_every_link_layer_octet(self) -> None: foreign = self.extract(fin=sample_path('arp.pcap'), nofile=True, store=True, engine='scapy') - self.assertEqual(len(native.frame), len(foreign.frame)) - for number, (mine, theirs) in enumerate(zip(native.frame, foreign.frame), start=1): + self.assertEqual(len(foreign.frame), len(ARP_ETHERNET_HEADERS)) + for number, (frame, header) in enumerate(zip(foreign.frame, ARP_ETHERNET_HEADERS), start=1): with self.subTest(frame=number): # Re-serialising scapy's frame reproduces the whole 60 octet frame, - # padded to the Ethernet minimum, and its first fourteen octets are - # the Ethernet header the default engine parsed into a layer of its - # own. That is what makes this a bytes-level parity check: both - # engines account for every captured octet, whatever they call the - # layers they split it into. + # padded to the Ethernet minimum, its first fourteen octets the + # Ethernet header as captured: the engine accounts for every + # captured octet, whatever it calls the layers it splits it into. # # #406: this comment used to say scapy "does not know this capture's # link type and hands back one opaque ``Raw`` layer". That was the # missing-layer-registry bug, not a property of the capture -- # arp.pcap is link type 1 and scapy now dissects it as - # ``Ethernet / ARP / Padding``. The assertions below are unchanged - # and still pass, because ``bytes()`` of a dissected frame is the - # same octet string as ``bytes()`` of an undissected one. - captured = bytes(theirs) + # ``Ethernet / ARP / Padding``. The assertions below still pass + # either way, because ``bytes()`` of a dissected frame is the same + # octet string as ``bytes()`` of an undissected one. + captured = bytes(frame) self.assertEqual(len(captured), 60) - self.assertEqual(captured[:14], bytes(mine['Ethernet'].packet.header)) + self.assertEqual(captured[:14], header) if __name__ == '__main__': diff --git a/tests/integration/test_reassembly_engine_parity.py b/tests/integration/test_reassembly_engine_parity.py index 07a6078138..478ea65fdb 100644 --- a/tests/integration/test_reassembly_engine_parity.py +++ b/tests/integration/test_reassembly_engine_parity.py @@ -1,23 +1,20 @@ # -*- coding: utf-8 -*- -"""The four toolkit adapters must describe an IPv6 fragment identically. +"""Each toolkit adapter describes an IPv6 fragment as :rfc:`8200` does. Each of :mod:`pcapkit.toolkit.pcap`, :mod:`pcapkit.toolkit.pcapng`, :mod:`pcapkit.toolkit.dpkt` and :mod:`pcapkit.toolkit.scapy` builds a :term:`reasm.ipv6.packet` from whatever object model its engine hands it, and each was only ever tested against itself. That is how #415 survived: for ``ipv6.pcap`` frame 13 the four reported ``ihl`` 48/48/40/40, ``len(header)`` -48/48/40/40 and ``tl`` 1496/1496/1488/1496 -- a *three*-way disagreement on -``tl``, with no two adapters agreeing on all three fields. +48/48/40/40 and ``tl`` 1496/1496/1488/1496, where the RFC calls for 40, 40 and +1488. -So the assertions here are on agreement rather than on constants: every adapter -is run over the same octets and the results are compared with each other. The -expected values are pinned as well, because agreement alone would be satisfied by -all four being wrong in the same way. +So every adapter is run over the same octets, and each one's result is checked +against the values the RFC calls for and the capture's own octets. The PCAP-NG adapter has no fragmented sample capture of its own, so the frames of -``ipv6.pcap`` are rewrapped into a PCAP-NG file here. That is deliberate: reading -two different captures would compare two different datagrams and prove nothing -about the adapters. +``ipv6.pcap`` are rewrapped into a PCAP-NG file here, so that every adapter is +checked over the same datagram. """ from __future__ import annotations @@ -195,40 +192,27 @@ def test_every_adapter_finds_the_same_four_fragments(self) -> None: self.assertEqual([(data.fo, len(data.payload)) for data in fragments], list(FRAGMENTS)) - def test_every_adapter_reports_the_same_ihl_header_and_tl(self) -> None: - """``ihl``, ``header`` and ``tl`` must not depend on the engine. + def test_every_adapter_reports_the_rfc_ihl_header_and_tl(self) -> None: + """``ihl``, ``header`` and ``tl`` leave the Fragment header out. - The three fields of #415, asserted both against each other and against - the values :rfc:`8200#section-4.5` calls for: the Fragment header is not - present in the reassembled packet, so none of the three may count its 8 - octets. + The three fields of #415, asserted for every adapter against the values + :rfc:`8200#section-4.5` calls for: the Fragment header is not present in + the reassembled packet, so none of the three may count its 8 octets. """ packets = self.adapter_packets() - for index, (offset, payload_len) in enumerate(FRAGMENTS): - described = {name: (data[index].ihl, data[index].header, data[index].tl) - for name, data in packets.items()} - - # a failure message naming the three numbers per adapter, since the - # raw ``header`` octets in ``described`` are unreadable in a diff - summary = {name: (ihl, len(header), tl) - for name, (ihl, header, tl) in described.items()} - - with self.subTest(fragment=offset): - # every adapter agrees ... - self.assertEqual(len(set(described.values())), 1, - f'adapters disagree on (ihl, len(header), tl): {summary}') - - # ... and what they agree on excludes the Fragment header - ihl, header, tl = next(iter(described.values())) - self.assertEqual(ihl, IPV6_HDR_LEN) - self.assertEqual(len(header), IPV6_HDR_LEN) - self.assertEqual(tl, IPV6_HDR_LEN + payload_len) - # the invariant the reassembly machinery relies on: it writes the - # payload into its datagram buffer over the span ``tl - ihl``, so - # a ``tl`` 8 octets too large leaves 8 stray zeroes per fragment - self.assertEqual(tl - ihl, payload_len) + for name, fragments in packets.items(): + self.assertEqual(len(fragments), len(FRAGMENTS), name) + for data, (offset, payload_len) in zip(fragments, FRAGMENTS): + with self.subTest(adapter=name, fragment=offset): + self.assertEqual(data.ihl, IPV6_HDR_LEN) + self.assertEqual(len(data.header), IPV6_HDR_LEN) + self.assertEqual(data.tl, IPV6_HDR_LEN + payload_len) + # the invariant the reassembly machinery relies on: it writes the + # payload into its datagram buffer over the span ``tl - ihl``, so + # a ``tl`` 8 octets too large leaves 8 stray zeroes per fragment + self.assertEqual(data.tl - data.ihl, payload_len) def test_the_fragment_header_octets_are_the_ones_left_out(self) -> None: """The 40 octets kept are the IPv6 header, not an arbitrary prefix. @@ -286,29 +270,29 @@ def test_no_engine_advertises_a_fragment_header_on_the_datagram(self) -> None: self.assertNotEqual(datagram.header[6], NH_IPV6_FRAG) self.assertEqual(int(datagram.id.proto), NH_UDP) - def test_every_engine_reassembles_the_same_payload(self) -> None: + def test_every_engine_reassembles_the_captured_payload(self) -> None: """The datagram itself, not only its header. ``scapy``'s ``tl`` counted the Fragment header while its ``ihl`` did not, so the reassembly machinery wrote each fragment's payload over a span 8 octets too wide and the datagram came out 4786 octets instead of 4778 -- - eight stray zeroes per fragment boundary, and a payload no other engine - agreed with. + eight stray zeroes per fragment boundary. Each engine's datagram is the + fragments' payloads as captured, read past the Ethernet, IPv6 and + Fragment headers of frames 13 to 16. """ - payloads = {} # type: dict[str, bytes] + captured = b''.join(frame[14 + IPV6_HDR_LEN + 8:] + for frame in pcap_frames(sample_path('ipv6.pcap'))[12:16]) + self.assertEqual(len(captured), sum(length for _, length in FRAGMENTS)) for engine in self.engines(): - extractor = self.extract(fin=sample_path('ipv6.pcap'), nofile=True, - store=False, ipv6=True, reassembly=True, - reasm_strict=True, engine=engine) - datagram, = extractor.reassembly.ipv6 - payloads[engine] = bytes(datagram.payload) - - for engine, payload in payloads.items(): with self.subTest(engine=engine): - self.assertEqual(len(payload), sum(length for _, length in FRAGMENTS)) + extractor = self.extract(fin=sample_path('ipv6.pcap'), nofile=True, + store=False, ipv6=True, reassembly=True, + reasm_strict=True, engine=engine) + datagram, = extractor.reassembly.ipv6 + payload = bytes(datagram.payload) self.assertEqual(len(payload), 4778) - self.assertEqual(len(set(payloads.values())), 1) + self.assertEqual(payload, captured) if __name__ == '__main__': diff --git a/tests/protocols/internet/test_ipv4_tso_total_length_1547_runtime.py b/tests/protocols/internet/test_ipv4_tso_total_length_1547_runtime.py index 895a759a86..ab127708d6 100644 --- a/tests/protocols/internet/test_ipv4_tso_total_length_1547_runtime.py +++ b/tests/protocols/internet/test_ipv4_tso_total_length_1547_runtime.py @@ -43,6 +43,8 @@ #: The flows every engine traces in ``tcp.pcap``, TSO or not. FLOWS = [(1, 2, 6, 7), (3, 4, 5)] +#: The frames of each TCP datagram every engine reassembles from ``tcp.pcap``, TSO or not. +DATAGRAMS = [(2, 6), (3,), (4, 5)] def _checksum(header: 'bytes') -> 'bytes': @@ -236,7 +238,7 @@ def test_a_total_length_past_the_capture_is_unchanged(self) -> None: @unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') class TSOTraceTests(unittest.TestCase): - """The default engine traces a TSO capture's IPv4 flow, as dpkt and scapy do.""" + """The default engine, dpkt and scapy each trace and reassemble a TSO capture's flows.""" def setUp(self) -> None: reimport_once_per_class(self) @@ -264,8 +266,6 @@ def _extract(self, fin: 'str', engine: 'str') -> 'Any': return extractor def test_flows_are_traced_and_reassembled(self) -> None: - baseline = self._extract(sample_path('tcp.pcap'), 'default') - expected_tcp = [tuple(datagram.index) for datagram in baseline.reassembly.tcp] for fix_checksum in (True, False): fin = self._write(f'tso-{fix_checksum}.pcap', *_tso_records(fix_checksum=fix_checksum)) for engine in ('default',) + ENGINES: @@ -273,7 +273,7 @@ def test_flows_are_traced_and_reassembled(self) -> None: extractor = self._extract(fin, engine) self.assertEqual([tuple(flow.index) for flow in extractor.trace.tcp], FLOWS) self.assertEqual([tuple(datagram.index) for datagram in extractor.reassembly.tcp], - expected_tcp) + DATAGRAMS) if __name__ == '__main__': diff --git a/tests/toolkit/test_dpkt_scapy_ipv4_header_checks_1607_unit.py b/tests/toolkit/test_dpkt_scapy_ipv4_header_checks_1607_unit.py index eca00caaa8..6fe29272ad 100644 --- a/tests/toolkit/test_dpkt_scapy_ipv4_header_checks_1607_unit.py +++ b/tests/toolkit/test_dpkt_scapy_ipv4_header_checks_1607_unit.py @@ -1,19 +1,18 @@ # -*- coding: utf-8 -*- -"""The dpkt and scapy engines reject the IPv4 headers the default engine rejects. C.f. #1607. +"""The IPv4 headers the default engine rejects, and what it reads behind them. C.f. #1607. The default engine's IPv4 parser rejects a header whose version is not 4, whose IHL is below 5 or runs past the captured octets, or whose options it rejects, and then dissects nothing past it: no IPv4 reassembly input, and no TCP behind -it. `dpkt`_ and `Scapy`_ decode such a header as IPv4 all the same, so their -engines handed on both. They now ask the default engine's own parser, as the -pypcapfile engine does (#1596), of every IPv4 header on the way to the TCP -segment, a tunnel's included (#1581). Neither engine checks the header checksum, -and nor does the default engine, so a bad one is still read. +it, a tunnel's header included (#1581). It does not check the header checksum, +so a bad one is still read. -Every capture is built here, so the module is in the unit tier. +The helpers here build those headers and read a capture of them with any +engine; :mod:`tests.toolkit.test_dpkt_tcp_ns_flags_1618_unit` and +:mod:`tests.toolkit.test_scapy_ipv6_fragment_and_encap_walk_1635_1636_unit` +import them. -.. _dpkt: https://dpkt.readthedocs.io -.. _Scapy: https://scapy.net +Every capture is built here, so the module is in the unit tier. """ from __future__ import annotations @@ -31,7 +30,7 @@ from tests._support import close_extractor, reimport_once_per_class from tests.foundation import _roundtrip as wire -from tests.toolkit.test_tunnelled_tcp_flow_key_1581_unit import INNER, OUTER, has +from tests.toolkit.test_tunnelled_tcp_flow_key_1581_unit import INNER, OUTER from tests.toolkit.test_tunnelled_tcp_flow_key_1581_unit import ip as tunnel_ip if TYPE_CHECKING: @@ -219,15 +218,13 @@ def spy_trace(this: 'Any', packet: 'Any') -> 'Any': @unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') -class DPKTScapyIPv4HeaderCheckTests(unittest.TestCase): - """dpkt and scapy hand on what the default engine does, whatever the IPv4 header.""" - - ENGINES = ('dpkt', 'scapy') +class DefaultEngineIPv4HeaderCheckTests(unittest.TestCase): + """The default engine reads nothing behind an IPv4 header it rejects.""" def setUp(self) -> None: reimport_once_per_class(self) - def test_rejected_ipv4_headers_agree_with_the_default_engine(self) -> None: + def test_nothing_is_read_behind_a_rejected_ipv4_header(self) -> None: built, tcp, ipv4_inputs = frames() base = read(self, 'default', built) # nothing is read behind a rejected header, outer or inner @@ -236,19 +233,10 @@ def test_rejected_ipv4_headers_agree_with_the_default_engine(self) -> None: self.assertEqual(len(base['ids']), len(tcp)) # and a rejected first header is no IPv4 input either self.assertEqual(sorted(base['ipv4']), ipv4_inputs) - for engine in self.ENGINES: - with self.subTest(engine=engine): - if not has(engine): - self.skipTest(f'{engine} not installed') - seen = read(self, engine, built) - for aspect in ('ipv4', 'tcp', 'trace', 'ids', 'flows'): - self.assertEqual(seen[aspect], base[aspect], aspect) - - def test_every_version_and_ihl_is_read_as_the_default_engine_reads_it(self) -> None: - # The shortcut that leaves a header of version 4 and IHL 5 unparsed - # answers as the parser does: for every first octet, the engines decode - # an IPv4 layer and accept it exactly when the default engine dissects - # one, with room for 40 octets of options and with none at all. + + def test_only_version_4_with_room_for_its_ihl_is_dissected(self) -> None: + # For every first octet, with room for 40 octets of options and with + # none at all. from pcapkit import extract from pcapkit.protocols.internet.ipv4 import IPv4 @@ -263,94 +251,6 @@ def test_every_version_and_ihl_is_read_as_the_default_engine_reads_it(self) -> N self.assertEqual(len(want), len(built)) self.assertEqual(sum(want), 11 + 1) # version 4: every IHL from 5 with room, IHL 5 without - for engine in self.ENGINES: - with self.subTest(engine=engine): - if not has(engine): - self.skipTest(f'{engine} not installed') - if engine == 'dpkt': - import dpkt - - from pcapkit.toolkit.dpkt import _ipv4_accepted - layers = [getattr(dpkt.ethernet.Ethernet(octets), 'ip', None) for octets in built] - else: - from scapy.layers.inet import IP - from scapy.layers.l2 import Ether - - from pcapkit.toolkit.scapy import _ipv4_accepted # type: ignore[no-redef] - layers = [Ether(octets).getlayer(IP) for octets in built] - with warnings.catch_warnings(): - warnings.simplefilter('ignore') - got = [layer is not None and _ipv4_accepted(layer) for layer in layers] - self.assertEqual([(first, len(rest)) for (first, rest), mine, theirs in zip(cases, got, want) - if mine != theirs], []) - - def test_an_ihl_below_5_warns_of_nothing_the_default_engine_does_not(self) -> None: - # Scapy dissects such a header; the parser is handed its 20 fixed octets, - # as the default engine is, rather than the fewer the IHL counts, which - # it would warn of reading past. - if not has('scapy'): - self.skipTest('scapy not installed') - from scapy.layers.inet import IP - from scapy.layers.l2 import Ether - - from pcapkit.toolkit import pypcapfile - from pcapkit.toolkit.scapy import _ipv4_accepted - - for ihl in range(5): - with self.subTest(ihl=ihl): - octets = ethernet(ipv4(segment(1), ihl=ihl)) - pypcapfile._default_accepts.cache_clear() - with warnings.catch_warnings(record=True) as caught: - warnings.simplefilter('always') - accepted = _ipv4_accepted(Ether(octets)[IP]) - self.assertFalse(accepted) - self.assertEqual([str(warning.message) for warning in caught], []) - - def test_the_parser_is_asked_only_of_a_header_it_could_reject(self) -> None: - # A plain header is not parsed; one with options is, once between the - # adapters that read the frame. - from pcapkit.const.reg.linktype import LinkType - from pcapkit.toolkit import pypcapfile - - plain = ethernet(ipv4(segment(1))) - tunnelled = ethernet(ipv4(ipv4(segment(2)), proto=4)) - with_options = ethernet(ipv4(segment(3), options=GOOD_OPTIONS['router alert'])) - for engine in self.ENGINES: - with self.subTest(engine=engine): - if not has(engine): - self.skipTest(f'{engine} not installed') - if engine == 'dpkt': - import dpkt - - from pcapkit.toolkit import dpkt as toolkit - - def adapt(octets: bytes) -> None: - packet = dpkt.ethernet.Ethernet(octets) - toolkit.ipv4_reassembly(packet, 1.0) - toolkit.tcp_reassembly(packet, 1.0) - toolkit.tcp_traceflow(packet, 1.0, data_link=LinkType.ETHERNET) - else: - from scapy.layers.l2 import Ether - - from pcapkit.toolkit import scapy as toolkit # type: ignore[no-redef] - - def adapt(octets: bytes) -> None: - packet = Ether(octets) - packet.time = 1.0 - toolkit.ipv4_reassembly(packet) - toolkit.tcp_reassembly(packet) - toolkit.tcp_traceflow(packet) - - pypcapfile._default_accepts.cache_clear() - real = pypcapfile.Protocol_IPv4 - with mock.patch.object(pypcapfile, 'Protocol_IPv4', side_effect=real) as parser: - adapt(plain) - adapt(tunnelled) - self.assertEqual(parser.call_count, 0) - adapt(with_options) - self.assertEqual([call.args[0] for call in parser.call_args_list], - [with_options[14:38]]) - if __name__ == '__main__': unittest.main() diff --git a/tests/toolkit/test_engine_pcap_trace_runtime.py b/tests/toolkit/test_engine_pcap_trace_runtime.py index bb6dc144fb..2f921f5f2e 100644 --- a/tests/toolkit/test_engine_pcap_trace_runtime.py +++ b/tests/toolkit/test_engine_pcap_trace_runtime.py @@ -5,9 +5,11 @@ length and its link type, so their flow tracing adapters carry a PCAP frame record beside the frame mapping they always reported, and :class:`~pcapkit.foundation.extraction.Extractor` no longer swaps the format for -JSON. The checks are the strongest available: every traced PCAP file is -byte-compared with the default engine's for the same input, and every JSON trace -with the one main wrote before. +JSON. Every traced PCAP file is checked against the capture it was traced from: +it holds its flow's records as captured, octets, original length and timestamp +alike, the timestamp in the trace's resolution. The records are read from the +capture by hand, so no engine's reading is the reference. Every JSON trace is +checked against the one main wrote before. The inputs no sample has are synthesised from :file:`tcp.pcap`: nanosecond PCAP and PCAP-NG (``if_tsresol=9``) captures, each with a timestamp that has @@ -52,6 +54,21 @@ #: none at all, which leaves a nanosecond count with a microsecond count's digits. EXTRA_NS = {'unaligned': 789, 'aligned': 0} +#: The flow labels of ``tcp.pcap`` made nanoseconds plus each of :data:`EXTRA_NS`: +#: the first frame's timestamp as a :obj:`float`, ``1500000000.000774789`` being +#: ``1500000000.0007749``. +NS_LABELS = { + 'unaligned': ['10.20.30.130_22-10.20.30.131_53406-1500000000.0007749', + 'fe80..a6.87f9.2793.16ee_51774-fe80..1ccd.7c77.bac7.46b7_22-1500000000.0024338'], + 'aligned': ['10.20.30.130_22-10.20.30.131_53406-1500000000.000774', + 'fe80..a6.87f9.2793.16ee_51774-fe80..1ccd.7c77.bac7.46b7_22-1500000000.002433'], +} + +#: The frames of each flow dpkt traces in ``tcp.pcap`` as ``LINKTYPE_IPV4``, and in +#: ``http6.cap`` as ``LINKTYPE_IPV6``, each with its Ethernet headers stripped. +RAW_TCP_FLOWS = [(1, 2, 6, 7)] +RAW_HTTP6_FLOWS = [tuple(range(1, 16)), tuple(range(16, 27))] + #: SHA-256 of each JSON flow trace of ``tcp.pcap``, as main wrote it at 6af87f608. JSON_TRACE_SHA256 = { 'dpkt': { @@ -83,6 +100,42 @@ def _records(source: 'str | bytes') -> 'tuple[bytes, list[list]]': return header, records +def _captured(source: 'str') -> 'list[list]': + """The ``[ts_sec, ts_nsec, orig_len, octets]`` of every frame of a little-endian + PCAP, or of the Enhanced Packet Blocks of a little-endian PCAP-NG, read by hand.""" + with open(source, 'rb') as file: + data = file.read() + if data[:4] in MAGIC.values(): + scale = 1 if data[:4] == MAGIC[True] else 1000 + return [[ts_sec, fraction * scale, orig_len, octets] + for ts_sec, fraction, orig_len, octets in _records(data)[1]] + assert data[8:12] == bytes.fromhex('4d3c2b1a'), f'{source}: not a little-endian PCAP or PCAP-NG' + records, resolutions, offset = [], [], 0 # type: list[list], list[int], int + while offset < len(data): + block_type, length = struct.unpack_from(' 'str': with open(path, 'wb') as file: file.write(header) @@ -114,7 +167,7 @@ def _write_pcapng(path: 'str', records: 'list[list]', *, tsresol: 'int') -> 'str @unittest.skipUnless(HAS_RUNTIME and ENGINES, 'runtime dependencies, dpkt or scapy not installed') class EnginePCAPTraceTests(unittest.TestCase): - """A traced flow is the same PCAP file whichever engine read the capture.""" + """Each engine writes a traced flow as a PCAP file of the flow's captured records.""" def setUp(self) -> None: reimport_once_per_class(self) @@ -149,18 +202,28 @@ def _flows(self, engine: 'str', fin: 'str', *, nanosecond: 'bool' = False, flows.append((flow.label, tuple(flow.index), file.read())) return flows - def _assert_same_files(self, fin: 'str', *, nanosecond: 'bool', - caught: 'list | None' = None) -> None: + def _assert_holds_the_records(self, fin: 'str', flows: 'list[tuple[str, tuple, bytes]]', *, + nanosecond: 'bool' = False) -> None: + """Each of ``flows`` is a PCAP file of its frames' records as ``fin`` holds them.""" + captured = _captured(fin) + self.assertTrue(flows) + for label, index, data in flows: + self.assertEqual(data[:4], MAGIC[nanosecond], label) + self.assertEqual(_records(data)[1], + [[ts_sec, ts_nsec if nanosecond else ts_nsec // 1000, orig_len, octets] + for ts_sec, ts_nsec, orig_len, octets in + (captured[number - 1] for number in index)], label) + + def _assert_every_engine_holds_the_records(self, fin: 'str', *, nanosecond: 'bool', + labels: 'list[str] | None' = None) -> None: + """Every engine's flows of ``fin`` hold their records, and are labelled ``labels`` if given.""" skip_if_stand_in(self, fin) - base = self._flows('default', fin, nanosecond=nanosecond) - self.assertTrue(base) for engine in ENGINES: with self.subTest(engine=engine): - flows = self._flows(engine, fin, nanosecond=nanosecond, caught=caught) - self.assertEqual([flow[:2] for flow in flows], [flow[:2] for flow in base]) - for (label, _, mine), (_, _, theirs) in zip(flows, base): - self.assertEqual(mine[:4], MAGIC[nanosecond], label) - self.assertEqual(mine, theirs, label) + flows = self._flows(engine, fin, nanosecond=nanosecond) + self._assert_holds_the_records(fin, flows, nanosecond=nanosecond) + if labels is not None: + self.assertEqual([label for label, _, _ in flows], labels) def _nanosecond_records(self, extra: 'int') -> 'list[list]': """``tcp.pcap``'s records, each microsecond fraction made nanoseconds plus ``extra``.""" @@ -168,16 +231,16 @@ def _nanosecond_records(self, extra: 'int') -> 'list[list]': return [[ts_sec, ts_usec * 1000 + extra, orig_len, octets] for ts_sec, ts_usec, orig_len, octets in records] - def test_traced_pcap_matches_the_default_engine_byte_for_byte(self) -> None: + def test_each_traced_pcap_holds_the_captured_records(self) -> None: for capture in CAPTURES: for nanosecond in (False, True): with self.subTest(capture=capture, nanosecond=nanosecond): - self._assert_same_files(sample_path(capture), nanosecond=nanosecond) + self._assert_every_engine_holds_the_records(sample_path(capture), nanosecond=nanosecond) def test_a_stand_in_capture_skips_rather_than_fails(self) -> None: # When the Wireshark download fails the generator writes a stand-in under the # upstream name, built from DHCP, so it has no TCP flow to trace (#1478). The - # comparison needs one, so on a stand-in it skips rather than fails. + # check needs one, so on a stand-in it skips rather than fails. spec = importlib.util.spec_from_file_location('_pcapng_generator_stand_in', PCAPNG_GENERATOR) assert spec is not None and spec.loader is not None generator = importlib.util.module_from_spec(spec) @@ -190,35 +253,37 @@ def test_a_stand_in_capture_skips_rather_than_fails(self) -> None: with open(fin, 'wb') as file: file.write(generator.build_many_interfaces()) with self.assertRaises(unittest.SkipTest): - self._assert_same_files(fin, nanosecond=False) + self._assert_every_engine_holds_the_records(fin, nanosecond=False) def test_a_nanosecond_pcap_keeps_its_resolution(self) -> None: - # The record fraction stays in the capture's own resolution, as the default - # engine's does. The labels agree too: the flow tracer is handed a float, - # not the Decimal dpkt reads a nanosecond PCAP's timestamp as (#1501). + # The record fraction stays in the capture's own resolution. The flow + # tracer is handed a float, not the Decimal dpkt reads a nanosecond PCAP's + # timestamp as (#1501), so the labels carry the float's digits. header, _ = _records(sample_path('tcp.pcap')) for case, extra in EXTRA_NS.items(): fin = _write(os.path.join(self.tmp, f'{case}.pcap'), MAGIC[True] + header[4:], self._nanosecond_records(extra)) for nanosecond in (False, True): with self.subTest(case=case, nanosecond=nanosecond): - self._assert_same_files(fin, nanosecond=nanosecond) + self._assert_every_engine_holds_the_records(fin, nanosecond=nanosecond, + labels=NS_LABELS[case]) @unittest.skipUnless(HAS_DPKT, 'dpkt not installed') - def test_a_raw_ip_link_type_traces_as_the_default_engine(self) -> None: + def test_a_raw_ip_link_type_is_traced(self) -> None: # With the Ethernet header stripped, dpkt reads each frame as a bare IP or # IP6, which names no ``ip`` or ``ip6`` layer, so it traced nothing (#1548). - # Every frame is kept, IPv6 ones under LINKTYPE_IPV4 included. - for capture, linktype in (('tcp.pcap', 228), ('http6.cap', 229)): + # The capture keeps every frame, its IPv6 ones under LINKTYPE_IPV4 + # included, and those are not traced. + for capture, linktype, flows in (('tcp.pcap', 228, RAW_TCP_FLOWS), ('http6.cap', 229, RAW_HTTP6_FLOWS)): header, records = _records(sample_path(capture)) fin = _write(os.path.join(self.tmp, f'raw-{linktype}.pcap'), header[:20] + struct.pack(' None: # ``if_tsresol=9``: an unaligned timestamp must keep its sub-microsecond @@ -229,11 +294,10 @@ def test_a_nanosecond_pcapng_keeps_its_resolution(self) -> None: self._nanosecond_records(extra), tsresol=9) for nanosecond in (False, True): with self.subTest(case=case, nanosecond=nanosecond): - self._assert_same_files(fin, nanosecond=nanosecond) + self._assert_every_engine_holds_the_records(fin, nanosecond=nanosecond) def test_a_cross_resolution_trace_rescales_the_fraction(self) -> None: - # Not only the same bytes as the default engine: the right ones. The first - # frame of tcp.pcap is at 1500000000.000774; the synthetic nanosecond + # The first frame of tcp.pcap is at 1500000000.000774; the synthetic nanosecond # captures put it at 1500000000.000774789. A trace in the other resolution # than its capture's scales the fraction, truncating nanoseconds (#1529). header, _ = _records(sample_path('tcp.pcap')) @@ -255,8 +319,8 @@ def test_a_cross_resolution_trace_rescales_the_fraction(self) -> None: def test_a_snapped_frame_keeps_its_original_length(self) -> None: # Both engines' readers report a record's original length, PCAP and - # PCAP-NG alike, so a frame the snapshot length cut short is written as - # the default engine writes it. + # PCAP-NG alike, so a frame the snapshot length cut short is written with + # the original length it was captured with. header, records = _records(sample_path('tcp.pcap')) longest = max(range(len(records)), key=lambda index: len(records[index][3])) records[longest][3] = records[longest][3][:-4] @@ -268,7 +332,7 @@ def test_a_snapped_frame_keeps_its_original_length(self) -> None: } for container, fin in captures.items(): with self.subTest(container=container): - self._assert_same_files(fin, nanosecond=False) + self._assert_every_engine_holds_the_records(fin, nanosecond=False) def test_no_engine_warns_about_original_lengths(self) -> None: # The frame record is complete, so there is nothing to warn about -- not on @@ -361,11 +425,11 @@ def __iter__(self) -> 'Any': hdr = header(buf) yield hdr.tv_sec + (hdr.tv_usec / divisor), file.read(hdr.caplen) - base = self._flows('default', sample_path('tcp.pcap')) caught = [] # type: list with mock.patch.object(dpkt.pcap, 'Reader', Renamed): flows = self._flows('dpkt', sample_path('tcp.pcap'), caught=caught) - self.assertEqual(flows, base) + # tcp.pcap captured every frame whole, so the original length lost is the one kept + self._assert_holds_the_records(sample_path('tcp.pcap'), flows) messages = [str(w.message) for w in caught if issubclass(w.category, DPKTWarning) and 'original length' in str(w.message)] self.assertEqual(len(messages), 1, messages) diff --git a/tests/toolkit/test_tcp_carrier_walk_1606_1617_unit.py b/tests/toolkit/test_tcp_carrier_walk_1606_1617_unit.py index 2a6d19b522..3521067ec0 100644 --- a/tests/toolkit/test_tcp_carrier_walk_1606_1617_unit.py +++ b/tests/toolkit/test_tcp_carrier_walk_1606_1617_unit.py @@ -1,21 +1,21 @@ # -*- coding: utf-8 -*- -"""The dpkt and scapy engines find TCP in IPv4 where the default engine does. C.f. #1606, #1617. +"""TCP carried in IPv4 behind an extension header or in a later fragment. C.f. #1606, #1617. The default engine dispatches an IPv4 packet's protocol through the registry IPv6 uses, so it reads an AH or IPv6 extension header carried directly in IPv4, and the TCP segment or IP-in-IP tunnel behind it, up to :data:`~pcapkit.protocols.internet.internet.EXTENSION_HEADER_LIMIT` headers in a -row. `dpkt`_ decodes protocol 0 as IPv4 and keeps protocols 43, 44, 60, 135 and -139 as octets, and `Scapy`_ keeps all of them as ``Raw`` but AH, behind which it -dissects no TCP; so neither engine handed that TCP on (#1606). Both now step over -the headers by the default engine's own rules, as the pypcapfile engine does -(:func:`pcapkit.toolkit.pypcapfile._past_extension_headers`). +row (#1606). A later IPv4 fragment carries no upper-layer header (:rfc:`791#section-3.2`), so the default engine reads no TCP in one, but `Scapy`_ binds IPv6 to IPv4 protocol 41, and AH to protocol 51, whatever the fragment offset, and its engine handed on the TCP it then dissected (#1617). It now stops at such a fragment. +`dpkt`_ decodes protocol 0 as IPv4, so a Hop-by-Hop header there can read as an +IPv4 packet dpkt cannot decode, or whose payload is no packet; neither ends the +extraction. + Every capture is built here, so the module is in the unit tier. .. _dpkt: https://dpkt.readthedocs.io @@ -41,9 +41,6 @@ RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) -#: The engines compared with the default engine. -ENGINES = ('dpkt', 'scapy') - def has(engine: str) -> bool: """Whether ``engine``'s third-party package is installed.""" @@ -217,66 +214,45 @@ def spy_trace(this: 'Any', packet: 'Any') -> 'Any': return seen +#: The flow of the second frame of the dpkt cases below, ``tcp(2)`` in plain IPv4. +NEXT_FLOW = ('192.0.2.1_1002-198.51.100.2_9-1.000001', (2,)) + + @unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') class TCPCarrierWalkTests(unittest.TestCase): - """dpkt and scapy hand on the TCP the default engine reads in IPv4, and no other.""" + """Where the engines read TCP in IPv4, and what dpkt survives on the way.""" def setUp(self) -> None: reimport_once_per_class(self) - def agree(self, cases: 'dict[str, tuple[bool, Callable[[int], bytes]]]') -> None: + def assert_reads(self, engine: str, cases: 'dict[str, tuple[bool, Callable[[int], bytes]]]') -> None: + """``engine`` reads TCP in exactly the frames of ``cases`` marked to be read.""" frames, reads = build(cases) - base = read(self, 'default', frames) - # what the default engine reads, as the cases say - self.assertEqual(sorted(base['tcp']), reads) - self.assertEqual(sorted(base['trace']), reads) - self.assertEqual(len(base['ids']), len(reads)) + seen = read(self, engine, frames) names = dict(enumerate(cases, start=1)) - for engine in ENGINES: - with self.subTest(engine=engine): - if not has(engine): - self.skipTest(f'{engine} not installed') - seen = read(self, engine, frames) - for aspect in ('tcp', 'trace'): - self.assertEqual( - [names[number] for number in sorted(set(seen[aspect]) ^ set(base[aspect]))], [], - f'{aspect}: frames read by one of {engine} and the default engine alone') - self.assertEqual(seen[aspect], base[aspect], aspect) - for aspect in ('ids', 'flows'): - self.assertEqual(seen[aspect], base[aspect], aspect) - - def test_tcp_behind_an_extension_header_in_ipv4_is_read_as_the_default_engine_reads_it(self) -> None: + for aspect in ('tcp', 'trace'): + self.assertEqual([names[number] for number in sorted(set(seen[aspect]) ^ set(reads))], [], + f'{aspect}: frames read, or left unread, against the cases') + self.assertEqual(len(seen['ids']), len(reads)) + self.assertEqual([index for _, index in seen['flows']], [(number,) for number in reads]) + + def test_the_default_engine_reads_tcp_behind_an_extension_header_in_ipv4(self) -> None: # #1606 - self.agree(EXTENSION_HEADERS) + self.assert_reads('default', EXTENSION_HEADERS) def test_no_tcp_is_read_in_a_later_ipv4_fragment(self) -> None: - # #1617 - self.agree(FRAGMENTS) - - def test_a_dpkt_packet_built_by_hand_is_walked_too(self) -> None: - # With no record octets attached, the IPv4 payload is DPKT's own - # re-serialisation of it, which here is the octets it was parsed from. - if not has('dpkt'): - self.skipTest('dpkt not installed') - import dpkt - - from pcapkit.toolkit import dpkt as toolkit - - segment = tcp(1, opts=b'\x02\x04\x05\xb4') - for name, packet in (('hop-by-hop', v4(options(6) + segment, 0)), - ('ah', v4(ah(6) + segment, 51)), - ('4in4 behind hop-by-hop', v4(options(4) + v4(segment, 6), 0))): - with self.subTest(case=name): - data = toolkit.tcp_reassembly(dpkt.ethernet.Ethernet(eth(packet)), 1.0) - self.assertIsNotNone(data) - self.assertEqual((data.header, bytes(data.payload)), (segment[:24], segment[24:])) + # #1617: Scapy dissected TCP behind a later fragment + for engine in ('default', 'scapy'): + with self.subTest(engine=engine): + if engine != 'default' and not has(engine): + self.skipTest(f'{engine} not installed') + self.assert_reads(engine, FRAGMENTS) def test_a_tunnel_dpkt_cannot_decode_ends_no_extraction(self) -> None: # dpkt 1.9.8 reads ``frag_off`` from whatever header follows an IPv6 # Fragment header (#1351), so the IPv6 packet behind this Hop-by-Hop - # header raises AttributeError. The next frame is still read, and the - # TCP in this one is read off the octets, as the default engine reads - # it (#1626). + # header raises AttributeError. The extraction goes on, and the next + # frame is still read (#1626). if not has('dpkt'): self.skipTest('dpkt not installed') import dpkt @@ -285,7 +261,8 @@ def test_a_tunnel_dpkt_cannot_decode_ends_no_extraction(self) -> None: with self.assertRaises(AttributeError): dpkt.ip6.IP6(inner) seen = read(self, 'dpkt', [eth(v4(options(41) + inner, 0)), eth(v4(tcp(2), 6))]) - self.assertEqual((sorted(seen['tcp']), sorted(seen['trace'])), ([1, 2], [1, 2])) + self.assertIn(2, seen['tcp']) + self.assertIn(NEXT_FLOW, seen['flows']) def test_a_dpkt_payload_that_is_no_packet_is_leaf_data(self) -> None: # dpkt decodes IPv4 protocol 0 as IPv4, so this Hop-by-Hop header and the @@ -305,12 +282,10 @@ def test_a_dpkt_payload_that_is_no_packet_is_leaf_data(self) -> None: info = toolkit.packet2dict(packet, 1.0, data_link=LinkType.ETHERNET) self.assertEqual(sorted(info['ETHERNET']['IP']['IP']['VRRP']), sorted(dpkt.vrrp.VRRP.__hdr_fields__)) - frames = [octets, eth(v4(tcp(2), 6))] - base = read(self, 'default', frames) - self.assertEqual((sorted(base['tcp']), sorted(base['trace'])), ([1, 2], [1, 2])) - seen = read(self, 'dpkt', frames) - for aspect in ('tcp', 'trace', 'ids', 'flows'): - self.assertEqual(seen[aspect], base[aspect], aspect) + # and the extraction goes on to the next frame + seen = read(self, 'dpkt', [octets, eth(v4(tcp(2), 6))]) + self.assertIn(2, seen['tcp']) + self.assertIn(NEXT_FLOW, seen['flows']) def test_a_dpkt_payload_that_is_no_packet_ends_the_verbose_chain(self) -> None: # The same frame as above, whose chain ``packet2chain`` prints for each diff --git a/tests/toolkit/test_tunnelled_tcp_flow_key_1581_unit.py b/tests/toolkit/test_tunnelled_tcp_flow_key_1581_unit.py index 2b31dd0e62..443b72d5bc 100644 --- a/tests/toolkit/test_tunnelled_tcp_flow_key_1581_unit.py +++ b/tests/toolkit/test_tunnelled_tcp_flow_key_1581_unit.py @@ -5,10 +5,8 @@ *first* IP layer, which in a tunnel is the tunnel's own: every connection through one tunnel that shares a port pair merged into one flow, under the tunnel's endpoints. The default engine, scapy and pyshark did so (scapy found the inner -IPv4 layer of 4in6 only), and dpkt found no TCP in a tunnel at all. Each now -keys it by the IP layer nearest above the TCP segment, as Wireshark's TCP -conversations do -- and dpkt enters only the tunnels the default engine -dissects, protocols 4 and 41, not the protocol 0 it also decodes as IPv4. +IPv4 layer of 4in6 only). Each now keys it by the IP layer nearest above the TCP +segment, as Wireshark's TCP conversations do. Every capture is built here, so the module is in the unit tier. pyshark runs only where the interpreter supports it, with tshark on the path. The @@ -47,8 +45,7 @@ #: Each tunnel, named inner version first: ``6in4`` is IPv6 carried in IPv4. KINDS = {'4in4': (4, 4), '6in4': (6, 4), '4in6': (4, 6), '6in6': (6, 6)} #: Engines that report dissected fields only: no TCP reassembly, and no TCP -#: octets in what they hand flow tracing (as ``NO_TCP_OCTETS`` in -#: :mod:`tests.foundation.engines.test_engine_agreement_runtime`). +#: octets in what they hand flow tracing. FIELDS_ONLY = frozenset(('pyshark',)) @@ -155,37 +152,13 @@ def spy_trace(this: 'Any', packet: 'Any') -> 'Any': @unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') class TunnelledTCPFlowKeyTests(unittest.TestCase): - """Every engine keys tunnelled TCP by the innermost IP header, as the default engine does.""" + """The default engine, scapy and pyshark key tunnelled TCP by the innermost IP header.""" - ENGINES = ('default', 'dpkt', 'scapy', 'pyshark') + ENGINES = ('default', 'scapy', 'pyshark') def setUp(self) -> None: reimport_once_per_class(self) - def agree(self, frames: 'list[bytes]', engines: 'tuple[str, ...]' = ENGINES[1:]) -> 'dict[str, Any]': - """Check each of ``engines`` hands on what the default engine does for ``frames``. - - That is every field of every TCP reassembly and flow tracing input, the - reassembled datagrams' ``id`` and the traced flows -- for an engine in - :data:`FIELDS_ONLY`, the flow tracing inputs less their octets, and the - flows. Returns the default engine's. - - """ - base = read(self, 'default', frames) - for engine in engines: - with self.subTest(engine=engine): - if not has(engine): - self.skipTest(f'{engine} not installed') - seen = read(self, engine, frames) - want = base - if engine in FIELDS_ONLY: - want = {'flows': base['flows'], 'trace': { - number: {key: value for key, value in record.items() if key not in ('header', 'payload')} - for number, record in base['trace'].items()}} - for aspect in want: - self.assertEqual(seen[aspect], want[aspect], aspect) - return base - def test_two_connections_through_one_tunnel_are_two_flows(self) -> None: for engine in self.ENGINES: for kind, (inner, _) in KINDS.items(): @@ -210,18 +183,16 @@ def test_two_connections_through_one_tunnel_are_two_flows(self) -> None: for record in seen['tcp'].values()}), pairs) def test_protocol_0_is_no_tunnel(self) -> None: - # dpkt decodes an IPv4 protocol 0 as IPv4, its IP_PROTO_IP; the default - # engine reads an IPv6 Hop-by-Hop Options header there, and so finds no - # TCP in the IPv4 packet behind it. + # The default engine reads an IPv6 Hop-by-Hop Options header at an IPv4 + # protocol 0, and so finds no TCP in the IPv4 packet behind it. frames = [frame(ip(4, ip(4, segment, *INNER[4][connection], 6), *OUTER[4], 0)) for connection, segment in enumerate((wire.tcp(b'', seq=100, syn=True), wire.tcp(b'hello', seq=101, ack=1)))] - base = self.agree(frames) + base = read(self, 'default', frames) self.assertEqual((base['trace'], base['flows'], base['ids']), ({}, [], [])) - def test_tunnelled_segments_agree_with_the_default_engine(self) -> None: - # The other engines hand reassembly and flow tracing every field the - # default engine does, however the tunnel is built. + def test_the_default_engine_traces_every_tunnelled_segment(self) -> None: + # However the tunnel is built. segment = wire.tcp(b'payload', seq=7, ack=1) # a Data Offset of 6 words in a 20-octet segment, which the TCP parser # rejects and the default engine reads out of the raw octets (#1518) @@ -242,18 +213,17 @@ def test_tunnelled_segments_agree_with_the_default_engine(self) -> None: frame(ip(6, hop_by_hop + ip(4, segment, *INNER[4][1], 6), *OUTER[6], 0)), tunnelled('4in4', wire.tcp(b'', seq=1, ack=1)) + bytes(6), ] - base = self.agree(frames) + base = read(self, 'default', frames) self.assertEqual(sorted(base['trace']), list(range(1, len(frames) + 1))) - def test_tunnelled_first_fragments_agree_with_the_default_engine(self) -> None: - # pyshark too, since it turns tshark's IP reassembly off (#1612) + def test_the_default_engine_traces_tunnelled_first_fragments(self) -> None: segment = wire.tcp(b'payload', seq=7, ack=1) frames = [ # the tunnel's first fragment, then the inner datagram's frame(ip(4, ip(4, segment, *INNER[4][1], 6), *OUTER[4], 4, mf=True)), frame(ip(4, ip(4, segment, *INNER[4][1], 6, mf=True), *OUTER[4], 4)), ] - base = self.agree(frames) + base = read(self, 'default', frames) self.assertEqual(sorted(base['trace']), [1, 2]) def test_the_default_engine_reads_the_inner_header(self) -> None: