From 71df9d4333c7804fbcc19ef38954e4e2464534a6 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Fri, 9 Oct 2026 23:55:00 -0400 Subject: [PATCH] fix(engines): take a scapy packet's link type and clock from its interface (#1503, #1517, #1549) - A Simple Packet Block has no timestamp, and scapy left the packet with the time it was built; the engine's PCAP-NG reader now sets it to 0, as the default engine reads it (#1503). - The flow tracer looked the link type up by the first layer's name, so a raw IPv4 packet (`IP`) raised MissingKeyError. Both readers now attach the interface's link type with `attach_linktype`, and `tcp_traceflow` uses it; only a packet built by hand is still looked up by name, with no default (#1517). - scapy ignores `if_tsoffset`; the PCAP-NG reader now reads option 14 of each interface itself and adds it to every timestamp on it (#1549). - The overrides hold on scapy 2.5, 2.7 and 2.8: `_check_interface_id` passes its result through (2.8 skips a block on a false one), the offsets are kept by the reader rather than in scapy's interface tuple, and a missing hook raises VersionError. - Document `packet2frame`, `attach_resolution` and `attach_linktype`; delete the #1503 Gap row of the engine agreement table. New runtime and unit tests fail on main; scapy, trace and agreement selections pass on scapy 2.7.0 and 2.8.0. --- docs/source/pcapkit/toolkit/3rdparty.rst | 6 + pcapkit/foundation/engines/scapy.py | 106 ++++++- pcapkit/toolkit/scapy.py | 69 +++-- .../engines/test_engine_agreement_runtime.py | 4 - .../engines/test_scapy_reader_hooks_unit.py | 119 ++++++++ .../test_scapy_reader_interfaces_runtime.py | 268 ++++++++++++++++++ tests/toolkit/test_scapy_linktype_unit.py | 76 +++++ 7 files changed, 613 insertions(+), 35 deletions(-) create mode 100644 tests/foundation/engines/test_scapy_reader_hooks_unit.py create mode 100644 tests/foundation/engines/test_scapy_reader_interfaces_runtime.py create mode 100644 tests/toolkit/test_scapy_linktype_unit.py diff --git a/docs/source/pcapkit/toolkit/3rdparty.rst b/docs/source/pcapkit/toolkit/3rdparty.rst index 90c4fe8289..c46c680642 100644 --- a/docs/source/pcapkit/toolkit/3rdparty.rst +++ b/docs/source/pcapkit/toolkit/3rdparty.rst @@ -33,6 +33,12 @@ Auxiliary Functions .. autofunction:: pcapkit.toolkit.scapy.packet2dict +.. autofunction:: pcapkit.toolkit.scapy.packet2frame + +.. autofunction:: pcapkit.toolkit.scapy.attach_resolution + +.. autofunction:: pcapkit.toolkit.scapy.attach_linktype + DPKT Tools ========== diff --git a/pcapkit/foundation/engines/scapy.py b/pcapkit/foundation/engines/scapy.py index fb5a790832..0cd46a6fba 100644 --- a/pcapkit/foundation/engines/scapy.py +++ b/pcapkit/foundation/engines/scapy.py @@ -37,10 +37,11 @@ warnings.filterwarnings('ignore', category=CryptographyDeprecationWarning) """ +import struct from typing import TYPE_CHECKING, cast from pcapkit.foundation.engines.engine import EngineBase -from pcapkit.utilities.exceptions import stacklevel +from pcapkit.utilities.exceptions import VersionError, stacklevel from pcapkit.utilities.logging import get_logger from pcapkit.utilities.warnings import AttributeWarning, warn @@ -58,17 +59,52 @@ #: :data:`pcapkit.utilities.logging.logger`. logger = get_logger(__name__) +#: The private methods of :class:`scapy.utils.PcapNgReader` that +#: :func:`_reader_type` overrides. +_READER_HOOKS = ('_read_block', '_read_block_shb', '_read_block_idb', '_check_interface_id') + + +def _read_tsoffset(options: 'bytes', endian: 'str') -> 'int': + """Read the ``if_tsoffset`` of a PCAP-NG Interface Description Block (#1549). + + Read here rather than by `Scapy`_'s ``_read_options``, whose result differs + by version: keyed by option code in 2.7 and 2.8, and in 2.5 by name, with + only ``tsresol`` and ``comment`` kept. + + Args: + options: The block's options area. + endian: :mod:`struct` byte order of the section, ``'<'`` or ``'>'``. + + Returns: + The offset (option 14), a signed count of seconds; 0 without one. + + .. _Scapy: https://scapy.net + + """ + while len(options) >= 4: + code, length = struct.unpack(f'{endian}HH', options[:4]) + if not code: # opt_endofopt + break + if code == 14 and length == 8 and len(options) >= 12: + return cast('int', struct.unpack(f'{endian}q', options[4:12])[0]) + options = options[4 + length + (-length % 4):] + return 0 + + def _reader_type() -> 'Type[PcapReader]': - """A :class:`scapy.utils.PcapReader` that stamps each packet with its resolution. + """A :class:`scapy.utils.PcapReader` that stamps each packet with its interface. `Scapy`_'s readers convert a record's timestamp to a :class:`~decimal.Decimal` and drop the resolution it was counted in, and an exact count of microseconds comes out with the same digits whether the capture counted microseconds or - nanoseconds. The PCAP frame record of a packet needs that resolution, so these - subclasses attach it with :func:`pcapkit.toolkit.scapy.attach_resolution`: - from the global header of a PCAP, and from the interface of each PCAP-NG - packet block, which :meth:`~scapy.utils.RawPcapNgReader._read_block` is the - last place to see. + nanoseconds. They drop the link type too, once it has picked the packet's + first layer, and a layer does not name its link type: a raw IPv4 packet's is + :class:`~scapy.layers.inet.IP` (#1517). The PCAP frame record and the flow + tracing input of a packet need both, so these subclasses attach them with + :func:`pcapkit.toolkit.scapy.attach_resolution` and + :func:`pcapkit.toolkit.scapy.attach_linktype`: from the global header of a + PCAP, and from the interface of each PCAP-NG packet block, which + :meth:`~scapy.utils.RawPcapNgReader._read_block` is the last place to see. The PCAP-NG subclass also starts each section with no interfaces, as the PCAP-NG specification defines a section: an interface ID counts the Interface @@ -76,6 +112,12 @@ def _reader_type() -> 'Type[PcapReader]': to one table across Section Header Blocks, so a later section's interface 0 resolved to the first section's, with its link type and resolution (#1522). + And it dates two kinds of packet as the default engine does. A Simple Packet + Block has no timestamp, and `Scapy`_ leaves its packet with the time it was + built, so it is set to 0 (#1503). `Scapy`_ ignores an interface's + ``if_tsoffset``, the seconds to add to each of its timestamps, so the + subclass reads it and adds it (#1549). + Built when called, so that importing this module does not import `Scapy`_, and so that it always subclasses the :class:`scapy.utils.PcapReader` of the moment. @@ -83,12 +125,23 @@ def _reader_type() -> 'Type[PcapReader]': Returns: A reader type, which opens a PCAP-NG file as its PCAP-NG alternative. + Raises: + VersionError: If :class:`scapy.utils.PcapNgReader` lacks a method in + :data:`_READER_HOOKS`, which these subclasses override. + .. _Scapy: https://scapy.net """ - from scapy.utils import PcapNgReader, PcapReader # isort:skip + from scapy.utils import EDecimal, PcapNgReader, PcapReader # isort:skip + + from pcapkit.toolkit.scapy import attach_linktype, attach_resolution # isort:skip - from pcapkit.toolkit.scapy import attach_resolution # isort:skip + # The PCAP-NG hooks overridden below are private to Scapy. One renamed away + # would leave its fix unapplied without a word, so a missing one is an error. + missing = [name for name in _READER_HOOKS if not callable(getattr(PcapNgReader, name, None))] + if missing: + raise VersionError(f"scapy.utils.PcapNgReader has no {', '.join(missing)}, which the Scapy " + "engine's reader overrides; this Scapy version is not supported") class _Closing: # ``sniff`` closes the reader at the end of the file, and the engine closes @@ -105,6 +158,7 @@ class _PcapReader(_Closing, PcapReader): def read_packet(self, *args: 'Any', **kwargs: 'Any') -> 'ScapyPacket': packet = super().read_packet(*args, **kwargs) attach_resolution(packet, 1_000_000_000 if self.nano else 1_000_000) + attach_linktype(packet, self.linktype) return packet class _PcapNgReader(_Closing, PcapNgReader): @@ -112,21 +166,55 @@ class _PcapNgReader(_Closing, PcapNgReader): alternative = _PcapReader _resolution = 1_000_000 + # each set from the packet block last read, before ``read_packet`` reads it + _linktype: 'int' + _timed: 'bool' + _offset: 'int' + _interface: 'int' + #: ``if_tsoffset`` of each interface of the section, by interface ID. + #: Kept here: what Scapy keeps of an interface differs by version. + _offsets: 'list[int]' def _read_block_shb(self, *args: 'Any', **kwargs: 'Any') -> 'None': super()._read_block_shb(*args, **kwargs) # a new section: its interface IDs restart from 0 (#1522) self.interfaces = [] + self._offsets = [] + + def _read_block_idb(self, block: 'bytes', *args: 'Any', **kwargs: 'Any') -> 'None': + super()._read_block_idb(block, *args, **kwargs) + # appended once Scapy has appended the interface, so the IDs match + self._offsets.append(_read_tsoffset(block[8:], self.endian)) + + def _check_interface_id(self, intid: 'int') -> 'Any': + # Each packet block's reader checks its interface ID here, a Simple + # Packet Block's as 0, and the metadata it returns does not keep it. + # Scapy 2.8 skips the block when this returns false, and before 2.8 + # it raises instead, so the result must be passed through. A block + # it rejects is never returned, so its ID is never read. + self._interface = intid + return super()._check_interface_id(intid) def _read_block(self, *args: 'Any', **kwargs: 'Any') -> 'Any': block = super()._read_block(*args, **kwargs) if block is not None: self._resolution = block[1].tsresol + self._linktype = block[1].linktype + # ``None`` for a Simple Packet Block, which has no timestamp (#1503) + self._timed = block[1].tshigh is not None + self._offset = self._offsets[self._interface] return block def read_packet(self, *args: 'Any', **kwargs: 'Any') -> 'ScapyPacket': packet = super().read_packet(*args, **kwargs) attach_resolution(packet, self._resolution) + attach_linktype(packet, self._linktype) + if not self._timed: + # no timestamp at all, which the default engine reads as 0, where + # Scapy leaves the time the packet was built (#1503) + packet.time = EDecimal(0) + elif self._offset: + packet.time += self._offset # if_tsoffset (#1549) return packet return _PcapReader diff --git a/pcapkit/toolkit/scapy.py b/pcapkit/toolkit/scapy.py index 9dcf82632e..1257552a37 100644 --- a/pcapkit/toolkit/scapy.py +++ b/pcapkit/toolkit/scapy.py @@ -66,7 +66,7 @@ from scapy.packet import Packet __all__ = [ - 'attach_resolution', 'packet2chain', 'packet2dict', 'packet2frame', + 'attach_linktype', 'attach_resolution', 'packet2chain', 'packet2dict', 'packet2frame', 'ipv4_reassembly', 'ipv6_reassembly', 'tcp_reassembly', 'tcp_traceflow' ] @@ -75,6 +75,10 @@ #: second, by :func:`attach_resolution`. RESOLUTION_ATTR = '__pcapkit_resolution__' +#: Attribute a packet's data link type is stashed under, as the capture's +#: ``LINKTYPE_*`` number, by :func:`attach_linktype`. +LINKTYPE_ATTR = '__pcapkit_linktype__' + def attach_resolution(packet: 'Packet', resolution: 'int') -> 'None': """Stash the timestamp resolution of a packet's capture on the packet. @@ -88,6 +92,18 @@ def attach_resolution(packet: 'Packet', resolution: 'int') -> 'None': setattr(packet, RESOLUTION_ATTR, resolution) +def attach_linktype(packet: 'Packet', linktype: 'int') -> 'None': + """Stash the data link type of a packet's capture on the packet. + + Args: + packet: Scapy packet. + linktype: ``LINKTYPE_*`` number: the PCAP global header's, or the + PCAP-NG interface's the packet was captured on. + + """ + setattr(packet, LINKTYPE_ATTR, linktype) + + def packet2chain(packet: 'Packet') -> 'str': """Fetch Scapy packet protocol chain. @@ -408,27 +424,36 @@ def tcp_traceflow(packet: 'Packet', *, count: 'int' = -1) -> 'TF_TCP_Packet | No ip = cast('IP', packet['IP']) if 'IP' in packet else cast('IPv6', packet['IPv6']) tcp = cast('TCP', packet['TCP']) - # NOTE: no default here, deliberately. ``get()`` with no default - # raises on an unresolvable name rather than minting one (#775). - # NULL and RAW are genuine DLTs -- BSD loopback and raw IP - # framing, respectively -- each meant to go with its own handler - # protocol class, so neither is an honest stand-in for "unknown link - # type" and this must not paper over the miss with either. An - # IP-rooted Scapy packet's ``(IP()/TCP()).name`` is ``'IP'``, which is - # not a LinkType member name, and raises. Note the asymmetry, which - # is not a choice made here: an IPv6-rooted packet's name uppercases to - # ``'IPV6'``, which *is* a member (``LinkType.IPV6``, 229), so it - # resolves silently -- to a DLT the caller never chose. Only the v4 name - # happens to miss. The ``KeyError`` - # from :meth:`LinkType.get` is caught and re-raised as - # :exc:`~pcapkit.utilities.exceptions.MissingKeyError` -- this - # package's own house exception for a lookup miss -- rather than - # letting it escape this public function. - name = packet.name.upper() - try: - protocol = Enum_LinkType.get(name) - except KeyError: - raise MissingKeyError(name) from None + # NOTE: the link type of the interface the packet was read from, which + # :class:`~pcapkit.foundation.engines.scapy.Scapy` attaches with + # :func:`attach_linktype` -- not one inferred from the first layer, + # which a raw IPv4 interface's packets share with no link type (#1517). + linktype = getattr(packet, LINKTYPE_ATTR, None) + if linktype is not None: + protocol = Enum_LinkType.get(linktype) + else: + # A packet built by hand carries none, so it is looked up by its + # first layer's name -- with no default, deliberately. ``get()`` + # with no default raises on an unresolvable name rather than minting + # one (#775). NULL and RAW are genuine DLTs -- BSD loopback and raw + # IP framing, respectively -- each meant to go with its own handler + # protocol class, so neither is an honest stand-in for "unknown link + # type" and this must not paper over the miss with either. An + # IP-rooted Scapy packet's ``(IP()/TCP()).name`` is ``'IP'``, which + # is not a LinkType member name, and raises. Note the asymmetry, + # which is not a choice made here: an IPv6-rooted packet's name + # uppercases to ``'IPV6'``, which *is* a member (``LinkType.IPV6``, + # 229), so it resolves silently -- to a DLT the caller never chose. + # Only the v4 name happens to miss. The ``KeyError`` from + # :meth:`LinkType.get` is caught and re-raised as + # :exc:`~pcapkit.utilities.exceptions.MissingKeyError` -- this + # package's own house exception for a lookup miss -- rather than + # letting it escape this public function. + name = packet.name.upper() + try: + protocol = Enum_LinkType.get(name) + except KeyError: + raise MissingKeyError(name) from None data = TF_TCP_Packet( # type: ignore[type-var] protocol=protocol, # data link type diff --git a/tests/foundation/engines/test_engine_agreement_runtime.py b/tests/foundation/engines/test_engine_agreement_runtime.py index 0689151950..73d7ec4347 100644 --- a/tests/foundation/engines/test_engine_agreement_runtime.py +++ b/tests/foundation/engines/test_engine_agreement_runtime.py @@ -394,10 +394,6 @@ class TestEngineAgreement(harness.RoundTripBase): ('INPUTS', 'DATAGRAMS'), ('extra, default has none', 'datagrams != '), ('test.pcapng/scapy/ipv4', 'test.pcapng/scapy/datagrams', 'many_interfaces.pcapng/scapy/ipv4', 'many_interfaces.pcapng/scapy/datagrams')), - Gap(1503, 'the scapy engine dates a Simple Packet Block, which carries no timestamp, with ' - 'the wall-clock time it was read (scapy leaves Packet.time at construction; ' - 'pcapkit/toolkit/scapy.py:186 forwards it), where default reports 0', - 'RECORDS', 'frame 4: timestamp ', ('test.pcapng/scapy/timestamps',)), ) def setUp(self) -> None: diff --git a/tests/foundation/engines/test_scapy_reader_hooks_unit.py b/tests/foundation/engines/test_scapy_reader_hooks_unit.py new file mode 100644 index 0000000000..f0eb2b46f5 --- /dev/null +++ b/tests/foundation/engines/test_scapy_reader_hooks_unit.py @@ -0,0 +1,119 @@ +# -*- coding: utf-8 -*- +"""The scapy engine's PCAP-NG reader overrides Scapy's private hooks safely. + +Its reader subclass of :mod:`pcapkit.foundation.engines.scapy` overrides private +methods of :class:`scapy.utils.RawPcapNgReader` whose contracts differ by Scapy +version, and ``sniff`` turns an exception raised by any of them into a warning +and an empty capture. So each contract is pinned here against a stand-in for +the parent method, whichever Scapy is installed: + +* ``_check_interface_id`` returns whether the interface exists since Scapy 2.8, + which then skips the block on a false result; before 2.8 it returns + :data:`None` and raises instead. An override that dropped the result made 2.8 + skip every packet block. +* Scapy 2.5 keeps an interface as ``(linktype, snaplen, tsresol)`` and 2.7 as + ``(linktype, snaplen, options)``, so the ``if_tsoffset`` of each interface is + kept by the reader itself, not written into Scapy's tuple. +* A hook Scapy no longer has is an error, never a fix left unapplied. + +""" +from __future__ import annotations + +import importlib.util +import struct +import unittest +from unittest import mock + +from tests._support import reimport_once_per_class + +HAS_SCAPY = importlib.util.find_spec('scapy') is not None +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None + for name in ('tbtrim', 'aenum', 'chardet', 'dictdumper')) + + +def _options(order: 'str', *options: 'tuple[int, bytes]') -> 'bytes': + """An options area, each value padded to 32 bits, then ``opt_endofopt``.""" + out = b'' + for code, value in options: + out += struct.pack(f'{order}HH', code, len(value)) + value + bytes(-len(value) % 4) + return out + struct.pack(f'{order}HH', 0, 0) + + +@unittest.skipUnless(HAS_RUNTIME and HAS_SCAPY, 'runtime dependencies or scapy not installed') +class ScapyReaderHookTests(unittest.TestCase): + """Each overridden hook keeps the contract of the Scapy it subclasses.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + @staticmethod + def _reader(order: 'str' = '<'): # type: ignore[no-untyped-def] + """A PCAP-NG reader of the engine's, opened on no file.""" + from pcapkit.foundation.engines.scapy import _reader_type + + reader_type = _reader_type().alternative + reader = object.__new__(reader_type) + reader.endian = order + reader.interfaces = [] + reader._offsets = [] # pylint: disable=protected-access + return reader + + def test_check_interface_id_passes_the_result_through(self) -> None: + from scapy.utils import RawPcapNgReader + + reader = self._reader() + for result in (True, False, None): + with self.subTest(result=result), \ + mock.patch.object(RawPcapNgReader, '_check_interface_id', autospec=True, + return_value=result) as parent: + self.assertIs(reader._check_interface_id(3), result) # pylint: disable=protected-access + parent.assert_called_once_with(reader, 3) + self.assertEqual(reader._interface, 3) # pylint: disable=protected-access + + def test_tsoffset_is_kept_beside_any_shape_of_interface(self) -> None: + from scapy.utils import RawPcapNgReader + + def parent(self: 'RawPcapNgReader', block: 'bytes', _: 'int') -> 'None': + self.interfaces.append(shape) + + idb = struct.pack(' None: + from pcapkit.foundation.engines.scapy import _read_tsoffset + + for order in ('<', '>'): + with self.subTest(order=order): + offset = struct.pack(f'{order}q', -1_000_000) + # after an option whose value needs padding + self.assertEqual(_read_tsoffset(_options(order, (2, b'eth0\x00'), (14, offset)), order), + -1_000_000) + self.assertEqual(_read_tsoffset(_options(order, (9, b'\x06')), order), 0) + self.assertEqual(_read_tsoffset(b'', order), 0) + # not 8 octets: not an offset + self.assertEqual(_read_tsoffset(_options(order, (14, offset[:4])), order), 0) + # nothing is read past opt_endofopt + self.assertEqual(_read_tsoffset(_options(order) + _options(order, (14, offset)), order), 0) + + def test_a_missing_hook_is_a_version_error(self) -> None: + import scapy.utils + + from pcapkit.foundation.engines.scapy import _READER_HOOKS, _reader_type + from pcapkit.utilities.exceptions import VersionError + + stand_in = type('PcapNgReader', (), {name: lambda self: None for name in _READER_HOOKS[1:]}) + with mock.patch.object(scapy.utils, 'PcapNgReader', stand_in), \ + self.assertRaisesRegex(VersionError, _READER_HOOKS[0]): + _reader_type() + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/foundation/engines/test_scapy_reader_interfaces_runtime.py b/tests/foundation/engines/test_scapy_reader_interfaces_runtime.py new file mode 100644 index 0000000000..ded33cc6f0 --- /dev/null +++ b/tests/foundation/engines/test_scapy_reader_interfaces_runtime.py @@ -0,0 +1,268 @@ +# -*- coding: utf-8 -*- +"""The scapy engine takes a packet's link type and clock from its interface. + +Three things :class:`scapy.utils.PcapNgReader` reads differently from the +default engine, all fixed in the reader subclass of +:mod:`pcapkit.foundation.engines.scapy`: + +* #1503 -- a Simple Packet Block carries no timestamp. Scapy leaves the packet + with the time it was built, i.e. the moment it was read; the default engine + reads 0. +* #1517 -- the flow tracer looked the link type up by the name of the packet's + first layer, and a raw IPv4 packet's is ``IP``, which names no link type, so + tracing raised :exc:`~pcapkit.utilities.exceptions.MissingKeyError`. +* #1549 -- Scapy ignores an interface's ``if_tsoffset``, so each timestamp on it + was short by the offset. + +The captures are synthesised from :file:`tcp.pcap`, whose frames 1, 2, 6 and 7 +are one IPv4 flow and 3, 4 and 5 one IPv6 flow, so the module belongs to the +fixture-dependent tier. + +""" +from __future__ import annotations + +import decimal +import importlib.util +import os +import struct +import tempfile +import unittest +import warnings + +from tests._support import close_extractor, reimport_once_per_class, sample_path + +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None + for name in ('tbtrim', 'aenum', 'chardet', 'dictdumper')) +HAS_SCAPY = importlib.util.find_spec('scapy') is not None + +#: ``LINKTYPE_ETHERNET`` and ``LINKTYPE_IPV4``. +ETHERNET, IPV4 = 1, 228 + +#: Frame numbers of :file:`tcp.pcap`'s two flows, as the default engine traces them. +FLOWS = [(1, 2, 6, 7), (3, 4, 5)] + + +def _block(order: 'str', kind: 'int', body: 'bytes') -> 'bytes': + body += bytes(-len(body) % 4) + length = len(body) + 12 + return struct.pack(f'{order}II', kind, length) + body + struct.pack(f'{order}I', length) + + +def _shb(order: 'str' = '<') -> 'bytes': + return _block(order, 0x0A0D0D0A, struct.pack(f'{order}IHHq', 0x1A2B3C4D, 1, 0, -1)) + + +def _idb(linktype: 'int', *, order: 'str' = '<', tsresol: 'int | None' = None, + tsoffset: 'int | None' = None) -> 'bytes': + options = b'' + if tsresol is not None: + options += struct.pack(f'{order}HHB', 9, 1, tsresol) + bytes(3) + if tsoffset is not None: + options += struct.pack(f'{order}HHq', 14, 8, tsoffset) + if options: + options += struct.pack(f'{order}HH', 0, 0) + return _block(order, 1, struct.pack(f'{order}HHI', linktype, 0, 0x40000) + options) + + +def _epb(interface: 'int', ticks: 'int', orig_len: 'int', octets: 'bytes', *, + order: 'str' = '<') -> 'bytes': + return _block(order, 6, struct.pack(f'{order}IIIII', interface, ticks >> 32, ticks & 0xFFFF_FFFF, + len(octets), orig_len) + octets) + + +def _pkt(interface: 'int', ticks: 'int', orig_len: 'int', octets: 'bytes', *, + order: 'str' = '<') -> 'bytes': + """An obsolete Packet Block, which names its interface in 16 bits.""" + return _block(order, 2, struct.pack(f'{order}HHIIII', interface, 0, ticks >> 32, ticks & 0xFFFF_FFFF, + len(octets), orig_len) + octets) + + +def _spb(orig_len: 'int', octets: 'bytes') -> 'bytes': + return _block('<', 3, struct.pack(' 'list[tuple[int, int, int, bytes]]': + """``(ts_sec, ts_usec, orig_len, octets)`` of each record of :file:`tcp.pcap`.""" + with open(sample_path('tcp.pcap'), 'rb') as file: + rest = file.read()[24:] + records = [] + while rest: + ts_sec, ts_usec, incl_len, orig_len = struct.unpack(' 'bool': + return octets[12:14] == b'\x08\x00' + + +@unittest.skipUnless(HAS_RUNTIME and HAS_SCAPY, 'runtime dependencies or scapy not installed') +class ScapyReaderInterfaceTests(unittest.TestCase): + """The scapy engine reads what the default engine reads, block for block.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + tmp = tempfile.TemporaryDirectory() # pylint: disable=consider-using-with + self.addCleanup(tmp.cleanup) + self.tmp = tmp.name + + def _write(self, name: 'str', data: 'bytes') -> 'str': + path = os.path.join(self.tmp, name) + with open(path, 'wb') as file: + file.write(data) + return path + + def _extract(self, fin: 'str', engine: 'str', **kwargs: 'object'): # type: ignore[no-untyped-def] + from pcapkit import extract + + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + extractor = extract(fin=fin, engine=engine, nofile=True, **kwargs) + self.addCleanup(close_extractor, extractor) + return extractor + + def _timestamps(self, fin: 'str') -> 'tuple[list, list]': + """Each frame's timestamp, from the default engine and from Scapy's ``Packet.time``.""" + default = self._extract(fin, 'default', store=True) + scapy = self._extract(fin, 'scapy', store=True) + return ([frame.info.timestamp_epoch for frame in default.frame], + [packet.time for packet in scapy.frame]) + + def _assert_traces_agree(self, fin: 'str') -> 'list[tuple[str, tuple, bytes]]': + """Trace both engines as PCAP and as JSON; return the default engine's PCAP flows.""" + result = [] # type: list[tuple[str, tuple, bytes]] + for trace_format in ('pcap', 'json'): + for nanosecond in (False, True): + with self.subTest(trace_format=trace_format, nanosecond=nanosecond): + flows = {} + for engine in ('default', 'scapy'): + extractor = self._extract(fin, engine, tcp=True, trace=True, trace_format=trace_format, + trace_fout=tempfile.mkdtemp(dir=self.tmp), + trace_nanosecond=nanosecond) + flows[engine] = [] + for flow in extractor.trace.tcp: + with open(flow.fpout, 'rb') as file: + flows[engine].append((flow.label, tuple(flow.index), file.read())) + self.assertEqual([index for _, index, _ in flows['default']], FLOWS) + if trace_format == 'pcap': + self.assertEqual(flows['scapy'], flows['default']) + if not nanosecond: + result = flows['default'] + else: + # the JSON files hold each engine's own frame mapping + self.assertEqual([flow[:2] for flow in flows['scapy']], + [flow[:2] for flow in flows['default']]) + return result + + def test_a_simple_packet_block_is_dated_0(self) -> None: + # #1503: tcp.pcap with frame 4, of the IPv6 flow, as a Simple Packet Block + blocks = [_shb(), _idb(ETHERNET)] + for number, (ts_sec, ts_usec, orig_len, octets) in enumerate(_records(), 1): + blocks.append(_spb(orig_len, octets) if number == 4 else + _epb(0, ts_sec * 10**6 + ts_usec, orig_len, octets)) + fin = self._write('spb.pcapng', b''.join(blocks)) + + default, scapy = self._timestamps(fin) + self.assertEqual(default[3], 0) + self.assertEqual(scapy, default) + + # and so does every adapter that reads the packet's time + from pcapkit.toolkit.scapy import packet2frame, tcp_reassembly, tcp_traceflow + + packet = self._extract(fin, 'scapy', store=True).frame[3] + self.assertEqual(tcp_reassembly(packet).timestamp, 0) # type: ignore[union-attr] + self.assertEqual(tcp_traceflow(packet).timestamp, 0) # type: ignore[union-attr] + record = packet2frame(packet) + self.assertEqual((record.frame_info.ts_sec, record.frame_info.ts_usec, record.time_epoch), (0, 0, 0)) + + flows = self._assert_traces_agree(fin) + # the IPv6 flow's second record is frame 4, at 0 s and 0 us + _, _, data = flows[1] + offset = 24 + 16 + struct.unpack_from(' None: + # #1549: two sections of opposite byte order; in each, an interface with + # an offset, one with a negative one and one with none + records = _records() + ipv4 = records[0][3][14:] + expected = [] # type: list[decimal.Decimal] + blocks = [] + for order in ('<', '>'): + blocks.append(_shb(order)) + blocks.append(_idb(ETHERNET, order=order)) + blocks.append(_idb(IPV4, order=order, tsresol=0x8A, tsoffset=100)) # 2^-10 s, +100 s + blocks.append(_idb(ETHERNET, order=order, tsoffset=-1_000_000)) + ts_sec, ts_usec, orig_len, octets = records[0] + ticks = ts_sec * 10**6 + ts_usec + ticks_bin = (ts_sec << 10) + 123 + blocks.extend([ + _epb(0, ticks, orig_len, octets, order=order), + _epb(1, ticks_bin, len(ipv4), ipv4, order=order), + _epb(2, ticks, orig_len, octets, order=order), + # an obsolete Packet Block names its interface in 16 bits + _pkt(1, ticks_bin, len(ipv4), ipv4, order=order), + _pkt(2, ticks, orig_len, octets, order=order), + ]) + micro = decimal.Decimal(ticks).scaleb(-6) + binary = decimal.Decimal(ticks_bin) / 1024 + 100 + expected.extend([micro, binary, micro - 1_000_000, binary, micro - 1_000_000]) + fin = self._write('tsoffset.pcapng', b''.join(blocks)) + + default, scapy = self._timestamps(fin) + self.assertEqual(default, expected) + self.assertEqual(scapy, default) + + def test_the_issues_tsoffset_value(self) -> None: + # #1549, as reported: 2^-10 s ticks plus 100 s, in a big-endian section + # after a little-endian one + ticks = 1_500_000_000_000_123 + _, _, orig_len, octets = _records()[0] + fin = self._write('issue.pcapng', b''.join([ + _shb('<'), _idb(ETHERNET), _epb(0, ticks, orig_len, octets), + _shb('>'), _idb(IPV4, order='>', tsresol=0x8A, tsoffset=100), + _epb(0, ticks, orig_len - 14, octets[14:], order='>'), + ])) + default, scapy = self._timestamps(fin) + self.assertEqual(default[1], decimal.Decimal('1464843750100.1201171875')) + self.assertEqual(scapy, default) + + def test_a_raw_ipv4_interface_traces_by_its_link_type(self) -> None: + # #1517: tcp.pcap with its IPv4 frames on a LINKTYPE_IPV4 interface + blocks = [_shb(), _idb(ETHERNET), _idb(IPV4)] + for ts_sec, ts_usec, orig_len, octets in _records(): + interface = 0 + if _is_ipv4(octets): + interface, octets, orig_len = 1, octets[14:], orig_len - 14 + blocks.append(_epb(interface, ts_sec * 10**6 + ts_usec, orig_len, octets)) + fin = self._write('raw.pcapng', b''.join(blocks)) + + flows = self._assert_traces_agree(fin) + self.assertEqual([struct.unpack_from(' None: + # #1517, in a PCAP: the global header's link type, for its IPv4 frames only + with open(sample_path('tcp.pcap'), 'rb') as file: + header = file.read(24) + out = [header[:20] + struct.pack(' None: + reimport_once_per_class(self) + + @staticmethod + def _tcp(network: 'type'): # type: ignore[no-untyped-def] + from scapy.layers.inet import TCP + from scapy.packet import Raw + + addresses = (('192.0.2.1', '198.51.100.1') if network.__name__ == 'IP' + else ('2001:db8::1', '2001:db8::2')) + packet = network(src=addresses[0], dst=addresses[1]) / TCP(sport=1234, dport=80, flags='S') / Raw(b'x') + return network(bytes(packet)) + + def test_an_attached_link_type_is_used(self) -> None: + from scapy.layers.inet import IP + from scapy.layers.inet6 import IPv6 + + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit.scapy import attach_linktype, tcp_traceflow + + for network, linktype, expected in ((IP, 228, LinkType.IPV4), (IP, 101, LinkType.RAW), + # by name, ``IPv6`` would resolve to IPV6 (229) + (IPv6, 101, LinkType.RAW)): + with self.subTest(network=network.__name__, linktype=linktype): + packet = self._tcp(network) + attach_linktype(packet, linktype) + flow = tcp_traceflow(packet, count=1) + assert flow is not None + self.assertIs(flow.protocol, expected) + + def test_a_packet_built_by_hand_is_looked_up_by_name(self) -> None: + from scapy.layers.inet import IP + from scapy.layers.inet6 import IPv6 + + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit.scapy import tcp_traceflow + from pcapkit.utilities.exceptions import MissingKeyError + + flow = tcp_traceflow(self._tcp(IPv6), count=1) + assert flow is not None + self.assertIs(flow.protocol, LinkType.IPV6) + # and still with no default (#775) + with self.assertRaises(MissingKeyError): + tcp_traceflow(self._tcp(IP), count=1) + + +if __name__ == '__main__': + unittest.main()