diff --git a/Pipfile b/Pipfile index 0efd5b3abd..42eed5e3ac 100644 --- a/Pipfile +++ b/Pipfile @@ -16,6 +16,12 @@ pyshark = "*" dpkt = "*" scapy = "*" cryptography = "*" +# 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 = "*"} 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..847f629800 100644 --- a/docs/source/pcapkit/foundation/engines/3rdparty.rst +++ b/docs/source/pcapkit/foundation/engines/3rdparty.rst @@ -71,3 +71,108 @@ 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`_ 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. + 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 661a686ac2..340dabeea9 100644 --- a/docs/source/pep.rst +++ b/docs/source/pep.rst @@ -110,15 +110,49 @@ 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``, 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 + 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. 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 + *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 @@ -128,6 +162,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..664824f212 --- /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 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 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 + # ``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..3c9b9b3b39 --- /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 does 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 fecf3b2f2e..a781a3bf83 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] @@ -194,6 +201,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 @@ -422,6 +431,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 @@ -808,10 +819,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.debug('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..681f26d031 --- /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 (``:``) separated 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..26a7a512dc --- /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 (``:``) separated 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..94b190fc35 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -88,12 +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", + "dpkt", "scapy", "pyshark", "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()