# -*- 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`.
.. _pcap-ct: https://pypi.org/project/pcap-ct/
"""
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 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, EngineWarning, warn
__all__ = ['PCAP_CT']
if TYPE_CHECKING:
from typing import Iterator, Optional
from pcap import pcap as Handle
from pcapkit.foundation.extraction import Extractor
#: A `pcap-ct`_ "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]
[docs]
class PCAP_CT(Engine['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 :mod:`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 a current
interpreter: 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 :file:`_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``. Measured on this host: the same
``libpcap`` 1.11.0b29 wheel loaded ``/usr/lib64/libpcap.so.1.5.3`` under one
interpreter and linuxbrew's 1.11.0 under another, purely because their
loader search paths differ.
Two consequences. 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 -- verified loading, and reporting libpcap 1.10.6.
Being the same interface, it has the same limits. Offline, :mod:`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 at the time of writing -- 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.
.. _pcap-ct: https://pypi.org/project/pcap-ct/
.. _libpcap: https://pypi.org/project/libpcap/
.. _PyPCAP: https://github.com/pynetwork/pypcap
Args:
extractor: :class:`~pcapkit.foundation.extraction.Extractor` instance.
"""
if TYPE_CHECKING:
import pcap
#: Engine extraction package.
_expkg: 'pcap'
#: 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 :mod:`pcap` import actually found, c.f. :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 :mod:`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 :mod:`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.
"""
wrong = _pcap_backend.wrong_backend_reason(cls.__engine_distribution__, found)
if wrong is not None:
return wrong
# ``missing`` separates "not installed", which the import test reports
# well, from a failure that would otherwise escape it -- the missing
# ``libpcap.so`` being the case this exists for.
if found.name is None and not found.missing:
return f'the `pcap` module is installed but unusable -- {found.failure}'
return None
[docs]
@classmethod
def unsupported_reason(cls) -> 'Optional[str]':
"""Why this engine cannot run here, or :data:`None` when it can.
Consulted by :meth:`pcapkit.foundation.extraction.Extractor.run` *before*
the import test, and it answers two questions the import test cannot.
**Which distribution owns :mod:`pcap`.** Upstream `PyPCAP`_ owns that name
just as legitimately, and ``import pcap._pcap`` -- what
:attr:`__engine_module__` names -- separates them only as long as upstream
keeps shipping 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.** This is the
important one, and it is measured rather than theoretical. `pcap-ct`_ needs
no compiler because it is :mod:`ctypes`, but at *runtime* it imports the
`libpcap`_ distribution, whose Linux loader calls
:func:`ctypes.util.find_library` and raises :exc:`OSError` --
``Cannot find libpcap.so library`` -- when there is none.
:exc:`OSError` 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 in either direction.**
Measured on 3.10.20 and 3.14.7: ``pcap-ct`` 1.3.0b3 with ``libpcap``
1.11.0b29 installs and reads :file:`examples/captures/in.pcap` identically
on both, so this engine covers the whole range the project supports. The
floor in the ``PCAP_CT`` extra's markers exists only because
:file:`pyproject.toml` still advertises ``requires-python >= 3.6``, and on
an interpreter below 3.10 the distributions simply will not be installed --
which the import test then reports correctly on its own.
Returns:
A phrase naming the real cause, or :data:`None`. An :mod:`pcap` that is
merely absent returns :data:`None`, since the import test reports that
case in its own words.
.. _PyPCAP: https://github.com/pynetwork/pypcap
.. _pcap-ct: https://pypi.org/project/pcap-ct/
.. _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.
Two distributions provide :mod:`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 :mod:`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 :meth:`Extractor.run
<pcapkit.foundation.extraction.Extractor.run>`, which degrades to
the default engine with a warning; constructing the engine
directly bypasses it, so this is the backstop for that path.
"""
# NOTE: the probe comes *before* ``import pcap``, and the order is
# load-bearing rather than stylistic. A bare ``import pcap`` here would
# raise ``OSError: Cannot find libpcap.so library`` on a host with no
# system libpcap, escaping both this constructor and ``Extractor``'s
# import test; ``probe`` performs the same import with that failure
# caught, so the check below can report it as a reason instead.
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 :class:`pcap.pcap` to extract PCAP files.
This method assigns :attr:`self._expkg <PCAP_CT._expkg>`
as :mod:`pcap`, :attr:`self._handle <PCAP_CT._handle>` as the
:class:`pcap.pcap` capture handle and :attr:`self._extmp
<PCAP_CT._extmp>` as an iterator over it.
Warns:
AttributeWarning: Warns under 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 currently does not
support such operations.
* 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 the vendored
: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
four frames and the same sub-second timestamps as
:class:`~pcapkit.foundation.engines.pcapng.PCAPNG`;
* against libpcap **1.5.3** it yields the four 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.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):
ext._flag_r = False
ext._reasm = ReassemblyManager(ipv4=None, ipv6=None, tcp=None)
warn("'Extractor(engine=pcap_ct)' 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=pcap_ct)' 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 ``pcap-ct`` falls through to opening the
# name as a live device when ``pcap_open_offline`` fails, and we do not
# want that attempt to request promiscuous mode. (That fallback is also
# why a bad path surfaces as ``OSError: ... No such device exists``,
# which is what the ``os.path.isfile`` check above pre-empts.)
self._handle = cast('Handle', self._expkg.pcap(name=ext._ifnm, promisc=False))
self._dlink = Enum_LinkType.get(self._handle.datalink())
# setup verbose handler
if ext._flag_v:
from pcapkit.toolkit.pcap_ct 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(self._handle)
[docs]
def read_frame(self) -> 'RawFrame':
"""Read frames with pcap-ct engine.
Returns:
The ``(timestamp, bytes)`` pair as yielded by :class:`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.
.. _pcap-ct: https://pypi.org/project/pcap-ct/
"""
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 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
[docs]
def close(self) -> 'None':
"""Close engine.
This method closes the underlying :class:`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 :meth:`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, and the flag also covers the case of an engine that was
never opened.
"""
if self._closed or self._handle is None:
return
self._closed = True
self._handle.close()