# -*- coding: utf-8 -*-
"""pcap-ct Support
=====================
.. module:: pcapkit.foundation.engines.pcap_ct
This module contains the implementation for `pcap-ct`_ engine
support, as is used by :class:`pcapkit.foundation.extraction.Extractor`.
"""
import os
from typing import TYPE_CHECKING, cast
from pcapkit.const.reg.linktype import LinkType as Enum_LinkType
from pcapkit.foundation.engines import _pcap_backend
from pcapkit.foundation.engines.engine import EngineBase
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, EngineWarning, warn
__all__ = ['PCAP_CT']
if TYPE_CHECKING:
from typing import Iterator, Optional, Protocol, Tuple, Type
from pcap import pcap as Handle
from pcapkit.foundation.extraction import Extractor
class _PCAPCTPackage(Protocol):
"""|pcap|_, as :class:`PCAP_CT` reads it through :attr:`PCAP_CT._expkg` (:issue:`1878`)."""
@property
def pcap(self) -> 'Type[Handle]': ...
#: A `pcap-ct`_ "frame" -- the ``(timestamp, bytes)`` pair that
#: ``pcap.pcap`` iteration yields. Deliberately *not* named ``Frame``,
#: so that it is not mistaken for
#: :class:`pcapkit.protocols.misc.pcap.frame.protocol.Frame`, which is what the
#: built-in engines return.
RawFrame = Tuple[float, bytes]
[docs]
class PCAP_CT(EngineBase['RawFrame']):
"""pcap-ct engine support.
`pcap-ct`_ is a :mod:`ctypes` reimplementation of the `PyPCAP`_ API on top of
the `libpcap`_ package, which supplies the :manpage:`libpcap(3)` bindings
(and ships a vendored copy of the library that it does not, by default, use
-- see the warning below). Both distributions install a
top-level |pcap|_ module and expose the same interface, so this engine is
a sibling of :class:`~pcapkit.foundation.engines.pypcap.PyPCAP` rather than a
replacement for it -- see :attr:`__engine_module__` for how the two are told
apart, and :doc:`the engine documentation
</pcapkit/foundation/engines/3rdparty>` for which one to install.
It exists because upstream `PyPCAP`_ cannot be installed on Python 3.12 or
newer: it ships no wheels and its ``pcap.c`` was pre-generated by
Cython 0.29.32, which does not compile against the Python 3.12+ C API.
`pcap-ct`_ and `libpcap`_ both publish ``py3-none-any`` wheels, so **installing**
this engine needs no compiler and no ``pcap.h``.
.. warning::
It does still need a **system** :manpage:`libpcap(3)` at *run* time, and
this is easy to get wrong: `libpcap`_ ships a vendored
:file:`libpcap.so` under ``_platform/{linux,macos,windows}/`` but does
**not** use it by default. Its :file:`libpcap.cfg` reads ``LIBPCAP = None``
as published, which sends its loader to
:func:`ctypes.util.find_library`, so the library actually mapped is the
host's ``libpcap.so.1``, and one wheel can load different libraries under
two interpreters whose loader search paths differ.
With no system :manpage:`libpcap(3)` at all,
``import pcap`` raises :exc:`OSError` -- not :exc:`ImportError` -- which
:meth:`unsupported_reason` exists partly to catch. And the
:manpage:`libpcap(3)` *version* in play is a property of the host rather
than of the wheel, which is why the PCAP-NG note on :meth:`run` does not
rely on it. Setting ``LIBPCAP = tcpdump`` in :file:`libpcap.cfg` selects the
vendored copy instead.
Being the same interface, it has the same limits. Offline, |pcap|_ 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.
.. important::
Both distributions are published as pre-releases only -- ``pcap-ct``
1.3.0b3 and ``libpcap`` 1.11.0b29 when measured -- and
``pcap-ct`` documents itself as tracking the `PyPCAP`_ **1.2.3** API
rather than 1.3.0. Every attribute this engine touches is present and
behaves identically on both (measured on ``pcap-ct`` 1.3.0b3 against
``pypcap`` 1.3.0), but that is a statement about the versions tested, not
a guarantee from upstream.
.. _libpcap: https://pypi.org/project/libpcap/
Args:
extractor: :class:`~pcapkit.foundation.extraction.Extractor` instance.
"""
if TYPE_CHECKING:
#: Engine extraction package, i.e. |pcap|_.
_expkg: '_PCAPCTPackage'
#: Capture handle, kept separately from the iterator it is read through
#: so that :meth:`close` does not depend on ``iter(handle) is handle``.
_handle: 'Handle'
#: 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'
#: What the |pcap|_ import actually found, cf. :attr:`backend`.
_backend: '_pcap_backend.Probe'
##########################################################################
# Defaults.
##########################################################################
#: Engine name.
__engine_name__ = 'PCAP_CT'
#: Engine module name. Note that this is ``pcap._pcap``, not |pcap|_:
#: `pcap-ct`_ and upstream `PyPCAP`_ both install a top-level |pcap|_,
#: so importing that name cannot tell which of the two is present, and
#: :meth:`Extractor.import_test
#: <pcapkit.foundation.extraction.Extractor.import_test>` decides engine
#: availability by import alone. ``pcap._pcap`` is the `pcap-ct`_
#: implementation module; upstream ships |pcap|_ as a single extension
#: module rather than a package, so the submodule import fails there and the
#: two engines stay distinguishable.
__engine_module__ = 'pcap._pcap'
#: Distribution this engine drives.
__engine_distribution__ = _pcap_backend.PCAP_CT
##########################################################################
# Class methods.
##########################################################################
@classmethod
def _reason_for(cls, found: '_pcap_backend.Probe') -> 'Optional[str]':
"""The verdict on an already-taken probe.
Split out from :meth:`unsupported_reason` so that the public hook keeps the
exact no-argument signature
:meth:`EngineBase.unsupported_reason
<pcapkit.foundation.engines.engine.EngineBase.unsupported_reason>` declares,
while :meth:`__init__` can pass the probe it already has. Probing twice is
not merely wasteful -- see
:func:`~pcapkit.foundation.engines._pcap_backend._purge` for why a repeated
import after a failed one does not reproduce the same error.
"""
return _pcap_backend.engine_reason(cls.__engine_distribution__, found)
[docs]
@classmethod
def unsupported_reason(cls) -> 'Optional[str]':
"""Why this engine cannot run here, or :data:`None` when it can.
Consulted when :class:`~pcapkit.foundation.extraction.Extractor` settles
its engine, *before* the import test, and answers what that test cannot.
**Which distribution owns** |pcap|_. Upstream `PyPCAP`_ owns that name
just as legitimately, and ``import pcap._pcap`` -- what
:attr:`__engine_module__` names -- separates them only while upstream
ships a single extension module. Asking
:func:`~pcapkit.foundation.engines._pcap_backend.probe` states the check
rather than inferring it from an import that happens to fail.
**Whether a system** :manpage:`libpcap(3)` **exists at all.** `pcap-ct`_
is :mod:`ctypes` and needs no compiler, but at runtime it imports the
`libpcap`_ distribution, whose Linux loader raises :exc:`OSError` --
``Cannot find libpcap.so library`` -- when there is none. That is not an
:exc:`ImportError`, so :meth:`Extractor.import_test
<pcapkit.foundation.extraction.Extractor.import_test>` lets it through and
the extraction dies rather than falling back. Reporting it here turns that
into the ordinary "engine unavailable" path.
There is deliberately **no Python version bound**: this engine reads
captures identically across the whole supported range. The floor in the
``PCAP_CT`` extra's markers exists only because :file:`pyproject.toml`
advertises ``requires-python >= 3.8``; below 3.10 the distributions are
not installed, which the import test reports on its own.
Returns:
A phrase naming the real cause, or :data:`None`. A |pcap|_ that is
merely absent returns :data:`None`, since the import test reports that
case in its own words.
.. _libpcap: https://pypi.org/project/libpcap/
"""
return cls._reason_for(_pcap_backend.probe())
##########################################################################
# Properties.
##########################################################################
@property
def dlink(self) -> 'Enum_LinkType':
"""Data link layer protocol, as reported by the capture handle."""
return self._dlink
@property
def backend(self) -> 'str':
"""The distribution, version and location this engine is driving.
More than one distribution provides |pcap|_, so "the pcap-ct engine failed" is
not a complete statement of what ran. This says which one it was, e.g.
``pcap-ct 1.3.0b3 (/.../pcap/__init__.py)``.
"""
return self._backend.describe()
##########################################################################
# Data models.
##########################################################################
[docs]
def __init__(self, extractor: 'Extractor') -> 'None':
"""Initialise the engine.
Warns:
EngineWarning: If both distributions that provide |pcap|_ are
installed. Only one of them can win the import -- measured on
Python 3.10 with both present, the ``pcap-ct`` package wins and
upstream's extension module is shadowed -- so the other is
unreachable, and nothing else would say why
``engine='pypcap'`` had stopped working.
Raises:
UnsupportedCall: If the environment cannot support this engine, for
either of the reasons :meth:`unsupported_reason` describes. That
hook is only consulted by
:class:`~pcapkit.foundation.extraction.Extractor`'s engine
selection, which degrades to the default engine with a warning;
constructing the engine directly bypasses it, so this is the
backstop for that path.
"""
# NOTE: probe before ``import pcap``: a bare import raises ``OSError`` on a host without
# libpcap, escaping ``Extractor``'s import test; ``probe`` reports it as a reason.
self._backend = _pcap_backend.probe()
collision = _pcap_backend.collision_reason(self._backend)
if collision is not None:
warn(f"'Extractor(engine=pcap_ct)': {collision}",
EngineWarning, stacklevel=stacklevel())
# Hand the probe over rather than letting it be taken again: a second
# import attempt after a failed one does not reproduce the same error, for
# the reason ``_pcap_backend._purge`` documents.
reason = self._reason_for(self._backend)
if reason is not None:
raise UnsupportedCall(
f"'Extractor(engine=pcap_ct)' requires 'pcap-ct': {reason}"
)
import pcap # isort:skip # safe: ``probe`` already imported it cleanly
self._expkg = pcap
self._handle = cast('Handle', None)
self._extmp = cast('Iterator[RawFrame]', None)
self._dlink = cast('Enum_LinkType', None)
self._closed = False
super().__init__(extractor)
##########################################################################
# Methods.
##########################################################################
[docs]
def run(self) -> 'None':
"""Call ``pcap.pcap`` to extract PCAP files.
This method assigns :attr:`self._expkg <PCAP_CT._expkg>`
as |pcap|_, :attr:`self._handle <PCAP_CT._handle>` as the
``pcap.pcap`` capture handle and :attr:`self._extmp
<PCAP_CT._extmp>` as an iterator over it.
Warns:
AttributeWarning: Warns under the following circumstances:
* if :attr:`self.extractor._exlyr <pcapkit.foundation.extraction.Extractor._exlyr>`
and/or :attr:`self.extractor._exptl <pcapkit.foundation.extraction.Extractor._exptl>`
is provided as the pcap-ct engine does not
support such operations.
* if :attr:`self.extractor._exctx <pcapkit.foundation.extraction.Extractor._exctx>`
is provided, as the pcap-ct engine does not parse with :mod:`pcapkit`'s
own protocol implementations.
* if reassembly and/or flow tracing is enabled, as the pcap-ct
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 even though
:manpage:`libpcap(3)` can read one -- see the note below.
UnsupportedCall: If the input is not a file on disk, as
:c:func:`pcap_open_offline` can only open a savefile by name.
Note:
The PCAP-NG rejection is a deliberate narrowing, and the reason is
that the alternative is *unpredictable* rather than merely limited.
:manpage:`libpcap(3)` can read a PCAP-NG savefile, but how well
depends on the version the host happens to provide -- which, per the
warning in this class's docstring, is not something the `libpcap`_
wheel pins. Both measured on
:file:`examples/captures/dhcp.pcapng`:
* against libpcap **1.11.0** it reads correctly, yielding the same
frames and the same sub-second timestamps as
:class:`~pcapkit.foundation.engines.pcapng.PCAPNG`;
* against libpcap **1.5.3** it yields the same frames but with
nonsense sub-second timestamps -- ``1102274184.0000002`` and the
like -- silently, with no error.
Even on a good version it cannot represent more than one interface:
:c:func:`pcap_datalink` returns a single link type for the whole file,
so a capture whose interfaces differ would have one interface's link
type applied to every frame.
:class:`~pcapkit.foundation.engines.pcapng.PCAPNG` reads the
per-interface blocks properly and does not depend on the host's
library at all, so PCAP-NG is routed there rather than read
approximately -- or wrongly -- here.
.. _libpcap: https://pypi.org/project/libpcap/
"""
from pcapkit.foundation.engines.pcap import PCAP # isort:skip
ext = self._extractor
if ext._exlyr != 'none' or ext._exptl != 'null':
warn("'Extractor(engine=pcap_ct)' does not support protocol and layer threshold; "
f"'layer={ext._exlyr}' and 'protocol={ext._exptl}' ignored",
AttributeWarning, stacklevel=stacklevel())
if ext._exctx:
warn("'Extractor(engine=pcap_ct)' does not parse with pcapkit's own protocol "
"implementations, so the parsing context supplied through "
"'context=' is ignored",
AttributeWarning, stacklevel=stacklevel())
if ext.magic_number not in PCAP.MAGIC_NUMBER:
raise FormatError(f'unsupported file format: {ext.magic_number!r}; '
'the pcap-ct engine reads PCAP savefiles only')
if not os.path.isfile(ext._ifnm):
raise UnsupportedCall(f"'Extractor(engine=pcap_ct)' 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 or ext._dtls):
ext._flag_r = False
ext._reasm = ReassemblyManager(ipv4=None, ipv6=None, tcp=None, dtls=None)
warn("'Extractor(engine=pcap_ct)' object does not support reassembly; "
f"so 'ipv4={ext._ipv4}', 'ipv6={ext._ipv6}', 'tcp={ext._tcp}' and "
f"'dtls={ext._dtls}' 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=pcap_ct)' object does not support flow tracing; "
f"so 'tcp={ext._tcp}' will be ignored", AttributeWarning, stacklevel=stacklevel())
# NOTE: ``promisc=False`` is defensive: ``pcap-ct`` falls back to a live device when
# ``pcap_open_offline`` fails, which must not request promiscuous mode.
self._handle = self._expkg.pcap(name=ext._ifnm, promisc=False)
self._dlink = Enum_LinkType.get(self._handle.datalink())
# setup verbose handler, unless the caller gave one of their own (#1874)
if ext._vfunc_default:
from pcapkit.toolkit.pcap_ct import packet2chain # isort:skip
ext._vfunc = lambda e, f: print(
f'Frame {e._frnum:>3d}: {packet2chain(cast("RawFrame", f)[1], data_link=self._dlink)}' # pylint: disable=protected-access
) # pylint: disable=logging-fstring-interpolation
# extract & analyse file
self._extmp = iter(self._handle)
[docs]
def read_frame(self) -> 'RawFrame':
"""Read frames with pcap-ct engine.
Returns:
The ``(timestamp, bytes)`` pair as yielded by ``pcap.pcap``.
Note:
The timestamp is seconds since the epoch as a :class:`float`.
`pcap-ct`_ opens savefiles with
:c:func:`pcap_open_offline_with_tstamp_precision` asking for
nanosecond precision, so a microsecond-resolution savefile is scaled
up by :manpage:`libpcap(3)` rather than truncated; the resulting
value matches the record header's ``ts_sec + ts_usec * 1e-6``
exactly for :file:`examples/captures/in.pcap`.
See Also:
Please refer to :meth:`PCAP.read_frame <pcapkit.foundation.engines.pcap.PCAP.read_frame>`
for more operational information.
"""
from pcapkit.toolkit.pcap_ct import packet2dict # isort:skip
ext = self._extractor
# fetch pcap-ct packet
frame = cast('RawFrame', next(self._extmp))
timestamp, packet = frame
# verbose output
ext._frnum += 1
ext._vfunc(ext, frame)
# write output
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
[docs]
def close(self) -> 'None':
"""Close engine.
This method closes the underlying ``pcap.pcap`` handle. It is
idempotent, as :meth:`Extractor._cleanup
<pcapkit.foundation.extraction.Extractor._cleanup>` and
:meth:`Extractor.__exit__ <pcapkit.foundation.extraction.Extractor.__exit__>`
may both reach it.
Note:
``pcap-ct`` 1.3.0b3's own ``pcap.pcap.close`` happens to tolerate
a second call, where upstream ``pypcap`` 1.3.0 segfaults on one. The
:attr:`_closed <PCAP_CT._closed>` flag is kept regardless: it is not
this engine's business to depend on a pre-release's undocumented
forgiveness. An engine that was never opened has no
:attr:`_handle <PCAP_CT._handle>` to close, and is skipped too.
"""
if self._closed or self._handle is None:
return
self._closed = True
self._handle.close()