From 097cb9453f17365dffdd91edfa068f3444d231f1 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Mon, 14 Sep 2026 22:45:46 -0400 Subject: [PATCH 1/3] engines: add pypcap and pypcapfile extraction engines Both candidates the Help Wanted page names are now selectable as engine='pypcap' and engine='pypcapfile', each with a matching pcapkit.toolkit module, a pyproject extra, docs and tests. They are registered as built-ins in Extractor.__engine__ rather than through Engine.__init_subclass__, because the public auto-registration path cannot work for an engine that ships with the library: pcapkit/__init__.py never imports pcapkit.foundation.engines, so the hook would not fire for a bare `import pcapkit` and engine='pypcap' would silently fall through to the built-in parser - the exact trap these engines are meant to avoid. ModuleDescriptor also keeps the third-party import lazy. The public path stays the documented route for user engines and keeps its own coverage. Capability gaps are surfaced rather than papered over. pypcap is a libpcap binding with no dissection: reassembly and flow tracing are switched off with a warning and the toolkit adapters raise UnsupportedCall; it is PCAP-only, since libpcap opens a PCAP-NG file happily and then yields zero frames, so the engine checks the magic number and raises FormatError instead of reporting an empty capture; and it needs a file on disk. pypcapfile decodes Ethernet/IPv4/TCP/UDP only, so IPv6 reassembly raises rather than returning None, which would be indistinguishable from "no fragment here". Neither library installs cleanly from PyPI on a current Python, and pep.rst says so instead of claiming otherwise: released pypcapfile 0.12.0 imports the `imp` module removed in 3.12, so the engine was verified against upstream master; and pypcap 1.3.0 ships pre-generated Cython C that no longer compiles against the 3.12+ C API, plus a setup.py that only searches fixed prefixes for pcap.h. Both were genuinely executed here, on a locally rebuilt wheel. Parity is asserted per frame - capture length, original length, timestamp and the full Ethernet header - across four captures, and each test proves the engine actually ran rather than being replaced by the fallback, by asserting no EngineWarning was raised and that the extractor reports the engine by name. One pre-existing crash had to be fixed for pypcapfile to work at all: Extractor(trace=True) died with AttributeError because the guard listed only pyshark and matched trace_format == 'pcap', while TraceFlow substitutes 'pcap' for None, so it never fired. It now matches ('pcap', 'cap', None), which incidentally fixes pyshark. dpkt and scapy remain broken there by design and are called out in a comment. --- Pipfile | 2 + docs/source/ext.rst | 42 +- .../pcapkit/foundation/engines/3rdparty.rst | 94 ++++ .../pcapkit/foundation/engines/index.rst | 31 +- docs/source/pcapkit/toolkit/3rdparty.rst | 82 ++++ docs/source/pep.rst | 52 +- pcapkit/foundation/engines/__init__.py | 8 +- pcapkit/foundation/engines/pypcap.py | 243 ++++++++++ pcapkit/foundation/engines/pypcapfile.py | 336 +++++++++++++ pcapkit/foundation/extraction.py | 32 +- pcapkit/interface/misc.py | 10 +- pcapkit/toolkit/__init__.py | 21 + pcapkit/toolkit/pypcap.py | 166 +++++++ pcapkit/toolkit/pypcapfile.py | 454 ++++++++++++++++++ pyproject.toml | 4 +- .../engines/test_new_engine_parity.py | 271 +++++++++++ .../foundation/engines/test_pypcap_engine.py | 293 +++++++++++ .../engines/test_pypcapfile_engine.py | 372 ++++++++++++++ tests/toolkit/test_pypcap_unit.py | 76 +++ tests/toolkit/test_pypcapfile_unit.py | 364 ++++++++++++++ 20 files changed, 2921 insertions(+), 32 deletions(-) create mode 100644 pcapkit/foundation/engines/pypcap.py create mode 100644 pcapkit/foundation/engines/pypcapfile.py create mode 100644 pcapkit/toolkit/pypcap.py create mode 100644 pcapkit/toolkit/pypcapfile.py create mode 100644 tests/foundation/engines/test_new_engine_parity.py create mode 100644 tests/foundation/engines/test_pypcap_engine.py create mode 100644 tests/foundation/engines/test_pypcapfile_engine.py create mode 100644 tests/toolkit/test_pypcap_unit.py create mode 100644 tests/toolkit/test_pypcapfile_unit.py diff --git a/Pipfile b/Pipfile index 0efd5b3abd..7c28605fee 100644 --- a/Pipfile +++ b/Pipfile @@ -16,6 +16,8 @@ pyshark = "*" dpkt = "*" scapy = "*" cryptography = "*" +pypcap = "*" +pypcapfile = "*" beautifulsoup4 = {extras = ["html5lib"],version = "*"} requests = {extras = ["socks"],version = "*"} autopep8 = "*" diff --git a/docs/source/ext.rst b/docs/source/ext.rst index 9a5df1bc79..4956018672 100644 --- a/docs/source/ext.rst +++ b/docs/source/ext.rst @@ -310,23 +310,39 @@ the network packets for further processing. The following table shows the available engines and the corresponding supported file formats: -+---------------------+-----------------------------------------------------+-------------------------------------+ -| Engine Type | Engine Class | Supported File Formats | -+=====================+=====================================================+=====================================+ -| | :class:`pcapkit.foundation.engines.pcap.PCAP` | PCAP only | -+ Built-in Engines +-----------------------------------------------------+-------------------------------------+ -| | :class:`pcapkit.foundation.engines.pcapng.PCAPNG` | PCAP-NG only | -+---------------------+-----------------------------------------------------+-------------------------------------+ -| | :class:`pcapkit.foundation.engines.scapy.Scapy` | all formats supported by `Scapy`_ | -+ +-----------------------------------------------------+-------------------------------------+ -| Third-party Engines | :class:`pcapkit.foundation.engines.dpkt.DPKT` | all formats supported by `DPKT`_ | -| +-----------------------------------------------------+-------------------------------------+ -| | :class:`pcapkit.foundation.engines.pyshark.PyShark` | all formats supported by `PyShark`_ | -+---------------------+-----------------------------------------------------+-------------------------------------+ ++---------------------+-----------------------------------------------------------+----------------------------------------+ +| Engine Type | Engine Class | Supported File Formats | ++=====================+===========================================================+========================================+ +| | :class:`pcapkit.foundation.engines.pcap.PCAP` | PCAP only | ++ Built-in Engines +-----------------------------------------------------------+----------------------------------------+ +| | :class:`pcapkit.foundation.engines.pcapng.PCAPNG` | PCAP-NG only | ++---------------------+-----------------------------------------------------------+----------------------------------------+ +| | :class:`pcapkit.foundation.engines.scapy.Scapy` | all formats supported by `Scapy`_ | ++ +-----------------------------------------------------------+----------------------------------------+ +| | :class:`pcapkit.foundation.engines.dpkt.DPKT` | all formats supported by `DPKT`_ | +| +-----------------------------------------------------------+----------------------------------------+ +| Third-party Engines | :class:`pcapkit.foundation.engines.pyshark.PyShark` | all formats supported by `PyShark`_ | +| +-----------------------------------------------------------+----------------------------------------+ +| | :class:`pcapkit.foundation.engines.pypcap.PyPCAP` | PCAP only, and from a file on disk | +| +-----------------------------------------------------------+----------------------------------------+ +| | :class:`pcapkit.foundation.engines.pypcapfile.PyPCAPFile` | PCAP only | ++---------------------+-----------------------------------------------------------+----------------------------------------+ + +.. note:: + + An engine is free to support less than :class:`~pcapkit.foundation.extraction.Extractor` + offers, and several do. `PyPCAP`_ performs no protocol dissection whatsoever, so + it supports neither reassembly nor flow tracing; `PyPCAPFile`_ has no IPv6 + decoder, so it supports IPv4 and TCP reassembly but not IPv6. What matters is + that the gap is *announced* -- each engine warns, or raises, for the capability + it cannot provide, rather than silently producing an empty result. See + :doc:`pcapkit/foundation/engines/index` for the full table. .. _Scapy: https://scapy.net .. _DPKT: https://dpkt.readthedocs.io .. _PyShark: https://kiminewt.github.io/pyshark +.. _PyPCAP: https://github.com/pynetwork/pypcap +.. _PyPCAPFile: https://github.com/kisom/pypcapfile Samples ~~~~~~~ diff --git a/docs/source/pcapkit/foundation/engines/3rdparty.rst b/docs/source/pcapkit/foundation/engines/3rdparty.rst index eb4273eaee..da90a35bf5 100644 --- a/docs/source/pcapkit/foundation/engines/3rdparty.rst +++ b/docs/source/pcapkit/foundation/engines/3rdparty.rst @@ -71,3 +71,97 @@ support, as is used by :class:`pcapkit.foundation.extraction.Extractor`. .. autoattribute:: _expkg .. autoattribute:: _extmp + +PyPCAP Support +============== + +.. module:: pcapkit.foundation.engines.pypcap + +This module contains the implementation for `PyPCAP`_ engine +support, as is used by :class:`pcapkit.foundation.extraction.Extractor`. + +.. _PyPCAP: https://github.com/pynetwork/pypcap + +.. important:: + + `PyPCAP`_ is a :manpage:`libpcap(3)` binding aimed primarily at live capture. + Offline it performs **no protocol dissection**: each frame is the + ``(timestamp, bytes)`` pair that :c:func:`pcap_next_ex` produced. Reassembly + and flow tracing are therefore unavailable and are disabled -- with an + :class:`~pcapkit.utilities.warnings.AttributeWarning` -- when requested. + + The engine also reads PCAP savefiles only, and only from a file on disk. + PCAP-NG is rejected with a + :exc:`~pcapkit.utilities.exceptions.FormatError`, because + :manpage:`libpcap(3)` opens such a file without complaint and then yields no + frames at all; and a non-file input is rejected with an + :exc:`~pcapkit.utilities.exceptions.UnsupportedCall`, because + :c:func:`pcap_open_offline` opens a savefile by *name*. + +.. autoclass:: pcapkit.foundation.engines.pypcap.PyPCAP + :no-members: + :show-inheritance: + + .. autoattribute:: __engine_name__ + .. autoattribute:: __engine_module__ + + .. autoproperty:: dlink + + .. automethod:: run + .. automethod:: read_frame + .. automethod:: close + + .. autoattribute:: _expkg + .. autoattribute:: _extmp + .. autoattribute:: _dlink + .. autoattribute:: _closed + +PyPCAPFile Support +================== + +.. module:: pcapkit.foundation.engines.pypcapfile + +This module contains the implementation for `PyPCAPFile`_ engine +support, as is used by :class:`pcapkit.foundation.extraction.Extractor`. + +.. _PyPCAPFile: https://github.com/kisom/pypcapfile + +.. important:: + + `PyPCAPFile`_ is a pure Python savefile reader that decodes Ethernet, IPv4, + TCP and UDP and nothing else. IPv6 reassembly is therefore unavailable and is + disabled -- with an :class:`~pcapkit.utilities.warnings.AttributeWarning` -- + when requested; IPv4 and TCP reassembly and TCP flow tracing remain + available. PCAP-NG is rejected with a + :exc:`~pcapkit.utilities.exceptions.FormatError`. + +.. autoclass:: pcapkit.foundation.engines.pypcapfile.PyPCAPFile + :no-members: + :show-inheritance: + + .. autoattribute:: __engine_name__ + .. autoattribute:: __engine_module__ + .. autoattribute:: LAYERS + + .. autoproperty:: dlink + + .. automethod:: run + .. automethod:: read_frame + + .. autoattribute:: _expkg + .. autoattribute:: _extmp + .. autoattribute:: _dlink + .. autoattribute:: _declf + +Internal Definitions +-------------------- + +.. autoclass:: pcapkit.foundation.engines.pypcapfile._NamedStream + :no-members: + :show-inheritance: + + .. autoattribute:: name + .. automethod:: read + +.. automethod:: pcapkit.foundation.engines.pypcapfile.PyPCAPFile._get_decoder +.. automethod:: pcapkit.foundation.engines.pypcapfile.PyPCAPFile._decode diff --git a/docs/source/pcapkit/foundation/engines/index.rst b/docs/source/pcapkit/foundation/engines/index.rst index 32a61fef98..efcfb3a5c9 100644 --- a/docs/source/pcapkit/foundation/engines/index.rst +++ b/docs/source/pcapkit/foundation/engines/index.rst @@ -6,7 +6,7 @@ Engine Support :mod:`pcapkit.foundation.engines` is a collection of engines support for :mod:`pcapkit`, including but not limited to the built-in PCAP and `PCAP-NG`_ file support, `Scapy`_, `PyShark`_, -and `DPKT`_ 3rd party engine support. +`DPKT`_, `PyPCAP`_ and `PyPCAPFile`_ 3rd party engine support. .. seealso:: @@ -44,6 +44,8 @@ class hierarchy of :mod:`pcapkit.foundation.engines`: Scapy DPKT PyShark + PyPCAP + PyPCAPFile end B --> third-party @@ -61,9 +63,36 @@ class hierarchy of :mod:`pcapkit.foundation.engines`: click Scapy "/pcapkit/foundation/engines/3rdparty.html#pcapkit.foundation.engines.scapy.Scapy" click DPKT "/pcapkit/foundation/engines/3rdparty.html#pcapkit.foundation.engines.dpkt.DPKT" click PyShark "/pcapkit/foundation/engines/3rdparty.html#pcapkit.foundation.engines.pyshark.PyShark" + click PyPCAP "/pcapkit/foundation/engines/3rdparty.html#pcapkit.foundation.engines.pypcap.PyPCAP" + click PyPCAPFile "/pcapkit/foundation/engines/3rdparty.html#pcapkit.foundation.engines.pypcapfile.PyPCAPFile" + +Not every engine can do everything :class:`~pcapkit.foundation.extraction.Extractor` +offers, and the ones that cannot say so rather than quietly doing less: + ++-----------------------------------------------------------------+---------------------------------------------------------------+ +| Engine | Gap, and how it is surfaced | ++=================================================================+===============================================================+ +| :class:`~pcapkit.foundation.engines.pyshark.PyShark` | no reassembly -- disabled with an | +| | :class:`~pcapkit.utilities.warnings.AttributeWarning` | ++-----------------------------------------------------------------+---------------------------------------------------------------+ +| :class:`~pcapkit.foundation.engines.pypcap.PyPCAP` | no protocol dissection at all, hence no reassembly and no | +| | flow tracing -- both disabled with an | +| | :class:`~pcapkit.utilities.warnings.AttributeWarning`; PCAP | +| | savefiles on disk only, otherwise | +| | :exc:`~pcapkit.utilities.exceptions.FormatError` / | +| | :exc:`~pcapkit.utilities.exceptions.UnsupportedCall` | ++-----------------------------------------------------------------+---------------------------------------------------------------+ +| :class:`~pcapkit.foundation.engines.pypcapfile.PyPCAPFile` | no IPv6 decoder, hence no IPv6 reassembly -- disabled with an | +| | :class:`~pcapkit.utilities.warnings.AttributeWarning`, and | +| | :func:`~pcapkit.toolkit.pypcapfile.ipv6_reassembly` raises; | +| | PCAP savefiles only, otherwise | +| | :exc:`~pcapkit.utilities.exceptions.FormatError` | ++-----------------------------------------------------------------+---------------------------------------------------------------+ .. _PCAP-NG: https://wiki.wireshark.org/Development/PcapNg .. _Scapy: https://scapy.net .. _DPKT: https://dpkt.readthedocs.io .. _PyShark: https://kiminewt.github.io/pyshark +.. _PyPCAP: https://github.com/pynetwork/pypcap +.. _PyPCAPFile: https://github.com/kisom/pypcapfile diff --git a/docs/source/pcapkit/toolkit/3rdparty.rst b/docs/source/pcapkit/toolkit/3rdparty.rst index 56d91bb999..18167dce0b 100644 --- a/docs/source/pcapkit/toolkit/3rdparty.rst +++ b/docs/source/pcapkit/toolkit/3rdparty.rst @@ -86,3 +86,85 @@ Auxiliary Functions ------------------- .. autofunction:: pcapkit.toolkit.pyshark.packet2dict + +PyPCAP Tools +============ + +.. module:: pcapkit.toolkit.pypcap + +:mod:`pcapkit.toolkit.pypcap` contains all you need for +:mod:`pcapkit` handy usage with `PyPCAP`_ engine. All reforming +functions returns with a flag to indicate if usable for +its caller. + +.. _PyPCAP: https://github.com/pynetwork/pypcap + +.. note:: + + `PyPCAP`_ performs no protocol dissection, so the reassembly and flow tracing + adapters below cannot be implemented. They are defined all the same, so that + reaching for one fails with an explanatory + :exc:`~pcapkit.utilities.exceptions.UnsupportedCall` rather than an + :exc:`ImportError`. + +.. autofunction:: pcapkit.toolkit.pypcap.ipv4_reassembly + +.. autofunction:: pcapkit.toolkit.pypcap.ipv6_reassembly + +.. autofunction:: pcapkit.toolkit.pypcap.tcp_reassembly + +.. autofunction:: pcapkit.toolkit.pypcap.tcp_traceflow + +Auxiliary Functions +------------------- + +.. autofunction:: pcapkit.toolkit.pypcap.packet2chain + +.. autofunction:: pcapkit.toolkit.pypcap.packet2dict + +PyPCAPFile Tools +================ + +.. module:: pcapkit.toolkit.pypcapfile + +:mod:`pcapkit.toolkit.pypcapfile` contains all you need for +:mod:`pcapkit` handy usage with `PyPCAPFile`_ engine. All reforming +functions returns with a flag to indicate if usable for +its caller. + +.. _PyPCAPFile: https://github.com/kisom/pypcapfile + +.. note:: + + `PyPCAPFile`_ has no IPv6 decoder, so :func:`~pcapkit.toolkit.pypcapfile.ipv6_reassembly` + raises :exc:`~pcapkit.utilities.exceptions.UnsupportedCall` rather than returning + :data:`None` -- which would be indistinguishable from "this frame carries no + IPv6 fragment". + +.. autofunction:: pcapkit.toolkit.pypcapfile.ipv4_reassembly + +.. autofunction:: pcapkit.toolkit.pypcapfile.ipv6_reassembly + +.. autofunction:: pcapkit.toolkit.pypcapfile.tcp_reassembly + +.. autofunction:: pcapkit.toolkit.pypcapfile.tcp_traceflow + +Auxiliary Functions +------------------- + +.. autofunction:: pcapkit.toolkit.pypcapfile.packet2timestamp + +.. autofunction:: pcapkit.toolkit.pypcapfile.ipv4_header + +.. autofunction:: pcapkit.toolkit.pypcapfile.packet2chain + +.. autofunction:: pcapkit.toolkit.pypcapfile.packet2dict + +Internal Definitions +-------------------- + +.. autodata:: pcapkit.toolkit.pypcapfile.TCP_MIN_HEADER_LEN + +.. autodata:: pcapkit.toolkit.pypcapfile.IPV4_FLAG_DF + +.. autodata:: pcapkit.toolkit.pypcapfile.IPV4_FLAG_MF diff --git a/docs/source/pep.rst b/docs/source/pep.rst index 576082ae45..3cc0e88efc 100644 --- a/docs/source/pep.rst +++ b/docs/source/pep.rst @@ -70,15 +70,43 @@ contributions are welcomed to integrate the logging system into PyPCAPKit. New Engines ----------- -Although PyPCAPKit already has support for some popular PCAP parsing libraries, -I'm expecting to extend the list of supported engines furthermore. The candidate -engines include: - -- `pypcap `__ -- `pycapfile `__ - .. note:: + **Done**, for both candidates. ``engine='pypcapfile'`` selects + :class:`pcapkit.foundation.engines.pypcapfile.PyPCAPFile` and + ``engine='pypcap'`` selects :class:`pcapkit.foundation.engines.pypcap.PyPCAP`; + each has a matching :mod:`pcapkit.toolkit` module + (:mod:`pcapkit.toolkit.pypcapfile`, :mod:`pcapkit.toolkit.pypcap`), a + ``pyproject.toml`` extra (``PyPCAPFile``, ``PyPCAP``, both in ``all``), docs + under :doc:`pcapkit/foundation/engines/index`, and tests under + ``tests/foundation/engines/`` and ``tests/toolkit/``. Both were verified + end-to-end against the sample captures: each agrees with the ``default`` engine + on frame count, per-record capture length, timestamp and Ethernet header. + + Neither library, though, is usable straight from PyPI on a current Python, and + both of the following are worth knowing before reaching for them: + + * **pypcapfile** -- the released 0.12.0 imports the :mod:`imp` module, removed + in Python 3.12, so it cannot be imported at all on 3.12 or newer. Upstream + ``master`` (0.12.1, unreleased) has fixed this, and that is what the engine + was verified against. The extra therefore installs a version that only works + on Python 3.10 and 3.11 until 0.12.1 is published. + * **pypcap** -- the 1.3.0 sdist ships a pre-generated ``pcap.c`` from an old + Cython, which no longer compiles against the Python 3.12+ C API + (``ob_digit``, ``_PyLong_AsByteArray``), and its ``setup.py`` searches a + fixed list of prefixes for ``pcap.h``. It builds once :program:`libpcap`'s + headers are visible under ``sys.prefix`` and ``pcap.c`` is regenerated with + Cython 3. That is a packaging problem upstream rather than an engine problem, + but it does mean ``pip install pypcapkit[PyPCAP]`` can fail to build. + + Both engines support less than the ``default`` engine does, deliberately and + noisily: `pypcap`_ performs no protocol dissection, so it disables reassembly + *and* flow tracing; `pypcapfile`_ has no IPv6 decoder, so it disables IPv6 + reassembly while keeping IPv4 and TCP. Each gap is announced through an + :class:`~pcapkit.utilities.warnings.AttributeWarning` or an outright exception + rather than by silently returning nothing -- + :doc:`pcapkit/foundation/engines/index` tabulates them. + The engine interface has since been refactored, so this no longer means adding handler methods to :class:`~pcapkit.foundation.extraction.Extractor`. A new engine subclasses :class:`pcapkit.foundation.engines.engine.Engine` and @@ -88,6 +116,16 @@ engines include: still apply is the unified auxiliary tools in :mod:`pcapkit.toolkit`, where each engine has a matching module. +Originally: although PyPCAPKit already has support for some popular PCAP parsing +libraries, I'm expecting to extend the list of supported engines furthermore. The +candidate engines include: + +- `pypcap `__ +- `pypcapfile `__ + +.. _pypcap: https://github.com/pynetwork/pypcap +.. _pypcapfile: https://github.com/kisom/pypcapfile + Test Cases ---------- diff --git a/pcapkit/foundation/engines/__init__.py b/pcapkit/foundation/engines/__init__.py index 760c87f3bf..bf68327fd0 100644 --- a/pcapkit/foundation/engines/__init__.py +++ b/pcapkit/foundation/engines/__init__.py @@ -7,8 +7,8 @@ :mod:`pcapkit.foundation.engines` is a collection of engines support for :mod:`pcapkit`, including but not limited to the built-in PCAP and `PCAP-NG`_ file support, :mod:`Scapy `, -:mod:`PyShark `, :mod:`DPKT ` 3rd party engine -support. +:mod:`PyShark `, :mod:`DPKT `, :mod:`PyPCAP ` +and :mod:`PyPCAPFile ` 3rd party engine support. .. _PCAPNG: https://wiki.wireshark.org/Development/PcapNg @@ -24,9 +24,11 @@ from pcapkit.foundation.engines.scapy import Scapy from pcapkit.foundation.engines.dpkt import DPKT from pcapkit.foundation.engines.pyshark import PyShark +from pcapkit.foundation.engines.pypcap import PyPCAP +from pcapkit.foundation.engines.pypcapfile import PyPCAPFile __all__ = [ 'PCAP', 'PCAPNG', - 'Scapy', 'DPKT', 'PyShark', + 'Scapy', 'DPKT', 'PyShark', 'PyPCAP', 'PyPCAPFile', ] diff --git a/pcapkit/foundation/engines/pypcap.py b/pcapkit/foundation/engines/pypcap.py new file mode 100644 index 0000000000..511f6d8afa --- /dev/null +++ b/pcapkit/foundation/engines/pypcap.py @@ -0,0 +1,243 @@ +# -*- coding: utf-8 -*- +"""PyPCAP Support +=================== + +.. module:: pcapkit.foundation.engines.pypcap + +This module contains the implementation for `PyPCAP`_ engine +support, as is used by :class:`pcapkit.foundation.extraction.Extractor`. + +.. _PyPCAP: https://github.com/pynetwork/pypcap + +""" +import os +from typing import TYPE_CHECKING, cast + +from pcapkit.const.reg.linktype import LinkType as Enum_LinkType +from pcapkit.foundation.engines.engine import EngineBase as Engine +from pcapkit.foundation.reassembly import ReassemblyManager +from pcapkit.foundation.traceflow import TraceFlowManager +from pcapkit.utilities.exceptions import FormatError, UnsupportedCall, stacklevel +from pcapkit.utilities.warnings import AttributeWarning, warn + +__all__ = ['PyPCAP'] + +if TYPE_CHECKING: + from typing import Iterator + + from pcap import pcap as Handle + + from pcapkit.foundation.extraction import Extractor + + #: A PyPCAP "frame": the ``(timestamp, bytes)`` pair that :class:`pcap.pcap` + #: iteration yields. Deliberately *not* named ``Frame``, so that it is not + #: mistaken for :class:`pcapkit.protocols.misc.pcap.frame.Frame`, which is + #: what the built-in engines return. + RawFrame = tuple[float, bytes] + + +class PyPCAP(Engine['RawFrame']): + """PyPCAP engine support. + + `PyPCAP`_ is a binding over :manpage:`libpcap(3)`, primarily aimed at live + capture. Offline it is a savefile reader and nothing more: iteration yields + the ``(timestamp, bytes)`` pair from :c:func:`pcap_next_ex` and no protocol + dissection is performed. Consequently this engine + + * returns each frame as a ``(timestamp, bytes)`` :obj:`tuple` rather than as + a parsed packet object, and + * disables both reassembly and flow tracing, warning as it does so, since + neither can be derived without an IP or TCP layer to read. + + It also requires the input to be a real file on disk, because + :c:func:`pcap_open_offline` opens by *name* -- there is no way to hand + :manpage:`libpcap(3)` an already-open Python stream. + + .. _PyPCAP: https://github.com/pynetwork/pypcap + + Args: + extractor: :class:`~pcapkit.foundation.extraction.Extractor` instance. + + """ + if TYPE_CHECKING: + import pcap + + #: Engine extraction package. + _expkg: 'pcap' + #: Engine extraction temporary storage. + _extmp: 'Iterator[RawFrame]' + #: Data link layer protocol, from the capture handle. + _dlink: 'Enum_LinkType' + #: Closed flag, so that the handle is not closed twice. + _closed: 'bool' + + ########################################################################## + # Defaults. + ########################################################################## + + #: Engine name. + __engine_name__ = 'PyPCAP' + + #: Engine module name. + __engine_module__ = 'pcap' + + ########################################################################## + # Properties. + ########################################################################## + + @property + def dlink(self) -> 'Enum_LinkType': + """Data link layer protocol, as reported by the capture handle.""" + return self._dlink + + ########################################################################## + # Data models. + ########################################################################## + + def __init__(self, extractor: 'Extractor') -> 'None': + import pcap # isort:skip + + self._expkg = pcap + self._extmp = cast('Iterator[RawFrame]', None) + self._dlink = cast('Enum_LinkType', None) + self._closed = False + + super().__init__(extractor) + + ########################################################################## + # Methods. + ########################################################################## + + def run(self) -> 'None': + """Call :class:`pcap.pcap` to extract PCAP files. + + This method assigns :attr:`self._expkg ` + as :mod:`pcap` and :attr:`self._extmp ` + as an iterator from :class:`pcap.pcap`. + + Warns: + AttributeWarning: Warns under following circumstances: + + * if :attr:`self.extractor._exlyr ` + and/or :attr:`self.extractor._exptl ` + is provided as the PyPCAP engine currently does not + support such operations. + * if reassembly and/or flow tracing is enabled, as the PyPCAP + engine performs no protocol dissection and so cannot support + either operation. + + Raises: + FormatError: If the file format is not supported, i.e., not a PCAP + file. PCAP-NG is rejected explicitly rather than left to + :manpage:`libpcap(3)`, which opens such a file without complaint + and then yields no frames at all. + UnsupportedCall: If the input is not a file on disk, as + :c:func:`pcap_open_offline` can only open a savefile by name. + + """ + from pcapkit.foundation.engines.pcap import PCAP # isort:skip + + ext = self._extractor + + if ext._exlyr != 'none' or ext._exptl != 'null': + warn("'Extractor(engine=pypcap)' does not support protocol and layer threshold; " + f"'layer={ext._exlyr}' and 'protocol={ext._exptl}' ignored", + AttributeWarning, stacklevel=stacklevel()) + + if ext.magic_number not in PCAP.MAGIC_NUMBER: + raise FormatError(f'unsupported file format: {ext.magic_number!r}; ' + 'the PyPCAP engine reads PCAP savefiles only') + + if not os.path.isfile(ext._ifnm): + raise UnsupportedCall(f"'Extractor(engine=pypcap)' requires a file on disk, " + f'but {ext._ifnm!r} is not one; libpcap opens savefiles ' + 'by name and cannot read an in-memory stream') + + if ext._flag_r and (ext._ipv4 or ext._ipv6 or ext._tcp): + ext._flag_r = False + ext._reasm = ReassemblyManager(ipv4=None, ipv6=None, tcp=None) + warn("'Extractor(engine=pypcap)' object dose not support reassembly; " + f"so 'ipv4={ext._ipv4}', 'ipv6={ext._ipv6}' and 'tcp={ext._tcp}' will be ignored", + AttributeWarning, stacklevel=stacklevel()) + + if ext._flag_t and ext._tcp: + ext._flag_t = False + ext._trace = TraceFlowManager(tcp=None) + warn("'Extractor(engine=pypcap)' object dose not support flow tracing; " + f"so 'tcp={ext._tcp}' will be ignored", AttributeWarning, stacklevel=stacklevel()) + + # NOTE: ``promisc=False`` is defensive only -- the offline branch of + # ``pcap.pcap`` ignores it, but should ``pcap_open_offline`` ever fail the + # constructor falls through to opening the name as a live device, and we + # do not want that attempt to request promiscuous mode. + handle = cast('Handle', self._expkg.pcap(name=ext._ifnm, promisc=False)) + self._dlink = Enum_LinkType.get(handle.datalink()) + + # setup verbose handler + if ext._flag_v: + from pcapkit.toolkit.pypcap import packet2chain # isort:skip + ext._vfunc = lambda e, f: print( + f'Frame {e._frnum:>3d}: {packet2chain(f[1], data_link=self._dlink)}' # pylint: disable=protected-access + ) # pylint: disable=logging-fstring-interpolation + + # extract & analyse file + self._extmp = iter(handle) + + def read_frame(self) -> 'RawFrame': + """Read frames with PyPCAP engine. + + Returns: + The ``(timestamp, bytes)`` pair as yielded by :class:`pcap.pcap`. + + See Also: + Please refer to :meth:`PCAP.read_frame ` + for more operational information. + + """ + from pcapkit.toolkit.pypcap import packet2dict # isort:skip + ext = self._extractor + + # fetch PyPCAP packet + frame = cast('RawFrame', next(self._extmp)) + timestamp, packet = frame + + # verbose output + ext._frnum += 1 + ext._vfunc(ext, frame) + + # write plist + frnum = f'Frame {ext._frnum}' + if not ext._flag_q: + info = packet2dict(packet, timestamp, data_link=self._dlink) + if ext._flag_f: + ofile = ext._ofile(f'{ext._ofnm}/{frnum}.{ext._fext}') + ofile(info, name=frnum) + else: + ext._ofile(info, name=frnum) + ofile = ext._ofile + ext._offmt = ofile.kind + + # NOTE: reassembly and flow tracing are disabled in ``run``, so there is + # deliberately no bookkeeping for either here. + + # record frames + if ext._flag_d: + ext._frame.append(frame) + + # return frame record + return frame + + def close(self) -> 'None': + """Close engine. + + This method closes the underlying :class:`pcap.pcap` handle. It is + idempotent, as :meth:`Extractor._cleanup + ` and + :meth:`Extractor.__exit__ ` + may both reach it, and :meth:`pcap.pcap.close` is not safe to call twice. + + """ + if self._closed or self._extmp is None: + return + self._closed = True + cast('Handle', self._extmp).close() diff --git a/pcapkit/foundation/engines/pypcapfile.py b/pcapkit/foundation/engines/pypcapfile.py new file mode 100644 index 0000000000..e95a884d87 --- /dev/null +++ b/pcapkit/foundation/engines/pypcapfile.py @@ -0,0 +1,336 @@ +# -*- coding: utf-8 -*- +"""PyPCAPFile Support +======================= + +.. module:: pcapkit.foundation.engines.pypcapfile + +This module contains the implementation for `PyPCAPFile`_ engine +support, as is used by :class:`pcapkit.foundation.extraction.Extractor`. + +.. _PyPCAPFile: https://github.com/kisom/pypcapfile + +""" +import struct +from typing import TYPE_CHECKING, cast + +from pcapkit.const.reg.linktype import LinkType as Enum_LinkType +from pcapkit.foundation.engines.engine import EngineBase as Engine +from pcapkit.foundation.reassembly import ReassemblyManager +from pcapkit.utilities.exceptions import FormatError, stacklevel +from pcapkit.utilities.warnings import AttributeWarning, warn + +__all__ = ['PyPCAPFile'] + +if TYPE_CHECKING: + from typing import Any, BinaryIO, Callable, Iterator, Optional + + from pcapfile.structs import pcap_packet as PCAPFilePacket + + from pcapkit.foundation.extraction import Extractor + + +class _NamedStream: + """Read-only proxy that gives a stream the ``name`` attribute. + + :func:`pcapfile.savefile.load_savefile` dereferences ``input_file.name`` + unconditionally, on the way into its trace helper. That is fine for the + :class:`~io.BufferedReader` :class:`~pcapkit.foundation.extraction.Extractor` + normally holds, but not for the :class:`~pcapkit.corekit.io.SeekableReader` + it substitutes when the caller supplied a non-seekable stream -- that class + exposes no ``name``, so the load would fail with :exc:`AttributeError` before + a single byte was read. + + Args: + stream: Underlying binary stream. + name: Name to report as :attr:`name`. + + """ + + def __init__(self, stream: 'BinaryIO', name: 'str') -> 'None': + self._stream = stream + #: Name of the underlying stream. + self.name = name + + def read(self, size: 'int' = -1) -> 'bytes': + """Read from the underlying stream. + + Args: + size: Number of bytes to read; all remaining bytes if negative. + + """ + return self._stream.read(size) + + +class PyPCAPFile(Engine['PCAPFilePacket']): + """PyPCAPFile engine support. + + `PyPCAPFile`_ is a pure Python savefile reader. It decodes Ethernet, IPv4, + TCP and UDP, and nothing else -- in particular there is no IPv6 decoder and + no PCAP-NG support. Consequently this engine + + * reads PCAP savefiles only, raising + :exc:`~pcapkit.utilities.exceptions.FormatError` on PCAP-NG, and + * disables IPv6 reassembly, warning as it does so, while leaving IPv4 and + TCP reassembly and TCP flow tracing in place. + + The engine stops decoding at the network layer rather than descending into + the transport layer. `PyPCAPFile`_ decoders *replace* the payload bytes of + the layer they decode, so descending further would discard the verbatim TCP + segment that :func:`~pcapkit.toolkit.pypcapfile.tcp_reassembly` needs in + order to report an exact header/payload split. + + .. _PyPCAPFile: https://github.com/kisom/pypcapfile + + Args: + extractor: :class:`~pcapkit.foundation.extraction.Extractor` instance. + + """ + if TYPE_CHECKING: + import pcapfile + + #: Engine extraction package. + _expkg: 'pcapfile' + #: Engine extraction temporary storage. + _extmp: 'Iterator[PCAPFilePacket]' + #: Data link layer protocol, from the savefile header. + _dlink: 'Enum_LinkType' + #: Link layer decoder for this savefile, or :data:`None` when + #: :mod:`pcapfile` has none for its link layer type. + _declf: 'Optional[Callable[..., Any]]' + + #: Number of layers to descend while decoding, i.e. link plus network. See + #: the class docstring for why this stops short of the transport layer. + LAYERS = 2 + + ########################################################################## + # Defaults. + ########################################################################## + + #: Engine name. + __engine_name__ = 'PyPCAPFile' + + #: Engine module name. + __engine_module__ = 'pcapfile' + + ########################################################################## + # Properties. + ########################################################################## + + @property + def dlink(self) -> 'Enum_LinkType': + """Data link layer protocol, as reported by the savefile header.""" + return self._dlink + + ########################################################################## + # Data models. + ########################################################################## + + def __init__(self, extractor: 'Extractor') -> 'None': + import pcapfile # isort:skip + import pcapfile.linklayer # isort:skip + import pcapfile.savefile # isort:skip + import pcapfile.structs # isort:skip + + self._expkg = pcapfile + self._extmp = cast('Iterator[PCAPFilePacket]', None) + self._dlink = cast('Enum_LinkType', None) + self._declf = None + + super().__init__(extractor) + + ########################################################################## + # Methods. + ########################################################################## + + def run(self) -> 'None': + """Call :func:`pcapfile.savefile.load_savefile` to extract PCAP files. + + This method assigns :attr:`self._expkg ` + as :mod:`pcapfile` and :attr:`self._extmp ` + as an iterator over the lazily generated savefile packets. + + The savefile is loaded with ``layers=0``, so that each frame arrives with + its bytes verbatim, and :meth:`read_frame` then decodes it to + :attr:`LAYERS` depth using :func:`pcapfile.linklayer.clookup` -- the very + call :mod:`pcapfile` makes internally. Doing it this way round is what + lets the link layer type be inspected (and reported on) *before* the + first frame is decoded. + + Warns: + AttributeWarning: Warns under following circumstances: + + * if :attr:`self.extractor._exlyr ` + and/or :attr:`self.extractor._exptl ` + is provided as the PyPCAPFile engine currently does not + support such operations. + * if IPv6 reassembly is enabled, as :mod:`pcapfile` has no IPv6 + decoder. + * if :mod:`pcapfile` has no decoder for the savefile's link layer + type, in which case frames are left undecoded. + + Raises: + FormatError: If the file format is not supported, i.e., not a PCAP + file. :mod:`pcapfile` reads libpcap savefiles only. + + """ + from pcapkit.foundation.engines.pcap import PCAP # isort:skip + + ext = self._extractor + pcapfile = self._expkg + + if ext._exlyr != 'none' or ext._exptl != 'null': + warn("'Extractor(engine=pypcapfile)' does not support protocol and layer threshold; " + f"'layer={ext._exlyr}' and 'protocol={ext._exptl}' ignored", + AttributeWarning, stacklevel=stacklevel()) + + if ext.magic_number not in PCAP.MAGIC_NUMBER: + raise FormatError(f'unsupported file format: {ext.magic_number!r}; ' + 'the PyPCAPFile engine reads PCAP savefiles only') + + if ext._flag_r and ext._ipv6: + ext._ipv6 = False + ext._reasm = ReassemblyManager(ipv4=ext._reasm.ipv4, ipv6=None, tcp=ext._reasm.tcp) + warn("'Extractor(engine=pypcapfile)' object dose not support IPv6 reassembly; " + "so 'ipv6=True' will be ignored", AttributeWarning, stacklevel=stacklevel()) + + sfile = pcapfile.savefile.load_savefile( + _NamedStream(ext._ifile, ext._ifnm), layers=0, lazy=True, + ) + self._dlink = Enum_LinkType.get(sfile.header.ll_type) + self._declf = self._get_decoder(sfile.header.ll_type) + + # setup verbose handler + if ext._flag_v: + from pcapkit.toolkit.pypcapfile import packet2chain # isort:skip + ext._vfunc = lambda e, f: print( + f'Frame {e._frnum:>3d}: {packet2chain(f, data_link=self._dlink)}' # pylint: disable=protected-access + ) # pylint: disable=logging-fstring-interpolation + + # extract & analyse file + self._extmp = iter(sfile.packets) + + def read_frame(self) -> 'PCAPFilePacket': + """Read frames with PyPCAPFile engine. + + Returns: + Parsed frame instance. + + See Also: + Please refer to :meth:`PCAP.read_frame ` + for more operational information. + + """ + from pcapkit.toolkit.pypcapfile import (ipv4_reassembly, packet2dict, tcp_reassembly, + tcp_traceflow) + ext = self._extractor + + # fetch PyPCAPFile packet + packet = self._decode(next(self._extmp), ext._frnum + 1) + + # verbose output + ext._frnum += 1 + ext._vfunc(ext, packet) + + # write plist + frnum = f'Frame {ext._frnum}' + if not ext._flag_q: + info = packet2dict(packet, data_link=self._dlink) + if ext._flag_f: + ofile = ext._ofile(f'{ext._ofnm}/{frnum}.{ext._fext}') + ofile(info, name=frnum) + else: + ext._ofile(info, name=frnum) + ofile = ext._ofile + ext._offmt = ofile.kind + + # record fragments + if ext._flag_r: + # NOTE: IPv6 reassembly is switched off in ``run``, as ``pcapfile`` + # cannot decode IPv6 at all. + if ext._ipv4: + data_ipv4 = ipv4_reassembly(packet, count=ext._frnum) + if data_ipv4 is not None: + ext._reasm.ipv4(data_ipv4) + if ext._tcp: + data_tcp = tcp_reassembly(packet, count=ext._frnum) + if data_tcp is not None: + ext._reasm.tcp(data_tcp) + + # trace flows + if ext._flag_t: + if ext._tcp: + data_tf_tcp = tcp_traceflow(packet, data_link=self._dlink, count=ext._frnum) + if data_tf_tcp is not None: + ext._trace.tcp(data_tf_tcp) + + # record frames + if ext._flag_d: + ext._frame.append(packet) + + # return frame record + return packet + + ########################################################################## + # Utilities. + ########################################################################## + + def _get_decoder(self, linktype: 'int') -> 'Optional[Callable[..., Any]]': + """Return the :mod:`pcapfile` link layer decoder for a link layer type. + + Args: + linktype: Link layer type code, from the savefile header. + + Returns: + The decoder class, or :data:`None` when :mod:`pcapfile` has none. + + Warns: + AttributeWarning: If no decoder is available, as frames will then be + left as raw bytes and no reassembly or flow tracing is possible. + + """ + try: + decoder = self._expkg.linklayer.clookup(linktype) + except IndexError: # malformed entry in ``pcapfile.linklayer.__LL_TYPES__`` + decoder = None + + if not callable(decoder): + warn(f'unrecognised link layer protocol: {self._dlink!r}; frames will be left ' + 'undecoded and all analysis functions ignored', AttributeWarning, + stacklevel=stacklevel()) + return None + return decoder + + def _decode(self, packet: 'PCAPFilePacket', frnum: 'int') -> 'PCAPFilePacket': + """Decode a raw savefile packet down to :attr:`LAYERS` depth. + + A new :class:`pcapfile.structs.pcap_packet` is built rather than the + given one mutated, so that a decoding failure leaves the original intact. + + Args: + packet: Undecoded savefile packet, i.e. as loaded with ``layers=0``. + frnum: Frame number, for the warning message below. + + Returns: + The decoded packet, or ``packet`` unchanged when it could not be + decoded. + + Warns: + AttributeWarning: If :mod:`pcapfile` could not decode the frame, e.g. + because the capture is truncated. One bad frame should not abort + the extraction, but it should not pass silently either. + + """ + if self._declf is None: + return packet + + try: + decoded = self._declf(packet.packet, layers=self.LAYERS - 1) + except (struct.error, AssertionError, ValueError, IndexError, KeyError) as error: + warn(f'Frame {frnum}: {self._dlink!r} decoding failed ({error!r}); ' + 'frame left undecoded', AttributeWarning, stacklevel=stacklevel()) + return packet + + return cast('PCAPFilePacket', self._expkg.structs.pcap_packet( + packet.header, packet.timestamp, packet.timestamp_us, + packet.capture_len, packet.packet_len, decoded, + )) diff --git a/pcapkit/foundation/extraction.py b/pcapkit/foundation/extraction.py index 676fe200dc..0f0902cf08 100644 --- a/pcapkit/foundation/extraction.py +++ b/pcapkit/foundation/extraction.py @@ -47,6 +47,7 @@ from typing import IO, Any, Callable, DefaultDict, Iterable, Mapping, Optional, Type, Union from dpkt.dpkt import Packet as DPKTPacket + from pcapfile.structs import pcap_packet as PCAPFilePacket from pyshark.packet.packet import Packet as PySharkPacket from scapy.packet import Packet as ScapyPacket from typing_extensions import Literal @@ -60,10 +61,16 @@ from pcapkit.protocols.protocol import ProtocolBase as Protocol Formats = Literal['pcap', 'json', 'tree', 'plist'] - Engines = Literal['default', 'pcapkit', 'dpkt', 'scapy', 'pyshark'] + # NOTE: this alias is duplicated verbatim in ``pcapkit.interface.misc``; both + # copies need updating when a new engine lands. The duplication predates the + # engines added here and is left as-is on purpose. + Engines = Literal['default', 'pcapkit', 'dpkt', 'scapy', 'pyshark', 'pypcap', 'pypcapfile'] Layers = Literal['link', 'internet', 'transport', 'application', 'none'] - Packet = Union[Frame, PCAPNG, ScapyPacket, DPKTPacket, PySharkPacket] + # NOTE: the PyPCAP engine performs no dissection, so its "packet" is the + # ``(timestamp, bytes)`` pair that ``pcap.pcap`` yields. + Packet = Union[Frame, PCAPNG, ScapyPacket, DPKTPacket, PySharkPacket, + PCAPFilePacket, tuple[float, bytes]] Protocols = Union[str, Protocol, Type[Protocol]] VerboseHandler = Callable[['Extractor', Packet], Any] @@ -190,6 +197,8 @@ class Extractor(Generic[_P]): 'scapy': ModuleDescriptor('pcapkit.foundation.engines.scapy', 'Scapy'), 'dpkt': ModuleDescriptor('pcapkit.foundation.engines.dpkt', 'DPKT'), 'pyshark': ModuleDescriptor('pcapkit.foundation.engines.pyshark', 'PyShark'), + 'pypcap': ModuleDescriptor('pcapkit.foundation.engines.pypcap', 'PyPCAP'), + 'pypcapfile': ModuleDescriptor('pcapkit.foundation.engines.pypcapfile', 'PyPCAPFile'), } # type: dict[str, ModuleDescriptor[Engine] | Type[Engine]] #: Reassembly support mapping for extracting frames. The values should be a tuple @@ -418,6 +427,8 @@ def run(self) -> 'None': # pylint: disable=inconsistent-return-statements * DPKT driver: :class:`pcapkit.foundation.engines.dpkt.DPKT` * Scapy driver: :class:`pcapkit.foundation.engines.scapy.Scapy` * PyShark driver: :class:`pcapkit.foundation.engines.pyshark.PyShark` + * PyPCAP driver: :class:`pcapkit.foundation.engines.pypcap.PyPCAP` + * PyPCAPFile driver: :class:`pcapkit.foundation.engines.pypcapfile.PyPCAPFile` Warns: pcapkit.utilities.warnings.EngineWarning: If the extraction engine is not @@ -790,10 +801,21 @@ def __init__(self, if trace: trace_obj_tcp = None - if self._exnam in ('pyshark',) and trace_format in ('pcap',): + # NOTE: these engines' flow tracing adapters report the frame as a + # plain :obj:`dict`, which the PCAP trace dumper cannot re-serialise + # -- it reaches for ``frame.packet``. ``None`` has to be caught along + # with ``'pcap'`` here, and replaced by a format that *can* take a + # mapping, because :meth:`TraceFlow.__init__ + # ` + # itself substitutes ``'pcap'`` for ``None``. + # + # The DPKT and Scapy engines report the frame as a mapping too and are + # deliberately *not* listed here: they are affected by the same defect + # on this revision, but they are outside the scope of this change. + if self._exnam in ('pyshark', 'pypcapfile') and trace_format in ('pcap', 'cap', None): warn(f"'Extractor(engine={self._exnam})' does not support 'trace_format={trace_format}'; " - "using 'trace_format=None' instead", FormatWarning, stacklevel=stacklevel()) - trace_format = None + "using 'trace_format=\"json\"' instead", FormatWarning, stacklevel=stacklevel()) + trace_format = 'json' if self._tcp: logger.info('TCP flow tracing enabled') diff --git a/pcapkit/interface/misc.py b/pcapkit/interface/misc.py index 07052f08fe..f827fad7e7 100644 --- a/pcapkit/interface/misc.py +++ b/pcapkit/interface/misc.py @@ -27,7 +27,9 @@ ByteOrder = Literal['little', 'big'] Formats = Literal['pcap', 'json', 'tree', 'plist'] - Engines = Literal['default', 'pcapkit', 'dpkt', 'scapy', 'pyshark'] + # NOTE: this alias duplicates the one in ``pcapkit.foundation.extraction``; + # both copies need updating when a new engine lands. + Engines = Literal['default', 'pcapkit', 'dpkt', 'scapy', 'pyshark', 'pypcap', 'pypcapfile'] __all__ = ['follow_tcp_stream'] @@ -72,7 +74,11 @@ def follow_tcp_stream(fin: 'Optional[str]' = None, verbose: 'bool' = False, List of extracted TCP streams. """ - if engine is not None and engine.lower() == 'pyshark': + # NOTE: both of these engines disable TCP flow tracing outright -- PyShark + # because ``pcapkit`` has no reassembly adapter for it, PyPCAP because it + # performs no protocol dissection at all -- so ``extraction.trace`` below + # would raise instead of yielding streams. + if engine is not None and engine.lower() in ('pyshark', 'pypcap'): warn(f'unsupported extraction engine: {engine}; fallback to default engine', EngineWarning, stacklevel=stacklevel()) engine = None diff --git a/pcapkit/toolkit/__init__.py b/pcapkit/toolkit/__init__.py index 6148c81862..d52d5542f1 100644 --- a/pcapkit/toolkit/__init__.py +++ b/pcapkit/toolkit/__init__.py @@ -38,6 +38,19 @@ # from pcapkit.toolkit.scapy import tcp_reassembly as scapy_tcp_reassembly # from pcapkit.toolkit.scapy import tcp_traceflow as scapy_tcp_traceflow +# # tools for PyPCAP engine +# from pcapkit.toolkit.pypcap import packet2chain as pypcap_packet2chain +# from pcapkit.toolkit.pypcap import packet2dict as pypcap_packet2dict + +# # tools for PyPCAPFile engine +# from pcapkit.toolkit.pypcapfile import packet2timestamp as pypcapfile_packet2timestamp +# from pcapkit.toolkit.pypcapfile import ipv4_header as pypcapfile_ipv4_header +# from pcapkit.toolkit.pypcapfile import packet2chain as pypcapfile_packet2chain +# from pcapkit.toolkit.pypcapfile import packet2dict as pypcapfile_packet2dict +# from pcapkit.toolkit.pypcapfile import ipv4_reassembly as pypcapfile_ipv4_reassembly +# from pcapkit.toolkit.pypcapfile import tcp_reassembly as pypcapfile_tcp_reassembly +# from pcapkit.toolkit.pypcapfile import tcp_traceflow as pypcapfile_tcp_traceflow + __all__ = [ # default engine 'ipv4_reassembly', 'ipv6_reassembly', 'tcp_reassembly', 'tcp_traceflow', @@ -54,4 +67,12 @@ # # Scapy engine # 'scapy_packet2chain', 'scapy_packet2dict', # 'scapy_ipv4_reassembly', 'scapy_ipv6_reassembly', 'scapy_tcp_reassembly', 'scapy_tcp_traceflow', + + # # PyPCAP engine + # 'pypcap_packet2chain', 'pypcap_packet2dict', + + # # PyPCAPFile engine + # 'pypcapfile_packet2timestamp', 'pypcapfile_ipv4_header', + # 'pypcapfile_packet2chain', 'pypcapfile_packet2dict', + # 'pypcapfile_ipv4_reassembly', 'pypcapfile_tcp_reassembly', 'pypcapfile_tcp_traceflow', ] diff --git a/pcapkit/toolkit/pypcap.py b/pcapkit/toolkit/pypcap.py new file mode 100644 index 0000000000..d3f2002df1 --- /dev/null +++ b/pcapkit/toolkit/pypcap.py @@ -0,0 +1,166 @@ +# -*- coding: utf-8 -*- +"""PyPCAP Tools +================= + +.. module:: pcapkit.toolkit.pypcap + +:mod:`pcapkit.toolkit.pypcap` contains all you need for +:mod:`pcapkit` handy usage with `PyPCAP`_ engine. All reforming +functions returns with a flag to indicate if usable for +its caller. + +.. _PyPCAP: https://github.com/pynetwork/pypcap + +.. important:: + + `PyPCAP`_ is a thin :manpage:`libpcap(3)` binding: it hands back the + ``(timestamp, bytes)`` pair that :c:func:`pcap_next_ex` produced and performs + **no protocol dissection whatsoever**. There is therefore no IP or TCP layer + for this module to read, and the reassembly and flow tracing adapters that + :mod:`pcapkit.toolkit.dpkt` and :mod:`pcapkit.toolkit.scapy` provide cannot + be implemented here. + + They are still defined below, but only so that reaching for one fails loudly + with :exc:`~pcapkit.utilities.exceptions.UnsupportedCall` rather than with an + :exc:`ImportError` that says nothing about why. Dissecting the raw bytes with + :mod:`pcapkit`'s own parsers would defeat the point of selecting a third-party + engine, so it is deliberately not done. + +""" +from typing import TYPE_CHECKING + +from pcapkit.utilities.exceptions import UnsupportedCall + +if TYPE_CHECKING: + from ipaddress import IPv4Address, IPv6Address + from typing import Any + + from pcapkit.const.reg.linktype import LinkType as Enum_LinkType + from pcapkit.foundation.reassembly.data.ip import Packet as IP_Packet + from pcapkit.foundation.reassembly.data.tcp import Packet as TCP_Packet + from pcapkit.foundation.traceflow.data.tcp import Packet as TF_TCP_Packet + +__all__ = [ + 'packet2chain', 'packet2dict', + 'ipv4_reassembly', 'ipv6_reassembly', 'tcp_reassembly', 'tcp_traceflow', +] + +#: Explanatory suffix shared by every unsupported adapter below. +_NO_DISSECTION = ("'pypcap' is a libpcap binding and performs no protocol " + "dissection, so there is no protocol layer to read") + + +def packet2chain(packet: 'bytes', *, data_link: 'Enum_LinkType') -> 'str': + """Fetch PyPCAP packet protocol chain. + + Args: + packet: Raw packet bytes, as returned by :class:`pcap.pcap` iteration. + data_link: Data link type, from the capture handle. + + Returns: + Colon (``:``) seperated list of protocol chain. + + Note: + As `PyPCAP`_ does not dissect the packet, the chain is only ever the + link layer type followed by ``Raw``, e.g. ``ETHERNET:Raw``. + + .. _PyPCAP: https://github.com/pynetwork/pypcap + + """ + return f'{data_link.name}:Raw' + + +def packet2dict(packet: 'bytes', timestamp: 'float', *, + data_link: 'Enum_LinkType') -> 'dict[str, Any]': + """Convert PyPCAP packet into :obj:`dict`. + + Args: + packet: Raw packet bytes, as returned by :class:`pcap.pcap` iteration. + timestamp: Timestamp of packet, as returned by :class:`pcap.pcap` iteration. + data_link: Data link type, from the capture handle. + + Returns: + Dict[str, Any]: A :obj:`dict` mapping of packet data. + + Note: + The mapping carries the captured bytes verbatim rather than a decoded + protocol tree, since `PyPCAP`_ does not decode anything. + + .. _PyPCAP: https://github.com/pynetwork/pypcap + + """ + return { + 'timestamp': timestamp, + 'packet': packet, + data_link.name: { + 'raw_len': len(packet), + 'raw': packet, + }, + } + + +def ipv4_reassembly(packet: 'bytes', *, count: 'int' = -1) -> 'IP_Packet[IPv4Address] | None': + """Make data for IPv4 reassembly. + + Args: + packet: Raw packet bytes. + count: Packet index. If not provided, default to ``-1``. + + Raises: + UnsupportedCall: Always, as `PyPCAP`_ provides no IPv4 layer to read. + + .. _PyPCAP: https://github.com/pynetwork/pypcap + + """ + raise UnsupportedCall(f'IPv4 reassembly is not supported by the PyPCAP engine: {_NO_DISSECTION}') + + +def ipv6_reassembly(packet: 'bytes', *, count: 'int' = -1) -> 'IP_Packet[IPv6Address] | None': + """Make data for IPv6 reassembly. + + Args: + packet: Raw packet bytes. + count: Packet index. If not provided, default to ``-1``. + + Raises: + UnsupportedCall: Always, as `PyPCAP`_ provides no IPv6 layer to read. + + .. _PyPCAP: https://github.com/pynetwork/pypcap + + """ + raise UnsupportedCall(f'IPv6 reassembly is not supported by the PyPCAP engine: {_NO_DISSECTION}') + + +def tcp_reassembly(packet: 'bytes', *, count: 'int' = -1) -> 'TCP_Packet | None': + """Make data for TCP reassembly. + + Args: + packet: Raw packet bytes. + count: Packet index. If not provided, default to ``-1``. + + Raises: + UnsupportedCall: Always, as `PyPCAP`_ provides no TCP layer to read. + + .. _PyPCAP: https://github.com/pynetwork/pypcap + + """ + raise UnsupportedCall(f'TCP reassembly is not supported by the PyPCAP engine: {_NO_DISSECTION}') + + +def tcp_traceflow(packet: 'bytes', timestamp: 'float', *, data_link: 'Enum_LinkType', + count: 'int' = -1) -> 'TF_TCP_Packet | None': + """Trace packet flow for TCP. + + Args: + packet: Raw packet bytes. + timestamp: Timestamp of the packet. + data_link: Data link layer protocol (from the capture handle). + count: Packet index. If not provided, default to ``-1``. + + Raises: + UnsupportedCall: Always, as `PyPCAP`_ provides no TCP layer to read. + + .. _PyPCAP: https://github.com/pynetwork/pypcap + + """ + raise UnsupportedCall(f'TCP flow tracing is not supported by the PyPCAP engine: {_NO_DISSECTION}') diff --git a/pcapkit/toolkit/pypcapfile.py b/pcapkit/toolkit/pypcapfile.py new file mode 100644 index 0000000000..018133eab3 --- /dev/null +++ b/pcapkit/toolkit/pypcapfile.py @@ -0,0 +1,454 @@ +# -*- coding: utf-8 -*- +"""PyPCAPFile Tools +===================== + +.. module:: pcapkit.toolkit.pypcapfile + +:mod:`pcapkit.toolkit.pypcapfile` contains all you need for +:mod:`pcapkit` handy usage with `PyPCAPFile`_ engine. All reforming +functions returns with a flag to indicate if usable for +its caller. + +.. _PyPCAPFile: https://github.com/kisom/pypcapfile + +.. important:: + + `PyPCAPFile`_ decodes Ethernet, IPv4, TCP and UDP and nothing else -- there + is no IPv6 decoder at all. :func:`ipv6_reassembly` therefore cannot be + implemented and raises :exc:`~pcapkit.utilities.exceptions.UnsupportedCall` + instead of quietly returning :data:`None`, which would be indistinguishable + from "this frame carries no IPv6 fragment". + + Note also that `PyPCAPFile`_ decoders *replace* the payload bytes of the layer + they decode. :class:`~pcapkit.foundation.engines.pypcapfile.PyPCAPFile` therefore + stops at the network layer, which keeps the transport segment verbatim so that + the header/payload split this module reports is exact. The transport header is + decoded on demand below, using `PyPCAPFile`_'s own + :class:`~pcapfile.protocols.transport.tcp.TCP` class rather than a + reimplementation. + +""" +import ipaddress +import struct +from typing import TYPE_CHECKING + +from pcapkit.const.reg.transtype import TransType as Enum_TransType +from pcapkit.foundation.reassembly.data.ip import Packet as IP_Packet +from pcapkit.foundation.reassembly.data.tcp import Packet as TCP_Packet +from pcapkit.foundation.traceflow.data.tcp import Packet as TF_TCP_Packet +from pcapkit.utilities.exceptions import UnsupportedCall + +if TYPE_CHECKING: + from ipaddress import IPv4Address, IPv6Address + from typing import Any, Optional + + from pcapfile.protocols.network.ip import IP + from pcapfile.structs import pcap_packet as Packet + + from pcapkit.const.reg.linktype import LinkType as Enum_LinkType + +__all__ = [ + 'packet2timestamp', 'ipv4_header', 'packet2chain', 'packet2dict', + 'ipv4_reassembly', 'ipv6_reassembly', 'tcp_reassembly', 'tcp_traceflow', +] + +#: Minimum length of a TCP header, i.e. the fixed part with no options. +TCP_MIN_HEADER_LEN = 20 + +#: IPv4 **DF** (don't fragment) bit, within the three-bit flags field. +IPV4_FLAG_DF = 0b010 + +#: IPv4 **MF** (more fragments) bit, within the three-bit flags field. +IPV4_FLAG_MF = 0b001 + + +def packet2timestamp(packet: 'Packet') -> 'float': + """Calculate the timestamp of a PyPCAPFile packet. + + Args: + packet: PyPCAPFile packet. + + Returns: + Timestamp of the packet, in seconds since the epoch. + + Note: + `PyPCAPFile`_ keeps the two halves of the per-packet timestamp apart, and + the sub-second half is nanoseconds rather than microseconds when the + savefile carries the nanosecond magic number -- which is recorded on the + savefile header as ``ns_resolution``. + + .. _PyPCAPFile: https://github.com/kisom/pypcapfile + + """ + divisor = 1_000_000_000 if packet.header[0].ns_resolution else 1_000_000 + return packet.timestamp + packet.timestamp_us / divisor + + +def ipv4_header(ipv4: 'IP') -> 'bytes': + """Rebuild the raw header bytes of a PyPCAPFile IPv4 packet. + + Args: + ipv4: PyPCAPFile IPv4 packet. + + Returns: + Raw IPv4 header bytes, options included. + + Note: + `PyPCAPFile`_ discards the header bytes once decoded, but it retains + *every* IPv4 header field plus the options blob, so this reconstruction + is byte-exact rather than approximate. + + .. _PyPCAPFile: https://github.com/kisom/pypcapfile + + """ + return struct.pack( + '!BBHHHBBHII', + (ipv4.v << 4) | ipv4.hl, # version and internet header length + ipv4.tos, # type of service + ipv4.len, # total length + ipv4.id, # identification + (ipv4.flags << 13) | ipv4.off, # flags and fragment offset + ipv4.ttl, # time to live + ipv4.p, # payload protocol type + ipv4.sum, # header checksum + ipv4.src, # source IP address + ipv4.dst, # destination IP address + ) + bytes(ipv4.opt) + + +def _is_raw(layer: 'Any') -> 'bool': + """Test if a decoded layer is in fact undecoded raw bytes. + + Args: + layer: Decoded layer, or raw bytes. + + """ + return isinstance(layer, (bytes, bytearray, memoryview)) + + +def _network(packet: 'Packet') -> 'Optional[IP]': + """Fetch the decoded IPv4 layer of a PyPCAPFile packet, if any. + + Args: + packet: PyPCAPFile packet. + + Returns: + The decoded :class:`~pcapfile.protocols.network.ip.IP` layer, or + :data:`None` when the frame was not decoded that far -- which is the + case for every non-IPv4 frame, `PyPCAPFile`_ having no other network + layer decoder. + + .. _PyPCAPFile: https://github.com/kisom/pypcapfile + + """ + link = packet.packet + if _is_raw(link): + return None + + payload = getattr(link, 'payload', None) + if type(payload).__name__ != 'IP': + return None + return payload + + +def _transport(ipv4: 'IP') -> 'Optional[bytes]': + """Fetch the verbatim TCP segment carried by a PyPCAPFile IPv4 packet. + + Args: + ipv4: PyPCAPFile IPv4 packet. + + Returns: + The raw TCP segment, or :data:`None` if the packet does not carry a + TCP payload long enough to hold a header. + + """ + if ipv4.p != Enum_TransType.TCP: + return None + + segment = ipv4.payload + if not _is_raw(segment): + return None + + segment = bytes(segment) + if len(segment) < TCP_MIN_HEADER_LEN: + return None + return segment + + +def _ethernet2dict(link: 'Any') -> 'dict[str, Any]': + """Convert a PyPCAPFile Ethernet frame into :obj:`dict`. + + Args: + link: PyPCAPFile Ethernet frame. + + """ + return { + 'dst': bytes(link.dst), + 'src': bytes(link.src), + 'type': link.type, + } + + +def _ipv4_2dict(ipv4: 'IP') -> 'dict[str, Any]': + """Convert a PyPCAPFile IPv4 packet into :obj:`dict`. + + Args: + ipv4: PyPCAPFile IPv4 packet. + + """ + return { + 'v': ipv4.v, + 'hl': ipv4.hl, + 'tos': ipv4.tos, + 'len': ipv4.len, + 'id': ipv4.id, + 'flags': ipv4.flags, + 'off': ipv4.off, + 'ttl': ipv4.ttl, + 'p': ipv4.p, + 'sum': ipv4.sum, + 'src': str(ipaddress.IPv4Address(ipv4.src)), + 'dst': str(ipaddress.IPv4Address(ipv4.dst)), + 'opt': bytes(ipv4.opt), + } + + +def _layer2dict(layer: 'Any') -> 'dict[str, Any]': + """Convert a decoded PyPCAPFile layer into :obj:`dict`, recursively. + + Args: + layer: Decoded PyPCAPFile layer, or raw bytes. + + """ + if _is_raw(layer): + raw = bytes(layer) + return {'raw_len': len(raw), 'raw': raw} + + name = type(layer).__name__ + if name == 'Ethernet': + dict_ = _ethernet2dict(layer) + elif name == 'IP': + dict_ = _ipv4_2dict(layer) + else: + dict_ = {field[0]: getattr(layer, field[0], None) + for field in getattr(type(layer), '_fields_', ())} + + payload = getattr(layer, 'payload', None) + if payload is not None: + dict_['Raw' if _is_raw(payload) else type(payload).__name__] = _layer2dict(payload) + return dict_ + + +def packet2chain(packet: 'Packet', *, data_link: 'Enum_LinkType') -> 'str': + """Fetch PyPCAPFile packet protocol chain. + + Args: + packet: PyPCAPFile packet. + data_link: Data link type, from the savefile header. + + Returns: + Colon (``:``) seperated list of protocol chain. + + Note: + The chain reports what `PyPCAPFile`_ actually decoded and no more, so it + ends in ``Raw`` -- the transport segment is left undecoded on purpose, + see the module notes. + + .. _PyPCAPFile: https://github.com/kisom/pypcapfile + + """ + layer = packet.packet + if _is_raw(layer): + return f'{data_link.name}:Raw' + + chain = [] # type: list[str] + while not _is_raw(layer): + chain.append(type(layer).__name__) + layer = getattr(layer, 'payload', b'') + chain.append('Raw') + return ':'.join(chain) + + +def packet2dict(packet: 'Packet', *, data_link: 'Enum_LinkType') -> 'dict[str, Any]': + """Convert PyPCAPFile packet into :obj:`dict`. + + Args: + packet: PyPCAPFile packet. + data_link: Data link type, from the savefile header. + + Returns: + Dict[str, Any]: A :obj:`dict` mapping of packet data. + + """ + return { + 'timestamp': packet2timestamp(packet), + 'capture_len': packet.capture_len, + 'packet_len': packet.packet_len, + data_link.name: _layer2dict(packet.packet), + } + + +def ipv4_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv4Address] | None': + """Make data for IPv4 reassembly. + + Args: + packet: PyPCAPFile packet. + count: Packet index. If not provided, default to ``-1``. + + Returns: + Data for IPv4 reassembly. + + * If the ``packet`` can be used for IPv4 reassembly. A packet can be reassembled + if it contains an IPv4 layer (:class:`pcapfile.protocols.network.ip.IP`) and the + **DF** flag is :data:`False`. + * If the ``packet`` can be reassembled, then the :obj:`dict` mapping of data for IPv4 + reassembly (:term:`reasm.ipv4.packet`) will be returned; otherwise, returns :data:`None`. + + See Also: + :class:`pcapkit.foundation.reassembly.ipv4.IPv4` + + """ + ipv4 = _network(packet) + if ipv4 is None: + return None + if ipv4.flags & IPV4_FLAG_DF: # dismiss not fragmented packet + return None + + header = ipv4_header(ipv4) + return IP_Packet( + bufid=( + ipaddress.IPv4Address(ipv4.src), # source IP address + ipaddress.IPv4Address(ipv4.dst), # destination IP address + ipv4.id, # identification + Enum_TransType.get(ipv4.p), # payload protocol type + ), + num=count, # original packet range number + fo=ipv4.off * 8, # fragment offset, in octets + ihl=len(header), # internet header length + mf=bool(ipv4.flags & IPV4_FLAG_MF), # more fragment flag + tl=ipv4.len, # total length, header includes + header=header, # raw bytes type header + payload=bytearray(ipv4.payload), # raw bytearray type payload + ) + + +def ipv6_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv6Address] | None': + """Make data for IPv6 reassembly. + + Args: + packet: PyPCAPFile packet. + count: Packet index. If not provided, default to ``-1``. + + Raises: + UnsupportedCall: Always, as `PyPCAPFile`_ has no IPv6 decoder. + + .. _PyPCAPFile: https://github.com/kisom/pypcapfile + + """ + raise UnsupportedCall('IPv6 reassembly is not supported by the PyPCAPFile engine: ' + "'pypcapfile' has no IPv6 decoder, so no IPv6 fragment header " + 'can be read') + + +def tcp_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'TCP_Packet | None': + """Make data for TCP reassembly. + + Args: + packet: PyPCAPFile packet. + count: Packet index. If not provided, default to ``-1``. + + Returns: + Data for TCP reassembly. + + * If the ``packet`` can be used for TCP reassembly. A packet can be reassembled + if it contains an IPv4 layer carrying a TCP segment. + * If the ``packet`` can be reassembled, then the :obj:`dict` mapping of data for TCP + reassembly (:term:`reasm.tcp.packet`) will be returned; otherwise, returns :data:`None`. + + See Also: + :class:`pcapkit.foundation.reassembly.tcp.TCP` + + """ + ipv4 = _network(packet) + if ipv4 is None: + return None + + segment = _transport(ipv4) + if segment is None: + return None + + # NOTE: imported only once there is something to decode, so that declining a + # frame does not require ``pcapfile`` to be installed. + from pcapfile.protocols.transport.tcp import TCP # isort:skip + + tcp = TCP(segment) + hdr_len = max(tcp.data_offset, TCP_MIN_HEADER_LEN) + payload = segment[hdr_len:] + + return TCP_Packet( + bufid=( + ipaddress.IPv4Address(ipv4.src), # source IP address + tcp.src_port, # source port + ipaddress.IPv4Address(ipv4.dst), # destination IP address + tcp.dst_port, # destination port + ), + num=count, # original packet range number + ack=tcp.acknum, # acknowledgement + dsn=tcp.seqnum, # data sequence number + rst=bool(tcp.rst), # reset connection flag + syn=bool(tcp.syn), # synchronise flag + fin=bool(tcp.fin), # finish flag + header=segment[:hdr_len], # raw bytes type header + payload=bytearray(payload), # raw bytearray type payload + first=tcp.seqnum, # this sequence number + last=tcp.seqnum + len(payload), # next (wanted) sequence number + len=len(payload), # payload length, header excludes + ) + + +def tcp_traceflow(packet: 'Packet', *, data_link: 'Enum_LinkType', + count: 'int' = -1) -> 'TF_TCP_Packet | None': + """Trace packet flow for TCP. + + Args: + packet: PyPCAPFile packet. + data_link: Data link layer protocol (from the savefile header). + count: Packet index. If not provided, default to ``-1``. + + Returns: + Data for TCP flow tracing. + + * If the ``packet`` can be used for TCP flow tracing. A packet can be traced + if it contains an IPv4 layer carrying a TCP segment. + * If the ``packet`` can be traced, then the :obj:`dict` mapping of data for TCP + flow tracing (:term:`trace.tcp.packet`) will be returned; otherwise, returns :data:`None`. + + See Also: + :class:`pcapkit.foundation.traceflow.tcp.TCP` + + """ + ipv4 = _network(packet) + if ipv4 is None: + return None + + segment = _transport(ipv4) + if segment is None: + return None + + # NOTE: imported only once there is something to decode, so that declining a + # frame does not require ``pcapfile`` to be installed. + from pcapfile.protocols.transport.tcp import TCP # isort:skip + + tcp = TCP(segment) + return TF_TCP_Packet( # type: ignore[type-var] + protocol=data_link, # data link type from savefile header + index=count, # frame number + frame=packet2dict(packet, data_link=data_link), # extracted packet + syn=bool(tcp.syn), # TCP synchronise (SYN) flag + fin=bool(tcp.fin), # TCP finish (FIN) flag + src=ipaddress.IPv4Address(ipv4.src), # source IP + dst=ipaddress.IPv4Address(ipv4.dst), # destination IP + srcport=tcp.src_port, # TCP source port + dstport=tcp.dst_port, # TCP destination port + timestamp=packet2timestamp(packet), # timestamp + ) diff --git a/pyproject.toml b/pyproject.toml index 9f2f6d9510..c18df2deec 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -88,12 +88,14 @@ crypto = [ "cryptography>=3.4" ] DPKT = [ "dpkt" ] Scapy = [ "scapy" ] PyShark = [ "pyshark" ] +PyPCAP = [ "pypcap" ] +PyPCAPFile = [ "pypcapfile" ] # for developers vendor = [ "requests[socks]", "beautifulsoup4[html5lib]" ] all = [ "emoji", "cryptography>=3.4", - "dpkt", "scapy", "pyshark", + "dpkt", "scapy", "pyshark", "pypcap", "pypcapfile", "requests[socks]", "beautifulsoup4[html5lib]", ] docs = [ diff --git a/tests/foundation/engines/test_new_engine_parity.py b/tests/foundation/engines/test_new_engine_parity.py new file mode 100644 index 0000000000..561a8e6bbf --- /dev/null +++ b/tests/foundation/engines/test_new_engine_parity.py @@ -0,0 +1,271 @@ +"""End-to-end agreement between the new engines and the ``default`` engine. + +: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. + +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. + +""" +from __future__ import annotations + +import importlib +import importlib.util +import unittest +from unittest import mock + +from tests._support import close_extractor, purge_modules, sample_path + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + + +def _importable(*modules: str) -> bool: + """Test if every named module can actually be imported. + + :func:`importlib.util.find_spec` is not enough for ``pypcapfile``: its released + 0.12.0 imports the :mod:`imp` module, removed in Python 3.12, so it can be + installed and still unusable. + + """ + for module in modules: + try: + importlib.import_module(module) + except ImportError: + return False + return True + + +HAS_PYPCAP = _importable('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): + def setUp(self) -> None: + purge_modules(['pcapkit']) + + def extract(self, engine: str, capture: str, **kwargs): + """Extract a capture, asserting that the requested engine really ran. + + Returns the extractor, having checked that + :class:`~pcapkit.foundation.extraction.Extractor` neither rewrote the engine + name nor warned about the engine being unavailable -- i.e. that the result + is not the built-in parser wearing the requested engine's label. + + """ + from pcapkit.interface import extract + from pcapkit.utilities.warnings import EngineWarning + + with mock.patch('pcapkit.foundation.extraction.warn') as warn: + extractor = extract(fin=sample_path(capture), fout='/tmp/parity-out', + format='tree', store=True, nofile=True, engine=engine, + **kwargs) + self.addCleanup(close_extractor, extractor) + + engines = [call.args[0] for call in warn.call_args_list + if len(call.args) > 1 and call.args[1] is EngineWarning] + self.assertEqual(engines, [], f'{engine!r} was replaced by the fallback engine') + self.assertEqual(extractor._exnam, engine) + return extractor + + ########################################################################## + # Proof that the engine ran. + ########################################################################## + + @unittest.skipUnless(HAS_PYPCAP, 'pypcap not installed') + def test_pypcap_engine_really_ran(self) -> None: + from pcapkit.foundation.engines.pypcap import PyPCAP + + extractor = self.extract('pypcap', 'in.pcap') + self.assertIsInstance(extractor.engine, PyPCAP) + self.assertEqual(extractor.engine.name, 'PyPCAP') + self.assertEqual(extractor.engine.module, 'pcap') + # a fallback would have produced ``Frame`` objects, not bare tuples + for frame in extractor.frame: + self.assertIsInstance(frame, tuple) + self.assertEqual(len(frame), 2) + + @unittest.skipUnless(HAS_PYPCAPFILE, 'pypcapfile not installed') + def test_pypcapfile_engine_really_ran(self) -> None: + from pcapfile.structs import pcap_packet + + from pcapkit.foundation.engines.pypcapfile import PyPCAPFile + + extractor = self.extract('pypcapfile', 'in.pcap') + self.assertIsInstance(extractor.engine, PyPCAPFile) + self.assertEqual(extractor.engine.name, 'PyPCAPFile') + self.assertEqual(extractor.engine.module, 'pcapfile') + 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') + self.assertEqual(mac(ethernet.dst), dst, + f'{capture} frame {index + 1}: ethernet dst') + self.assertEqual(mac(ethernet.src), 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. + ########################################################################## + + @unittest.skipUnless(HAS_PYPCAPFILE, 'pypcapfile not installed') + def test_pypcapfile_reports_no_ipv6_reassembly(self) -> None: + engine = self.extract('pypcapfile', 'ipv6.pcap', reassembly=True, ip=True) + self.assertIsNone(engine.reassembly.ipv6) + # ...whereas the default engine does reassemble it + 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 + + engine = self.extract('pypcap', 'tcp.pcap', reassembly=True, ip=True, tcp=True, + trace=True) + with self.assertRaises(UnsupportedCall): + engine.reassembly # pylint: disable=pointless-statement + with self.assertRaises(UnsupportedCall): + engine.trace # pylint: disable=pointless-statement + + ########################################################################## + # PCAP-NG. + ########################################################################## + + @unittest.skipUnless(HAS_PYPCAP, 'pypcap not installed') + def test_pypcap_rejects_pcapng_rather_than_reporting_zero_frames(self) -> None: + from pcapkit.utilities.exceptions import FormatError + + # libpcap opens a PCAP-NG savefile without complaint and then yields no + # frames at all, which would look like an empty capture; the engine gates + # on the magic number so that it looks like an error instead. + with self.assertRaises(FormatError): + self.extract('pypcap', 'test.pcapng') + + @unittest.skipUnless(HAS_PYPCAPFILE, 'pypcapfile not installed') + def test_pypcapfile_rejects_pcapng(self) -> None: + from pcapkit.utilities.exceptions import FormatError + + with self.assertRaises(FormatError): + self.extract('pypcapfile', 'test.pcapng') + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/foundation/engines/test_pypcap_engine.py b/tests/foundation/engines/test_pypcap_engine.py new file mode 100644 index 0000000000..de5e2ecfd6 --- /dev/null +++ b/tests/foundation/engines/test_pypcap_engine.py @@ -0,0 +1,293 @@ +"""Unit tests for :mod:`pcapkit.foundation.engines.pypcap`. + +The engine itself is exercised with a stand-in for :class:`pcap.pcap`, so that the +routing decisions -- format gating, capability warnings, output, storage, close -- +are covered whether or not `pypcap`_ is installed. End-to-end agreement with the +``default`` engine lives in :mod:`tests.foundation.engines.test_new_engine_parity`. + +.. _pypcap: https://github.com/pynetwork/pypcap + +""" +from __future__ import annotations + +import importlib.util +import io +import os +import tempfile +import types +import unittest +from unittest import mock + +from tests._support import purge_modules + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + +#: Magic number of a little-endian, microsecond-resolution PCAP savefile. +PCAP_MAGIC = b'\xd4\xc3\xb2\xa1' +#: Magic number of a PCAP-NG section header block. +PCAPNG_MAGIC = b'\x0a\x0d\x0d\x0a' + + +class OutputSink: + """Stand-in for a :class:`dictdumper.dumper.Dumper`, recording what it is handed.""" + + kind = 'unit' + + def __init__(self) -> None: + self.paths: list[str] = [] + self.records: list[tuple[object, str | None]] = [] + + def __call__(self, *args, **kwargs): + if len(args) == 1 and isinstance(args[0], str) and not kwargs: + self.paths.append(args[0]) + return self + self.records.append((args[0] if args else None, kwargs.get('name'))) + return self + + +class FakeHandle: + """Stand-in for :class:`pcap.pcap`, yielding ``(timestamp, bytes)`` pairs.""" + + def __init__(self, frames=None, datalink: int = 1) -> None: + self.frames = list(frames if frames is not None else [(1.5, b'payload')]) + self._iter = None + self._datalink = datalink + self.closed = 0 + self.setup = 0 + + def datalink(self) -> int: + return self._datalink + + def __iter__(self): + self.setup += 1 + self._iter = iter(self.frames) + return self + + def __next__(self): + return next(self._iter) + + def close(self) -> None: + self.closed += 1 + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class PyPCAPEngineTests(unittest.TestCase): + def setUp(self) -> None: + purge_modules(['pcapkit']) + + handle, path = tempfile.mkstemp(suffix='.pcap') + os.close(handle) + self.addCleanup(os.unlink, path) + self.ifnm = path + + def make_extractor(self, **overrides): + sink = OutputSink() + reasm = types.SimpleNamespace(ipv4=mock.Mock(), ipv6=mock.Mock(), tcp=mock.Mock()) + trace = types.SimpleNamespace(tcp=mock.Mock()) + values = { + '_ifile': io.BytesIO(b'capture'), + '_ifnm': self.ifnm, + '_ofile': sink, + '_ofnm': 'out', + '_fext': 'json', + '_offmt': None, + '_flag_q': False, + '_flag_f': False, + '_flag_v': False, + '_flag_r': False, + '_flag_t': False, + '_flag_d': True, + '_ipv4': False, + '_ipv6': False, + '_tcp': False, + '_reasm': reasm, + '_trace': trace, + '_frame': [], + '_frnum': 0, + '_exlyr': 'none', + '_exptl': 'null', + '_vfunc': mock.Mock(), + 'magic_number': PCAP_MAGIC, + } + values.update(overrides) + return types.SimpleNamespace(**values), sink + + def engine(self, extractor): + from pcapkit.foundation.engines.pypcap import PyPCAP + + engine = PyPCAP.__new__(PyPCAP) + engine._expkg = types.SimpleNamespace(pcap=FakeHandle) + engine._extmp = None + engine._dlink = None + engine._closed = False + engine._extractor = extractor + return engine + + ########################################################################## + # run() + ########################################################################## + + def test_run_opens_handle_and_records_link_type(self) -> None: + from pcapkit.const.reg.linktype import LinkType + + extractor, _ = self.make_extractor() + engine = self.engine(extractor) + + handle = FakeHandle() + with mock.patch.object(engine._expkg, 'pcap', return_value=handle) as ctor: + engine.run() + + ctor.assert_called_once_with(name=self.ifnm, promisc=False) + self.assertIs(engine._extmp, handle) + self.assertEqual(handle.setup, 1) + self.assertEqual(engine.dlink, LinkType.ETHERNET) + + def test_run_warns_on_layer_and_protocol_threshold(self) -> None: + from pcapkit.utilities.warnings import AttributeWarning + + # NOTE: ``BaseWarning.__init__`` installs an ``ignore`` filter for its own + # category outside development mode, so the warning never reaches + # ``assertWarns``; the ``warn`` call itself is what can be observed. + for overrides in ({'_exlyr': 'internet'}, {'_exptl': 'tcp'}): + with self.subTest(overrides=overrides): + extractor, _ = self.make_extractor(**overrides) + engine = self.engine(extractor) + with mock.patch.object(engine._expkg, 'pcap', return_value=FakeHandle()): + with mock.patch('pcapkit.foundation.engines.pypcap.warn') as warn: + engine.run() + self.assertEqual(warn.call_count, 1) + self.assertIn('protocol and layer threshold', warn.call_args.args[0]) + self.assertIs(warn.call_args.args[1], AttributeWarning) + + def test_run_rejects_pcapng_rather_than_yielding_no_frames(self) -> None: + from pcapkit.utilities.exceptions import FormatError + + extractor, _ = self.make_extractor(magic_number=PCAPNG_MAGIC) + engine = self.engine(extractor) + with mock.patch.object(engine._expkg, 'pcap', return_value=FakeHandle()) as ctor: + with self.assertRaises(FormatError): + engine.run() + ctor.assert_not_called() + + def test_run_rejects_input_that_is_not_a_file_on_disk(self) -> None: + from pcapkit.utilities.exceptions import UnsupportedCall + + extractor, _ = self.make_extractor(_ifnm=os.path.join(self.ifnm, 'nope.pcap')) + engine = self.engine(extractor) + with mock.patch.object(engine._expkg, 'pcap', return_value=FakeHandle()) as ctor: + with self.assertRaises(UnsupportedCall): + engine.run() + ctor.assert_not_called() + + def test_run_disables_reassembly_and_flow_tracing(self) -> None: + from pcapkit.utilities.warnings import AttributeWarning + + extractor, _ = self.make_extractor(_flag_r=True, _flag_t=True, _ipv4=True, + _ipv6=True, _tcp=True) + engine = self.engine(extractor) + with mock.patch.object(engine._expkg, 'pcap', return_value=FakeHandle()): + with mock.patch('pcapkit.foundation.engines.pypcap.warn') as warn: + engine.run() + + messages = [call.args[0] for call in warn.call_args_list + if call.args[1] is AttributeWarning] + self.assertTrue(any('reassembly' in message for message in messages), messages) + self.assertTrue(any('flow tracing' in message for message in messages), messages) + + self.assertFalse(extractor._flag_r) + self.assertFalse(extractor._flag_t) + self.assertIsNone(extractor._reasm.ipv4) + self.assertIsNone(extractor._reasm.ipv6) + self.assertIsNone(extractor._reasm.tcp) + self.assertIsNone(extractor._trace.tcp) + + def test_run_leaves_disabled_reassembly_and_tracing_untouched(self) -> None: + extractor, _ = self.make_extractor(_flag_r=True, _flag_t=True) + original_reasm, original_trace = extractor._reasm, extractor._trace + engine = self.engine(extractor) + with mock.patch.object(engine._expkg, 'pcap', return_value=FakeHandle()): + engine.run() + + # no protocol was requested, so there is nothing to warn about or replace + self.assertTrue(extractor._flag_r) + self.assertTrue(extractor._flag_t) + self.assertIs(extractor._reasm, original_reasm) + self.assertIs(extractor._trace, original_trace) + + def test_run_installs_verbose_handler_reporting_the_raw_chain(self) -> None: + extractor, _ = self.make_extractor(_flag_v=True) + engine = self.engine(extractor) + with mock.patch.object(engine._expkg, 'pcap', return_value=FakeHandle()): + engine.run() + + with mock.patch('builtins.print') as printer: + extractor._frnum = 3 + extractor._vfunc(extractor, (1.5, b'payload')) + printer.assert_called_once() + self.assertIn('ETHERNET:Raw', printer.call_args.args[0]) + + ########################################################################## + # read_frame() + ########################################################################## + + def prepared(self, **overrides): + extractor, sink = self.make_extractor(**overrides) + engine = self.engine(extractor) + handle = FakeHandle(frames=[(1.5, b'payload'), (2.5, b'second')]) + with mock.patch.object(engine._expkg, 'pcap', return_value=handle): + engine.run() + return extractor, sink, engine + + def test_read_frame_returns_the_pair_and_routes_output_and_storage(self) -> None: + extractor, sink, engine = self.prepared() + + frame = engine.read_frame() + + self.assertEqual(frame, (1.5, b'payload')) + self.assertEqual(extractor._frnum, 1) + extractor._vfunc.assert_called_once_with(extractor, frame) + self.assertEqual(sink.records[-1][1], 'Frame 1') + self.assertEqual(sink.records[-1][0]['timestamp'], 1.5) + self.assertEqual(sink.records[-1][0]['packet'], b'payload') + self.assertEqual(sink.records[-1][0]['ETHERNET'], {'raw_len': 7, 'raw': b'payload'}) + self.assertEqual(extractor._offmt, 'unit') + self.assertEqual(extractor._frame, [frame]) + + self.assertEqual(engine.read_frame(), (2.5, b'second')) + self.assertEqual(extractor._frnum, 2) + with self.assertRaises(StopIteration): + engine.read_frame() + + def test_read_frame_splits_files_when_asked(self) -> None: + extractor, sink, engine = self.prepared(_flag_f=True) + engine.read_frame() + self.assertEqual(sink.paths[-1], 'out/Frame 1.json') + + def test_read_frame_writes_nothing_and_stores_nothing_when_disabled(self) -> None: + extractor, sink, engine = self.prepared(_flag_q=True, _flag_d=False) + engine.read_frame() + self.assertEqual(sink.records, []) + self.assertEqual(sink.paths, []) + self.assertEqual(extractor._frame, []) + self.assertIsNone(extractor._offmt) + + ########################################################################## + # close() + ########################################################################## + + def test_close_is_idempotent_and_tolerates_an_unopened_engine(self) -> None: + extractor, _, engine = self.prepared() + handle = engine._extmp + + engine.close() + engine.close() + self.assertEqual(handle.closed, 1) + + extractor, _ = self.make_extractor() + unopened = self.engine(extractor) + unopened.close() # must not raise + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/foundation/engines/test_pypcapfile_engine.py b/tests/foundation/engines/test_pypcapfile_engine.py new file mode 100644 index 0000000000..7487e9bec0 --- /dev/null +++ b/tests/foundation/engines/test_pypcapfile_engine.py @@ -0,0 +1,372 @@ +"""Unit tests for :mod:`pcapkit.foundation.engines.pypcapfile`. + +The engine is exercised against stand-ins for :func:`pcapfile.savefile.load_savefile` +and :func:`pcapfile.linklayer.clookup`, so that the routing decisions -- format +gating, the IPv6 capability gap, per-frame decoding and its failure path, output, +reassembly, tracing, storage -- are covered whether or not `pypcapfile`_ is +installed. End-to-end agreement with the ``default`` engine lives in +:mod:`tests.foundation.engines.test_new_engine_parity`. + +.. _pypcapfile: https://github.com/kisom/pypcapfile + +""" +from __future__ import annotations + +import importlib.util +import io +import struct +import types +import unittest +from unittest import mock + +from tests._support import purge_modules + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + +#: Magic number of a little-endian, microsecond-resolution PCAP savefile. +PCAP_MAGIC = b'\xd4\xc3\xb2\xa1' +#: Magic number of a PCAP-NG section header block. +PCAPNG_MAGIC = b'\x0a\x0d\x0d\x0a' + + +class OutputSink: + """Stand-in for a :class:`dictdumper.dumper.Dumper`, recording what it is handed.""" + + kind = 'unit' + + def __init__(self) -> None: + self.paths: list[str] = [] + self.records: list[tuple[object, str | None]] = [] + + def __call__(self, *args, **kwargs): + if len(args) == 1 and isinstance(args[0], str) and not kwargs: + self.paths.append(args[0]) + return self + self.records.append((args[0] if args else None, kwargs.get('name'))) + return self + + +class FakePacket: + """Stand-in for :class:`pcapfile.structs.pcap_packet`.""" + + def __init__(self, header, timestamp, timestamp_us, capture_len, packet_len, packet) -> None: + self.header = header + self.timestamp = timestamp + self.timestamp_us = timestamp_us + self.capture_len = capture_len + self.packet_len = packet_len + self.packet = packet + + +class FakeSaveFile: + """Stand-in for :class:`pcapfile.savefile.pcap_savefile`.""" + + def __init__(self, packets, ll_type: int = 1, ns_resolution: bool = False) -> None: + self.header = types.SimpleNamespace(ll_type=ll_type, ns_resolution=ns_resolution) + self.packets = packets + + +class FakeDecoded: + """Stand-in for a decoded link layer frame.""" + + def __init__(self, packet, layers=0) -> None: + self.raw = packet + self.layers = layers + self.payload = b'decoded-payload' + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class PyPCAPFileEngineTests(unittest.TestCase): + def setUp(self) -> None: + purge_modules(['pcapkit']) + + def make_extractor(self, **overrides): + sink = OutputSink() + reasm = types.SimpleNamespace(ipv4=mock.Mock(), ipv6=mock.Mock(), tcp=mock.Mock()) + trace = types.SimpleNamespace(tcp=mock.Mock()) + values = { + '_ifile': io.BytesIO(b'capture'), + '_ifnm': 'capture.pcap', + '_ofile': sink, + '_ofnm': 'out', + '_fext': 'json', + '_offmt': None, + '_flag_q': False, + '_flag_f': False, + '_flag_v': False, + '_flag_r': False, + '_flag_t': False, + '_flag_d': True, + '_ipv4': False, + '_ipv6': False, + '_tcp': False, + '_reasm': reasm, + '_trace': trace, + '_frame': [], + '_frnum': 0, + '_exlyr': 'none', + '_exptl': 'null', + '_vfunc': mock.Mock(), + 'magic_number': PCAP_MAGIC, + } + values.update(overrides) + return types.SimpleNamespace(**values), sink + + def engine(self, extractor, savefile=None, decoder=FakeDecoded): + from pcapkit.foundation.engines.pypcapfile import PyPCAPFile + + if savefile is None: + savefile = FakeSaveFile([FakePacket(None, 1, 500000, 7, 7, b'payload')]) + + engine = PyPCAPFile.__new__(PyPCAPFile) + engine._expkg = types.SimpleNamespace( + savefile=types.SimpleNamespace(load_savefile=mock.Mock(return_value=savefile)), + linklayer=types.SimpleNamespace(clookup=mock.Mock(return_value=decoder)), + structs=types.SimpleNamespace(pcap_packet=FakePacket), + ) + engine._extmp = None + engine._dlink = None + engine._declf = None + engine._extractor = extractor + return engine + + ########################################################################## + # run() + ########################################################################## + + def test_run_loads_savefile_undecoded_and_records_link_type(self) -> None: + from pcapkit.const.reg.linktype import LinkType + + extractor, _ = self.make_extractor() + engine = self.engine(extractor) + engine.run() + + load = engine._expkg.savefile.load_savefile + load.assert_called_once() + self.assertEqual(load.call_args.kwargs, {'layers': 0, 'lazy': True}) + stream = load.call_args.args[0] + self.assertEqual(stream.name, 'capture.pcap') + self.assertEqual(stream.read(3), b'cap') + + self.assertEqual(engine.dlink, LinkType.ETHERNET) + self.assertIs(engine._declf, FakeDecoded) + + def test_run_warns_on_layer_and_protocol_threshold(self) -> None: + from pcapkit.utilities.warnings import AttributeWarning + + # NOTE: ``BaseWarning.__init__`` installs an ``ignore`` filter for its own + # category outside development mode, so the warning never reaches + # ``assertWarns``; the ``warn`` call itself is what can be observed. + for overrides in ({'_exlyr': 'transport'}, {'_exptl': 'udp'}): + with self.subTest(overrides=overrides): + extractor, _ = self.make_extractor(**overrides) + engine = self.engine(extractor) + with mock.patch('pcapkit.foundation.engines.pypcapfile.warn') as warn: + engine.run() + self.assertEqual(warn.call_count, 1) + self.assertIn('protocol and layer threshold', warn.call_args.args[0]) + self.assertIs(warn.call_args.args[1], AttributeWarning) + + def test_run_rejects_pcapng(self) -> None: + from pcapkit.utilities.exceptions import FormatError + + extractor, _ = self.make_extractor(magic_number=PCAPNG_MAGIC) + engine = self.engine(extractor) + with self.assertRaises(FormatError): + engine.run() + engine._expkg.savefile.load_savefile.assert_not_called() + + def test_run_disables_ipv6_reassembly_but_keeps_ipv4_and_tcp(self) -> None: + from pcapkit.utilities.warnings import AttributeWarning + + extractor, _ = self.make_extractor(_flag_r=True, _ipv4=True, _ipv6=True, _tcp=True) + ipv4, tcp = extractor._reasm.ipv4, extractor._reasm.tcp + engine = self.engine(extractor) + with mock.patch('pcapkit.foundation.engines.pypcapfile.warn') as warn: + engine.run() + + messages = [call.args[0] for call in warn.call_args_list + if call.args[1] is AttributeWarning] + self.assertTrue(any('IPv6 reassembly' in message for message in messages), messages) + + self.assertTrue(extractor._flag_r) + self.assertFalse(extractor._ipv6) + self.assertTrue(extractor._ipv4) + self.assertIs(extractor._reasm.ipv4, ipv4) + self.assertIs(extractor._reasm.tcp, tcp) + self.assertIsNone(extractor._reasm.ipv6) + + def test_run_leaves_reassembly_alone_when_ipv6_was_not_requested(self) -> None: + extractor, _ = self.make_extractor(_flag_r=True, _ipv4=True, _tcp=True) + original = extractor._reasm + engine = self.engine(extractor) + engine.run() + self.assertIs(extractor._reasm, original) + + def test_run_warns_and_gives_up_decoding_for_an_unknown_link_layer(self) -> None: + from pcapkit.utilities.warnings import AttributeWarning + + for decoder in (None, 'not-callable'): + with self.subTest(decoder=decoder): + extractor, _ = self.make_extractor() + engine = self.engine(extractor, decoder=decoder) + with mock.patch('pcapkit.foundation.engines.pypcapfile.warn') as warn: + engine.run() + self.assertIsNone(engine._declf) + self.assertIn('unrecognised link layer protocol', warn.call_args.args[0]) + self.assertIs(warn.call_args.args[1], AttributeWarning) + + def test_run_survives_the_malformed_upstream_link_layer_table(self) -> None: + # ``pcapfile.linklayer.__LL_TYPES__`` carries a three-element entry for + # LINKTYPE_IEEE802_11_RADIOTAP, so ``clookup`` raises IndexError for it. + extractor, _ = self.make_extractor() + engine = self.engine(extractor) + engine._expkg.linklayer.clookup = mock.Mock(side_effect=IndexError('tuple index')) + with mock.patch('pcapkit.foundation.engines.pypcapfile.warn'): + engine.run() + self.assertIsNone(engine._declf) + + def test_run_installs_verbose_handler_reporting_the_chain(self) -> None: + extractor, _ = self.make_extractor(_flag_v=True) + engine = self.engine(extractor) + engine.run() + + packet = FakePacket(types.SimpleNamespace(ns_resolution=False), 1, 0, 7, 7, b'raw') + with mock.patch('builtins.print') as printer: + extractor._frnum = 4 + extractor._vfunc(extractor, packet) + printer.assert_called_once() + self.assertIn('ETHERNET:Raw', printer.call_args.args[0]) + + ########################################################################## + # read_frame() + ########################################################################## + + def prepared(self, packets=None, decoder=FakeDecoded, **overrides): + if packets is None: + packets = [FakePacket(types.SimpleNamespace(ns_resolution=False), + 1, 500000, 7, 7, b'payload')] + extractor, sink = self.make_extractor(**overrides) + engine = self.engine(extractor, savefile=FakeSaveFile(iter(packets)), decoder=decoder) + with mock.patch('pcapkit.foundation.engines.pypcapfile.warn'): + engine.run() + return extractor, sink, engine + + def test_read_frame_decodes_to_layers_depth_and_routes_output(self) -> None: + extractor, sink, engine = self.prepared() + + with mock.patch('pcapkit.toolkit.pypcapfile.packet2dict', return_value={'ok': True}): + frame = engine.read_frame() + + self.assertIsInstance(frame.packet, FakeDecoded) + self.assertEqual(frame.packet.raw, b'payload') + self.assertEqual(frame.packet.layers, engine.LAYERS - 1) + self.assertEqual((frame.timestamp, frame.timestamp_us), (1, 500000)) + self.assertEqual((frame.capture_len, frame.packet_len), (7, 7)) + + self.assertEqual(extractor._frnum, 1) + extractor._vfunc.assert_called_once_with(extractor, frame) + self.assertEqual(sink.records[-1], ({'ok': True}, 'Frame 1')) + self.assertEqual(extractor._offmt, 'unit') + self.assertEqual(extractor._frame, [frame]) + + with self.assertRaises(StopIteration): + engine.read_frame() + + def test_read_frame_leaves_the_frame_alone_with_no_decoder(self) -> None: + extractor, _, engine = self.prepared(decoder=None, _flag_q=True) + frame = engine.read_frame() + self.assertEqual(frame.packet, b'payload') + + def test_read_frame_warns_and_falls_back_when_decoding_fails(self) -> None: + from pcapkit.utilities.warnings import AttributeWarning + + for error in (struct.error('unpack'), AssertionError('not ipv4'), + ValueError('bad'), IndexError('short'), KeyError('missing')): + with self.subTest(error=type(error).__name__): + extractor, _, engine = self.prepared( + decoder=mock.Mock(side_effect=error), _flag_q=True, + ) + with mock.patch('pcapkit.foundation.engines.pypcapfile.warn') as warn: + frame = engine.read_frame() + self.assertEqual(frame.packet, b'payload') + self.assertIn('decoding failed', warn.call_args.args[0]) + self.assertIn('Frame 1', warn.call_args.args[0]) + self.assertIs(warn.call_args.args[1], AttributeWarning) + + def test_read_frame_splits_files_when_asked(self) -> None: + extractor, sink, engine = self.prepared(_flag_f=True) + with mock.patch('pcapkit.toolkit.pypcapfile.packet2dict', return_value={}): + engine.read_frame() + self.assertEqual(sink.paths[-1], 'out/Frame 1.json') + + def test_read_frame_routes_reassembly_and_tracing(self) -> None: + extractor, _, engine = self.prepared(_flag_q=True, _flag_r=True, _flag_t=True, + _ipv4=True, _tcp=True, _flag_d=False) + with mock.patch('pcapkit.toolkit.pypcapfile.ipv4_reassembly', return_value='ipv4') as v4: + with mock.patch('pcapkit.toolkit.pypcapfile.tcp_reassembly', return_value='tcp'): + with mock.patch('pcapkit.toolkit.pypcapfile.tcp_traceflow', + return_value='trace') as trace: + frame = engine.read_frame() + + v4.assert_called_once_with(frame, count=1) + trace.assert_called_once_with(frame, data_link=engine.dlink, count=1) + extractor._reasm.ipv4.assert_called_once_with('ipv4') + extractor._reasm.tcp.assert_called_once_with('tcp') + extractor._trace.tcp.assert_called_once_with('trace') + # IPv6 is unreachable for this engine, so it must never be consulted + extractor._reasm.ipv6.assert_not_called() + self.assertEqual(extractor._frame, []) + + def test_read_frame_skips_reassembly_and_tracing_when_adapters_decline(self) -> None: + extractor, _, engine = self.prepared(_flag_q=True, _flag_r=True, _flag_t=True, + _ipv4=True, _tcp=True) + with mock.patch('pcapkit.toolkit.pypcapfile.ipv4_reassembly', return_value=None): + with mock.patch('pcapkit.toolkit.pypcapfile.tcp_reassembly', return_value=None): + with mock.patch('pcapkit.toolkit.pypcapfile.tcp_traceflow', return_value=None): + engine.read_frame() + + extractor._reasm.ipv4.assert_not_called() + extractor._reasm.tcp.assert_not_called() + extractor._trace.tcp.assert_not_called() + + def test_read_frame_ignores_reassembly_and_tracing_when_flags_are_off(self) -> None: + extractor, _, engine = self.prepared(_flag_q=True, _flag_r=False, _flag_t=False, + _ipv4=True, _tcp=True) + with mock.patch('pcapkit.toolkit.pypcapfile.ipv4_reassembly', return_value='unused'): + with mock.patch('pcapkit.toolkit.pypcapfile.tcp_reassembly', return_value='unused'): + with mock.patch('pcapkit.toolkit.pypcapfile.tcp_traceflow', return_value='unused'): + engine.read_frame() + + extractor._reasm.ipv4.assert_not_called() + extractor._trace.tcp.assert_not_called() + + def test_read_frame_ignores_protocols_that_were_not_requested(self) -> None: + extractor, _, engine = self.prepared(_flag_q=True, _flag_r=True, _flag_t=True, + _ipv4=False, _tcp=False) + with mock.patch('pcapkit.toolkit.pypcapfile.ipv4_reassembly', return_value='unused'): + with mock.patch('pcapkit.toolkit.pypcapfile.tcp_reassembly', return_value='unused'): + with mock.patch('pcapkit.toolkit.pypcapfile.tcp_traceflow', return_value='unused'): + engine.read_frame() + + extractor._reasm.ipv4.assert_not_called() + extractor._reasm.tcp.assert_not_called() + extractor._trace.tcp.assert_not_called() + + ########################################################################## + # _NamedStream + ########################################################################## + + def test_named_stream_proxies_reads_and_carries_a_name(self) -> None: + from pcapkit.foundation.engines.pypcapfile import _NamedStream + + stream = _NamedStream(io.BytesIO(b'abcdef'), 'given.pcap') + self.assertEqual(stream.name, 'given.pcap') + self.assertEqual(stream.read(2), b'ab') + self.assertEqual(stream.read(), b'cdef') + self.assertEqual(stream.read(), b'') + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/toolkit/test_pypcap_unit.py b/tests/toolkit/test_pypcap_unit.py new file mode 100644 index 0000000000..ecde79236f --- /dev/null +++ b/tests/toolkit/test_pypcap_unit.py @@ -0,0 +1,76 @@ +"""Unit tests for :mod:`pcapkit.toolkit.pypcap`. + +`pypcap`_ hands back raw bytes, so this toolkit has only the two auxiliary +functions; the reassembly and flow tracing adapters exist purely to refuse +loudly, and that refusal is asserted here so it cannot regress into a silent +:data:`None`. + +.. _pypcap: https://github.com/pynetwork/pypcap + +""" +from __future__ import annotations + +import importlib.util +import unittest + +from tests._support import purge_modules + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class PyPCAPToolkitTests(unittest.TestCase): + def setUp(self) -> None: + purge_modules(['pcapkit']) + + def test_packet2chain_reports_the_link_layer_and_raw(self) -> None: + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit.pypcap import packet2chain + + self.assertEqual(packet2chain(b'\x00' * 20, data_link=LinkType.ETHERNET), + 'ETHERNET:Raw') + self.assertEqual(packet2chain(b'', data_link=LinkType.RAW), 'RAW:Raw') + + def test_packet2dict_carries_the_bytes_verbatim(self) -> None: + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit.pypcap import packet2dict + + info = packet2dict(b'abcd', 1.25, data_link=LinkType.ETHERNET) + self.assertEqual(info, { + 'timestamp': 1.25, + 'packet': b'abcd', + 'ETHERNET': {'raw_len': 4, 'raw': b'abcd'}, + }) + + def test_unsupported_adapters_refuse_loudly(self) -> None: + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit.pypcap import (ipv4_reassembly, ipv6_reassembly, tcp_reassembly, + tcp_traceflow) + from pcapkit.utilities.exceptions import UnsupportedCall + + for name, call in ( + ('ipv4_reassembly', lambda: ipv4_reassembly(b'raw', count=1)), + ('ipv6_reassembly', lambda: ipv6_reassembly(b'raw', count=1)), + ('tcp_reassembly', lambda: tcp_reassembly(b'raw', count=1)), + ('tcp_traceflow', lambda: tcp_traceflow(b'raw', 1.0, + data_link=LinkType.ETHERNET, count=1)), + ): + with self.subTest(function=name): + with self.assertRaises(UnsupportedCall) as caught: + call() + self.assertIn('no protocol dissection', str(caught.exception)) + + def test_module_exports_the_documented_surface(self) -> None: + from pcapkit.toolkit import pypcap + + self.assertEqual(sorted(pypcap.__all__), [ + 'ipv4_reassembly', 'ipv6_reassembly', 'packet2chain', 'packet2dict', + 'tcp_reassembly', 'tcp_traceflow', + ]) + for name in pypcap.__all__: + self.assertTrue(callable(getattr(pypcap, name)), name) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/toolkit/test_pypcapfile_unit.py b/tests/toolkit/test_pypcapfile_unit.py new file mode 100644 index 0000000000..b42b381466 --- /dev/null +++ b/tests/toolkit/test_pypcapfile_unit.py @@ -0,0 +1,364 @@ +"""Unit tests for :mod:`pcapkit.toolkit.pypcapfile`. + +The adapters that only walk decoded layers are tested against stand-ins, so they +run without `pypcapfile`_ installed. The ones that reconstruct or re-split raw +bytes -- :func:`~pcapkit.toolkit.pypcapfile.ipv4_header`, +:func:`~pcapkit.toolkit.pypcapfile.tcp_reassembly` and +:func:`~pcapkit.toolkit.pypcapfile.tcp_traceflow` -- are tested against +`pypcapfile`_'s real decoders and real packet bytes, because "byte-exact" is the +claim being made and a stand-in cannot check it. + +.. _pypcapfile: https://github.com/kisom/pypcapfile + +""" +from __future__ import annotations + +import importlib.util +import ipaddress +import struct +import types +import unittest + +from tests._support import purge_modules + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + + +def _has_pypcapfile() -> bool: + """Test if :mod:`pcapfile` is importable. + + A plain :func:`importlib.util.find_spec` is not enough: the released + ``pypcapfile`` 0.12.0 imports the :mod:`imp` module, which was removed in + Python 3.12, so the distribution can be present and still unusable. + + """ + try: + importlib.import_module('pcapfile.savefile') + importlib.import_module('pcapfile.protocols.transport.tcp') + except ImportError: + return False + return True + + +HAS_PYPCAPFILE = _has_pypcapfile() + + +def make_ipv4(payload: bytes, *, src: str = '10.1.1.2', dst: str = '10.1.1.3', + protocol: int = 6, ident: int = 4660, flags: int = 0, offset: int = 0, + options: bytes = b'') -> bytes: + """Build a raw IPv4 packet, options included.""" + assert len(options) % 4 == 0, 'IPv4 options must be a whole number of words' + ihl = 5 + len(options) // 4 + header = struct.pack( + '!BBHHHBBHII', + (4 << 4) | ihl, + 0x10, + 20 + len(options) + len(payload), + ident, + (flags << 13) | offset, + 64, + protocol, + 0xBEEF, + int(ipaddress.IPv4Address(src)), + int(ipaddress.IPv4Address(dst)), + ) + options + return header + payload + + +def make_tcp(payload: bytes, *, src_port: int = 51000, dst_port: int = 22, + seq: int = 1000, ack: int = 2000, flags: int = 0x12, + options: bytes = b'') -> bytes: + """Build a raw TCP segment, options included.""" + assert len(options) % 4 == 0, 'TCP options must be a whole number of words' + offset = 5 + len(options) // 4 + header = struct.pack('!HHIIBBHHH', src_port, dst_port, seq, ack, + offset << 4, flags, 8192, 0xCAFE, 0) + options + return header + payload + + +class FakeIP: + """Stand-in for :class:`pcapfile.protocols.network.ip.IP`.""" + + # NOTE: the class must be *named* ``IP``, since that is how the toolkit + # recognises a decoded network layer. + def __init__(self, payload, **fields) -> None: + self.v = 4 + self.hl = 5 + self.tos = 0x10 + self.len = 20 + (len(payload) if isinstance(payload, bytes) else 0) + self.id = 4660 + self.flags = 0 + self.off = 0 + self.ttl = 64 + self.p = 6 + self.sum = 0xBEEF + self.src = int(ipaddress.IPv4Address('10.1.1.2')) + self.dst = int(ipaddress.IPv4Address('10.1.1.3')) + self.opt = b'' + self.payload = payload + self.__dict__.update(fields) + + +FakeIP.__name__ = 'IP' +FakeIP.__qualname__ = 'IP' + + +class FakeEthernet: + """Stand-in for :class:`pcapfile.protocols.linklayer.ethernet.Ethernet`.""" + + def __init__(self, payload) -> None: + self.dst = bytearray(b'\x40\x33\x1a\xd1\x85\x1c') + self.src = bytearray(b'\xa4\x5e\x60\xd9\x6b\x97') + self.type = 0x0800 + self.payload = payload + + +FakeEthernet.__name__ = 'Ethernet' +FakeEthernet.__qualname__ = 'Ethernet' + + +def make_packet(layer, *, timestamp: int = 1511106545, timestamp_us: int = 471719, + ns_resolution: bool = False, capture_len: int = 86, packet_len: int = 86): + """Build a stand-in for :class:`pcapfile.structs.pcap_packet`.""" + return types.SimpleNamespace( + header=[types.SimpleNamespace(ns_resolution=ns_resolution)], + timestamp=timestamp, + timestamp_us=timestamp_us, + capture_len=capture_len, + packet_len=packet_len, + packet=layer, + ) + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class PyPCAPFileToolkitTests(unittest.TestCase): + def setUp(self) -> None: + purge_modules(['pcapkit']) + + ########################################################################## + # Auxiliary functions. + ########################################################################## + + def test_packet2timestamp_honours_the_nanosecond_flag(self) -> None: + from pcapkit.toolkit.pypcapfile import packet2timestamp + + micro = make_packet(b'raw', timestamp=1000, timestamp_us=500000) + self.assertEqual(packet2timestamp(micro), 1000.5) + + nano = make_packet(b'raw', timestamp=1000, timestamp_us=500000000, + ns_resolution=True) + self.assertEqual(packet2timestamp(nano), 1000.5) + + def test_packet2chain_walks_the_decoded_layers(self) -> None: + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit.pypcapfile import packet2chain + + self.assertEqual( + packet2chain(make_packet(b'raw'), data_link=LinkType.ETHERNET), + 'ETHERNET:Raw', + ) + self.assertEqual( + packet2chain(make_packet(FakeEthernet(b'raw')), data_link=LinkType.ETHERNET), + 'Ethernet:Raw', + ) + self.assertEqual( + packet2chain(make_packet(FakeEthernet(FakeIP(b'segment'))), + data_link=LinkType.ETHERNET), + 'Ethernet:IP:Raw', + ) + + def test_packet2dict_reports_the_decoded_tree(self) -> None: + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit.pypcapfile import packet2dict + + info = packet2dict(make_packet(FakeEthernet(FakeIP(b'segment'))), + data_link=LinkType.ETHERNET) + self.assertEqual(info['timestamp'], 1511106545.471719) + self.assertEqual(info['capture_len'], 86) + self.assertEqual(info['packet_len'], 86) + + ethernet = info['ETHERNET'] + self.assertEqual(ethernet['src'], b'\xa4\x5e\x60\xd9\x6b\x97') + self.assertEqual(ethernet['dst'], b'\x40\x33\x1a\xd1\x85\x1c') + self.assertEqual(ethernet['type'], 0x0800) + + network = ethernet['IP'] + self.assertEqual(network['src'], '10.1.1.2') + self.assertEqual(network['dst'], '10.1.1.3') + self.assertEqual(network['p'], 6) + self.assertEqual(network['Raw'], {'raw_len': 7, 'raw': b'segment'}) + + def test_packet2dict_falls_back_to_ctypes_fields_for_unknown_layers(self) -> None: + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit.pypcapfile import packet2dict + + class Wifi: + _fields_ = [('flags', None), ('missing', None)] + + def __init__(self) -> None: + self.flags = 3 + self.payload = None + + info = packet2dict(make_packet(Wifi()), data_link=LinkType.IEEE802_11) + self.assertEqual(info[LinkType.IEEE802_11.name], {'flags': 3, 'missing': None}) + + def test_packet2dict_and_chain_tolerate_an_undecoded_frame(self) -> None: + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit.pypcapfile import packet2chain, packet2dict + + info = packet2dict(make_packet(b'\x01\x02'), data_link=LinkType.ETHERNET) + self.assertEqual(info['ETHERNET'], {'raw_len': 2, 'raw': b'\x01\x02'}) + self.assertEqual(packet2chain(make_packet(bytearray(b'\x01')), + data_link=LinkType.ETHERNET), 'ETHERNET:Raw') + + ########################################################################## + # Reassembly and flow tracing. + ########################################################################## + + def test_ipv6_reassembly_refuses_loudly(self) -> None: + from pcapkit.toolkit.pypcapfile import ipv6_reassembly + from pcapkit.utilities.exceptions import UnsupportedCall + + with self.assertRaises(UnsupportedCall) as caught: + ipv6_reassembly(make_packet(b'raw'), count=1) + self.assertIn('no IPv6 decoder', str(caught.exception)) + + def test_ipv4_reassembly_declines_undecoded_and_unfragmented_frames(self) -> None: + from pcapkit.toolkit.pypcapfile import ipv4_reassembly + + # undecoded link layer + self.assertIsNone(ipv4_reassembly(make_packet(b'raw'), count=1)) + # decoded link layer, undecoded network layer (e.g. IPv6) + self.assertIsNone(ipv4_reassembly(make_packet(FakeEthernet(b'raw')), count=1)) + # decoded IPv4, but the DF flag is set + packet = make_packet(FakeEthernet(FakeIP(b'segment', flags=0b010))) + self.assertIsNone(ipv4_reassembly(packet, count=1)) + + def test_ipv4_reassembly_reports_offsets_in_octets(self) -> None: + from pcapkit.const.reg.transtype import TransType + from pcapkit.toolkit.pypcapfile import ipv4_reassembly + + ipv4 = FakeIP(b'fragment-payload', flags=0b001, off=185, len=36) + data = ipv4_reassembly(make_packet(FakeEthernet(ipv4)), count=7) + + self.assertIsNotNone(data) + self.assertEqual(data.num, 7) + self.assertEqual(data.fo, 185 * 8) + self.assertEqual(data.ihl, 20) + self.assertTrue(data.mf) + self.assertEqual(data.tl, 36) + self.assertEqual(data.payload, bytearray(b'fragment-payload')) + self.assertEqual(data.bufid, ( + ipaddress.IPv4Address('10.1.1.2'), + ipaddress.IPv4Address('10.1.1.3'), + 4660, + TransType.TCP, + )) + + def test_ipv4_reassembly_counts_options_into_the_header_length(self) -> None: + from pcapkit.toolkit.pypcapfile import ipv4_reassembly + + ipv4 = FakeIP(b'payload', flags=0b001, hl=6, opt=b'\x01\x01\x01\x00') + data = ipv4_reassembly(make_packet(FakeEthernet(ipv4)), count=1) + self.assertEqual(data.ihl, 24) + self.assertEqual(len(data.header), 24) + self.assertEqual(data.header[20:], b'\x01\x01\x01\x00') + + def test_tcp_adapters_decline_non_tcp_and_truncated_segments(self) -> None: + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit.pypcapfile import tcp_reassembly, tcp_traceflow + + for label, packet in ( + ('undecoded link', make_packet(b'raw')), + ('undecoded network', make_packet(FakeEthernet(b'raw'))), + ('not tcp', make_packet(FakeEthernet(FakeIP(b'x' * 40, p=17)))), + ('too short', make_packet(FakeEthernet(FakeIP(b'short')))), + ('already decoded', make_packet(FakeEthernet(FakeIP(object())))), + ): + with self.subTest(case=label): + self.assertIsNone(tcp_reassembly(packet, count=1)) + self.assertIsNone(tcp_traceflow(packet, data_link=LinkType.ETHERNET, count=1)) + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +@unittest.skipUnless(HAS_PYPCAPFILE, 'pypcapfile not installed or not importable') +class PyPCAPFileToolkitAgainstRealDecodersTests(unittest.TestCase): + """Tests that need :mod:`pcapfile`'s own decoders to be meaningful.""" + + def setUp(self) -> None: + purge_modules(['pcapkit']) + + def test_ipv4_header_reconstruction_is_byte_exact(self) -> None: + from pcapfile.protocols.network.ip import IP + + from pcapkit.toolkit.pypcapfile import ipv4_header + + for label, options in (('no options', b''), ('with options', b'\x94\x04\x00\x00')): + with self.subTest(case=label): + raw = make_ipv4(b'payload' * 4, options=options) + ihl = 20 + len(options) + decoded = IP(raw) + self.assertEqual(ipv4_header(decoded), raw[:ihl]) + + def test_ipv4_header_reconstruction_survives_fragment_flags(self) -> None: + from pcapfile.protocols.network.ip import IP + + from pcapkit.toolkit.pypcapfile import ipv4_header + + raw = make_ipv4(b'x' * 32, flags=0b001, offset=185) + self.assertEqual(ipv4_header(IP(raw)), raw[:20]) + + def test_tcp_reassembly_splits_the_real_segment_exactly(self) -> None: + from pcapfile.protocols.network.ip import IP + + from pcapkit.toolkit.pypcapfile import tcp_reassembly + + options = b'\x02\x04\x05\xb4' + segment = make_tcp(b'SSH-2.0-OpenSSH_9.3\r\n', flags=0b00010011, options=options) + packet = make_packet(FakeEthernet(IP(make_ipv4(segment)))) + + data = tcp_reassembly(packet, count=3) + self.assertIsNotNone(data) + self.assertEqual(data.num, 3) + self.assertEqual(data.header, segment[:24]) + self.assertEqual(data.header[20:], options) + self.assertEqual(bytes(data.payload), b'SSH-2.0-OpenSSH_9.3\r\n') + self.assertEqual(data.len, 21) + self.assertEqual(data.dsn, 1000) + self.assertEqual(data.ack, 2000) + self.assertEqual(data.first, 1000) + self.assertEqual(data.last, 1021) + self.assertTrue(data.syn) + self.assertTrue(data.fin) + self.assertFalse(data.rst) + self.assertEqual(data.bufid, ( + ipaddress.IPv4Address('10.1.1.2'), 51000, + ipaddress.IPv4Address('10.1.1.3'), 22, + )) + + def test_tcp_traceflow_reports_the_flow_endpoints(self) -> None: + from pcapfile.protocols.network.ip import IP + + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit.pypcapfile import tcp_traceflow + + segment = make_tcp(b'body', flags=0b00000010) + packet = make_packet(FakeEthernet(IP(make_ipv4(segment)))) + + data = tcp_traceflow(packet, data_link=LinkType.ETHERNET, count=5) + self.assertIsNotNone(data) + self.assertEqual(data.protocol, LinkType.ETHERNET) + self.assertEqual(data.index, 5) + self.assertTrue(data.syn) + self.assertFalse(data.fin) + self.assertEqual(data.src, ipaddress.IPv4Address('10.1.1.2')) + self.assertEqual(data.dst, ipaddress.IPv4Address('10.1.1.3')) + self.assertEqual(data.srcport, 51000) + self.assertEqual(data.dstport, 22) + self.assertEqual(data.timestamp, 1511106545.471719) + self.assertEqual(data.frame['ETHERNET']['IP']['src'], '10.1.1.2') + + +if __name__ == '__main__': + unittest.main() From 1cbb9e726bc9426d324eace23c5e41b037258a6a Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 15 Sep 2026 04:02:03 +0000 Subject: [PATCH 2/3] =?UTF-8?q?Fix=20typos:=20dose=E2=86=92does,=20seperat?= =?UTF-8?q?ed=E2=86=92separated?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: JarryShaw <15666417+JarryShaw@users.noreply.github.com> --- pcapkit/foundation/engines/pypcap.py | 4 ++-- pcapkit/foundation/engines/pypcapfile.py | 2 +- pcapkit/toolkit/pypcap.py | 2 +- pcapkit/toolkit/pypcapfile.py | 2 +- 4 files changed, 5 insertions(+), 5 deletions(-) diff --git a/pcapkit/foundation/engines/pypcap.py b/pcapkit/foundation/engines/pypcap.py index 511f6d8afa..664824f212 100644 --- a/pcapkit/foundation/engines/pypcap.py +++ b/pcapkit/foundation/engines/pypcap.py @@ -156,14 +156,14 @@ def run(self) -> 'None': if ext._flag_r and (ext._ipv4 or ext._ipv6 or ext._tcp): ext._flag_r = False ext._reasm = ReassemblyManager(ipv4=None, ipv6=None, tcp=None) - warn("'Extractor(engine=pypcap)' object dose not support reassembly; " + warn("'Extractor(engine=pypcap)' object does not support reassembly; " f"so 'ipv4={ext._ipv4}', 'ipv6={ext._ipv6}' and 'tcp={ext._tcp}' will be ignored", AttributeWarning, stacklevel=stacklevel()) if ext._flag_t and ext._tcp: ext._flag_t = False ext._trace = TraceFlowManager(tcp=None) - warn("'Extractor(engine=pypcap)' object dose not support flow tracing; " + warn("'Extractor(engine=pypcap)' object does not support flow tracing; " f"so 'tcp={ext._tcp}' will be ignored", AttributeWarning, stacklevel=stacklevel()) # NOTE: ``promisc=False`` is defensive only -- the offline branch of diff --git a/pcapkit/foundation/engines/pypcapfile.py b/pcapkit/foundation/engines/pypcapfile.py index e95a884d87..3c9b9b3b39 100644 --- a/pcapkit/foundation/engines/pypcapfile.py +++ b/pcapkit/foundation/engines/pypcapfile.py @@ -190,7 +190,7 @@ def run(self) -> 'None': if ext._flag_r and ext._ipv6: ext._ipv6 = False ext._reasm = ReassemblyManager(ipv4=ext._reasm.ipv4, ipv6=None, tcp=ext._reasm.tcp) - warn("'Extractor(engine=pypcapfile)' object dose not support IPv6 reassembly; " + warn("'Extractor(engine=pypcapfile)' object does not support IPv6 reassembly; " "so 'ipv6=True' will be ignored", AttributeWarning, stacklevel=stacklevel()) sfile = pcapfile.savefile.load_savefile( diff --git a/pcapkit/toolkit/pypcap.py b/pcapkit/toolkit/pypcap.py index d3f2002df1..681f26d031 100644 --- a/pcapkit/toolkit/pypcap.py +++ b/pcapkit/toolkit/pypcap.py @@ -58,7 +58,7 @@ def packet2chain(packet: 'bytes', *, data_link: 'Enum_LinkType') -> 'str': data_link: Data link type, from the capture handle. Returns: - Colon (``:``) seperated list of protocol chain. + Colon (``:``) separated list of protocol chain. Note: As `PyPCAP`_ does not dissect the packet, the chain is only ever the diff --git a/pcapkit/toolkit/pypcapfile.py b/pcapkit/toolkit/pypcapfile.py index 018133eab3..26a7a512dc 100644 --- a/pcapkit/toolkit/pypcapfile.py +++ b/pcapkit/toolkit/pypcapfile.py @@ -247,7 +247,7 @@ def packet2chain(packet: 'Packet', *, data_link: 'Enum_LinkType') -> 'str': data_link: Data link type, from the savefile header. Returns: - Colon (``:``) seperated list of protocol chain. + Colon (``:``) separated list of protocol chain. Note: The chain reports what `PyPCAPFile`_ actually decoded and no more, so it From 48628d43072f0541195c61a2138f7187f59812d7 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Tue, 15 Sep 2026 01:59:22 -0400 Subject: [PATCH 3/3] build: keep pypcap out of the `all` extra, which broke five workflows `pypcap` 1.3.0 publishes no wheels at all -- only an sdist -- so installing it compiles a C extension against libpcap and needs both the headers and the shared library. Listing it in `all` therefore made `pip install pypcapkit[all]` demand a compiler and libpcap development files from every user, on every platform. CI proved it before any user could: deploy-pages failed on this PR at `Getting requirements to build wheel ... error`, with pypcap's setup.py saying Found pcap headers in /Applications/Xcode.app/.../MacOSX26.sdk/usr/include/pcap.h None of the following found: ['libpcap.a', 'libpcap.so', 'libpcap.dylib', 'wpcap.lib'] The macOS runner has the header and no library, which is exactly the case that fails. deploy-pages was only the first to run: `.[all]` is also installed by cron-conda (x3), cron-vendor and create-release (x2), so the release and conda paths would have broken on merge too. `pypcap` stays available as its own opt-in extra, `pip install pypcapkit[PyPCAP]`, and is documented as needing libpcap present. `pypcapfile` is sdist-only as well but pure Python, so it installs anywhere and stays in `all`. Also dropped `pypcap` from the Pipfile's dev-packages, where it would have broken `pipenv install --dev` -- and `pipenv lock`, which has to build its metadata -- on any machine without libpcap. This one included: it has no pcap.h either. The docs already warned that `pip install pypcapkit[PyPCAP]` can fail to build while pep.rst simultaneously claimed both extras were in `all`; that contradiction is now resolved in favour of what the packaging actually does. Verified: `pip install --dry-run '.[all]'` on this libpcap-less host now resolves and would install only pypcapfile. Unit tier 438 passed, 14 skipped (the pypcap/pypcapfile-gated ones); engines and toolkit 74 passed, 14 skipped. --- Pipfile | 6 +++++- docs/source/pcapkit/foundation/engines/3rdparty.rst | 11 +++++++++++ docs/source/pep.rst | 10 ++++++++-- pyproject.toml | 10 +++++++++- 4 files changed, 33 insertions(+), 4 deletions(-) diff --git a/Pipfile b/Pipfile index 7c28605fee..42eed5e3ac 100644 --- a/Pipfile +++ b/Pipfile @@ -16,7 +16,11 @@ pyshark = "*" dpkt = "*" scapy = "*" cryptography = "*" -pypcap = "*" +# NB: ``pypcap`` is deliberately absent. It builds a C extension against +# libpcap, so listing it here would break ``pipenv install --dev`` -- and even +# ``pipenv lock``, which has to build its metadata -- on any machine without the +# libpcap development files. Install it by hand to work on that engine: +# ``pipenv run pip install pypcap``. pypcapfile = "*" beautifulsoup4 = {extras = ["html5lib"],version = "*"} requests = {extras = ["socks"],version = "*"} diff --git a/docs/source/pcapkit/foundation/engines/3rdparty.rst b/docs/source/pcapkit/foundation/engines/3rdparty.rst index da90a35bf5..847f629800 100644 --- a/docs/source/pcapkit/foundation/engines/3rdparty.rst +++ b/docs/source/pcapkit/foundation/engines/3rdparty.rst @@ -82,6 +82,17 @@ support, as is used by :class:`pcapkit.foundation.extraction.Extractor`. .. _PyPCAP: https://github.com/pynetwork/pypcap +.. important:: + + `PyPCAP`_ publishes no wheels, so installing it compiles a C extension and + needs both the :manpage:`libpcap(3)` headers and its shared library on the + system -- a header alone is not enough. It is therefore **not** part of the + ``all`` extra, and is installed on its own once libpcap is available: + + .. code-block:: shell + + pip install pypcapkit[PyPCAP] + .. important:: `PyPCAP`_ is a :manpage:`libpcap(3)` binding aimed primarily at live capture. diff --git a/docs/source/pep.rst b/docs/source/pep.rst index 3cc0e88efc..d81d1afde6 100644 --- a/docs/source/pep.rst +++ b/docs/source/pep.rst @@ -77,7 +77,8 @@ New Engines ``engine='pypcap'`` selects :class:`pcapkit.foundation.engines.pypcap.PyPCAP`; each has a matching :mod:`pcapkit.toolkit` module (:mod:`pcapkit.toolkit.pypcapfile`, :mod:`pcapkit.toolkit.pypcap`), a - ``pyproject.toml`` extra (``PyPCAPFile``, ``PyPCAP``, both in ``all``), docs + ``pyproject.toml`` extra (``PyPCAPFile``, which ``all`` includes, and + ``PyPCAP``, which it deliberately does not -- see below), docs under :doc:`pcapkit/foundation/engines/index`, and tests under ``tests/foundation/engines/`` and ``tests/toolkit/``. Both were verified end-to-end against the sample captures: each agrees with the ``default`` engine @@ -97,7 +98,12 @@ New Engines fixed list of prefixes for ``pcap.h``. It builds once :program:`libpcap`'s headers are visible under ``sys.prefix`` and ``pcap.c`` is regenerated with Cython 3. That is a packaging problem upstream rather than an engine problem, - but it does mean ``pip install pypcapkit[PyPCAP]`` can fail to build. + but it does mean ``pip install pypcapkit[PyPCAP]`` can fail to build. Since + there is no wheel to fall back on, the extra is kept **out of** ``all``: + otherwise ``pip install pypcapkit[all]`` would demand a compiler and the + libpcap development files from every user, and it broke the docs, conda and + release workflows -- all of which install ``.[all]`` -- on the macOS runner, + where :file:`pcap.h` is present but no ``libpcap.dylib`` is. Both engines support less than the ``default`` engine does, deliberately and noisily: `pypcap`_ performs no protocol dissection, so it disables reassembly diff --git a/pyproject.toml b/pyproject.toml index c18df2deec..94b190fc35 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -88,14 +88,22 @@ crypto = [ "cryptography>=3.4" ] DPKT = [ "dpkt" ] Scapy = [ "scapy" ] PyShark = [ "pyshark" ] +# pypcap ships no wheels, so this compiles a C extension against libpcap and +# needs both its headers and its shared library present; deliberately kept out +# of ``all`` for that reason, c.f. the note below PyPCAP = [ "pypcap" ] PyPCAPFile = [ "pypcapfile" ] # for developers vendor = [ "requests[socks]", "beautifulsoup4[html5lib]" ] +# NB: ``pypcap`` is *not* included here. It is an sdist-only C extension, so +# ``pip install pypcapkit[all]`` would require a compiler and the libpcap +# development files on every platform, and fails where only the header is +# present -- as on the macOS runners. Install it explicitly with +# ``pip install pypcapkit[PyPCAP]`` once libpcap is available. all = [ "emoji", "cryptography>=3.4", - "dpkt", "scapy", "pyshark", "pypcap", "pypcapfile", + "dpkt", "scapy", "pyshark", "pypcapfile", "requests[socks]", "beautifulsoup4[html5lib]", ] docs = [