# -*- coding: utf-8 -*-
"""PCAP-NG Support
=====================
.. module:: pcapkit.foundation.engines.pcapng
This module contains the implementation for PCAP-NG file extraction
support, as is used by :class:`pcapkit.foundation.extraction.Extractor`.
"""
import collections
import os
from typing import TYPE_CHECKING, cast
from pcapkit.const.pcapng.block_type import BlockType as Enum_BlockType
from pcapkit.corekit.infoclass import Info, info_final
from pcapkit.corekit.io import SeekableReader
from pcapkit.dumpkit.pcap import PCAPIO
from pcapkit.foundation.engines.engine import EngineBase
from pcapkit.protocols.misc.pcapng import PCAPNG as P_PCAPNG
from pcapkit.protocols.schema.misc.pcapng import (BLOCK_MINIMUM, BYTE_ORDER_MAGIC,
GENERIC_BLOCK_MINIMUM)
from pcapkit.utilities.exceptions import FormatError, stacklevel
from pcapkit.utilities.logging import get_logger
from pcapkit.utilities.warnings import DeprecatedFormatWarning, warn
__all__ = ['PCAPNG']
#: logging.Logger: Module-level logger, a child of the package-wide
#: :data:`pcapkit.utilities.logging.logger`.
logger = get_logger(__name__)
if TYPE_CHECKING:
from typing import Literal, Optional
from pcapkit.foundation.extraction import Extractor
from pcapkit.protocols.data.misc.pcapng import PCAPNG as Data_PCAPNG
from pcapkit.protocols.data.misc.pcapng import CustomBlock as Data_CustomBlock
from pcapkit.protocols.data.misc.pcapng import \
DecryptionSecretsBlock as Data_DecryptionSecretsBlock
from pcapkit.protocols.data.misc.pcapng import \
InterfaceDescriptionBlock as Data_InterfaceDescriptionBlock
from pcapkit.protocols.data.misc.pcapng import \
InterfaceStatisticsBlock as Data_InterfaceStatisticsBlock
from pcapkit.protocols.data.misc.pcapng import NameResolutionBlock as Data_NameResolutionBlock
from pcapkit.protocols.data.misc.pcapng import SectionHeaderBlock as Data_SectionHeaderBlock
from pcapkit.protocols.data.misc.pcapng import \
SystemdJournalExportBlock as Data_SystemdJournalExportBlock
from pcapkit.protocols.data.misc.pcapng import UnknownBlock as Data_UnknownBlock
[docs]
@info_final
class Context(Info):
"""Per-section context of a PCAP-NG file: its section header and the blocks read under it."""
#: Section header.
section: 'Data_SectionHeaderBlock'
def __post_init__(self) -> None:
"""Post initialisation hook."""
self.__update__(
interfaces=[],
#packets=[],
names=[],
journals=[],
secrets=[],
custom=[],
statistics=[],
unknown=[],
)
if TYPE_CHECKING:
#: Interface descriptions.
interfaces: 'list[Data_InterfaceDescriptionBlock]'
#: Packets.
#packets: 'list[Data_PacketBlock | Data_SimplePacketBlock | Data_EnhancedPacketBlock]'
#: Name resolution records.
names: 'list[Data_NameResolutionBlock]'
#: :manpage:`systemd(1)` journal export records.
journals: 'list[Data_SystemdJournalExportBlock]'
#: Decryption secrets.
secrets: 'list[Data_DecryptionSecretsBlock]'
#: Custom blocks.
custom: 'list[Data_CustomBlock]'
#: Interface statistics.
statistics: 'list[Data_InterfaceStatisticsBlock]'
#: Unknown blocks.
unknown: 'list[Data_UnknownBlock]'
def __init__(self, section: 'Data_SectionHeaderBlock') -> 'None': ...
[docs]
class PCAPNG(EngineBase[P_PCAPNG]):
"""PCAP-NG file extraction support.
Args:
extractor: :class:`~pcapkit.foundation.extraction.Extractor` instance.
"""
if TYPE_CHECKING:
#: Current context.
_ctx: 'Context'
#: File context storage.
_ctx_list: 'list[Context]'
#: Number of non-packet blocks of each kind written so far, counted
#: across the whole file rather than per section.
_block_count: 'collections.Counter[str]'
#: The PCAP-NG magic number this engine claims: the Block Type of a Section
#: Header Block, read from its one home,
#: :class:`~pcapkit.const.pcapng.block_type.BlockType` (:issue:`1739`). It
#: reads the same in either byte order.
MAGIC_NUMBER = (
Enum_BlockType.Section_Header_Block.to_bytes(4, 'big'),
)
#: A Section Header Block's Byte-Order Magic, as the octets read, mapped to
#: the byte order it declares for its section; built from the one home of
#: the value, :data:`pcapkit.protocols.schema.misc.pcapng.BYTE_ORDER_MAGIC`
#: (:issue:`1739`).
_BYTE_ORDER_OCTETS = {
BYTE_ORDER_MAGIC.to_bytes(4, 'big'): 'big',
BYTE_ORDER_MAGIC.to_bytes(4, 'little'): 'little',
} # type: dict[bytes, Literal['big', 'little']]
#: Block types that carry a captured packet, mapped to the tag used when
#: reporting them. Parsing any of them resolves the interface the packet was
#: captured on, so the enclosing section must describe at least one
#: interface before such a block can be read at all.
PACKET_BLOCK_TYPES = {
Enum_BlockType.Simple_Packet_Block: 'SPB',
Enum_BlockType.Enhanced_Packet_Block: 'EPB',
Enum_BlockType.Packet_Block: 'Packet',
} # type: dict[int, str]
##########################################################################
# Defaults.
##########################################################################
#: Engine name.
__engine_name__ = 'PCAP-NG'
#: Engine module name.
__engine_module__ = 'pcapkit.protocols.misc.pcapng'
##########################################################################
# Data models.
##########################################################################
def __init__(self, extractor: 'Extractor') -> 'None':
self._ctx = None # type: ignore[assignment]
self._ctx_list = []
self._block_count = collections.Counter()
super().__init__(extractor)
##########################################################################
# Methods.
##########################################################################
[docs]
def run(self) -> 'None':
"""Start extraction.
This method is the entry point for PCAP-NG file extraction. It will
directly extract the first block, which should be a section header
block, and then save the related information into the internal
context storage.
Raises:
FormatError: If the first block is not a section header block, or if
the output format is PCAP.
"""
ext = self._extractor
self._bound_read_ahead()
shb = P_PCAPNG(ext._ifile, num=0, sct=1, ctx=None)
if shb.info.type != Enum_BlockType.Section_Header_Block:
raise FormatError(f'PCAP-NG: [SHB] invalid block type: {shb.info.type!r}')
self._ctx = Context(cast('Data_SectionHeaderBlock', shb.info))
self._ctx_list.append(self._ctx)
shb._ctx = self._ctx
logger.debug('PCAP-NG section %d header read', len(self._ctx_list))
# NOTE: a PCAP file has one link type and one timestamp resolution,
# where a PCAP-NG file declares them per interface, in blocks that follow.
if not ext._flag_q and isinstance(ext._ofile, type) and issubclass(ext._ofile, PCAPIO):
raise FormatError('PCAP-NG: PCAP output is not supported for a PCAP-NG input')
ext._open_output()
self._write_file(shb.info, name=f'Section Header {len(self._ctx_list)}')
[docs]
def read_frame(self) -> 'P_PCAPNG':
"""Read frames.
This method performs the following tasks:
- read the next block from input file;
- check if the block is a packet block;
- if not, save the block into the internal context storage and repeat;
- if yes, save the related information into the internal context storage;
- write the parsed block into output file.
- reassemble IP and/or TCP fragments;
- trace TCP flows if any;
- record frame information if any.
Returns:
Parsed PCAP-NG block.
"""
from pcapkit.toolkit.pcapng import (ipv4_reassembly, ipv6_reassembly, tcp_reassembly,
tcp_traceflow)
ext = self._extractor
while True:
# a packet block resolves the interface it was captured on while it
# is being parsed, so a section that has not described one yet has to
# be rejected before the block is read, not after
self._check_packet_block_context()
# read next block
self._bound_read_ahead()
block = P_PCAPNG(ext._ifile, num=ext._frnum+1, sct=len(self._ctx_list),
ctx=self._ctx, layer=ext._exlyr, protocol=ext._exptl,
__context__=ext._exctx,
__packet__={
'snaplen': self._get_snaplen(),
})
# check block type
if block.info.type == Enum_BlockType.Section_Header_Block:
self._ctx = Context(cast('Data_SectionHeaderBlock', block.info))
self._ctx_list.append(self._ctx)
block._ctx = self._ctx
self._write_file(block.info, name=f'Section Header {len(self._ctx_list)}')
elif block.info.type == Enum_BlockType.Interface_Description_Block:
self._ctx.interfaces.append(cast('Data_InterfaceDescriptionBlock', block.info))
self._write_file(block.info, name=self._block_name('Interface Description'))
elif block.info.type == Enum_BlockType.Name_Resolution_Block:
self._ctx.names.append(cast('Data_NameResolutionBlock', block.info))
self._write_file(block.info, name=self._block_name('Name Resolution'))
elif block.info.type == Enum_BlockType.systemd_Journal_Export_Block:
self._ctx.journals.append(cast('Data_SystemdJournalExportBlock', block.info))
self._write_file(block.info, name=self._block_name('systemd Journal Export'))
elif block.info.type == Enum_BlockType.Decryption_Secrets_Block:
self._ctx.secrets.append(cast('Data_DecryptionSecretsBlock', block.info))
self._write_file(block.info, name=self._block_name('Decryption Secrets'))
elif block.info.type == Enum_BlockType.Interface_Statistics_Block:
# NOTE: The interface ID is bounds-checked while the block is
# parsed, by ``PCAPNG._get_interface``. Re-checking it here
# cannot fire: parsing an ISB resolves its interface to read the
# block's timestamp, so a block that reaches this branch has
# already been validated.
isb_info = cast('Data_InterfaceStatisticsBlock', block.info)
self._ctx.statistics.append(isb_info)
self._write_file(isb_info, name=self._block_name('Interface Statistics'))
elif block.info.type in (Enum_BlockType.Custom_Block_that_rewriters_can_copy_into_new_files,
Enum_BlockType.Custom_Block_that_rewriters_should_not_copy_into_new_files):
self._ctx.custom.append(cast('Data_CustomBlock', block.info))
self._write_file(block.info, name=self._block_name('Custom'))
elif block.info.type == Enum_BlockType.Enhanced_Packet_Block:
# NOTE: as for the ISB above, the interface ID has already been
# bounds-checked by ``PCAPNG._get_interface`` while parsing.
break
elif block.info.type == Enum_BlockType.Simple_Packet_Block:
# an SPB has no interface ID field, so it refers to the interface
# described by the section's first IDB; a section with several
# interfaces is legal and merely means the other interfaces have
# to use an EPB instead (Section 4.4 of the PCAP-NG specification)
if not self._ctx.interfaces:
raise FormatError('PCAP-NG: [SPB] section has no interface description block')
break
elif block.info.type == Enum_BlockType.Packet_Block:
# NOTE: as for the ISB above, the interface ID has already been
# bounds-checked by ``PCAPNG._get_interface`` while parsing.
warn('PCAP-NG: [Packet] deprecated block type', DeprecatedFormatWarning,
stacklevel=stacklevel())
break
else:
self._ctx.unknown.append(cast('Data_UnknownBlock', block.info))
self._write_file(block.info, name=self._block_name('Unknown'))
# increment frame number
ext._frnum += 1
# verbose output
ext._vfunc(ext, block)
# write output
self._write_file(block.info, name=f'Frame {ext._frnum}')
# record fragments
if ext._flag_r:
if ext._ipv4:
data_ipv4 = ipv4_reassembly(block)
if data_ipv4 is not None:
ext._reasm.ipv4(data_ipv4)
if ext._ipv6:
data_ipv6 = ipv6_reassembly(block)
if data_ipv6 is not None:
ext._reasm.ipv6(data_ipv6)
if ext._tcp:
data_tcp = tcp_reassembly(block)
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(block, nanosecond=block.nanosecond)
if data_tf_tcp is not None:
ext._trace.tcp(data_tf_tcp)
# record blocks
if ext._flag_d:
ext._frame.append(block)
# return block record
return block
##########################################################################
# Utilities.
##########################################################################
def _block_name(self, kind: 'str') -> 'str':
"""Name the next non-packet block of the given kind.
The per-section context lists (e.g. ``Context.interfaces``) restart
with every Section Header Block, so numbering a block by their length
gives the first IDB of every section the same name. The output dumpers
key by name, so a repeated name overwrites the earlier block (#1256).
The number is therefore counted across the whole file instead; for a
single-section file it is the same number as before.
Args:
kind: The kind of the block, e.g. ``'Interface Description'``.
Returns:
The block name, ``'<kind> <n>'`` with ``n`` starting at 1.
"""
self._block_count[kind] += 1
return f'{kind} {self._block_count[kind]}'
def _write_file(self, block: 'Data_PCAPNG', *, name: 'str') -> 'None':
"""Write the parsed block into output file.
Args:
block: The parsed block.
name: The name of the block.
"""
ext = self._extractor
if ext._flag_q:
return
if ext._flag_f:
ofile = ext._ofile(f'{ext._ofnm}/{name}.{ext._fext}')
ofile(block, name=name)
else:
ext._ofile(block, name=name)
ofile = ext._ofile
ext._offmt = ofile.kind
def _peek_block_type(self) -> 'Optional[int]':
"""Read the block type of the next block without consuming it.
The block type is the first 32-bit field of every block, read in the
byte order declared by the enclosing section header block (SHB).
Returns:
Block type of the next block, or :obj:`None` if the input file does
not have a whole block type field left to read.
"""
buffer = self._peek(4)
if len(buffer) < 4:
return None
return int.from_bytes(buffer, self._ctx.section.byteorder)
def _peek(self, size: 'int') -> 'bytes':
"""Read the next ``size`` octets of the input without consuming them.
Args:
size: Number of octets wanted.
Returns:
The octets, fewer than ``size`` only at the end of the input.
``peek`` makes at most one read, which a pipe may answer short of ``size``
mid-capture, so a short answer is read again and the position put back
(:issue:`1586`).
"""
ifile = self._extractor._ifile # pylint: disable=protected-access
buffer = ifile.peek(size)[:size]
if len(buffer) < size:
position = ifile.tell()
buffer = ifile.read(size)
ifile.seek(position, os.SEEK_SET)
return buffer
def _bound_read_ahead(self) -> 'None':
"""Have a non-seekable input read ahead to the end of the next block, no further.
A block is measured, and then parsed, from as far as
:meth:`SeekableReader.seek <pcapkit.corekit.io.SeekableReader.seek>` reads
ahead, so the read has to cover all the block's parse reads, however long --
the buffer grows to hold them. A read past that waits, on a live capture, for
the blocks after: a buffer's worth held each block back until that much more
had arrived (:issue:`1586`). A bogus length is read as far as the stream goes,
which is as far as a file read goes too.
How far is enough is the block's Total Length, in the section's byte order
or, for a section header block, the one its Byte-Order Magic declares. Some
kinds of block are parsed past it, and so are read as far as from a file:
one shorter than its own fixed fields, or not a whole number of words,
keeps the buffer's worth; a Decryption Secrets Block reads as far as its
Secrets Length runs.
"""
# pylint: disable=protected-access
ifile = self._extractor._ifile
if not isinstance(ifile, SeekableReader):
return
size = ifile._buffer_size
head = self._peek(12)
if head[:4] == self.MAGIC_NUMBER[0]:
byteorder = self._BYTE_ORDER_OCTETS.get(head[8:12])
else:
byteorder = self._ctx.section.byteorder if self._ctx is not None else None
if byteorder is not None and len(head) >= 8:
block_type = int.from_bytes(head[:4], byteorder)
length = int.from_bytes(head[4:8], byteorder)
if length >= BLOCK_MINIMUM.get(block_type, GENERIC_BLOCK_MINIMUM) and length % 4 == 0:
size = length
if block_type == Enum_BlockType.Decryption_Secrets_Block:
secrets = int.from_bytes(self._peek(16)[12:16], byteorder)
size = max(size, BLOCK_MINIMUM[block_type] + secrets + -secrets % 4)
ifile._read_ahead_size = size
def _check_packet_block_context(self) -> 'None':
"""Reject a packet block in a section that describes no interface.
Every packet block -- Simple Packet Block (SPB), Enhanced Packet Block
(EPB) and the obsolete Packet Block -- resolves the interface it was
captured on while it is being parsed, from the Interface Description
Blocks (IDB) of its section. A section that carries a packet block
before any IDB is therefore unparseable rather than merely invalid, so
the check has to run before the block is read.
Raises:
FormatError: If the next block is a packet block while the current
section describes no interface.
"""
if self._ctx.interfaces:
return
block_type = self._peek_block_type()
if block_type is None:
return
tag = self.PACKET_BLOCK_TYPES.get(block_type)
if tag is not None:
raise FormatError(f'PCAP-NG: [{tag}] section has no interface description block')
def _get_snaplen(self) -> 'int':
"""Get snapshot length from the current context.
This method is used for providing the snapshot length to the ``__packet__``
argument when parsing a Simple Packet Block (SPB).
Notes:
If there is no interface, or interface 0's snaplen is zero, which
means no limit, return ``0xFFFF_FFFF_FFFF_FFFF``. A zero passed
through reads as a limit of no octets, and the block's padding
then becomes packet data (:issue:`1558`).
"""
if self._ctx.interfaces and self._ctx.interfaces[0].snaplen:
return self._ctx.interfaces[0].snaplen
return 0xFFFF_FFFF_FFFF_FFFF