# -*- 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`.
.. seealso::
:mod:`pcapkit.foundation.engines.pcap_ct` is the engine for `pcap-ct`_, an
independent reimplementation of the same interface that installs on Python
3.12 and newer, where upstream `PyPCAP`_ does not.
.. _PyPCAP: https://github.com/pynetwork/pypcap
.. _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__ = ['PyPCAP']
if TYPE_CHECKING:
from typing import Iterator, Optional
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]
[docs]
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.
.. important::
This engine is upstream `PyPCAP`_ specifically, which 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
3.12+ C API. `pcap-ct`_ is an independent reimplementation of the same
interface that does install there, and
:class:`~pcapkit.foundation.engines.pcap_ct.PCAP_CT` is the engine for it.
Because `pcap-ct`_ installs the same top-level :mod:`pcap` module, this
engine checks which distribution it got and refuses the other one rather
than running under the wrong name -- see :meth:`__init__`.
.. _PyPCAP: https://github.com/pynetwork/pypcap
.. _pcap-ct: https://pypi.org/project/pcap-ct/
Args:
extractor: :class:`~pcapkit.foundation.extraction.Extractor` instance.
Raises:
UnsupportedCall: If the installed :mod:`pcap` is `pcap-ct`_ rather than
upstream `PyPCAP`_.
"""
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'
#: What the :mod:`pcap` import actually found, c.f. :attr:`backend`.
_backend: '_pcap_backend.Probe'
##########################################################################
# Defaults.
##########################################################################
#: Engine name.
__engine_name__ = 'PyPCAP'
#: Engine module name. Note that this cannot separate the two distributions
#: that own it -- see :func:`pcapkit.foundation.engines._pcap_backend.probe`
#: and :meth:`unsupported_reason`, which is where that is done.
__engine_module__ = 'pcap'
#: Distribution this engine drives.
__engine_distribution__ = _pcap_backend.PYPCAP
##########################################################################
# 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.wrong_backend_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 by :meth:`pcapkit.foundation.extraction.Extractor.run` *before*
the import test, because the import test cannot answer this question:
``import pcap`` succeeds when `pcap-ct`_ is installed and upstream
`PyPCAP`_ is not, so the guard is satisfied and this engine would run on a
distribution it does not drive -- reporting ``PyPCAP`` for work that
``pcap-ct`` did.
The condition is deliberately *not* a Python version ceiling, even though
upstream cannot be installed on 3.12 or newer. What matters is which
distribution is actually present: a version check would refuse a
hypothetical upstream build that someone got working on a newer
interpreter, and would say nothing useful on 3.10 and 3.11, where both
distributions install happily and either could be the one in place.
Returns:
A phrase naming the real cause, or :data:`None`. An absent :mod:`pcap`
returns :data:`None`, since the import test reports that case in its
own words and duplicating it would produce two warnings for one
problem.
.. _PyPCAP: https://github.com/pynetwork/pypcap
.. _pcap-ct: https://pypi.org/project/pcap-ct/
"""
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 PyPCAP engine failed" is
not a complete statement of what ran. This says which one it was, e.g.
``pypcap 1.3.0 (/.../pcap.cpython-310-x86_64-linux-gnu.so)``.
"""
return self._backend.describe()
##########################################################################
# Data models.
##########################################################################
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, so the other is
shadowed and unreachable -- a state worth naming, because
``engine=`` then selects an engine that cannot run and nothing
else would say why.
Raises:
UnsupportedCall: If the installed :mod:`pcap` is `pcap-ct`_ rather
than upstream `PyPCAP`_. This is the same condition
:meth:`unsupported_reason` reports, kept here as well because that
hook is only consulted by
:meth:`Extractor.run <pcapkit.foundation.extraction.Extractor.run>`;
constructing the engine directly bypasses it, and running
`pcap-ct`_ under the ``PyPCAP`` name would misreport what did the
work. Via ``Extractor`` the hook fires first and degrades to the
default engine with a warning, so this is a backstop rather than
the usual path.
.. _PyPCAP: https://github.com/pynetwork/pypcap
.. _pcap-ct: https://pypi.org/project/pcap-ct/
"""
# NOTE: the probe comes *before* ``import pcap`` so that the two engines
# behave the same way here. It matters more for
# :class:`~pcapkit.foundation.engines.pcap_ct.PCAP_CT`, whose import can
# raise ``OSError`` rather than ``ImportError``, but a shared order is one
# fewer difference between them.
self._backend = _pcap_backend.probe()
collision = _pcap_backend.collision_reason(self._backend)
if collision is not None:
warn(f"'Extractor(engine=pypcap)': {collision}",
EngineWarning, stacklevel=stacklevel())
# Hand the probe over rather than letting it be taken again -- see
# ``_pcap_backend._purge`` for why a repeated import is not free.
reason = self._reason_for(self._backend)
if reason is not None:
raise UnsupportedCall(
f"'Extractor(engine=pypcap)' requires upstream 'pypcap': {reason}"
)
import pcap # isort:skip # safe: ``probe`` already imported it cleanly
self._expkg = pcap
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 <PyPCAP._expkg>`
as :mod:`pcap` and :attr:`self._extmp <PyPCAP._extmp>`
as an iterator from :class:`pcap.pcap`.
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 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)
[docs]
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 <pcapkit.foundation.engines.pcap.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
[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, 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()