# -*- coding: utf-8 -*-
"""PCAP Dumper
=================
.. module:: pcapkit.dumpkit.pcap
:mod:`pcapkit.dumpkit.pcap` is the dumper for :mod:`pcapkit` implementation,
specifically for PCAP format, like those described in
:mod:`dictdumper`.
"""
import decimal
import struct
import sys
from typing import TYPE_CHECKING
from pcapkit.dumpkit.common import DumperBase
from pcapkit.protocols.data.misc.pcap.header import Header as Data_Header
from pcapkit.protocols.misc.pcap.header import Header
from pcapkit.protocols.schema.misc.pcap.header import RESOLUTION
if TYPE_CHECKING:
from enum import IntEnum as StdlibIntEnum
from typing import IO, Any, Optional
from aenum import IntEnum as AenumIntEnum
from typing_extensions import Literal
from pcapkit.const.reg.linktype import LinkType as Enum_LinkType
from pcapkit.corekit.version import VersionInfo
from pcapkit.protocols.data.misc.pcap.frame import Frame as Data_Frame
from pcapkit.protocols.data.misc.pcap.frame import FrameInfo as Data_FrameInfo
__all__ = [
'PCAPIO',
]
#: Per-byte-order record header packers for the four ``uint32`` fields of a PCAP
#: record header -- ``ts_sec``, ``ts_usec``, ``incl_len``, ``orig_len``. Keyed by
#: the byte order of the global header this dumper wrote, so that the records
#: agree with the magic number a reader will find in front of them.
_RECORD_HEADER = {
'little': struct.Struct('<IIII'),
'big': struct.Struct('>IIII'),
}
#: Truncation mask for those four fields. :class:`~pcapkit.corekit.fields.numbers.UInt32Field`,
#: the field class that would otherwise pack them, masks to the field width in
#: :meth:`~pcapkit.corekit.fields.numbers.NumberField.pre_process` rather than
#: rejecting an out-of-range value -- a ``ts_sec`` of ``2**32 + 5`` is written as
#: ``5``. :func:`struct.pack` raises instead, so the mask is applied here to keep
#: the two spellings writing the same octets for every input.
_UINT32_MASK = 0xFFFF_FFFF
def _truncates_to(scaled: 'int', den: 'int', fraction: 'int') -> 'bool':
"""Whether ``fraction`` is ``scaled / den`` truncated to a whole unit.
Args:
scaled: numerator of a frame's offset into its second, in units
den: denominator of that offset
fraction: a record header's ``ts_usec`` field
Returns:
:data:`True` if ``fraction <= scaled / den < fraction + 1``.
"""
return fraction * den <= scaled < (fraction + 1) * den
[docs]
class PCAPIO(DumperBase):
"""PCAP file dumper.
Args:
fname: output file name
protocol: data link type
byteorder: header byte order
nanosecond: nanosecond-resolution file flag
**kwargs: further global header fields, i.e. ``version``, ``thiszone``,
``sigfigs`` and ``snaplen``, and arbitrary keyword arguments
.. versionchanged:: 1.5.0
The global header takes ``version``, ``thiszone``, ``sigfigs`` and
``snaplen`` from ``**kwargs``, so ``format='pcap'`` output keeps the
input's values; 1.4.x wrote the defaults (:issue:`1165`,
:issue:`1255`).
"""
if TYPE_CHECKING:
#: PCAP file global header.
_ghdr: 'Data_Header'
#: Record header packer, in the global header's byte order.
_rechdr: 'struct.Struct'
#: Units per second of a record's fraction, in this file's resolution.
_tsres: 'int'
#: Units per second of a record's fraction, in the other resolution.
_tsalt: 'int'
##########################################################################
# Properties.
##########################################################################
@property
def kind(self) -> 'Literal["pcap"]':
"""File format of current dumper."""
return 'pcap'
##########################################################################
# Data models.
##########################################################################
[docs]
def __init__(self, fname: 'str', *, protocol: 'Enum_LinkType | StdlibIntEnum | AenumIntEnum | str | int',
byteorder: 'Literal["big", "little"]' = sys.byteorder,
nanosecond: 'bool' = False, **kwargs: 'Any') -> 'None': # pylint: disable=arguments-differ
"""Initialise dumper.
Args:
fname: output file name
protocol: data link type
byteorder: header byte order
nanosecond: nanosecond-resolution file flag
**kwargs: further global header fields, i.e. ``version``,
``thiszone``, ``sigfigs`` and ``snaplen``, and arbitrary keyword
arguments
"""
#: int: Frame counter.
self._fnum = 1
# NOTE: Both of these only record how the dumper was configured -- the
# values that shape the output reach it through :meth:`self._dump_header
# <_dump_header>`'s own arguments, and are readable afterwards from
# :attr:`self._ghdr <_ghdr>`. They are kept because they are part of the
# instance surface a subclass may already read.
#: bool: Nanosecond-resolution file flag.
self._nsec = nanosecond
#: Enum_LinkType | StdlibIntEnum | AenumIntEnum | str | int: Data link type.
self._link = protocol
super().__init__(fname, protocol=protocol, byteorder=byteorder, nanosecond=nanosecond, **kwargs)
def __call__(self, value: 'Data_Frame', name: 'Optional[str]' = None) -> 'PCAPIO':
"""Dump a new frame.
Args:
value: content to be dumped
name: name of current content block
Returns:
The dumper class itself (to support chain calling).
"""
with open(self._file, 'ab') as file:
self._append_value(value, file, name or '')
return self
##########################################################################
# Utilities.
##########################################################################
def _dump_header(self, *, protocol: 'Enum_LinkType | StdlibIntEnum | AenumIntEnum | str | int', # pylint: disable=arguments-differ,too-many-arguments
byteorder: 'Literal["big", "little"]' = sys.byteorder, nanosecond: 'bool' = False,
version: 'tuple[int, int] | VersionInfo' = (2, 4),
thiszone: 'int' = 0, sigfigs: 'int' = 0, snaplen: 'int' = 0x40_000,
**kwargs: 'Any') -> 'None': # pylint: disable=unused-argument
"""Initially dump file heads and tails.
Args:
protocol: data link type
byteorder: header byte order
nanosecond: nanosecond-resolution file flag
version: format version, as ``(major, minor)``
thiszone: GMT to local correction
sigfigs: accuracy of timestamps
snaplen: max length of captured packets, in octets
**kwargs: arbitrary keyword arguments
"""
header = Header(
network=protocol,
byteorder=byteorder,
nanosecond=nanosecond,
version=version,
thiszone=thiszone,
sigfigs=sigfigs,
snaplen=snaplen,
)
packet = header.data
with open(self._file, 'wb') as file:
file.write(packet)
self._ghdr = header.info
#: struct.Struct: Packer for the record header preceding each frame, in the
#: byte order of the global header just written. Taken from
#: :attr:`self._ghdr <_ghdr>` rather than from the ``byteorder`` argument
#: because :class:`~pcapkit.protocols.misc.pcap.header.Header` is what
#: validates and normalises it.
self._rechdr = _RECORD_HEADER[self._ghdr.magic_number.byteorder]
#: int: Units per second of the fraction this file's records carry, and
#: of the one the other resolution would, taken from :attr:`self._ghdr
#: <_ghdr>` for the same reason as :attr:`self._rechdr <_rechdr>`.
self._tsres = RESOLUTION[self._ghdr.magic_number.nanosecond]
self._tsalt = RESOLUTION[not self._ghdr.magic_number.nanosecond]
[docs]
def _append_value(self, value: 'Data_Frame', file: 'IO[bytes]', name: 'str') -> 'None': # pylint: disable=unused-argument
"""Call this function to write contents.
Args:
value: content to be dumped
file: output file
name: name of current content block
Notes:
A PCAP record is a 16-octet header followed by the packet octets, and
both are already in hand: the header fields are ``value.frame_info``,
with the fraction in this file's resolution (see
:meth:`self._make_timestamp <_make_timestamp>`), and the octets are
exactly ``value.packet``. So this writes them directly, rather than
handing them to
:class:`~pcapkit.protocols.misc.pcap.frame.Frame`, whose constructor
packs the record and then **dissects it again** through the whole
protocol stack to arrive at bytes it was given. That round trip
measured about 82% of the cost of a flow-traced extraction -- ``http.pcap``,
1117 frames, best of 7: 2319 ms with the rebuild against 1263 ms
without, over a 1030 ms untraced baseline.
Skipping it is not only cheaper. The re-dissection would re-emit
every parse warning the frame had already produced once, and warn
about payloads it has no business parsing at all -- writing a
3-octet payload would raise ``SchemaWarning: packet length < 0: -3``
from a dumper that only has to copy it.
``incl_len`` is written as ``len(value.packet)``, the octets that
follow it, not as ``value.frame_info.incl_len``. The two are equal
for every record a file holds whole. A record whose captured length
runs past the end of its file, or a PCAP-NG block whose captured
length runs past the block (:issue:`1405`), is *kept as captured*
by its reader: the declared length stays in ``frame_info``, so the
source rebuilds byte-exactly, and ``packet`` is the octets present.
Copying that declared length would claim octets the trace does not
hold, and a reader would run off the end of the file (:issue:`1541`).
"""
# NOTE: The payload is read before the metadata so that a caller passing a
# mapping rather than a dissected frame -- which the flow-tracing adapters
# of several engines do -- still fails naming ``packet``.
# :mod:`pcapkit.foundation.extraction` substitutes a
# dict-capable trace format on the strength of that error.
packet = value.packet
frame_info = value.frame_info
ts_sec, ts_frac = self._make_timestamp(frame_info, value.time_epoch)
file.write(self._rechdr.pack(ts_sec & _UINT32_MASK,
ts_frac & _UINT32_MASK,
len(packet) & _UINT32_MASK,
frame_info.orig_len & _UINT32_MASK) + packet)
self._fnum += 1
[docs]
def _make_timestamp(self, frame_info: 'Data_FrameInfo',
time_epoch: 'decimal.Decimal | float | int') -> 'tuple[int, int]':
"""Record header seconds and fraction, in this file's resolution.
Args:
frame_info: the frame's record header fields
time_epoch: the frame's UNIX timestamp
Returns:
The ``ts_sec`` and ``ts_usec`` fields to write.
Notes:
``frame_info`` does not say which resolution its ``ts_usec`` is in.
A PCAP frame carries its file's, and one converted from PCAP-NG
carries its interface's (cf. :func:`~pcapkit.toolkit.pcapng.block2frame`),
either of which may differ from this file's (:issue:`1500`).
``time_epoch`` settles it: the fraction is in whichever resolution
truncates the frame's offset into its second to exactly that value.
The fraction is written as is when that is this file's resolution,
so a record traced at its own resolution keeps its octets, and also
when neither resolution fits, as for a frame built by hand. When it
is the other resolution, the fraction is recomputed from
``time_epoch`` at this file's -- microseconds times 1000, or
nanoseconds divided by 1000 and **truncated** rather than rounded.
That is how :func:`~pcapkit.toolkit.pcapng.block2frame` and
:meth:`Frame.make <pcapkit.protocols.misc.pcap.frame.Frame.make>`
reduce a timestamp to a fraction, and how libpcap scales a
nanosecond file down (``sf-pcap.c``). It also keeps the second:
999,999,999 ns truncates to 999,999 us, where rounding would reach
a whole second. A recomputed fraction of a second or more is carried
into ``ts_sec``, so that multiplying by 1000 cannot overflow the
field.
"""
ts_sec = frame_info.ts_sec
ts_usec = frame_info.ts_usec
# NOTE: In integers rather than in :class:`decimal.Decimal`, whose
# context precision would round a long fraction, e.g. from an interface
# with a binary ``if_tsresol``.
num, den = decimal.Decimal(time_epoch).as_integer_ratio()
offset = num - ts_sec * den # ``time_epoch - ts_sec``, over ``den``
if (_truncates_to(offset * self._tsres, den, ts_usec)
or not _truncates_to(offset * self._tsalt, den, ts_usec)):
return ts_sec, ts_usec
carry, ts_frac = divmod(offset * self._tsres // den, self._tsres)
return ts_sec + carry, ts_frac