# -*- 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`.
.. note::
Constructing this engine imports ``scapy.all``, which populates `Scapy`_'s
layer registries -- see :meth:`Scapy.__init__` for why anything narrower
silently returns undissected frames.
That load reaches `Scapy`_'s TLS layer through :mod:`scapy.layers.dcerpc`,
which triggers a ``CryptographyDeprecationWarning`` from |cryptography|_
about finite-field Diffie-Hellman, a code path :mod:`pcapkit` never
executes. It subclasses :exc:`UserWarning`, not :exc:`DeprecationWarning`,
so Python's default filters show it.
:mod:`pcapkit` leaves it unfiltered: it is `Scapy`_'s to emit and the
consumer's to silence, as for every other category (see
:mod:`pcapkit.utilities.warnings`), and hiding it would suppress the only
advance notice that a future |cryptography|_ release breaks `Scapy`_'s TLS
layer. To silence it::
import warnings
from cryptography.utils import CryptographyDeprecationWarning
warnings.filterwarnings('ignore', category=CryptographyDeprecationWarning)
"""
import struct
from typing import TYPE_CHECKING, cast
from pcapkit.corekit.schema.padding import pad_length
from pcapkit.foundation.engines.engine import EngineBase
from pcapkit.foundation.reassembly import ReassemblyManager
from pcapkit.protocols.misc.pcap.frame.schema import RECORD_HEADER_LEN
from pcapkit.protocols.misc.pcap.header.schema import RESOLUTION
from pcapkit.protocols.misc.pcapng.schema 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 typing import Any, Iterator, Protocol, Type
from scapy.packet import Packet as ScapyPacket
from scapy.plist import PacketList
from scapy.utils import PcapReader
from pcapkit.foundation.extraction import Extractor
class _ScapyPackage(Protocol):
"""``scapy.all``, as :class:`Scapy` reads it through :attr:`Scapy._expkg` (:issue:`1878`).
Typed by scapy itself, which ships its own type information.
"""
def sniff(self, *args: 'Any', **kwargs: 'Any') -> 'PacketList': ...
#: 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 (:issue:`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 (:issue:`1717`).
_READ_CHUNK = 0x40000
def _read_tsoffset(options: 'bytes', endian: 'str') -> 'int':
"""Read the ``if_tsoffset`` of a PCAP-NG Interface Description Block (:issue:`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.
"""
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 + pad_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` (:issue:`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
``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
resolves to the first section's, with its link type and resolution (:issue:`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 (:issue:`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 (:issue:`1549`).
Both subclasses read each record whole (:issue:`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.
"""
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``
# cuts 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. ``scapy.all``.
_expkg: '_ScapyPackage'
#: 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: ``scapy.all`` fills the layer registries scapy dispatches on, which the sniffing
# submodule alone leaves empty (#406); imported here to keep ``import pcapkit`` cheap.
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
``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; or if DTLS reassembly is enabled, for the
same reason.
"""
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())
if ext._flag_r and ext._dtls:
ext._dtls = False
ext._reasm = ReassemblyManager(ipv4=ext._reasm.ipv4, ipv6=ext._reasm.ipv6,
tcp=ext._reasm.tcp, dtls=None)
warn("'Extractor(engine=scapy)' object does not support DTLS reassembly, "
"which reads pcapkit's own DTLS layer; so 'dtls=True' will be ignored",
AttributeWarning, stacklevel=stacklevel())
# setup verbose handler, unless the caller gave one of their own (#1874)
if ext._vfunc_default:
from pcapkit.toolkit.scapy import packet2chain # isort:skip
ext._vfunc = lambda e, f: print(
f'Frame {e._frnum:>3d}: {packet2chain(cast("ScapyPacket", f))}' # pylint: disable=protected-access
)
# extract global header
ext.record_header()
# extract & analyse file
logger.debug('scapy: sniffing %s', ext._ifnm)
# NOTE: opened here rather than by ``sniff(offline=...)`` so packets keep the capture's
# timestamp resolution (#1507); closed here too in case ``sniff`` raised first.
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