# -*- coding: utf-8 -*-
"""Scapy Support
===================
.. module:: pcapkit.foundation.engines.scapy
This module contains the implementation for `Scapy`_ engine
support, as is used by :class:`pcapkit.foundation.extraction.Extractor`.
.. _Scapy: https://scapy.net
.. note::
Constructing this engine imports :mod:`scapy.all`, which is what populates
`Scapy`_'s layer registries -- see :meth:`Scapy.__init__` for why anything
narrower silently returns undissected frames.
One side effect is worth knowing about in advance: :mod:`scapy.all` loads
:mod:`scapy.layers.dcerpc`, which reaches `Scapy`_'s TLS layer and there
triggers a ``CryptographyDeprecationWarning`` from :mod:`cryptography` about
finite-field Diffie-Hellman. It is emitted by any complete registry load, not
by :mod:`scapy.all` in particular, and it concerns a key-exchange code path
:mod:`pcapkit` never executes -- but it subclasses :exc:`UserWarning`, not
:exc:`DeprecationWarning`, so Python's default filters show it.
:mod:`pcapkit` deliberately does not filter it away. The warning is `Scapy`_'s
to emit and the consumer's to silence, on the same footing as every other
category (see :mod:`pcapkit.utilities.warnings`); hiding a third-party
deprecation notice from inside a constructor would suppress the only advance
warning that a future :mod:`cryptography` release breaks `Scapy`_'s TLS layer.
To silence it, filter it as usual::
import warnings
from cryptography.utils import CryptographyDeprecationWarning
warnings.filterwarnings('ignore', category=CryptographyDeprecationWarning)
"""
import struct
from typing import TYPE_CHECKING, cast
from pcapkit.foundation.engines.engine import EngineBase
from pcapkit.protocols.schema.misc.pcap.frame import RECORD_HEADER_LEN
from pcapkit.protocols.schema.misc.pcap.header import RESOLUTION
from pcapkit.protocols.schema.misc.pcapng import DEFAULT_RESOLUTION
from pcapkit.utilities.exceptions import VersionError, stacklevel
from pcapkit.utilities.logging import get_logger
from pcapkit.utilities.warnings import AttributeWarning, warn
__all__ = ['Scapy']
if TYPE_CHECKING:
from types import ModuleType
from typing import Any, Iterator, Type
from scapy.packet import Packet as ScapyPacket
from scapy.utils import PcapReader
from pcapkit.foundation.extraction import Extractor
#: logging.Logger: Module-level logger, a child of the package-wide
#: :data:`pcapkit.utilities.logging.logger`.
logger = get_logger(__name__)
#: The private methods of :class:`scapy.utils.PcapNgReader` that
#: :func:`_reader_type` overrides.
_READER_HOOKS = ('_read_block', '_read_block_shb', '_read_block_idb', '_check_interface_id')
#: The most octets a record can hold, as its captured length is 32 bits in a
#: PCAP record header and in every PCAP-NG packet block. As the ``size`` a
#: packet is read with, it cuts nothing (#1717).
_CAPLEN_MAX = 0xFFFF_FFFF
#: The most octets of a PCAP record read in one call, so that a forged captured
#: length costs only the octets the file holds, as in Scapy 2.8 (#1717).
_READ_CHUNK = 0x40000
def _read_tsoffset(options: 'bytes', endian: 'str') -> 'int':
"""Read the ``if_tsoffset`` of a PCAP-NG Interface Description Block (#1549).
Read here rather than by `Scapy`_'s ``_read_options``, whose result differs
by version: keyed by option code in 2.7 and 2.8, and in 2.5 by name, with
only ``tsresol`` and ``comment`` kept.
Args:
options: The block's options area.
endian: :mod:`struct` byte order of the section, ``'<'`` or ``'>'``.
Returns:
The offset (option 14), a signed count of seconds; 0 without one.
.. _Scapy: https://scapy.net
"""
while len(options) >= 4:
code, length = struct.unpack(f'{endian}HH', options[:4])
if not code: # opt_endofopt
break
if code == 14 and length == 8 and len(options) >= 12:
return cast('int', struct.unpack(f'{endian}q', options[4:12])[0])
options = options[4 + length + (-length % 4):]
return 0
def _reader_type() -> 'Type[PcapReader]':
"""A :class:`scapy.utils.PcapReader` that stamps each packet with its interface.
`Scapy`_'s readers convert a record's timestamp to a :class:`~decimal.Decimal`
and drop the resolution it was counted in, and an exact count of microseconds
comes out with the same digits whether the capture counted microseconds or
nanoseconds. They drop the link type too, once it has picked the packet's
first layer, and a layer does not name its link type: a raw IPv4 packet's is
:class:`~scapy.layers.inet.IP` (#1517). The PCAP frame record and the flow
tracing input of a packet need both, so these subclasses attach them with
:func:`pcapkit.toolkit.scapy.attach_resolution` and
:func:`pcapkit.toolkit.scapy.attach_linktype`: from the global header of a
PCAP, and from the interface of each PCAP-NG packet block, which
:meth:`~scapy.utils.RawPcapNgReader._read_block` is the last place to see.
The PCAP-NG subclass also starts each section with no interfaces, as the
PCAP-NG specification defines a section: an interface ID counts the Interface
Description Blocks of its own section only. `Scapy`_'s reader keeps appending
to one table across Section Header Blocks, so a later section's interface 0
resolved to the first section's, with its link type and resolution (#1522).
And it dates packets that `Scapy`_ misdates. A Simple Packet
Block has no timestamp, and `Scapy`_ leaves its packet with the time it was
built, so it is set to 0 (#1503). `Scapy`_ ignores an interface's
``if_tsoffset``, the seconds to add to each of its timestamps, so the
subclass reads it and adds it (#1549).
Both subclasses read each record whole (#1717). ``sniff`` reads a packet
with ``recv()``, whose ``size`` defaults to `Scapy`_'s ``MTU``, 65,535
octets, and every reader cuts the record to ``size``, so ``recv`` defaults
to :data:`_CAPLEN_MAX` instead, as does a PCAP-NG block, which `Scapy`_ 2.6
reads at ``MTU`` whatever the ``size``. `Scapy`_ 2.8's PCAP reader also
reads at most ``MTU * 4`` octets of a record, 262,140, and skips the rest,
so the PCAP subclass reads the record itself, :data:`_READ_CHUNK` octets at
a time.
Built when called, so that importing this module does not import `Scapy`_,
and so that it always subclasses the :class:`scapy.utils.PcapReader` of the
moment.
Returns:
A reader type, which opens a PCAP-NG file as its PCAP-NG alternative.
Raises:
VersionError: If :class:`scapy.utils.PcapNgReader` lacks a method in
:data:`_READER_HOOKS`, which these subclasses override.
.. _Scapy: https://scapy.net
"""
from scapy.error import warning # isort:skip
from scapy.utils import EDecimal, PcapNgReader, PcapReader, RawPcapReader # isort:skip
from pcapkit.toolkit.scapy import attach_linktype, attach_resolution # isort:skip
# The PCAP-NG hooks overridden below are private to Scapy. One renamed away
# would leave its fix unapplied without a word, so a missing one is an error.
missing = [name for name in _READER_HOOKS if not callable(getattr(PcapNgReader, name, None))]
if missing:
raise VersionError(f"scapy.utils.PcapNgReader has no {', '.join(missing)}, which the Scapy "
"engine's reader overrides; this Scapy version is not supported")
class _Closing:
# ``sniff`` closes the reader at the end of the file, and the engine closes
# it again in case ``sniff`` raised first; only the first close is real,
# since closing a gzip reader twice raises
_closed = False
def close(self) -> 'None':
if not self._closed:
self._closed = True
super().close() # type: ignore[misc]
class _Uncapped(PcapReader):
# ``sniff`` calls ``recv()``, and Scapy's default ``size`` of ``MTU``
# cut every record to 65,535 octets (#1717); an explicit one is kept
def recv(self, size: 'int' = _CAPLEN_MAX, **kwargs: 'Any') -> 'ScapyPacket': # type: ignore[override]
return super().recv(size, **kwargs)
class _PcapRecords(RawPcapReader):
# Scapy 2.8 reads at most ``MTU * 4`` octets of a record and skips the
# rest (#1717), so the record is read here: whole, but in chunks as 2.8
# reads it. ``PcapReader.read_packet`` calls ``super(PcapReader,
# self)._read_packet``, which looks in the classes after ``PcapReader``
# in the reader's MRO, so this one is a base listed after it.
def _read_packet(self, size: 'int' = _CAPLEN_MAX) -> 'tuple[bytes, Any]':
header = self.f.read(RECORD_HEADER_LEN)
if len(header) < RECORD_HEADER_LEN:
raise EOFError
sec, usec, caplen, wirelen = struct.unpack(f'{self.endian}IIII', header)
chunks = [] # type: list[bytes]
remaining = caplen
while remaining:
chunk = self.f.read(min(remaining, _READ_CHUNK))
if not chunk:
# the file ends inside the record, which Scapy 2.8 warns of
warning('Pcap: packet has been truncated')
break
chunks.append(chunk)
remaining -= len(chunk)
return (b''.join(chunks)[:size],
self.PacketMetadata(sec=sec, usec=usec, wirelen=wirelen, caplen=caplen))
class _PcapReader(_Closing, _Uncapped, PcapReader, _PcapRecords):
def read_packet(self, *args: 'Any', **kwargs: 'Any') -> 'ScapyPacket':
packet = super().read_packet(*args, **kwargs)
attach_resolution(packet, RESOLUTION[bool(self.nano)])
attach_linktype(packet, self.linktype)
return packet
class _PcapNgReader(_Closing, _Uncapped, PcapNgReader):
# the metaclass makes this ``_PcapReader``'s alternative in turn
alternative = _PcapReader
_resolution = DEFAULT_RESOLUTION
# each set from the packet block last read, before ``read_packet`` reads it
_linktype: 'int'
_timed: 'bool'
_offset: 'int'
_interface: 'int'
#: ``if_tsoffset`` of each interface of the section, by interface ID.
#: Kept here: what Scapy keeps of an interface differs by version.
_offsets: 'list[int]'
def _read_block_shb(self, *args: 'Any', **kwargs: 'Any') -> 'None':
super()._read_block_shb(*args, **kwargs)
# a new section: its interface IDs restart from 0 (#1522)
self.interfaces = []
self._offsets = []
def _read_block_idb(self, block: 'bytes', *args: 'Any', **kwargs: 'Any') -> 'None':
super()._read_block_idb(block, *args, **kwargs)
# appended once Scapy has appended the interface, so the IDs match
self._offsets.append(_read_tsoffset(block[8:], self.endian))
def _check_interface_id(self, intid: 'int') -> 'Any':
# Each packet block's reader checks its interface ID here, a Simple
# Packet Block's as 0, and the metadata it returns does not keep it.
# Scapy 2.8 skips the block when this returns false, and before 2.8
# it raises instead, so the result must be passed through. A block
# it rejects is never returned, so its ID is never read.
self._interface = intid
return super()._check_interface_id(intid)
def _read_block(self, size: 'int' = _CAPLEN_MAX) -> 'Any':
# Scapy 2.6 reads each block with ``_read_block()``, so at ``MTU``
# whatever the packet's ``size`` (#1717)
block = super()._read_block(size)
if block is not None:
self._resolution = block[1].tsresol
self._linktype = block[1].linktype
# ``None`` for a Simple Packet Block, which has no timestamp (#1503)
self._timed = block[1].tshigh is not None
self._offset = self._offsets[self._interface]
return block
def read_packet(self, *args: 'Any', **kwargs: 'Any') -> 'ScapyPacket':
packet = super().read_packet(*args, **kwargs)
attach_resolution(packet, self._resolution)
attach_linktype(packet, self._linktype)
if not self._timed:
# no timestamp at all, so 0, where Scapy leaves the time the
# packet was built (#1503)
packet.time = EDecimal(0)
elif self._offset:
packet.time += self._offset # if_tsoffset (#1549)
return packet
return _PcapReader
[docs]
class Scapy(EngineBase['ScapyPacket']):
"""Scapy engine support.
Args:
extractor: :class:`~pcapkit.foundation.extraction.Extractor` instance.
"""
if TYPE_CHECKING:
#: Engine extraction package, i.e. :mod:`scapy.all`.
_expkg: 'ModuleType'
#: Engine extraction temporary storage.
_extmp: 'Iterator[ScapyPacket]'
##########################################################################
# Defaults.
##########################################################################
#: Engine name.
__engine_name__ = 'Scapy'
#: Engine module name.
__engine_module__ = 'scapy'
##########################################################################
# Data models.
##########################################################################
[docs]
def __init__(self, extractor: 'Extractor') -> 'None':
"""Initialise the engine.
Args:
extractor: :class:`~pcapkit.foundation.extraction.Extractor` instance.
"""
# NOTE: :mod:`scapy.all`, not :mod:`scapy.sendrecv`, and the difference is
# load-bearing rather than cosmetic. Scapy dispatches on two registries that
# exist only as *import side effects* of its layer modules: ``conf.l2types``,
# mapping a capture's link type onto a link-layer class, and the
# ``bind_layers`` payload table. ``scapy/__init__.py`` fills in neither, so
# importing just the sniffing submodule leaves both empty --
# :class:`~scapy.utils.PcapReader` then cannot map even link type 1 (plain
# Ethernet), writes ``unknown LL type [1]/[0x1]`` to stderr and returns every
# frame as one opaque :class:`~scapy.packet.Raw` layer. Nothing raises, so the
# engine would deliver no dissection whatsoever and announce it only on
# stderr (#406).
#
# Naming the layer modules individually is not a cheaper way to the same
# place: it repairs the link layer and leaves the payload table short, so
# ``dhcp.pcapng`` still comes back as ``Ethernet / IP / UDP / Raw`` rather
# than ``... / UDP / BOOTP / DHCP options``. Enumerating them here would also
# go stale silently -- a link or payload type added by a later scapy would
# regress to ``Raw`` with no error -- whereas :mod:`scapy.all` defers to
# ``conf.load_layers``, which scapy itself keeps current. The cost is the
# layer loading, not the façade: importing :mod:`scapy.layers.all` alone
# measures the same, so there is no complete-but-cheaper option to prefer.
#
# This stays inside ``__init__`` rather than moving to module scope so that
# ``import pcapkit`` does not pay for it; only callers who actually select
# this engine do. And it is bound to :attr:`_expkg`, the attribute
# :meth:`run` calls ``sniff`` on, rather than left as a bare side-effecting
# import -- an import whose value is used cannot be dropped later as unused,
# which is how this class of bug comes back.
from scapy import all as scapy # isort:skip
self._expkg = scapy
self._extmp = cast('Iterator[ScapyPacket]', None)
super().__init__(extractor)
##########################################################################
# Methods.
##########################################################################
[docs]
def run(self) -> 'None':
"""Call :func:`scapy.sendrecv.sniff` to extract PCAP files.
This method assigns :attr:`self._extmp <Scapy._extmp>` as an iterator
from :func:`scapy.sendrecv.sniff`, reached through
:attr:`self._expkg <Scapy._expkg>` -- which :meth:`__init__` binds to
:mod:`scapy.all`, since that is the import that populates the layer
registries ``sniff`` needs to dissect anything.
The global header is parsed and dumped first, by
:meth:`self.extractor.record_header <pcapkit.foundation.extraction.Extractor.record_header>`.
Warns:
AttributeWarning: 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 Scapy engine does not support such operations;
or if :attr:`self.extractor._exctx <pcapkit.foundation.extraction.Extractor._exctx>`
is provided, as the Scapy engine does not parse with :mod:`pcapkit`'s own
protocol implementations.
"""
ext = self._extractor
if ext._exlyr != 'none' or ext._exptl != 'null':
warn("'Extractor(engine=scapy)' 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=scapy)' does not parse with pcapkit's own protocol "
"implementations, so the parsing context supplied through "
"'context=' is ignored",
AttributeWarning, stacklevel=stacklevel())
# setup verbose handler
if ext._flag_v:
from pcapkit.toolkit.scapy import packet2chain # isort:skip
ext._vfunc = lambda e, f: print(
f'Frame {e._frnum:>3d}: {packet2chain(f)}' # pylint: disable=protected-access
)
# extract global header
ext.record_header()
# extract & analyse file
logger.debug('scapy: sniffing %s', ext._ifnm)
# NOTE: the reader is opened here rather than by ``sniff(offline=...)`` so
# that every packet carries its capture's timestamp resolution, which
# Scapy's own readers discard (#1507). Labelled with the file name, as
# ``offline=`` labels it, so ``Packet.sniffed_on`` is unchanged. ``sniff``
# closes it at the end of the file, as it closes the reader ``offline=``
# opens; it is closed here as well in case ``sniff`` raised first, which
# the reader takes as a no-op once closed.
reader = _reader_type()(ext._ifnm)
try:
self._extmp = iter(self._expkg.sniff(opened_socket={reader: ext._ifnm}))
finally:
reader.close()
[docs]
def read_frame(self) -> 'ScapyPacket':
"""Read frames with Scapy engine.
Returns:
Parsed frame instance.
See Also:
Please refer to :meth:`PCAP.read_frame <pcapkit.foundation.engines.pcap.PCAP.read_frame>`
for more operational information.
"""
from pcapkit.toolkit.scapy import (ipv4_reassembly, ipv6_reassembly, packet2dict,
tcp_reassembly, tcp_traceflow)
ext = self._extractor
# fetch Scapy packet
packet = next(self._extmp)
# verbose output
ext._frnum += 1
ext._vfunc(ext, packet)
# write output
frnum = f'Frame {ext._frnum}'
if not ext._flag_q:
info = packet2dict(packet)
if ext._flag_f:
ofile = ext._ofile(f'{ext._ofnm}/{frnum}.{ext._fext}')
ofile(info, name=frnum)
else:
ext._ofile(info, name=frnum)
# record fragments
if ext._flag_r:
if ext._ipv4:
data_ipv4 = ipv4_reassembly(packet, count=ext._frnum)
if data_ipv4 is not None:
ext._reasm.ipv4(data_ipv4)
if ext._ipv6:
data_ipv6 = ipv6_reassembly(packet, count=ext._frnum)
if data_ipv6 is not None:
ext._reasm.ipv6(data_ipv6)
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, count=ext._frnum)
if data_tf_tcp is not None:
ext._trace.tcp(data_tf_tcp)
# record frames
if ext._flag_d:
# setattr(packet, 'packet2dict', packet2dict)
# setattr(packet, 'packet2chain', packet2chain)
ext._frame.append(packet)
# return frame record
return packet