# -*- 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
from typing import TYPE_CHECKING
from pcapkit.corekit.io import SeekableReader
from pcapkit.foundation.engines.engine import EngineBase
from pcapkit.protocols.misc.pcap.frame import Frame
from pcapkit.protocols.misc.pcap.header import Header
from pcapkit.protocols.schema.misc.pcap.frame import RECORD_HEADER_LEN
from pcapkit.protocols.schema.misc.pcap.header import MAGIC_NUMBER as Schema_MAGIC_NUMBER
from pcapkit.utilities.logging import get_logger
__all__ = ['PCAP']
if TYPE_CHECKING:
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__)
[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.schema.misc.pcap.header.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;
- trace TCP flows if any;
- record frame :class:`~pcapkit.corekit.infoclass.Info` object to frame storage.
Returns:
Parsed frame instance.
"""
from pcapkit.toolkit.pcap import (ipv4_reassembly, ipv6_reassembly, tcp_reassembly,
tcp_traceflow)
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: a PCAP writer for its
# octets and record header, the others for its fields, which only
# the info itself tells apart from the option lists it holds
# (GitHub issue #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)
# 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)
ifile._read_ahead_size = size