# -*- coding: utf-8 -*-
"""PCAP Support
==================
.. module:: pcapkit.foundation.engines.pcap
This module contains the implementation for PCAP file extraction
support, as is used by :class:`pcapkit.foundation.extraction.Extractor`.
"""
import io
import sys
from typing import TYPE_CHECKING
from pcapkit.corekit.io import SeekableReader
from pcapkit.corekit.module import ModuleDescriptor
from pcapkit.foundation.engines.engine import EngineBase
from pcapkit.protocols.misc.pcap.frame.protocol import Frame
from pcapkit.protocols.misc.pcap.frame.schema import RECORD_HEADER_LEN
from pcapkit.protocols.misc.pcap.header.protocol import Header
from pcapkit.protocols.misc.pcap.header.schema import MAGIC_NUMBER as Schema_MAGIC_NUMBER
from pcapkit.utilities.logging import get_logger
__all__ = ['PCAP']
if TYPE_CHECKING:
from typing import Any
from pcapkit.const.reg.linktype import LinkType as Enum_LinkType
from pcapkit.corekit.version import VersionInfo
#: logging.Logger: Module-level logger, a child of the package-wide
#: :data:`pcapkit.utilities.logging.logger`.
logger = get_logger(__name__)
#: The toolkit functions :meth:`PCAP.read_frame` hands each frame to, resolved
#: there rather than imported, so their first import holds
#: :data:`~pcapkit.utilities.importlib.import_lock` (:issue:`2771`).
_TOOLKIT = tuple(ModuleDescriptor('pcapkit.toolkit.pcap', name) for name in (
'dtls_reassembly', 'ipv4_reassembly', 'ipv6_reassembly', 'tcp_reassembly', 'tcp_traceflow',
)) # type: tuple[ModuleDescriptor[Any], ...]
[docs]
class PCAP(EngineBase[Frame]):
"""PCAP file extraction support.
Args:
extractor: :class:`~pcapkit.foundation.extraction.Extractor` instance.
"""
if TYPE_CHECKING:
#: Global header.
_gbhdr: 'Header'
#: Version info.
_vinfo: 'VersionInfo'
#: Data link layer protocol.
_dlink: 'Enum_LinkType'
#: Nanosecond flag.
_nnsec: 'bool'
#: The PCAP magic numbers this engine claims, read from their one home,
#: :data:`pcapkit.protocols.misc.pcap.header.schema.MAGIC_NUMBER`
#: (:issue:`1739`).
MAGIC_NUMBER = tuple(Schema_MAGIC_NUMBER)
##########################################################################
# Defaults.
##########################################################################
#: Engine name.
__engine_name__ = 'PCAP'
#: Engine module name.
__engine_module__ = 'pcapkit.protocols.misc.pcap'
##########################################################################
# Properties.
##########################################################################
@property
def header(self) -> 'Header':
"""Global header."""
return self._gbhdr
@property
def version(self) -> 'VersionInfo':
"""Version of input PCAP file."""
return self._vinfo
@property
def dlink(self) -> 'Enum_LinkType':
"""Data link layer protocol."""
return self._dlink
@property
def nanosecond(self) -> 'bool':
"""Nanosecond flag."""
return self._nnsec
##########################################################################
# Methods.
##########################################################################
[docs]
def run(self) -> 'None':
"""Start extraction.
This method is the entry point for PCAP file extraction. It will start
the extraction process by parsing the PCAP global header and then halt
the extraction process until the
:meth:`self.extractor.record_frames <pcapkit.foundation.extraction.Extractor.record_frames>`
method is called.
The method will parse the PCAP global header and save the parsed result
as :attr:`self.header <header>`. Information such as PCAP version, data
link layer protocol type and nanosecond flag are also saved on the
current :class:`PCAP` engine instance.
For output, the method will create the output writer from the parsed
PCAP global header, via :meth:`Extractor._open_output
<pcapkit.foundation.extraction.Extractor._open_output>`, and dump the
header under the name of ``Global Header`` -- except to a PCAP writer,
which has already written it.
"""
# pylint: disable=attribute-defined-outside-init,protected-access
ext = self._extractor
self._gbhdr = Header(ext._ifile)
self._vinfo = self._gbhdr.version
self._dlink = self._gbhdr.protocol
self._nnsec = self._gbhdr.nanosecond
logger.debug('PCAP global header: version %s, link layer %s, nanosecond %s',
self._vinfo, self._dlink.name, self._nnsec)
if ext._flag_q:
return
ext._open_output(protocol=self._dlink, byteorder=self._gbhdr.byteorder,
nanosecond=self._nnsec, version=self._vinfo, thiszone=self._gbhdr.info.thiszone,
sigfigs=self._gbhdr.info.sigfigs, snaplen=self._gbhdr.info.snaplen)
if ext._flag_f:
ofile = ext._ofile(f'{ext._ofnm}/Global Header.{ext._fext}')
else:
ofile = ext._ofile
# NOTE: a PCAP writer has already written the global header on creation.
if ofile.kind != 'pcap':
ofile(self._gbhdr.info, name='Global Header')
ext._offmt = ofile.kind
[docs]
def read_frame(self) -> 'Frame':
"""Read frames.
This method performs the following operations:
- extract frames and each layer of packets;
- make :class:`~pcapkit.corekit.infoclass.Info` object out of frame properties;
- write to output file with corresponding dumper;
- reassemble IP and/or TCP datagram, and/or DTLS handshake message;
- trace TCP flows if any;
- record frame :class:`~pcapkit.corekit.infoclass.Info` object to frame storage.
Returns:
Parsed frame instance.
"""
dtls_reassembly, ipv4_reassembly, ipv6_reassembly, tcp_reassembly, tcp_traceflow = (
descriptor.klass for descriptor in _TOOLKIT)
ext = self._extractor
# read frame header
self._bound_read_ahead()
frame = Frame(ext._ifile, num=ext._frnum+1, header=self._gbhdr.info,
layer=ext._exlyr, protocol=ext._exptl, nanosecond=self._nnsec,
__context__=ext._exctx)
ext._frnum += 1
# verbose output
ext._vfunc(ext, frame)
# write output
frnum = f'Frame {ext._frnum}'
if not ext._flag_q:
# NOTE: every writer takes the frame's info, since only the info itself tells its
# fields apart from the option lists it holds (#1484).
info = frame.info
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(frame)
if data_ipv4 is not None:
ext._reasm.ipv4(data_ipv4)
if ext._ipv6:
data_ipv6 = ipv6_reassembly(frame)
if data_ipv6 is not None:
ext._reasm.ipv6(data_ipv6)
if ext._tcp:
data_tcp = tcp_reassembly(frame)
if data_tcp is not None:
ext._reasm.tcp(data_tcp)
if ext._dtls:
data_dtls = dtls_reassembly(frame)
if data_dtls is not None:
for packet_dtls in data_dtls:
ext._reasm.dtls(packet_dtls)
# trace flows
if ext._flag_t:
if ext._tcp:
data_tf_tcp = tcp_traceflow(frame, data_link=self._dlink)
if data_tf_tcp is not None:
ext._trace.tcp(data_tf_tcp)
# record frames
if ext._flag_d:
ext._frame.append(frame)
# return frame record
return frame
##########################################################################
# Utilities.
##########################################################################
def _bound_read_ahead(self) -> 'None':
"""Have a non-seekable input read ahead to the end of the next record, no further.
A record is measured by reading ahead (see :meth:`SeekableReader.seek
<pcapkit.corekit.io.SeekableReader.seek>`), and then read again from its first
octet. A read past it waits, on a live capture, for the records after: a
buffer's worth held each frame back until that much more had arrived. A read
short of it lets a record longer than the buffer slide out of it before it is
read again (:issue:`1586`). So the read covers the record header and the
``incl_len`` octets it declares, in the global header's byte order -- as far as
the stream goes, for a bogus length, which is as far as a file read goes too.
"""
# pylint: disable=protected-access
ifile = self._extractor._ifile
if not isinstance(ifile, SeekableReader):
return
position = ifile.tell()
head = ifile.read(RECORD_HEADER_LEN)
ifile.seek(position, io.SEEK_SET)
size = RECORD_HEADER_LEN
if len(head) == RECORD_HEADER_LEN:
size += int.from_bytes(head[8:12], self._gbhdr.byteorder)
# NOTE: no buffer holds more than :data:`sys.maxsize` octets; the reader
# moves past the rest of a longer record itself (#1975).
ifile._read_ahead_size = min(size, sys.maxsize)