# -*- coding: utf-8 -*-
# mypy: disable-error-code=dict-item
"""PCAP-NG File Format
=========================
.. module:: pcapkit.protocols.misc.pcapng
:mod:`pcapkit.protocols.misc.pcapng` contains
:class:`~pcapkit.protocols.misc.pcapng.PCAPNG`,
which implements an extractor for PCAP-NG file format [*]_, and the
:class:`~pcapkit.protocols.misc.pcapng.PacketDirection`,
:class:`~pcapkit.protocols.misc.pcapng.PacketReception` and
:class:`~pcapkit.protocols.misc.pcapng.WireGuardKeyLabel` enumerations.
.. [*] https://www.ietf.org/archive/id/draft-tuexen-opsawg-pcapng-03.html
.. versionremoved:: 1.5.0
``pcapkit.protocols.misc.pcapng.TLSKeyLabel``, with no alias; use
:class:`pcapkit.const.pcapng.tls_key_label.TLSKeyLabel` instead
(:issue:`886`, :issue:`1744`).
.. versionchanged:: 1.5.0
A block rebuilds as it was parsed, byte for byte: its lengths, padding,
byte order, flag bits and repeated options are kept, and so is a block
that is unknown, malformed, shorter than its fixed fields or cut short
(:issue:`1272`, :issue:`1273`, :issue:`1276`, :issue:`1277`,
:issue:`1278`, :issue:`1279`, :issue:`1280`, :issue:`1281`,
:issue:`1282`, :issue:`1283`, :issue:`1284`, :issue:`1382`,
:issue:`1383`, :issue:`1384`, :issue:`1405`, :issue:`1413`,
:issue:`1414`, :issue:`1422`, :issue:`1470`). A packet block keeps the
octets it captured (:issue:`646`); an option running past its block's
option area leaves the whole area kept as captured in ``options_raw``,
with a :exc:`~pcapkit.utilities.warnings.ProtocolWarning`, where it was
read as zeros (:issue:`594`, :issue:`1325`); a truncated capture does not
raise :exc:`ValueError` (:issue:`678`); and ``timestamp_epoch`` is not
shifted by the host's timezone (:issue:`361`).
"""
import base64
import builtins
import collections
import collections.abc
import datetime
import decimal
import enum
import io
import math
import platform
import re
import struct
import sys
import textwrap
import time
from typing import TYPE_CHECKING, cast, overload
from pcapkit.const.pcapng.block_type import BlockType as Enum_BlockType
from pcapkit.const.pcapng.filter_type import FilterType as Enum_FilterType
from pcapkit.const.pcapng.hash_algorithm import HashAlgorithm as Enum_HashAlgorithm
from pcapkit.const.pcapng.option_type import OptionType as Enum_OptionType
from pcapkit.const.pcapng.record_type import RecordType as Enum_RecordType
from pcapkit.const.pcapng.secrets_type import SecretsType as Enum_SecretsType
from pcapkit.const.pcapng.tls_key_label import TLSKeyLabel as Enum_TLSKeyLabel
from pcapkit.const.pcapng.verdict_type import VerdictType as Enum_VerdictType
from pcapkit.const.reg.linktype import LinkType as Enum_LinkType
from pcapkit.corekit.enum import EnumLookup
from pcapkit.corekit.fields.strings import DecodedString
from pcapkit.corekit.infoclass import OrderedMultiInfo
from pcapkit.corekit.io import advance
from pcapkit.corekit.module import ModuleDescriptor
from pcapkit.corekit.multidict import OrderedMultiDict
from pcapkit.corekit.version import VersionInfo
from pcapkit.protocols.data.misc.pcapng import PCAPNG as Data_PCAPNG
from pcapkit.protocols.data.misc.pcapng import CommentOption as Data_CommentOption
from pcapkit.protocols.data.misc.pcapng import CustomBlock as Data_CustomBlock
from pcapkit.protocols.data.misc.pcapng import CustomOption as Data_CustomOption
from pcapkit.protocols.data.misc.pcapng import DecryptionSecretsBlock as Data_DecryptionSecretsBlock
from pcapkit.protocols.data.misc.pcapng import DSBSecrets as Data_DSBSecrets
from pcapkit.protocols.data.misc.pcapng import EndOfOption as Data_EndOfOption
from pcapkit.protocols.data.misc.pcapng import EndRecord as Data_EndRecord
from pcapkit.protocols.data.misc.pcapng import EnhancedPacketBlock as Data_EnhancedPacketBlock
from pcapkit.protocols.data.misc.pcapng import EPB_DropCountOption as Data_EPB_DropCountOption
from pcapkit.protocols.data.misc.pcapng import EPB_FlagsOption as Data_EPB_FlagsOption
from pcapkit.protocols.data.misc.pcapng import EPB_HashOption as Data_EPB_HashOption
from pcapkit.protocols.data.misc.pcapng import EPB_PacketIDOption as Data_EPB_PacketIDOption
from pcapkit.protocols.data.misc.pcapng import EPB_QueueOption as Data_EPB_QueueOption
from pcapkit.protocols.data.misc.pcapng import EPB_VerdictOption as Data_EPB_VerdictOption
from pcapkit.protocols.data.misc.pcapng import IF_DescriptionOption as Data_IF_DescriptionOption
from pcapkit.protocols.data.misc.pcapng import IF_EUIAddrOption as Data_IF_EUIAddrOption
from pcapkit.protocols.data.misc.pcapng import IF_FCSLenOption as Data_IF_FCSLenOption
from pcapkit.protocols.data.misc.pcapng import IF_FilterOption as Data_IF_FilterOption
from pcapkit.protocols.data.misc.pcapng import IF_HardwareOption as Data_IF_HardwareOption
from pcapkit.protocols.data.misc.pcapng import IF_IPv4AddrOption as Data_IF_IPv4AddrOption
from pcapkit.protocols.data.misc.pcapng import IF_IPv6AddrOption as Data_IF_IPv6AddrOption
from pcapkit.protocols.data.misc.pcapng import IF_MACAddrOption as Data_IF_MACAddrOption
from pcapkit.protocols.data.misc.pcapng import IF_NameOption as Data_IF_NameOption
from pcapkit.protocols.data.misc.pcapng import IF_OSOption as Data_IF_OSOption
from pcapkit.protocols.data.misc.pcapng import IF_RxSpeedOption as Data_IF_RxSpeedOption
from pcapkit.protocols.data.misc.pcapng import IF_SpeedOption as Data_IF_SpeedOption
from pcapkit.protocols.data.misc.pcapng import IF_TSOffsetOption as Data_IF_TSOffsetOption
from pcapkit.protocols.data.misc.pcapng import IF_TSResolOption as Data_IF_TSResolOption
from pcapkit.protocols.data.misc.pcapng import IF_TxSpeedOption as Data_IF_TxSpeedOption
from pcapkit.protocols.data.misc.pcapng import IF_TZoneOption as Data_IF_TZoneOption
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 IPv4Record as Data_IPv4Record
from pcapkit.protocols.data.misc.pcapng import IPv6Record as Data_IPv6Record
from pcapkit.protocols.data.misc.pcapng import ISB_EndTimeOption as Data_ISB_EndTimeOption
from pcapkit.protocols.data.misc.pcapng import ISB_FilterAcceptOption as Data_ISB_FilterAcceptOption
from pcapkit.protocols.data.misc.pcapng import ISB_IFDropOption as Data_ISB_IFDropOption
from pcapkit.protocols.data.misc.pcapng import ISB_IFRecvOption as Data_ISB_IFRecvOption
from pcapkit.protocols.data.misc.pcapng import ISB_OSDropOption as Data_ISB_OSDropOption
from pcapkit.protocols.data.misc.pcapng import ISB_StartTimeOption as Data_ISB_StartTimeOption
from pcapkit.protocols.data.misc.pcapng import ISB_UsrDelivOption as Data_ISB_UsrDelivOption
from pcapkit.protocols.data.misc.pcapng import NameResolutionBlock as Data_NameResolutionBlock
from pcapkit.protocols.data.misc.pcapng import NameResolutionRecord as Data_NameResolutionRecord
from pcapkit.protocols.data.misc.pcapng import NS_DNSIP4AddrOption as Data_NS_DNSIP4AddrOption
from pcapkit.protocols.data.misc.pcapng import NS_DNSIP6AddrOption as Data_NS_DNSIP6AddrOption
from pcapkit.protocols.data.misc.pcapng import NS_DNSNameOption as Data_NS_DNSNameOption
from pcapkit.protocols.data.misc.pcapng import Option as Data_Option
from pcapkit.protocols.data.misc.pcapng import PACK_FlagsOption as Data_PACK_FlagsOption
from pcapkit.protocols.data.misc.pcapng import PACK_HashOption as Data_PACK_HashOption
from pcapkit.protocols.data.misc.pcapng import PacketBlock as Data_PacketBlock
from pcapkit.protocols.data.misc.pcapng import SectionHeaderBlock as Data_SectionHeaderBlock
from pcapkit.protocols.data.misc.pcapng import SimplePacketBlock as Data_SimplePacketBlock
from pcapkit.protocols.data.misc.pcapng import \
SystemdJournalExportBlock as Data_SystemdJournalExportBlock
from pcapkit.protocols.data.misc.pcapng import TLSKeyLog as Data_TLSKeyLog
from pcapkit.protocols.data.misc.pcapng import UnknownBlock as Data_UnknownBlock
from pcapkit.protocols.data.misc.pcapng import UnknownOption as Data_UnknownOption
from pcapkit.protocols.data.misc.pcapng import UnknownRecord as Data_UnknownRecord
from pcapkit.protocols.data.misc.pcapng import UnknownSecrets as Data_UnknownSecrets
from pcapkit.protocols.data.misc.pcapng import WireGuardKeyLog as Data_WireGuardKeyLog
from pcapkit.protocols.data.misc.pcapng import ZigBeeAPSKey as Data_ZigBeeAPSKey
from pcapkit.protocols.data.misc.pcapng import ZigBeeNWKKey as Data_ZigBeeNWKKey
from pcapkit.protocols.data.protocol import Packet as Data_Packet
from pcapkit.protocols.protocol import ProtocolBase
from pcapkit.protocols.schema.misc.pcap.header import RESOLUTION
from pcapkit.protocols.schema.misc.pcapng import (BLOCK_MINIMUM, BYTE_ORDER_MAGIC,
DEFAULT_RESOLUTION, GENERIC_BLOCK_MINIMUM)
from pcapkit.protocols.schema.misc.pcapng import PCAPNG as Schema_PCAPNG
from pcapkit.protocols.schema.misc.pcapng import BlockType as Schema_BlockType
from pcapkit.protocols.schema.misc.pcapng import CommentOption as Schema_CommentOption
from pcapkit.protocols.schema.misc.pcapng import CustomBlock as Schema_CustomBlock
from pcapkit.protocols.schema.misc.pcapng import CustomOption as Schema_CustomOption
from pcapkit.protocols.schema.misc.pcapng import \
DecryptionSecretsBlock as Schema_DecryptionSecretsBlock
from pcapkit.protocols.schema.misc.pcapng import DSBSecrets as Schema_DSBSecrets
from pcapkit.protocols.schema.misc.pcapng import EndOfOption as Schema_EndOfOption
from pcapkit.protocols.schema.misc.pcapng import EndRecord as Schema_EndRecord
from pcapkit.protocols.schema.misc.pcapng import EnhancedPacketBlock as Schema_EnhancedPacketBlock
from pcapkit.protocols.schema.misc.pcapng import EPB_DropCountOption as Schema_EPB_DropCountOption
from pcapkit.protocols.schema.misc.pcapng import EPB_FlagsOption as Schema_EPB_FlagsOption
from pcapkit.protocols.schema.misc.pcapng import EPB_HashOption as Schema_EPB_HashOption
from pcapkit.protocols.schema.misc.pcapng import EPB_PacketIDOption as Schema_EPB_PacketIDOption
from pcapkit.protocols.schema.misc.pcapng import EPB_QueueOption as Schema_EPB_QueueOption
from pcapkit.protocols.schema.misc.pcapng import EPB_VerdictOption as Schema_EPB_VerdictOption
from pcapkit.protocols.schema.misc.pcapng import IF_DescriptionOption as Schema_IF_DescriptionOption
from pcapkit.protocols.schema.misc.pcapng import IF_EUIAddrOption as Schema_IF_EUIAddrOption
from pcapkit.protocols.schema.misc.pcapng import IF_FCSLenOption as Schema_IF_FCSLenOption
from pcapkit.protocols.schema.misc.pcapng import IF_FilterOption as Schema_IF_FilterOption
from pcapkit.protocols.schema.misc.pcapng import IF_HardwareOption as Schema_IF_HardwareOption
from pcapkit.protocols.schema.misc.pcapng import IF_IPv4AddrOption as Schema_IF_IPv4AddrOption
from pcapkit.protocols.schema.misc.pcapng import IF_IPv6AddrOption as Schema_IF_IPv6AddrOption
from pcapkit.protocols.schema.misc.pcapng import IF_MACAddrOption as Schema_IF_MACAddrOption
from pcapkit.protocols.schema.misc.pcapng import IF_NameOption as Schema_IF_NameOption
from pcapkit.protocols.schema.misc.pcapng import IF_OSOption as Schema_IF_OSOption
from pcapkit.protocols.schema.misc.pcapng import IF_RxSpeedOption as Schema_IF_RxSpeedOption
from pcapkit.protocols.schema.misc.pcapng import IF_SpeedOption as Schema_IF_SpeedOption
from pcapkit.protocols.schema.misc.pcapng import IF_TSOffsetOption as Schema_IF_TSOffsetOption
from pcapkit.protocols.schema.misc.pcapng import IF_TSResolOption as Schema_IF_TSResolOption
from pcapkit.protocols.schema.misc.pcapng import IF_TxSpeedOption as Schema_IF_TxSpeedOption
from pcapkit.protocols.schema.misc.pcapng import IF_TZoneOption as Schema_IF_TZoneOption
from pcapkit.protocols.schema.misc.pcapng import \
InterfaceDescriptionBlock as Schema_InterfaceDescriptionBlock
from pcapkit.protocols.schema.misc.pcapng import \
InterfaceStatisticsBlock as Schema_InterfaceStatisticsBlock
from pcapkit.protocols.schema.misc.pcapng import IPv4Record as Schema_IPv4Record
from pcapkit.protocols.schema.misc.pcapng import IPv6Record as Schema_IPv6Record
from pcapkit.protocols.schema.misc.pcapng import ISB_EndTimeOption as Schema_ISB_EndTimeOption
from pcapkit.protocols.schema.misc.pcapng import \
ISB_FilterAcceptOption as Schema_ISB_FilterAcceptOption
from pcapkit.protocols.schema.misc.pcapng import ISB_IFDropOption as Schema_ISB_IFDropOption
from pcapkit.protocols.schema.misc.pcapng import ISB_IFRecvOption as Schema_ISB_IFRecvOption
from pcapkit.protocols.schema.misc.pcapng import ISB_OSDropOption as Schema_ISB_OSDropOption
from pcapkit.protocols.schema.misc.pcapng import ISB_StartTimeOption as Schema_ISB_StartTimeOption
from pcapkit.protocols.schema.misc.pcapng import ISB_UsrDelivOption as Schema_ISB_UsrDelivOption
from pcapkit.protocols.schema.misc.pcapng import NameResolutionBlock as Schema_NameResolutionBlock
from pcapkit.protocols.schema.misc.pcapng import NameResolutionRecord as Schema_NameResolutionRecord
from pcapkit.protocols.schema.misc.pcapng import NS_DNSIP4AddrOption as Schema_NS_DNSIP4AddrOption
from pcapkit.protocols.schema.misc.pcapng import NS_DNSIP6AddrOption as Schema_NS_DNSIP6AddrOption
from pcapkit.protocols.schema.misc.pcapng import NS_DNSNameOption as Schema_NS_DNSNameOption
from pcapkit.protocols.schema.misc.pcapng import PACK_FlagsOption as Schema_PACK_FlagsOption
from pcapkit.protocols.schema.misc.pcapng import PACK_HashOption as Schema_PACK_HashOption
from pcapkit.protocols.schema.misc.pcapng import PacketBlock as Schema_PacketBlock
from pcapkit.protocols.schema.misc.pcapng import SectionHeaderBlock as Schema_SectionHeaderBlock
from pcapkit.protocols.schema.misc.pcapng import SimplePacketBlock as Schema_SimplePacketBlock
from pcapkit.protocols.schema.misc.pcapng import \
SystemdJournalExportBlock as Schema_SystemdJournalExportBlock
from pcapkit.protocols.schema.misc.pcapng import TLSKeyLog as Schema_TLSKeyLog
from pcapkit.protocols.schema.misc.pcapng import UnknownBlock as Schema_UnknownBlock
from pcapkit.protocols.schema.misc.pcapng import UnknownOption as Schema_UnknownOption
from pcapkit.protocols.schema.misc.pcapng import UnknownRecord as Schema_UnknownRecord
from pcapkit.protocols.schema.misc.pcapng import UnknownSecrets as Schema_UnknownSecrets
from pcapkit.protocols.schema.misc.pcapng import WireGuardKeyLog as Schema_WireGuardKeyLog
from pcapkit.protocols.schema.misc.pcapng import ZigBeeAPSKey as Schema_ZigBeeAPSKey
from pcapkit.protocols.schema.misc.pcapng import ZigBeeNWKKey as Schema_ZigBeeNWKKey
from pcapkit.protocols.schema.misc.pcapng import zigbee_key_held
from pcapkit.protocols.schema.schema import Schema
from pcapkit.utilities.compat import StrEnum, localcontext
from pcapkit.utilities.exceptions import (BaseError, EnumValueError, FormatError, ProtocolError,
StreamEOFError, UnsupportedCall, stacklevel)
from pcapkit.utilities.warnings import (AttributeWarning, DeprecatedFormatWarning, ProtocolWarning,
RegistryWarning, warn)
__all__ = ['PCAPNG']
if TYPE_CHECKING:
from datetime import datetime as dt_type
from datetime import timedelta, timezone
from decimal import Decimal
from enum import IntEnum as StdlibEnum
from ipaddress import IPv4Address, IPv4Interface, IPv6Address, IPv6Interface
from typing import (IO, Any, Callable, Counter, DefaultDict, Optional, Tuple, Type, TypeVar,
Union)
from aenum import IntEnum as AenumEnum
from mypy_extensions import DefaultArg, KwArg, NamedArg
from typing_extensions import Literal
from pcapkit.foundation.engines.pcapng import Context
from pcapkit.protocols.schema.misc.pcapng import Option as Schema_Option
from pcapkit.protocols.schema.schema import Schema
Packet = Union[Data_EnhancedPacketBlock, Data_SimplePacketBlock, Data_PacketBlock]
Option = OrderedMultiDict[Enum_OptionType, Data_Option]
Record = OrderedMultiDict[Enum_RecordType, Data_NameResolutionRecord]
BlockParser = Callable[[Schema_BlockType, NamedArg(Schema_PCAPNG, 'header')], Data_PCAPNG]
BlockConstructor = Callable[[DefaultArg(Optional[Data_PCAPNG]),
KwArg(Any)], Schema_BlockType]
OptionParser = Callable[[Schema_Option, NamedArg(Option, 'options')], Data_Option]
OptionConstructor = Callable[[Enum_OptionType, DefaultArg(Optional[Data_Option]),
KwArg(Any)], Schema_Option]
RecordParser = Callable[[Schema_NameResolutionRecord, NamedArg(Record, 'records')], Data_NameResolutionRecord]
RecordConstructor = Callable[[Enum_RecordType, DefaultArg(Optional[Data_NameResolutionRecord]),
KwArg(Any)], Schema_NameResolutionRecord]
SecretsParser = Callable[[Schema_DSBSecrets, NamedArg(Schema_DecryptionSecretsBlock, 'block')], Data_DSBSecrets]
SecretsConstructor = Callable[[Enum_SecretsType, DefaultArg(Optional[Data_DSBSecrets]),
KwArg(Any)], Schema_DSBSecrets]
_FlagsEnum = TypeVar('_FlagsEnum', 'PacketDirection', 'PacketReception')
_S = TypeVar('_S', bound='Schema | bytes')
# check Python version
py37 = ((version_info := sys.version_info).major >= 3 and version_info.minor >= 7)
py38 = ((version_info := sys.version_info).major >= 3 and version_info.minor >= 8)
# Ethernet address pattern
PAT_MAC_ADDR = re.compile(rb'(?i)(?:[0-9a-f]{2}[:-]){5}[0-9a-f]{2}')
# EUI address pattern
PAT_EUI_ADDR = re.compile(rb'(?i)(?:[0-9a-f]{2}[:-]){7}[0-9a-f]{2}')
def _option_key(code: 'Enum_OptionType') -> 'Union[Enum_OptionType, Tuple[str, int]]':
"""Return a collision-free key for PCAP-NG option registries.
Args:
code: The option code.
Returns:
A collision-free key for the option registries.
"""
if code.opt_name:
return (code.opt_name, code.opt_value)
return code
def _utf8_length(text: 'str') -> 'int':
"""Return the number of octets a UTF-8 :class:`~pcapkit.corekit.fields.strings.StringField` packs ``text`` as.
Args:
text: String value; a :class:`~pcapkit.corekit.fields.strings.DecodedString`
packs as the octets it was decoded from.
Returns:
Length of the encoded value, in octets.
"""
if isinstance(text, DecodedString):
return len(text.raw)
return len(text.encode('utf-8'))
def _join_names(names: 'list[str]') -> 'str':
"""Join NRB record names into zero-terminated name resolution data.
Args:
names: Host names; a :class:`~pcapkit.corekit.fields.strings.DecodedString`
contributes the octets it was decoded from.
Returns:
The name resolution data, as a
:class:`~pcapkit.corekit.fields.strings.DecodedString` when any name
carries octets that its text does not encode back to.
"""
names = list(names) or ['']
text = ''.join(f'{name}\x00' for name in names)
raw = b''.join((name.raw if isinstance(name, DecodedString) else name.encode('utf-8')) + b'\x00'
for name in names)
if text.encode('utf-8') == raw:
return text
return DecodedString(text, raw)
def _unassigned_member(enum_cls: 'Type[_FlagsEnum]', value: 'int') -> '_FlagsEnum':
"""Build an unassigned ``epb_flags`` sub-field member, absent from the lookup tables.
Args:
enum_cls: :class:`PacketDirection` or :class:`PacketReception`.
value: The member's value.
Returns:
The unassigned member.
"""
obj = int.__new__(enum_cls, value)
obj._name_ = 'Unassigned' # pylint: disable=protected-access
obj._value_ = value # pylint: disable=protected-access
return obj
[docs]
class PacketDirection(EnumLookup, enum.IntEnum):
"""Packet direction for ``epb_flags`` options.
Shares the :class:`~pcapkit.corekit.enum.EnumLookup` lookup contract that
every non-registry enumeration follows (:issue:`877`). Its
``_missing_`` resolves the one value the specification leaves undefined,
``0b11``, to an unassigned member, so a capture carrying it still parses
and rebuilds (:issue:`1382`).
"""
#: Information not available.
UNKNOWN = 0b00
#: Inbound packet.
INBOUND = 0b01
#: Outbound packet.
OUTBOUND = 0b10
[docs]
@classmethod
def _missing_(cls, value: 'object') -> 'PacketDirection':
"""Lookup function used when value is not found.
Args:
value: Value to get enum item.
Returns:
An unassigned member, absent from the lookup tables.
Raises:
EnumValueError: If ``value`` does not fit the 2-bit sub-field.
"""
if not (isinstance(value, int) and 0b00 <= value <= 0b11):
raise EnumValueError(f'{value!r} is not a valid {cls.__name__}')
return _unassigned_member(cls, value)
[docs]
class PacketReception(EnumLookup, enum.IntEnum):
"""Reception type for ``epb_flags`` options.
Shares the :class:`~pcapkit.corekit.enum.EnumLookup` lookup contract that
every non-registry enumeration follows (:issue:`877`). Its
``_missing_`` resolves the values the specification leaves undefined,
``0b101`` to ``0b111``, to an unassigned member, so a capture carrying one
still parses and rebuilds (:issue:`1382`).
"""
#: Not specified.
UNKNOWN = 0b000
#: Unicast.
UNICAST = 0b001
#: Multicast.
MULTICAST = 0b010
#: Broadcast.
BROADCAST = 0b011
#: Promiscuous.
PROMISCUOUS = 0b100
[docs]
@classmethod
def _missing_(cls, value: 'object') -> 'PacketReception':
"""Lookup function used when value is not found.
Args:
value: Value to get enum item.
Returns:
An unassigned member, absent from the lookup tables.
Raises:
EnumValueError: If ``value`` does not fit the 3-bit sub-field.
"""
if not (isinstance(value, int) and 0b000 <= value <= 0b111):
raise EnumValueError(f'{value!r} is not a valid {cls.__name__}')
return _unassigned_member(cls, value)
# NOTE: RFC 9850 :rfc:`9850#section-4.2` makes the TLS key labels an IANA
# registry ("TLS SSLKEYLOGFILE Labels", Specification Required) rather than a
# closed, hand-picked set, so the TLS key label enumeration is the generated
# :class:`pcapkit.const.pcapng.tls_key_label.TLSKeyLabel`, imported above as
# ``Enum_TLSKeyLabel`` like its :mod:`pcapkit.const.pcapng` siblings. This
# module does not re-export it under the unaliased name (#1744).
# :class:`WireGuardKeyLabel` below stays hand-written: it is a closed set
# (draft-ietf-opsawg-pcapng-06 section 4.7's "is one of" four names), not a
# registry, so filing it under :mod:`pcapkit.const` would misrepresent it as
# one. The ``Enum_*`` alias convention covers imports of const enums, so it does
# not apply to a local definition.
[docs]
class WireGuardKeyLabel(EnumLookup, StrEnum):
"""WireGuard key log label.
Shares the :class:`~pcapkit.corekit.enum.EnumLookup` lookup contract that
every non-registry enumeration follows (:issue:`877`); it
defines neither ``get`` nor ``_missing_`` of its own.
"""
LOCAL_STATIC_PRIVATE_KEY = 'LOCAL_STATIC_PRIVATE_KEY'
REMOTE_STATIC_PUBLIC_KEY = 'REMOTE_STATIC_PUBLIC_KEY'
LOCAL_EPHEMERAL_PRIVATE_KEY = 'LOCAL_EPHEMERAL_PRIVATE_KEY'
PRESHARED_KEY = 'PRESHARED_KEY'
[docs]
class PCAPNG(ProtocolBase[Data_PCAPNG, Schema_PCAPNG],
schema=Schema_PCAPNG, data=Data_PCAPNG):
"""PCAP-NG file block extractor.
The class supports parsing of the following protocols, which are
registered in the :attr:`self.__proto__ <pcapkit.protocols.misc.pcapng.PCAPNG.__proto__>`
attribute:
.. list-table::
:header-rows: 1
* - Index
- Protocol
* - :attr:`pcapkit.const.reg.linktype.LinkType.NULL`
- :class:`pcapkit.protocols.link.loopback.Loopback`
* - :attr:`pcapkit.const.reg.linktype.LinkType.ETHERNET`
- :class:`pcapkit.protocols.link.ethernet.Ethernet`
* - :attr:`pcapkit.const.reg.linktype.LinkType.LOOP`
- :class:`pcapkit.protocols.link.loopback.Loopback`
* - :attr:`pcapkit.const.reg.linktype.LinkType.IPV4`
- :class:`pcapkit.protocols.internet.ipv4.IPv4`
* - :attr:`pcapkit.const.reg.linktype.LinkType.IPV6`
- :class:`pcapkit.protocols.internet.ipv6.IPv6`
:attr:`pcapkit.const.reg.linktype.LinkType.RAW` has no entry of its own:
its data is dispatched as ``IPV4`` or ``IPV6`` by the version nibble of
its first octet, see :meth:`_dispatch_linktype`.
The class supports parsing of the following block types, which
are registered in the :attr:`self.__block__ <pcapkit.protocols.misc.pcapng.PCAPNG.__block__>`
attribute:
.. list-table::
:header-rows: 1
* - Block Type
- Block Parser
- Block Constructor
* - :attr:`~pcapkit.const.pcapng.block_type.BlockType.Section_Header_Block`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_block_shb`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_block_shb`
* - :attr:`~pcapkit.const.pcapng.block_type.BlockType.Interface_Description_Block`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_block_idb`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_block_idb`
* - :attr:`~pcapkit.const.pcapng.block_type.BlockType.Enhanced_Packet_Block`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_block_epb`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_block_epb`
* - :attr:`~pcapkit.const.pcapng.block_type.BlockType.Simple_Packet_Block`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_block_spb`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_block_spb`
* - :attr:`~pcapkit.const.pcapng.block_type.BlockType.Name_Resolution_Block`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_block_nrb`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_block_nrb`
* - :attr:`~pcapkit.const.pcapng.block_type.BlockType.Interface_Statistics_Block`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_block_isb`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_block_isb`
* - :attr:`~pcapkit.const.pcapng.block_type.BlockType.systemd_Journal_Export_Block`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_block_systemd`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_block_systemd`
* - :attr:`~pcapkit.const.pcapng.block_type.BlockType.Decryption_Secrets_Block`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_block_dsb`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_block_dsb`
* - :attr:`~pcapkit.const.pcapng.block_type.BlockType.Custom_Block_that_rewriters_can_copy_into_new_files`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_block_cb`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_block_cb`
* - :attr:`~pcapkit.const.pcapng.block_type.BlockType.Custom_Block_that_rewriters_should_not_copy_into_new_files`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_block_cb`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_block_cb`
* - :attr:`~pcapkit.const.pcapng.block_type.BlockType.Packet_Block`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_block_packet`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_block_packet`
The class supports parsing of the following option types, which
are registered in the :attr:`self.__option__ <pcapkit.protocols.misc.pcapng.PCAPNG.__option__>`
attribute:
.. list-table::
:header-rows: 1
* - Option Type
- Option Parser
- Option Constructor
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.opt_endofopt`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_endofopt`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_endofopt`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.opt_comment`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_comment`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_comment`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.opt_custom_2988`,
:attr:`~pcapkit.const.pcapng.option_type.OptionType.opt_custom_2989`,
:attr:`~pcapkit.const.pcapng.option_type.OptionType.opt_custom_19372`,
:attr:`~pcapkit.const.pcapng.option_type.OptionType.opt_custom_19373`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_custom`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_custom`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_name`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_if_name`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_if_name`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_description`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_if_description`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_if_description`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_IPv4addr`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_if_ipv4`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_if_ipv4`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_IPv6addr`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_if_ipv6`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_if_ipv6`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_MACaddr`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_if_mac`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_if_mac`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_EUIaddr`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_if_eui`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_if_eui`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_speed`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_if_speed`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_if_speed`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_tsresol`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_if_tsresol`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_if_tsresol`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_tzone`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_if_tzone`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_if_tzone`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_filter`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_if_filter`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_if_filter`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_os`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_if_os`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_if_os`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_fcslen`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_if_fcslen`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_if_fcslen`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_tsoffset`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_if_tsoffset`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_if_tsoffset`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_hardware`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_if_hardware`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_if_hardware`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_txspeed`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_if_txspeed`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_if_txspeed`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_rxspeed`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_if_rxspeed`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_if_rxspeed`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.epb_flags`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_epb_flags`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_epb_flags`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.epb_hash`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_epb_hash`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_epb_hash`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.epb_dropcount`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_epb_dropcount`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_epb_dropcount`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.epb_packetid`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_epb_packetid`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_epb_packetid`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.epb_queue`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_epb_queue`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_epb_queue`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.epb_verdict`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_epb_verdict`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_epb_verdict`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.ns_dnsname`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_ns_dnsname`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_ns_dnsname`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.ns_dnsIP4addr`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_ns_dnsipv4`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_ns_dnsipv4`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.ns_dnsIP6addr`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_ns_dnsipv6`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_ns_dnsipv6`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.isb_starttime`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_isb_starttime`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_isb_starttime`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.isb_endtime`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_isb_endtime`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_isb_endtime`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.isb_ifrecv`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_isb_ifrecv`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_isb_ifrecv`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.isb_ifdrop`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_isb_ifdrop`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_isb_ifdrop`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.isb_filteraccept`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_isb_filteraccept`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_isb_filteraccept`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.isb_osdrop`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_isb_osdrop`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_isb_osdrop`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.isb_usrdeliv`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_isb_usrdeliv`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_isb_usrdeliv`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.pack_flags`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_pack_flags`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_pack_flags`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.pack_hash`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_option_pack_hash`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_option_pack_hash`
The class supports parsing of the following name resolution
record types, which are registered in the :attr:`self.__record__ <pcapkit.protocols.misc.pcapng.PCAPNG.__record__>`
attribute:
.. list-table::
:header-rows: 1
* - Record Type
- Record Parser
- Record Constructor
* - :attr:`~pcapkit.const.pcapng.record_type.RecordType.nrb_record_end`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_record_end`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_record_end`
* - :attr:`~pcapkit.const.pcapng.record_type.RecordType.nrb_record_ipv4`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_record_ipv4`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_record_ipv4`
* - :attr:`~pcapkit.const.pcapng.record_type.RecordType.nrb_record_ipv6`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_record_ipv6`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_record_ipv6`
The class supports parsing of the following decryption secrets
types, which are registered in the :attr:`self.__secrets__ <pcapkit.protocols.misc.pcapng.PCAPNG.__secrets__>`
attribute:
.. list-table::
:header-rows: 1
* - Secrets Type
- Secrets Parser
- Secrets Constructor
* - :attr:`~pcapkit.const.pcapng.secrets_type.SecretsType.TLS_Key_Log`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_secrets_tls`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_secrets_tls`
* - :attr:`~pcapkit.const.pcapng.secrets_type.SecretsType.WireGuard_Key_Log`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_secrets_wireguard`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_secrets_wireguard`
* - :attr:`~pcapkit.const.pcapng.secrets_type.SecretsType.ZigBee_NWK_Key`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_secrets_zigbee_nwk`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_secrets_zigbee_nwk`
* - :attr:`~pcapkit.const.pcapng.secrets_type.SecretsType.ZigBee_APS_Key`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._read_secrets_zigbee_aps`
- :meth:`~pcapkit.protocols.misc.pcapng.PCAPNG._make_secrets_zigbee_aps`
"""
if TYPE_CHECKING:
#: PCAP-NG context manager.
_ctx: 'Optional[Context]'
#: PCAP-NG block type.
_type: Enum_BlockType
#: PCAP-NG block byte order.
_byte: 'Literal["little", "big"]'
#: Blocks that carry a captured packet.
PACKET_TYPES = (Enum_BlockType.Enhanced_Packet_Block,
Enum_BlockType.Simple_Packet_Block,
Enum_BlockType.Packet_Block)
#: Packet blocks whose fixed fields span 32 octets, so that a shorter Block
#: Total Length is kept as captured (see :meth:`_read_block_short`).
SHORT_PACKET_TYPES = (Enum_BlockType.Enhanced_Packet_Block,
Enum_BlockType.Packet_Block)
#: Header blocks mapped to the octets their fixed fields span, so that a
#: shorter Block Total Length is kept as captured (see
#: :meth:`_read_block_short_header`). The lengths are read from their one
#: home, :data:`pcapkit.protocols.schema.misc.pcapng.BLOCK_MINIMUM`
#: (:issue:`1739`).
SHORT_HEADER_TYPES = {
block: BLOCK_MINIMUM[block] for block in (Enum_BlockType.Section_Header_Block,
Enum_BlockType.Interface_Description_Block)
}
#: Blocks that resolve an interface of their section, mapped to the tag used
#: when reporting an interface ID that names no such interface. A Simple
#: Packet Block carries no interface ID field and always refers to the
#: section's first interface, so it is listed here too.
INTERFACE_ID_BLOCK_TAGS = {
Enum_BlockType.Enhanced_Packet_Block: 'EPB',
Enum_BlockType.Simple_Packet_Block: 'SPB',
Enum_BlockType.Packet_Block: 'Packet',
Enum_BlockType.Interface_Statistics_Block: 'ISB',
} # type: dict[int, str]
##########################################################################
# Defaults.
##########################################################################
#: Protocol index mapping for decoding the next layer; see
#: :meth:`self._decode_next_layer <pcapkit.protocols.protocol.Protocol._decode_next_layer>`
#: and :meth:`self._import_next_layer <pcapkit.protocols.protocol.Protocol._import_next_layer>`.
#: Each value is a :class:`~pcapkit.corekit.module.ModuleDescriptor` naming the
#: module and class, or a :class:`~pcapkit.protocols.protocol.Protocol` subclass.
__proto__ = collections.defaultdict(
lambda: ModuleDescriptor('pcapkit.protocols.misc.raw', 'Raw'),
{
Enum_LinkType.NULL: ModuleDescriptor('pcapkit.protocols.link', 'Loopback'),
Enum_LinkType.ETHERNET: ModuleDescriptor('pcapkit.protocols.link', 'Ethernet'),
Enum_LinkType.LOOP: ModuleDescriptor('pcapkit.protocols.link', 'Loopback'),
Enum_LinkType.IPV4: ModuleDescriptor('pcapkit.protocols.internet', 'IPv4'),
Enum_LinkType.IPV6: ModuleDescriptor('pcapkit.protocols.internet', 'IPv6'),
},
) # type: DefaultDict[Enum_LinkType | int, ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]]
#: Block type to method mapping. A :obj:`str` value names the
#: ``_read_block_${name}`` and ``_make_block_${name}`` methods, which fall
#: back to ``_read_block_unknown`` and ``_make_block_unknown`` when not
#: defined, as an unregistered code does; any other value is a
#: ``(parser, constructor)`` pair of callables.
__block__ = collections.defaultdict(
lambda: 'unknown',
{
Enum_BlockType.Section_Header_Block: 'shb',
Enum_BlockType.Interface_Description_Block: 'idb',
Enum_BlockType.Enhanced_Packet_Block: 'epb',
Enum_BlockType.Simple_Packet_Block: 'spb',
Enum_BlockType.Name_Resolution_Block: 'nrb',
Enum_BlockType.Interface_Statistics_Block: 'isb',
Enum_BlockType.systemd_Journal_Export_Block: 'systemd',
Enum_BlockType.Decryption_Secrets_Block: 'dsb',
Enum_BlockType.Custom_Block_that_rewriters_can_copy_into_new_files: 'cb',
Enum_BlockType.Custom_Block_that_rewriters_should_not_copy_into_new_files: 'cb',
Enum_BlockType.Packet_Block: 'packet',
},
) # type: DefaultDict[Enum_BlockType | int, str | tuple[BlockParser, BlockConstructor]]
#: Block option type to method mapping. A :obj:`str` value names the
#: ``_read_option_${name}`` and ``_make_option_${name}`` methods, which fall
#: back to ``_read_option_unknown`` and ``_make_option_unknown`` when not
#: defined, as an unregistered code does; any other value is a
#: ``(parser, constructor)`` pair of callables.
__option__ = collections.defaultdict(
lambda: 'unknown',
{
_option_key(Enum_OptionType.opt_endofopt): 'endofopt',
_option_key(Enum_OptionType.opt_comment): 'comment',
_option_key(Enum_OptionType.opt_custom_2988): 'custom',
_option_key(Enum_OptionType.opt_custom_2989): 'custom',
_option_key(Enum_OptionType.opt_custom_19372): 'custom',
_option_key(Enum_OptionType.opt_custom_19373): 'custom',
_option_key(Enum_OptionType.if_name): 'if_name',
_option_key(Enum_OptionType.if_description): 'if_description',
_option_key(Enum_OptionType.if_IPv4addr): 'if_ipv4',
_option_key(Enum_OptionType.if_IPv6addr): 'if_ipv6',
_option_key(Enum_OptionType.if_MACaddr): 'if_mac',
_option_key(Enum_OptionType.if_EUIaddr): 'if_eui',
_option_key(Enum_OptionType.if_speed): 'if_speed',
_option_key(Enum_OptionType.if_tsresol): 'if_tsresol',
_option_key(Enum_OptionType.if_tzone): 'if_tzone',
_option_key(Enum_OptionType.if_filter): 'if_filter',
_option_key(Enum_OptionType.if_os): 'if_os',
_option_key(Enum_OptionType.if_fcslen): 'if_fcslen',
_option_key(Enum_OptionType.if_tsoffset): 'if_tsoffset',
_option_key(Enum_OptionType.if_hardware): 'if_hardware',
_option_key(Enum_OptionType.if_txspeed): 'if_txspeed',
_option_key(Enum_OptionType.if_rxspeed): 'if_rxspeed',
_option_key(Enum_OptionType.epb_flags): 'epb_flags',
_option_key(Enum_OptionType.epb_hash): 'epb_hash',
_option_key(Enum_OptionType.epb_dropcount): 'epb_dropcount',
_option_key(Enum_OptionType.epb_packetid): 'epb_packetid',
_option_key(Enum_OptionType.epb_queue): 'epb_queue',
_option_key(Enum_OptionType.epb_verdict): 'epb_verdict',
_option_key(Enum_OptionType.ns_dnsname): 'ns_dnsname',
_option_key(Enum_OptionType.ns_dnsIP4addr): 'ns_dnsipv4',
_option_key(Enum_OptionType.ns_dnsIP6addr): 'ns_dnsipv6',
_option_key(Enum_OptionType.isb_starttime): 'isb_starttime',
_option_key(Enum_OptionType.isb_endtime): 'isb_endtime',
_option_key(Enum_OptionType.isb_ifrecv): 'isb_ifrecv',
_option_key(Enum_OptionType.isb_ifdrop): 'isb_ifdrop',
_option_key(Enum_OptionType.isb_filteraccept): 'isb_filteraccept',
_option_key(Enum_OptionType.isb_osdrop): 'isb_osdrop',
_option_key(Enum_OptionType.isb_usrdeliv): 'isb_usrdeliv',
_option_key(Enum_OptionType.pack_flags): 'pack_flags',
_option_key(Enum_OptionType.pack_hash): 'pack_hash',
},
) # type: DefaultDict[Union[Enum_OptionType, Tuple[str, int]], str | tuple[OptionParser, OptionConstructor]]
#: Name Resolution Block record type to method mapping. A :obj:`str` value
#: names the ``_read_record_${name}`` and ``_make_record_${name}`` methods,
#: which fall back to ``_read_record_unknown`` and ``_make_record_unknown``
#: when not defined, as an unregistered code does; any other value is a
#: ``(parser, constructor)`` pair of callables.
__record__ = collections.defaultdict(
lambda: 'unknown',
{
Enum_RecordType.nrb_record_end: 'end',
Enum_RecordType.nrb_record_ipv4: 'ipv4',
Enum_RecordType.nrb_record_ipv6: 'ipv6',
},
) # type: DefaultDict[Enum_RecordType | int, str | tuple[RecordParser, RecordConstructor]]
#: Decryption secrets type to method mapping. A :obj:`str` value names the
#: ``_read_secrets_${name}`` and ``_make_secrets_${name}`` methods, which
#: fall back to ``_read_secrets_unknown`` and ``_make_secrets_unknown`` when
#: not defined, as an unregistered code does; any other value is a
#: ``(parser, constructor)`` pair of callables.
__secrets__ = collections.defaultdict(
lambda: 'unknown',
{
Enum_SecretsType.TLS_Key_Log: 'tls',
Enum_SecretsType.WireGuard_Key_Log: 'wireguard',
Enum_SecretsType.ZigBee_NWK_Key: 'zigbee_nwk',
Enum_SecretsType.ZigBee_APS_Key: 'zigbee_aps',
},
) # type: DefaultDict[Enum_SecretsType | int, str | tuple[SecretsParser, SecretsConstructor]]
##########################################################################
# Properties.
##########################################################################
@property
def name(self) -> 'str':
"""Name of corresponding protocol."""
if self._info.type not in self.PACKET_TYPES:
return f'PCAP-NG {self._info.type!r} - Section {self._sect}'
return f'Frame {self._sect}-{self._fnum}'
@property
def length(self) -> 'int':
"""Block total length of corresponding protocol.
Note:
This is the wire's *Block Total Length* -- the whole block, trailing
length field included -- and not a header length. The two are the
same thing for a protocol whose payload runs to the end of its
buffer, which is why
:attr:`ProtocolBase.length <pcapkit.protocols.protocol.ProtocolBase.length>`
does not distinguish them, but a PCAP-NG block carries a trailer.
:attr:`self.packet <packet>` is overridden accordingly.
"""
return self._info.length
# NOTE: A plain property, where the inherited one is a
# :func:`~pcapkit.utilities.compat.cached_property`. That is deliberate: the
# inherited one caches because it *reads the stream*, and a second read would
# consume octets that are no longer there, whereas this one only walks buffers
# the schema layer has already filled, at the cost of a handful of dict
# lookups.
#
# Caching it would make a second ``unpack`` on the same instance return the
# *first* call's octets, because :meth:`self.unpack <unpack>` reports the
# payload through this property and ``get_payload`` would never be reached.
# Nothing in the tree calls ``unpack`` twice on one instance
# (``__post_init__`` is its only caller), so this keeps an invariant rather
# than fixing a bug. A data descriptor also wins over ``__dict__``, so a
# stale entry left by the inherited ``cached_property`` cannot shadow this
# one either.
@property
def packet(self) -> 'Data_Packet':
"""Header and payload octets of the current block.
A PCAP-NG block is not a header followed by a payload. The captured
octets sit in the *middle* of a packet block, ahead of the option list
and the trailing Block Total Length, so no single split point expresses
the shape and the inherited
:attr:`ProtocolBase.packet <pcapkit.protocols.protocol.ProtocolBase.packet>`
-- which reads :attr:`self.length <length>` octets of header and takes
everything after as payload -- cannot produce it. Since
:attr:`self.length <length>` is the Block Total Length, that split would
put the entire block in the header and leave the payload empty, so
``packet`` would be ``b''`` while ``captured_len`` declared the octets.
The payload is therefore the block schema's
:attr:`~pcapkit.protocols.schema.schema.Schema.__payload__` field, which
is where the schema layer put the captured octets, and the header is the
block octets ahead of it. The block types that carry captured
octets -- ``PACKET_TYPES``, i.e. the Enhanced
Packet Block, the Simple Packet Block and the obsolete Packet Block --
each place the payload at a different offset, so the offset is summed
from the octets the schema actually unpacked rather than hard-coded per
block type. A block declaring no payload field is all header and no
payload.
"""
block = self.__header__.block
payload_name = self._get_payload_name(block)
if payload_name is None:
return Data_Packet(header=self._data, payload=b'')
# The octets ahead of the payload field, in wire order: the outer
# schema's fields up to the one holding the block -- its 4-octet Block
# Type -- and then the block's own fields up to the payload. The block is
# found by identity rather than by name so that renaming the field
# cannot silently fold the whole block body into the header; if it is not
# found at all, the header loses the Block Type and nothing else, and the
# payload stays exact either way.
names = list(self.__header__.__fields__)
stop = next((idx for idx, name in enumerate(names)
if getattr(self.__header__, name, None) is block), 0)
block_names = list(block.__fields__)
# Safe: ``_get_payload_name`` returned a name only because it is a field.
block_stop = block_names.index(payload_name)
header = bytearray()
for name in names[:stop]:
header += self.__header__.__buffer__.get(name, b'')
for name in block_names[:block_stop]:
header += block.__buffer__.get(name, b'')
return Data_Packet(
header=bytes(header),
payload=block.get_payload(payload_name),
)
@property
def context(self) -> 'Context':
"""Context of current PCAP-NG block."""
if self._ctx is None:
raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute 'context'")
return self._ctx
@property
def byteorder(self) -> 'Literal["big", "little"]':
"""Byte order of the current block."""
return self._byte
@property
def nanosecond(self) -> 'bool':
"""Whether the timestamp is in nanoseconds."""
return self.ts_resolution > RESOLUTION[False]
@property
def ts_resolution(self) -> 'int':
"""Timestamp resolution of the current block, in units per second."""
if self._ctx is None:
#raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute 'ts_resolution'")
warn(f"'{self.__class__.__name__}' object has no attribute 'ts_resolution'",
AttributeWarning, stacklevel=stacklevel())
return DEFAULT_RESOLUTION
info = cast('Packet', self._info)
return self._get_resolution(info.interface_id)
@property
def ts_offset(self) -> 'int':
"""Timestamp offset of the current block, in seconds."""
if self._ctx is None:
#raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute 'ts_offset'")
warn(f"'{self.__class__.__name__}' object has no attribute 'ts_offset'",
AttributeWarning, stacklevel=stacklevel())
return 0
info = cast('Packet', self._info)
return self._get_offset(info.interface_id)
@property
def ts_timezone(self) -> 'timezone':
"""Timezone of the current block.
Defaults to UTC when the capture names no ``if_tzone``, for the reason
given in :meth:`self._get_timezone <_get_timezone>`.
"""
if self._ctx is None:
#raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute 'ts_timezone'")
warn(f"'{self.__class__.__name__}' object has no attribute 'ts_timezone'",
AttributeWarning, stacklevel=stacklevel())
return datetime.timezone.utc
info = cast('Packet', self._info)
return self._get_timezone(info.interface_id)
@property
def linktype(self) -> 'Enum_LinkType':
"""Data link layer protocol type.
Raises:
UnsupportedCall: If current block is not a valid packet block, i.e.,
EPB, SPB or obsolete Packet Block.
"""
if self._ctx is None or self._type not in self.PACKET_TYPES:
raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute 'linktype'")
info = cast('Packet', self._info)
return self._get_linktype(info.interface_id)
@property
def block(self) -> 'Enum_BlockType':
"""PCAP-NG block type."""
return self._type
##########################################################################
# Methods.
##########################################################################
[docs]
@classmethod
def register(cls, code: 'Enum_LinkType', protocol: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]') -> 'None': # type: ignore[override]
r"""Register a new protocol class.
Notes:
The fully qualified class name of the new protocol class
should be ``{protocol.module}.{protocol.name}``.
Arguments:
code: protocol code as in :class:`~pcapkit.const.reg.linktype.LinkType`
protocol: module descriptor or a
:class:`~pcapkit.protocols.protocol.Protocol` subclass
Raises:
pcapkit.utilities.exceptions.RegistryError: If ``protocol`` is not a
class, or not a :class:`~pcapkit.protocols.protocol.Protocol` subclass.
Warns:
pcapkit.utilities.warnings.RegistryWarning: If this link type is
already registered against PCAP-NG blocks, naming the displaced
entry and its replacement so a caller can tell *what* was lost.
Fires only when the incumbent differs from the replacement, and
an unresolved module descriptor counts as different from the
class it names; handing back the stored entry, or a descriptor
equal to it, is a silent no-op that keeps it. This registry is separate from the PCAP one, so
:func:`~pcapkit.foundation.registry.protocols.register_linktype`
writing to both cannot make either warn about the other. See
:meth:`ProtocolBase.register
<pcapkit.protocols.protocol.ProtocolBase.register>` for the
shipped descriptor a restore puts back (:issue:`1504`).
"""
incumbent = cls.__proto__.get(code)
entry = cls._next_layer_entry(code, protocol)
if entry is None:
return
if incumbent is not None and incumbent is not entry:
warn(f'protocol {code} already registered, overwriting '
f'{incumbent!r} with {entry!r}', RegistryWarning)
cls.__proto__[code] = entry
[docs]
@classmethod
def register_block(cls, code: 'Enum_BlockType', meth: 'str | tuple[BlockParser, BlockConstructor]') -> 'None':
"""Register a block parser.
Args:
code: PCAP-NG block type code.
meth: Method name or callable to parse and/or construct the block.
Warns:
RegistryWarning: If a different entry is already registered for
``code``; it is overwritten. Handing back the entry already
stored, or one equal to it, is a silent no-op. The entry
``code`` shipped with is accepted like any other, so an
override can be undone (:issue:`1362`, :issue:`1363`).
"""
incumbent = cls.__block__.get(code)
if incumbent is not None and incumbent == meth:
return
if incumbent is not None:
warn(f'PCAP-NG: [Type {code}] block already registered', RegistryWarning)
cls.__block__[code] = meth
[docs]
@classmethod
def register_option(cls, code: 'Enum_OptionType', meth: 'str | tuple[OptionParser, OptionConstructor]') -> 'None':
"""Register an option parser.
Args:
code: PCAP-NG option type code.
meth: Method name or callable to parse and/or construct the option.
Warns:
RegistryWarning: If a different entry is already registered for
``code``; it is overwritten. Handing back the entry already
stored, or one equal to it, is a silent no-op. The entry
``code`` shipped with is accepted like any other, so an
override can be undone (:issue:`1362`, :issue:`1363`).
"""
code_key = _option_key(code)
incumbent = cls.__option__.get(code_key)
if incumbent is not None and incumbent == meth:
return
if incumbent is not None:
warn(f'PCAP-NG: [Option {code}] option already registered', RegistryWarning)
cls.__option__[code_key] = meth
[docs]
@classmethod
def register_record(cls, code: 'Enum_RecordType', meth: 'str | tuple[RecordParser, RecordConstructor]') -> 'None':
"""Register a name resolution record parser.
Args:
code: PCAP-NG name resolution record type code.
meth: Method name or callable to parse and/or construct the name resolution record.
Warns:
RegistryWarning: If a different entry is already registered for
``code``; it is overwritten. Handing back the entry already
stored, or one equal to it, is a silent no-op. The entry
``code`` shipped with is accepted like any other, so an
override can be undone (:issue:`1362`, :issue:`1363`).
"""
incumbent = cls.__record__.get(code)
if incumbent is not None and incumbent == meth:
return
if incumbent is not None:
warn(f'PCAP-NG: [Type {code}] name resolution record already registered', RegistryWarning)
cls.__record__[code] = meth
[docs]
@classmethod
def register_secrets(cls, code: 'Enum_SecretsType', meth: 'str | tuple[SecretsParser, SecretsConstructor]') -> 'None':
"""Register a decryption secrets parser.
Args:
code: PCAP-NG decryption secrets type code.
meth: Method name or callable to parse and/or construct the decryption secrets.
Warns:
RegistryWarning: If a different entry is already registered for
``code``; it is overwritten. Handing back the entry already
stored, or one equal to it, is a silent no-op. The entry
``code`` shipped with is accepted like any other, so an
override can be undone (:issue:`1362`, :issue:`1363`).
"""
incumbent = cls.__secrets__.get(code)
if incumbent is not None and incumbent == meth:
return
if incumbent is not None:
warn(f'PCAP-NG: [Secrets {code}] decryption secrets already registered', RegistryWarning)
cls.__secrets__[code] = meth
[docs]
def index(self, name: 'str | ProtocolBase | Type[ProtocolBase]') -> 'int':
"""Call :meth:`ProtoChain.index <pcapkit.corekit.protochain.ProtoChain.index>`.
Args:
name: ``name`` to be searched
Returns:
First index of ``name``.
Raises:
IndexNotFound: if ``name`` is not present
"""
return self._protos.index(name)
[docs]
def pack(self, **kwargs: 'Any') -> 'bytes':
"""Pack (construct) packet data.
Args:
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed packet data.
Notes:
We used a special keyword argument ``__packet__`` to pass the
global packet data to underlying methods. This is useful when
the packet data is not available in the current instance.
As in :meth:`Protocol.pack <pcapkit.protocols.protocol.Protocol.pack>`,
the header is made on a copy outside construction, so the instance
keeps its own header, block type, byte order and option counts
(:issue:`1769`).
"""
# pylint: disable=protected-access
target = self._packing_target()
if target is not self:
# NOTE: The copy would share the instance's counter, which
# :meth:`_make_pcapng_options` increments in place.
target._opt = collections.Counter()
target.__header__ = target.make(**kwargs)
packet = kwargs.get('__packet__', {}) # packet data
if target._ctx is not None:
target._byte = target._ctx.section.byteorder
packet['byteorder'] = target._byte
elif target._type == Enum_BlockType.Section_Header_Block:
# NOTE: A section header starts its own section, so the byte order
# is the one :meth:`_make_block_shb` chose rather than the host's.
packet.setdefault('byteorder', target._byte)
return target.__header__.pack(packet)
[docs]
def unpack(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_PCAPNG':
"""Unpack (parse) packet data.
Args:
length: Length of packet data.
**kwargs: Arbitrary keyword arguments.
Returns:
Parsed packet data.
Notes:
We used a special keyword argument ``__packet__`` to pass the
global packet data to underlying methods. This is useful when
the packet data is not available in the current instance.
"""
if cast('Optional[Schema_PCAPNG]', self.__header__) is None:
packet = kwargs.get('__packet__', {}) # packet data
# NOTE: Reached on the parsing path only, with no flag needed: the
# construction path sets ``__header__`` from ``make`` while packing
# its own buffer, so the branch above is already false there. A
# ``_read`` guard would be a branch that can never be taken.
self._check_block_floor(length)
if self._ctx is not None:
self._byte = self._ctx.section.byteorder
packet['byteorder'] = self._byte
# NOTE: A Simple Packet Block's data is ``min(original_len,
# snaplen)`` octets, with the snaplen of interface 0.
# A snaplen of zero means no limit.
interfaces = getattr(self._ctx, 'interfaces', None)
if interfaces and interfaces[0].snaplen:
packet.setdefault('snaplen', interfaces[0].snaplen)
# NOTE: A zero snaplen handed in, e.g. interface 0's passed through,
# means no limit as well; kept, it reads as a limit of no octets and
# the block's padding becomes packet data (#1558).
if 'snaplen' in packet and not packet['snaplen']:
del packet['snaplen']
self.__header__ = cast('Schema_PCAPNG', self.__schema__.unpack(self._file, length, packet)) # type: ignore[call-arg,misc]
data = self.read(length, **kwargs)
# NOTE: One source of truth for the captured octets. Reading them through
# the :attr:`packet` property means the value ``ProtocolBase.__init__``
# injects from ``self.packet.payload`` is the value computed here, rather
# than a second attempt at it.
data.__update__(packet=self.packet.payload)
return data
[docs]
def read(self, length: 'Optional[int]' = None, *, _read: 'bool' = True,
_seek_set: 'int' = 0, **kwargs: 'Any') -> 'Data_PCAPNG':
r"""Read PCAP-NG file blocks.
Structure of PCAP-NG file blocks:
.. code-block:: text
1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
0 | Block Type |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
4 | Block Total Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
8 / Block Body /
/ variable length, padded to 32 bits /
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Block Total Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Args:
length: Length of data to be read.
\_read: If the class is called in a parsing scenario.
\_seek_set: File offset before reading.
**kwargs: Arbitrary keyword arguments.
Returns:
Parsed packet data.
"""
schema = self.__header__
if schema.block.length < GENERIC_BLOCK_MINIMUM or schema.block.length % 4 != 0:
raise ProtocolError(f'PCAP-NG: [Block {schema.type}] invalid length: {schema.block.length}')
self._type = schema.type
if schema.type == Enum_BlockType.Section_Header_Block:
self._sect += 1
self._ctx = None
name = self._lookup_registry(self.__block__, schema.type)
if schema.type in self.SHORT_PACKET_TYPES and schema.block.length < BLOCK_MINIMUM[schema.type]:
meth = cast('BlockParser', self._read_block_short)
elif schema.block.length < self.SHORT_HEADER_TYPES.get(schema.type, 0):
meth = cast('BlockParser', self._read_block_short_header)
elif isinstance(name, str):
meth_name = f'_read_block_{name}'
meth = cast('BlockParser',
getattr(self, meth_name, self._read_block_unknown))
else:
meth = name[0]
try:
block = meth(schema.block, header=schema)
except BaseError as error:
# NOTE: Fields past the end of a block the capture ended inside are
# read as zeros, and a value its parser refuses is an artefact of
# that, not of the block; it is kept as captured instead.
if not _read or self._captured(_seek_set) >= schema.block.length:
raise
block = self._read_block_cut(schema, _seek_set, error)
# NOTE: An option or record area holding a malformed entry is kept as
# the octets captured, as one :obj:`bytes` item (see
# :class:`~pcapkit.protocols.schema.misc.pcapng.OptionAreaField`), and
# parses as no options. The octets are kept on the block as
# ``options_raw`` or ``records_raw``, which the block's ``make`` writes
# back verbatim (#1325).
for name in ('records', 'options'):
area = getattr(schema.block, name, None)
if isinstance(area, list) and len(area) == 1 and isinstance(area[0], bytes):
block.__update__([(f'{name}_raw', area[0])])
if not _read:
# move backward to the beginning of the packet
self._file.seek(0, io.SEEK_SET)
else:
# NOTE: We create a copy of the block data here for parsing
# scenarios to keep the original packet data intact.
seek_cur = _seek_set + block.length
# move backward to the beginning of the block
self._file.seek(_seek_set, io.SEEK_SET)
#: bytes: Raw block data.
#
# NOTE: Asked for no further than one read can go, :data:`sys.maxsize`,
# past which the read raises :exc:`OverflowError` instead of coming
# back short as the truncation handling below expects (#1975).
bound = min(schema.block.length, sys.maxsize)
self._data = self._read_fileng(bound)
# NOTE: Block Total Length is cross-checked against its own trailing
# copy and never against the file, so on a capture cut short it runs
# past the real end -- and seeking past the end is legal and silent.
# The *next* block read would then measure a negative remainder,
# since ``pcapkit.corekit.decorators.prepare`` derives it as the
# end of the stream less the current position, and a negative
# ``__length__`` is what ``pcapng_block_selector`` hands to
# :class:`~pcapkit.corekit.fields.misc.SchemaField`, which fails with
# a bare ``ValueError: read length must be non-negative or -1``. That
# would cost the whole extraction rather than the one truncated
# block, the opposite of the `#431
# <https://github.com/JarryShaw/PyPCAPKit/issues/431>`__ accommodation
# for damaged input. ``_read_fileng`` stops at the end of the file, so
# the octets it returned are the authority on where the block really
# finishes.
read = len(self._data)
if read < schema.block.length:
warn(f'PCAP-NG: [Block {schema.type}] block length '
f'{schema.block.length} exceeds the {read} octet(s) left in '
f'the file; block truncated', ProtocolWarning,
stacklevel=stacklevel())
seek_cur = min(seek_cur, _seek_set + read)
# NOTE: The fields read past the end of the capture are zeros,
# and an option, record or packet the end cut through is only
# its first octets, so the block cannot be rebuilt from them at
# the length it declares. The octets captured are kept as
# ``__truncated_raw__`` instead, which :meth:`__post_init__` parses
# again in place of building the block (#1470).
block.__update__([('__truncated_raw__', self._data)])
# NOTE: A read that came back whole at the bound stopped there, not
# at the end of the file: the block runs on past it, and the next one
# starts where its Block Total Length says, so the stream is moved
# over the rest (#1975).
if read == bound < schema.block.length:
advance(self._file, schema.block.length - bound)
else:
# move backward to the beginning of next block
self._file.seek(seek_cur, io.SEEK_SET)
#: io.BytesIO: Source packet stream.
self._file = io.BytesIO(self._data)
return block
[docs]
def make(self,
type: 'Enum_BlockType | StdlibEnum | AenumEnum | str | int' = Enum_BlockType.Simple_Packet_Block,
type_default: 'Optional[int]' = None,
type_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long
type_reversed: 'bool' = False,
block: 'bytes | Data_PCAPNG | Schema_BlockType | dict[str, Any]' = b'',
**kwargs: 'Any') -> 'Schema_PCAPNG':
"""Make PCAP-NG block data.
Args:
type: Block type.
type_default: Default block type.
type_namespace: Block type namespace.
type_reversed: Whether to reverse block type namespace.
block: Block data.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed packet data.
"""
type_val = cast('Enum_BlockType',
self._make_index(type, type_default, namespace=type_namespace,
reversed=type_reversed, pack=False))
self._type = type_val
if isinstance(block, bytes):
block_val = block # type: bytes | Schema_BlockType
elif isinstance(block, (dict, Data_PCAPNG)):
name = self._lookup_registry(self.__block__, type_val)
if isinstance(name, str):
meth_name = f'_make_block_{name}'
meth = cast('BlockConstructor',
getattr(self, meth_name, self._make_block_unknown))
else:
meth = name[1]
if isinstance(block, dict):
block_val = meth(**block)
else:
block_val = meth(block)
elif isinstance(block, Schema):
block_val = block
else:
raise ProtocolError(f'PCAP-NG: [Type {type_val}] invalid format')
return Schema_PCAPNG(
type=type_val,
block=block_val,
)
##########################################################################
# Data models.
##########################################################################
@overload # pragma: no cover
def __post_init__(self, file: 'IO[bytes] | bytes', length: 'Optional[int]' = ..., *, # pylint: disable=arguments-differ
num: 'int', sct: 'int', ctx: 'Context', **kwargs: 'Any') -> 'None': ...
@overload # pragma: no cover
def __post_init__(self, *, num: 'int', sct: 'int', ctx: 'Context', # pylint: disable=arguments-differ
**kwargs: 'Any') -> 'None': ...
[docs]
def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None, length: 'Optional[int]' = None, *, # pylint: disable=arguments-differ
num: 'int', sct: 'int', ctx: 'Context', **kwargs: 'Any') -> 'None':
"""Initialisation.
Args:
file: Source packet stream.
length: Length of packet data.
num: Frame index number.
sct: Section index number.
ctx: Section context of the PCAP-NG file.
**kwargs: Arbitrary keyword arguments.
Notes:
For the first block, ``num`` will be set to ``0`` and ctx as :obj:`None`,
such that we can be sure that the first block is the section header block.
See Also:
For construction arguments, please refer to :meth:`make`.
"""
#: int: Section index number.
self._sect = sct
#: int: Block index number.
self._fnum = num
#: pcapkit.foundation.engines.pcapng.Context: Context of the PCAP-NG file.
self._ctx = ctx
#: collections.Counter: Counter for option types.
self._opt = collections.Counter() # type: Counter[Enum_OptionType]
# NOTE: A block the capture ended inside rebuilds as the octets that
# were captured, parsed again as they were the first time (see
# :meth:`read`), rather than at the length it declares (#1470).
block = kwargs.get('block')
if file is None and isinstance(block, collections.abc.Mapping) and '__truncated_raw__' in block:
file = bytes(block['__truncated_raw__'])
if file is None:
_read = False
#: bytes: Raw packet data.
self._data = self.pack(**kwargs)
#: io.BytesIO: Source packet stream.
self._file = io.BytesIO(self._data)
# NOTE: The block just built is parsed again below on this same
# instance, and the parse counts each option afresh against its
# "only one" guard -- so the count left by the make pass must not
# carry over into it.
self._opt = collections.Counter()
else:
_read = True
#: io.BytesIO: Source packet stream.
self._file = io.BytesIO(file) if isinstance(file, bytes) else file
_seek_set = self._file.tell()
#: pcapkit.corekit.infoclass.Info: Parsed packet data.
self._info = self.unpack(length, _read=_read, _seek_set=_seek_set, **kwargs)
[docs]
def __length_hint__(self) -> 'Literal[12]':
"""Return an estimated length for the object."""
return GENERIC_BLOCK_MINIMUM # type: ignore[return-value]
# NOTE: This is a hack to make the ``__index__`` method work both as a
# class method and an instance method.
[docs]
def __index__(self: 'Optional[PCAPNG]' = None) -> 'int': # type: ignore[override]
"""Index of the block.
Args:
self: :class:`PCAPNG` object or :obj:`None`.
Returns:
If the object is initialised, i.e. ``self._fnum`` exists, and is of a
packet block (EPB, SPB or Packet), returns the block index number of
itself; else raises
:exc:`~pcapkit.utilities.exceptions.UnsupportedCall`.
Raises:
UnsupportedCall: If called on the class, or on a block that is not a
packet block.
"""
if self is None or self._info.type not in self.PACKET_TYPES:
raise UnsupportedCall("'PCAPNG' object cannot be interpreted as an integer")
return self._fnum
##########################################################################
# Utilities.
##########################################################################
[docs]
def _check_block_floor(self, length: 'Optional[int]') -> 'None':
"""Reject a tail too short to hold any block at all.
Args:
length: Length of packet data, as the caller declared it, or
:obj:`None` to measure what is left in
:attr:`self._file <pcapkit.protocols.protocol.Protocol._file>`.
Raises:
StreamEOFError: If fewer than twelve octets are left to read.
A PCAP-NG block is twelve octets at its smallest -- Block Type, Block
Total Length and the trailing copy of it, over an empty body -- which is
what :meth:`__length_hint__` reports. A tail shorter than that is not a
block, so parsing one out of it can only invent fields from
:meth:`FieldBase.unpack <pcapkit.corekit.fields.field.FieldBase.unpack>`'s
zero padding, and once the four octets of :attr:`PCAPNG.type
<pcapkit.protocols.schema.misc.pcapng.PCAPNG.type>` have been padded out
of nothing ``__length__`` is negative -- which
:func:`~pcapkit.protocols.schema.misc.pcapng.pcapng_block_selector` hands
to :class:`~pcapkit.corekit.fields.misc.SchemaField`, where
:meth:`io.RawIOBase.read` raises the bare ``ValueError`` of :issue:`678`.
Reporting the end of the stream instead is what keeps the :issue:`431` accommodation:
:exc:`~pcapkit.utilities.exceptions.StreamEOFError` is an
:exc:`EOFError`, which :meth:`Extractor.record_frames
<pcapkit.foundation.extraction.Extractor.record_frames>` already catches
to stop with the frames read so far, so -- unlike a refusal inside
:meth:`FieldBase.unpack <pcapkit.corekit.fields.field.FieldBase.unpack>`,
which has no catch point above it and would abort the whole extraction --
this costs no frame that was actually in the file. It is the same signal
:func:`~pcapkit.corekit.decorators.prepare` raises for a *derived*
remainder of exactly zero; a remainder of one, two or three octets
passes that check and is caught here instead.
Note:
This runs on the parsing path only, since :meth:`unpack` reaches it
only when no schema has been built yet and the construction path
builds one from :meth:`make` while packing its own buffer. That is
what it should do: a constructed block short enough to trip this floor
is already reported by :meth:`read`'s own Block Total Length check,
which names the length and so says more than an end-of-stream would.
Measuring an undeclared length needs the stream to seek, and nothing
guards that here because :meth:`__post_init__` has already called
:meth:`~io.IOBase.tell` on it to record ``_seek_set``: a stream that
could not seek never reaches this method.
"""
if length is None:
current = self._file.tell()
length = self._file.seek(0, io.SEEK_END) - current
self._file.seek(current, io.SEEK_SET)
if length < GENERIC_BLOCK_MINIMUM:
raise StreamEOFError(f'PCAP-NG: block truncated: {length} octet(s) left, '
f'fewer than the {GENERIC_BLOCK_MINIMUM} a block needs', quiet=True)
def _captured(self, offset: 'int') -> 'int':
"""Count the octets left in the stream from ``offset`` on.
Args:
offset: File offset of the block.
Returns:
Octets from ``offset`` to the end of
:attr:`self._file <pcapkit.protocols.protocol.Protocol._file>`.
"""
current = self._file.tell()
end = self._file.seek(0, io.SEEK_END)
self._file.seek(current, io.SEEK_SET)
return end - offset
def _read_block_cut(self, header: 'Schema_PCAPNG', offset: 'int',
error: 'BaseError') -> 'Data_UnknownBlock':
"""Read a PCAP-NG block the capture ended inside, whose parser refused it.
Args:
header: Parsed PCAP-NG header schema.
offset: File offset of the block.
error: What the block's parser raised.
Returns:
Parsed packet data, with the octets captured after the Block Total
Length as its body. :meth:`read` keeps the block's octets as
``__truncated_raw__`` too, which a rebuild parses again, so that it
rebuilds as captured (:issue:`1470`).
"""
warn(f'PCAP-NG: [Block {header.type}] block truncated; kept as captured: {error}',
ProtocolWarning, stacklevel=stacklevel())
current = self._file.tell()
self._file.seek(offset + 8, io.SEEK_SET)
# NOTE: bounded as :meth:`read` bounds the block's read (#1975).
body = self._read_fileng(min(header.block.length - GENERIC_BLOCK_MINIMUM, sys.maxsize))
self._file.seek(current, io.SEEK_SET)
return Data_UnknownBlock(
type=header.type,
length=header.block.length,
body=body,
)
[docs]
def _get_payload(self) -> 'bytes':
"""Get payload of :attr:`self.__header__ <pcapkit.protocols.protocol.Protocol.__header__>`.
Returns:
Payload of :attr:`self.__header__ <pcapkit.protocols.protocol.Protocol.__header__>` as :obj:`bytes`.
See Also:
This is a wrapper function for :meth:`pcapkit.protocols.schema.schema.Schema.get_payload`.
Note:
A packet block kept as captured has no payload (see
:meth:`_read_block_short`).
"""
if isinstance(self.__header__.block, Schema_UnknownBlock):
return b''
return self.__header__.block.get_payload()
@staticmethod
def _get_payload_name(block: 'Schema_BlockType | bytes') -> 'Optional[str]':
"""Get the name of the field carrying a block's captured packet octets.
Args:
block: Parsed block schema.
Returns:
The name of the block schema's payload field, or :obj:`None` when the
block declares none -- which is every block type outside
:attr:`PACKET_TYPES <PCAPNG.PACKET_TYPES>`, and also a block whose
body was handed over as raw :obj:`bytes` at construction time rather
than as a schema.
"""
name = getattr(block, '__payload__', None)
if name is None or name not in getattr(block, '__fields__', {}):
return None
return name
@staticmethod
def _get_local_timezone() -> 'timezone':
"""Get local timezone.
Warning:
Not used when reconstructing a block timestamp, and must not be:
a PCAP-NG timestamp is an offset from the UNIX epoch, so mixing the
*reading* host's zone into it makes one file parse to different
instants on different machines.
:meth:`self._get_timezone <_get_timezone>` returns
:attr:`datetime.timezone.utc` instead when the capture names no
``if_tzone``.
It also answers for *today* rather than for the capture's own
instant -- ``America/New_York`` reports ``-04:00`` all year, so a
December timestamp would get EDT's offset and not EST's -- which is
a second reason it cannot stand in for a capture's timezone.
"""
tzinfo = datetime.datetime.now(datetime.timezone.utc).astimezone().tzinfo
if tzinfo is None:
return datetime.timezone.utc
return cast('timezone', tzinfo)
def _get_interface(self, interface_id: 'int') -> 'Data_InterfaceDescriptionBlock':
"""Interface description that ``interface_id`` names.
Args:
interface_id: Interface ID that the current block associates with.
Returns:
Interface Description Block (IDB) of the current section that
``interface_id`` identifies.
Raises:
FormatError: If the current section describes no such interface.
Note:
The PCAP-NG specification requires a block's interface ID to name an
Interface Description Block of the block's own section, so an ID that
names none is a malformed file rather than a programming error --
hence :exc:`~pcapkit.utilities.exceptions.FormatError` rather than
the bare :exc:`IndexError` that indexing the list would raise.
"""
interfaces = self._ctx.interfaces
if not 0 <= interface_id < len(interfaces):
tag = self.INTERFACE_ID_BLOCK_TAGS.get(self._type, f'Block {self._type}')
raise FormatError(f'PCAP-NG: [{tag}] invalid interface ID: {interface_id}')
return interfaces[interface_id]
def _get_resolution(self, interface_id: 'int' = 0) -> 'int':
"""Timestamp resolution of the current block, in units per second.
Args:
interface_id: Interface ID that the current block associates with.
Returns:
Timestamp resolution of the current block, in units per second.
"""
if self._ctx is None:
# raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute '_get_resolution'")
warn(f"'{self.__class__.__name__}' object has no attribute '_get_resolution'",
AttributeWarning, stacklevel=stacklevel())
return DEFAULT_RESOLUTION
options = self._get_interface(interface_id).options
tsresol = cast('Optional[Data_IF_TSResolOption]',
options.get(Enum_OptionType.if_tsresol))
if tsresol is None:
return DEFAULT_RESOLUTION
return tsresol.resolution
def _get_offset(self, interface_id: 'int' = 0) -> 'int':
"""Timestamp offset of the current block, in seconds.
Args:
interface_id: Interface ID that the current block associates with.
Returns:
Timestamp offset of the current block, in seconds.
"""
if self._ctx is None:
# raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute '_get_offset'")
warn(f"'{self.__class__.__name__}' object has no attribute '_get_offset'",
AttributeWarning, stacklevel=stacklevel())
return 0
options = self._get_interface(interface_id).options
tsoffset = cast('Optional[Data_IF_TSOffsetOption]',
options.get(Enum_OptionType.if_tsoffset))
if tsoffset is None:
return 0
return tsoffset.offset
[docs]
def _get_timezone(self, interface_id: 'int' = 0) -> 'timezone':
"""Timezone of the current block.
Args:
interface_id: Interface ID that the current block associates with.
Returns:
Timezone of the current block.
"""
# NOTE: UTC, not the host's timezone, is the default in both branches
# below. A capture carries no timezone of its own unless it says so, and
# draft-ietf-opsawg-pcapng-02 §4.3 defines the block timestamp as an
# offset from 1970-01-01 00:00:00 UTC with no timezone term at all, so
# substituting the *reading* machine's zone would make the same file
# parse to different instants on different hosts. ``_get_resolution``
# and ``_get_offset`` fall back to the format's own defaults in this
# situation; this is the matching one.
if self._ctx is None:
# raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute '_get_timezone'")
warn(f"'{self.__class__.__name__}' object has no attribute '_get_timezone'",
AttributeWarning, stacklevel=stacklevel())
return datetime.timezone.utc
options = self._get_interface(interface_id).options
tzone = cast('Optional[Data_IF_TZoneOption]',
options.get(Enum_OptionType.if_tzone))
# an ``if_tzone`` kept as captured (a :class:`~datetime.timedelta` of 24
# hours or more) names no zone a datetime can be rendered in
if tzone is None or not isinstance(tzone.timezone, datetime.timezone):
return datetime.timezone.utc
return tzone.timezone
def _get_linktype(self, interface_id: 'int' = 0) -> 'Enum_LinkType':
"""Data link layer protocol type.
Args:
interface_id: Interface ID that the current block associates with.
Returns:
Data link layer protocol type.
Raises:
UnsupportedCall: If current block is not a valid packet block, i.e.,
EPB, SPB or obsolete Packet Block.
"""
if self._ctx is None or self._type not in self.PACKET_TYPES:
raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute '_get_linktype'")
return self._get_interface(interface_id).linktype
def _read_timestamp(self, timestamp_high: 'int', timestamp_low: 'int', *,
interface_id: 'int' = 0) -> 'tuple[dt_type, Decimal]':
"""Read timestamp.
Args:
timestamp_high: Higher 32-bit integer value of timestamp.
timestamp_low: Lower 32-bit integer value of timestamp.
interface_id: Interface ID that the current block associates with.
Returns:
Tuple of timestamp in :class:`~datetime.datetime` object with
timezone information and :class:`decimal.Decimal` object since
UNIX-Epoch in UTC timezone.
"""
tzone = self._get_timezone(interface_id)
timestamp_raw = (timestamp_high << 32) | timestamp_low
with localcontext(prec=64):
# NOTE: This is the UTC epoch the docstring above promises, with no
# ``tzone.utcoffset(None)`` added. The block
# timestamp is an offset from the UNIX epoch, so it is
# timezone-independent by definition (draft-ietf-opsawg-pcapng-02
# §4.3 spells out the arithmetic and has no timezone term), and the
# same draft §4.2 says of ``if_tzone`` that it "SHOULD NOT be used"
# and of ``if_tsoffset`` that it is "not intended to be used as an
# offset between local time and UTC". A shifted epoch would also
# contradict ``ts_datetime`` below, which is built from the
# unshifted value.
timestamp_epoch = decimal.Decimal(timestamp_raw) / self._get_resolution(interface_id) + \
self._get_offset(interface_id)
ts_ratio = timestamp_epoch.as_integer_ratio()
try:
# NOTE: ``tzone`` survives only here, as the zone the instant is
# *rendered* in -- which is all a ``tzinfo`` can affect on an aware
# datetime. It never moves the instant itself.
ts_datetime = datetime.datetime.fromtimestamp(ts_ratio[0] / ts_ratio[1], tzone)
# an epoch past what ``datetime`` or the platform's ``time_t`` holds,
# e.g. from an extreme ``if_tsoffset``, raises one of these
except (ValueError, OverflowError, OSError):
warn(f'PCAP-NG: [Block {self._type}] invalid timestamp: {timestamp_epoch}',
ProtocolWarning, stacklevel=stacklevel())
ts_datetime = datetime.datetime.fromtimestamp(0, datetime.timezone.utc)
return (ts_datetime, timestamp_epoch)
[docs]
@classmethod
def _make_data(cls, data: 'Data_PCAPNG') -> 'dict[str, Any]': # type: ignore[override]
"""Create key-value pairs from ``data`` for protocol construction.
Args:
data: protocol data
Returns:
Key-value pairs for protocol construction.
"""
return {
'type': data.type,
'block': data,
}
@staticmethod
def _block_area(block: 'Data_PCAPNG', name: 'str') -> 'Any':
"""Option or record area of a parsed block, for its ``make``.
Args:
block: Block data model.
name: ``'options'`` or ``'records'``.
Returns:
The area's octets if :meth:`read` kept it as captured, as
``<name>_raw``; otherwise the parsed ``<name>``.
"""
return getattr(block, f'{name}_raw', getattr(block, name))
@staticmethod
def _read_padding(schema: 'Schema', data: 'Any', name: 'str' = 'padding',
key: 'Optional[str]' = None, *, nonzero: 'bool' = True) -> 'None':
"""Keep a padding field of a parsed schema on its data model.
Args:
schema: Parsed block, option or record schema.
data: Its data model, updated in place.
name: Name of the padding field on ``schema``.
key: Name to keep it as on ``data``; defaults to ``name``.
nonzero: Keep the octets only when they are not all zeros -- the
form a fresh build writes -- rather than whenever there are any.
The octets are kept as ``key`` only when there is something a rebuild
would otherwise lose, so the data model of well-formed input gains no
key (:issue:`1437`). :meth:`_make_padding` writes them back.
"""
padding = schema.__dict__.get(name)
if isinstance(padding, bytes) and (any(padding) if nonzero else padding):
data.__update__([(key or name, padding)])
@staticmethod
def _make_padding(schema: '_S', source: 'Any', name: 'str' = 'padding',
key: 'Optional[str]' = None) -> '_S':
"""Write the padding octets :meth:`_read_padding` kept back onto a schema.
Args:
schema: Constructed block, option or record schema, updated in place.
source: Data model or keyword arguments it was made from, if any.
name: Name of the padding field on ``schema``.
key: Name the octets are kept as on ``source``; defaults to ``name``.
Returns:
The ``schema`` given. Without kept octets its padding stays zeros.
"""
if source is None or isinstance(schema, bytes):
return schema
padding = source.get(key or name)
if isinstance(padding, (bytes, bytearray)) and name in schema.__fields__:
setattr(schema, name, bytes(padding))
return schema
def _make_options_trailer(self, block: 'Optional[Data_PCAPNG]',
options: 'list[Schema_Option | bytes]') -> 'int':
"""Append the octets a parsed block held past its ``opt_endofopt``.
Args:
block: Block data model, if any.
options: Constructed options, extended in place.
Returns:
Number of octets appended.
An option area ends at its ``opt_endofopt`` and the octets after it are
read as padding. :meth:`_read_padding` keeps them as ``padding_opts``,
and they are written back after the options, verbatim (:issue:`1437`).
"""
trailer = block.get('padding_opts') if block is not None else None
if not isinstance(trailer, bytes) or not trailer:
return 0
options.append(trailer)
return len(trailer)
@classmethod
def _make_packet_data(cls, block: 'Data_PCAPNG') -> 'bytes | ProtocolBase':
"""Create the packet data of an EPB, SPB or PB from its data model.
Args:
block: Block data model.
Returns:
The captured packet octets the block was parsed from, as
:attr:`~pcapkit.protocols.data.protocol.Protocol.packet` holds them,
cut to the block's ``captured_len``; or, for a data model that
carries none, the payload rebuilt from its decoded next layer.
Note:
The cut is what keeps a block with ``captured_len`` of zero intact:
its ``packet`` can hold octets past the payload (:issue:`1275`), and
writing those back would grow the block.
The fallback is lossy. A next layer rebuilt from its own data model
drops anything it did not decode, such as Ethernet pad octets, so
the block it yields can be shorter than its ``captured_len`` says.
"""
packet = block.get('packet')
if isinstance(packet, bytes):
return packet[:block.captured_len]
return cls._make_payload(block)
def _make_timestamp(self, timestamp: 'Optional[float | Decimal | dt_type | int]' = None, *,
interface_id: 'int' = 0) -> 'tuple[int, int]':
"""Make timestamp.
Args:
timestamp: Timestamp in seconds since UNIX-Epoch, in UTC, i.e. as
:meth:`self._read_timestamp <_read_timestamp>` returns it.
interface_id: Interface ID that the current block associates with.
Returns:
Tuple of timestamp in higher and lower 32-bit integer value,
scaled by the interface's ``if_tsresol`` and shifted by its
``if_tsoffset``.
Notes:
No timezone conversion happens here, and none should: the field this
builds is defined as an offset from the UNIX epoch, and the read
side applies no timezone shift for it to mirror.
"""
with localcontext(prec=64):
if timestamp is None:
if py37 and self.nanosecond:
timestamp = decimal.Decimal(time.time_ns()) / 1_000_000_000
else:
timestamp = decimal.Decimal(time.time())
else:
if isinstance(timestamp, datetime.datetime):
timestamp = timestamp.timestamp()
timestamp = decimal.Decimal(timestamp)
ts_info = int(
(timestamp - self._get_offset(interface_id)) * self._get_resolution(interface_id)
)
return (ts_info >> 32) & 0xFFFF_FFFF, ts_info & 0xFFFF_FFFF
def _read_mac_addr(self, addr: 'bytes') -> 'str':
"""Read MAC address.
Args:
addr: MAC address.
Returns:
Colon (``:``) separated *hex* encoded MAC address.
"""
if py38:
_addr = addr.hex(':')
else:
_addr = ':'.join(textwrap.wrap(addr.hex(), 2))
return _addr
def _read_eui_addr(self, addr: 'bytes') -> 'str':
"""Read EUI address.
Args:
addr: EUI address.
Returns:
Colon (``:``) separated *hex* encoded EUI address.
"""
if py38:
_addr = addr.hex(':')
else:
_addr = ':'.join(textwrap.wrap(addr.hex(), 2))
return _addr
[docs]
def _make_mac_addr(self, addr: 'str | bytes | bytearray') -> 'bytes':
"""Make MAC address.
Args:
addr: MAC address.
Returns:
MAC address packed as 6 octets.
Raises:
ProtocolError: If ``addr`` is not a ``:``- or ``-``-separated
*hex* encoded MAC address.
"""
_addr = addr.encode() if isinstance(addr, str) else addr
if PAT_MAC_ADDR.fullmatch(_addr) is not None:
return bytes.fromhex(_addr.replace(b':', b'').replace(b'-', b'').decode())
raise ProtocolError(f'invalid MAC address: {addr!r}')
def _make_eui_addr(self, addr: 'str | bytes | bytearray') -> 'bytes':
"""Make EUI address.
Args:
addr: EUI address.
Returns:
EUI address packed as 8 octets.
Raises:
ProtocolError: If ``addr`` is not a ``:``- or ``-``-separated
*hex* encoded EUI-64 address.
"""
_addr = addr.encode() if isinstance(addr, str) else addr
if PAT_EUI_ADDR.fullmatch(_addr) is not None:
return bytes.fromhex(_addr.replace(b':', b'').replace(b'-', b'').decode())
raise ProtocolError(f'invalid EUI address: {addr!r}')
[docs]
def _decode_next_layer(self, dict_: 'Data_PCAPNG', proto: 'Optional[int]' = None,
length: 'Optional[int]' = None, *, packet: 'Optional[dict[str, Any]]' = None) -> 'Data_PCAPNG': # pylint: disable=arguments-differ
r"""Decode next layer protocol.
Arguments:
dict\_: info buffer
proto: next layer protocol index
length: valid (*non-padding*) length
packet: packet info (passed from :meth:`self.unpack <pcapkit.protocols.protocol.Protocol.unpack>`)
Returns:
Current protocol with packet extracted.
Notes:
We added a new key ``__next_type__`` to ``dict_`` to store the
next layer protocol type, and a new key ``__next_name__`` to
store the next layer protocol name. :meth:`Info.to_dict
<pcapkit.corekit.infoclass.Info.to_dict>` carries both, and the
dumpers leave them out.
We also added a new key ``protocols`` to ``dict_`` to store the
protocol chain of the current packet (frame).
``proto`` is dispatched as :meth:`_dispatch_linktype` says, so
``LINKTYPE_RAW`` data is read as IPv4 or IPv6 (:issue:`1502`).
"""
proto = self._dispatch_linktype(proto, length)
next_ = cast('ProtocolBase', self._import_next_layer(proto, length, packet=packet)) # type: ignore[misc,call-arg,redundant-cast]
info, chain = next_.info, next_.protochain
# make next layer protocol name
layer = next_.info_name
# proto = next_.__class__.__name__
# write info and protocol chain into dict
dict_.__update__({
layer: info,
'protocols': chain.chain if chain else '',
'__next_type__': type(next_),
'__next_name__': layer,
})
self._next = next_ # pylint: disable=attribute-defined-outside-init
self._protos = chain # pylint: disable=attribute-defined-outside-init
return dict_
[docs]
def _dispatch_linktype(self, proto: 'Optional[int]', length: 'Optional[int]' = None) -> 'Optional[int]':
"""Link type a packet block's data is dispatched under.
Arguments:
proto: link type of the interface the block was captured on
length: valid (*non-padding*) length
Returns:
For :attr:`LinkType.RAW <pcapkit.const.reg.linktype.LinkType.RAW>`
data whose first octet carries IP version 4 or 6,
:attr:`LinkType.IPV4 <pcapkit.const.reg.linktype.LinkType.IPV4>` or
:attr:`LinkType.IPV6 <pcapkit.const.reg.linktype.LinkType.IPV6>`;
otherwise ``proto``.
Note:
As :meth:`Frame._dispatch_linktype
<pcapkit.protocols.misc.pcap.frame.Frame._dispatch_linktype>` does
for PCAP: ``LINKTYPE_RAW`` (101) data is IPv4 or IPv6 by its version
nibble, an entry registered under ``RAW`` itself is used as it
stands, and data of any other version is kept as
:class:`~pcapkit.protocols.misc.raw.Raw` (:issue:`1502`).
"""
if proto != Enum_LinkType.RAW or Enum_LinkType.RAW in self.__proto__:
return proto
data = self._get_payload()[:length]
version = data[0] >> 4 if data else None
if version == 4:
return Enum_LinkType.IPV4
if version == 6:
return Enum_LinkType.IPV6
return proto
[docs]
def _read_block_unknown(self, schema: 'Schema_UnknownBlock', *,
header: 'Schema_PCAPNG') -> 'Data_UnknownBlock':
"""Read unknown PCAP-NG block.
Args:
schema: Parsed block schema.
header: Parsed PCAP-NG header schema.
Returns:
Parsed packet data.
"""
data = Data_UnknownBlock(
type=header.type,
length=schema.length,
body=schema.body,
)
return data
[docs]
def _read_block_shb(self, schema: 'Schema_SectionHeaderBlock', *,
header: 'Schema_PCAPNG') -> 'Data_SectionHeaderBlock':
"""Read PCAP-NG section header block (SHB).
Structure of Section Header Block:
.. code-block:: text
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
0 | Block Type = 0x0A0D0D0A |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
4 | Block Total Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
8 | Byte-Order Magic |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
12 | Major Version | Minor Version |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
16 | |
| Section Length |
| |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
24 / /
/ Options (variable) /
/ /
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Block Total Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Args:
schema: Parsed block schema.
header: Parsed PCAP-NG header schema.
Returns:
Parsed packet data.
"""
self._byte = schema.byteorder
data = Data_SectionHeaderBlock(
type=header.type,
length=schema.length,
byteorder=schema.byteorder,
version=VersionInfo(
major=schema.major,
minor=schema.minor,
),
section_length=schema.section_length,
options=self._read_pcapng_options(schema.options),
)
# NOTE: The octets after ``opt_endofopt``, if any (#1437).
self._read_padding(schema, data, 'padding', 'padding_opts', nonzero=False)
return data
[docs]
def _read_block_idb(self, schema: 'Schema_InterfaceDescriptionBlock', *,
header: 'Schema_PCAPNG') -> 'Data_InterfaceDescriptionBlock':
"""Read PCAP-NG interface description block (IDB).
Structure of Interface Description Block:
.. code-block:: text
1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
0 | Block Type = 0x00000001 |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
4 | Block Total Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
8 | LinkType | Reserved |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
12 | SnapLen |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
16 / /
/ Options (variable) /
/ /
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Block Total Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Args:
schema: Parsed block schema.
header: Parsed PCAP-NG header schema.
Returns:
Parsed packet data.
"""
data = Data_InterfaceDescriptionBlock(
type=header.type,
length=schema.length,
linktype=schema.linktype,
snaplen=schema.snaplen,
options=self._read_pcapng_options(schema.options),
)
# NOTE: The reserved octets, if they are not all zeros (#1437).
self._read_padding(schema, data, 'reserved')
# NOTE: The octets after ``opt_endofopt``, if any (#1437).
self._read_padding(schema, data, 'padding', 'padding_opts', nonzero=False)
return data
[docs]
def _read_block_epb(self, schema: 'Schema_EnhancedPacketBlock', *,
header: 'Schema_PCAPNG') -> 'Data_EnhancedPacketBlock':
"""Read PCAP-NG enhanced packet block (EPB).
Structure of Enhanced Packet Block:
.. code-block:: text
1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
0 | Block Type = 0x00000006 |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
4 | Block Total Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
8 | Interface ID |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
12 | Timestamp (High) |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
16 | Timestamp (Low) |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
20 | Captured Packet Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
24 | Original Packet Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
28 / /
/ Packet Data /
/ variable length, padded to 32 bits /
/ /
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
/ /
/ Options (variable) /
/ /
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Block Total Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Args:
schema: Parsed block schema.
header: Parsed PCAP-NG header schema.
Returns:
Parsed packet data.
"""
timestamp, timestamp_epoch = self._read_timestamp(schema.timestamp_high, schema.timestamp_low,
interface_id=schema.interface_id)
data = Data_EnhancedPacketBlock(
type=header.type,
length=schema.length,
section_number=self._sect,
number=self._fnum,
interface_id=schema.interface_id,
timestamp=timestamp,
timestamp_epoch=timestamp_epoch,
captured_len=schema.captured_len,
original_len=schema.original_len,
options=self._read_pcapng_options(schema.options),
)
# NOTE: The padding of the packet data to 32 bits, if it is not all
# zeros (#1437).
self._read_padding(schema, data, 'padding_data')
# NOTE: The octets after ``opt_endofopt``, if any (#1437).
self._read_padding(schema, data, 'padding_opts', 'padding_opts', nonzero=False)
# NOTE: A captured length running past the block is kept as declared,
# and the packet data is the octets the block holds (#1405).
if schema.captured_len > len(schema.packet_data):
warn(f'PCAP-NG: [Block {header.type}] captured length {schema.captured_len} runs past '
f'the block, which holds {len(schema.packet_data)} octet(s); kept as captured',
ProtocolWarning, stacklevel=stacklevel())
return self._decode_next_layer(data, self._get_linktype(schema.interface_id),
schema.captured_len) # type: ignore[return-value]
[docs]
def _read_block_spb(self, schema: 'Schema_SimplePacketBlock', *,
header: 'Schema_PCAPNG') -> 'Data_SimplePacketBlock':
"""Read PCAP-NG simple packet block (SPB).
Structure of Simple Packet Block:
.. code-block:: text
1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
0 | Block Type = 0x00000003 |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
4 | Block Total Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
8 | Original Packet Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
12 / /
/ Packet Data /
/ variable length, padded to 32 bits /
/ /
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Block Total Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Args:
schema: Parsed block schema.
header: Parsed PCAP-NG header schema.
Returns:
Parsed packet data.
"""
data = Data_SimplePacketBlock(
type=header.type,
length=schema.length,
section_number=self._sect,
number=self._fnum,
captured_len=len(schema.packet_data),
original_len=schema.original_len,
)
# NOTE: The padding of the packet data to 32 bits, if it is not all
# zeros (#1437).
self._read_padding(schema, data, 'padding', 'padding_data')
# NOTE: A block holding more packet data than the snaplen of interface 0
# allows keeps the excess (#1384); the specification says the
# snaplen determines the length, so the excess is reported.
snaplen = self._ctx.interfaces[0].snaplen if self._ctx is not None and self._ctx.interfaces else 0
if snaplen and data.captured_len > snaplen:
warn(f'PCAP-NG: [Block {header.type}] packet data of {data.captured_len} octet(s) '
f'exceeds the snaplen of {snaplen}; kept as captured',
ProtocolWarning, stacklevel=stacklevel())
# NOTE: Likewise for more packet data than the original length, padding
# aside (#1414).
if data.captured_len > (data.original_len + 3) // 4 * 4:
warn(f'PCAP-NG: [Block {header.type}] packet data of {data.captured_len} octet(s) '
f'exceeds the original length of {data.original_len}; kept as captured',
ProtocolWarning, stacklevel=stacklevel())
return self._decode_next_layer(data, self._get_linktype(0), data.captured_len) # type: ignore[return-value]
[docs]
def _read_block_nrb(self, schema: 'Schema_NameResolutionBlock', *,
header: 'Schema_PCAPNG') -> 'Data_NameResolutionBlock':
"""Read PCAP-NG name resolution block (NRB).
Structure of Name Resolution Block:
.. code-block:: text
1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
0 | Block Type = 0x00000004 |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
4 | Block Total Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
8 | Record Type | Record Value Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
12 / Record Value /
/ variable length, padded to 32 bits /
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
. .
. . . . other records . . . .
. .
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Record Type = nrb_record_end | Record Value Length = 0 |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
/ /
/ Options (variable) /
/ /
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Block Total Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Args:
schema: Parsed block schema.
header: Parsed PCAP-NG header schema.
Returns:
Parsed packet data.
"""
data = Data_NameResolutionBlock(
type=header.type,
length=schema.length,
records=self._read_nrb_records(schema.records),
options=self._read_pcapng_options(schema.options),
)
# NOTE: The octets after ``opt_endofopt``, if any (#1437).
self._read_padding(schema, data, 'padding', 'padding_opts', nonzero=False)
return data
[docs]
def _read_block_isb(self, schema: 'Schema_InterfaceStatisticsBlock', *,
header: 'Schema_PCAPNG') -> 'Data_InterfaceStatisticsBlock':
"""Read PCAP-NG interface statistics block (ISB).
Structure of Interface Statistics Block:
.. code-block:: text
1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
0 | Block Type = 0x00000005 |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
4 | Block Total Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
8 | Interface ID |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
12 | Timestamp (High) |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
16 | Timestamp (Low) |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
20 / /
/ Options (variable) /
/ /
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Block Total Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Args:
schema: Parsed block schema.
header: Parsed PCAP-NG header schema.
Returns:
Parsed packet data.
"""
timestamp, timestamp_epoch = self._read_timestamp(schema.timestamp_high, schema.timestamp_low,
interface_id=schema.interface_id)
self._isb_interface_id = schema.interface_id
data = Data_InterfaceStatisticsBlock(
type=header.type,
length=schema.length,
interface_id=schema.interface_id,
timestamp=timestamp,
timestamp_epoch=timestamp_epoch,
options=self._read_pcapng_options(schema.options),
)
# NOTE: The octets after ``opt_endofopt``, if any (#1437).
self._read_padding(schema, data, 'padding', 'padding_opts', nonzero=False)
return data
[docs]
def _read_block_systemd(self, schema: 'Schema_SystemdJournalExportBlock', *,
header: 'Schema_PCAPNG') -> 'Data_SystemdJournalExportBlock':
"""Read PCAP-NG :manpage:`systemd(1)` journal export block.
Structure of :manpage:`systemd(1)` Journal Export Block:
.. code-block:: text
1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
0 | Block Type = 0x00000009 |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
4 | Block Total Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
8 / /
/ Journal Entry /
/ variable length, padded to 32 bits /
/ /
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Block Total Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Args:
schema: Parsed block schema.
header: Parsed PCAP-NG header schema.
Returns:
Parsed packet data.
"""
entries = schema.data
# NOTE: A malformed entry is kept as the octets captured, as one
# :obj:`bytes` item, and parses as no entries. The octets are kept on
# the block as ``entry_raw``, which the block's ``make`` writes back
# verbatim (#1406).
kept = len(entries) == 1 and isinstance(entries[0], bytes)
data = Data_SystemdJournalExportBlock(
type=header.type,
length=schema.length,
# NOTE: each entry as an OrderedMultiInfo, which ``Info`` does not
# do for a mapping held in a tuple (#1484).
data=() if kept else tuple(OrderedMultiInfo(entry) for entry in entries), # type: ignore[arg-type]
)
if kept:
data.__update__([('entry_raw', entries[0])])
return data
[docs]
def _read_block_dsb(self, schema: 'Schema_DecryptionSecretsBlock', *,
header: 'Schema_PCAPNG') -> 'Data_DecryptionSecretsBlock':
"""Read PCAP-NG decryption secrets block (DSB).
Structure of Decryption Secrets Block:
.. code-block:: text
1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
0 | Block Type = 0x0000000A |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
4 | Block Total Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
8 | Secrets Type |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
12 | Secrets Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
16 / /
/ Secrets Data /
/ (variable length, padded to 32 bits) /
/ /
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
/ /
/ Options (variable) /
/ /
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
/ Block Total Length /
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Args:
schema: Parsed block schema.
header: Parsed PCAP-NG header schema.
Returns:
Parsed packet data.
"""
name = self._lookup_registry(self.__secrets__, schema.secrets_type)
if isinstance(name, str):
meth_name = f'_read_secrets_{name}'
meth = cast('SecretsParser',
getattr(self, meth_name, self._read_secrets_unknown))
else:
meth = name[0]
secrets_data = meth(schema.secrets_data, block=schema)
data = Data_DecryptionSecretsBlock(
type=header.type,
length=schema.length,
secrets_type=schema.secrets_type,
secrets_length=schema.secrets_length,
secrets_data=secrets_data,
options=self._read_pcapng_options(schema.options),
)
# NOTE: The padding of the secrets to 32 bits, if it is not all zeros
# (#1437).
self._read_padding(schema, data, 'padding_data')
# NOTE: The octets after ``opt_endofopt``, if any (#1437).
self._read_padding(schema, data, 'padding_opts', 'padding_opts', nonzero=False)
return data
[docs]
def _read_block_cb(self, schema: 'Schema_CustomBlock', *,
header: 'Schema_PCAPNG') -> 'Data_CustomBlock':
"""Read PCAP-NG custom block (CB).
Structure of Custom Block:
.. code-block:: text
1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
0 | Block Type = 0x00000BAD or 0x40000BAD |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
4 | Block Total Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
8 | Private Enterprise Number (PEN) |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
12 / /
/ Custom Data /
/ variable length, padded to 32 bits /
/ /
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
/ /
/ Options (variable) /
/ /
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Block Total Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Args:
schema: Parsed block schema.
header: Parsed PCAP-NG header schema.
Returns:
Parsed packet data.
"""
data = Data_CustomBlock(
type=header.type,
length=schema.length,
pen=schema.pen,
data=schema.data,
)
return data
[docs]
def _read_block_packet(self, schema: 'Schema_PacketBlock', *,
header: 'Schema_PCAPNG') -> 'Data_PacketBlock':
"""Read PCAP-NG packet block (obsolete).
Structure of Packet Block:
.. code-block:: text
1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
0 | Block Type = 0x00000002 |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
4 | Block Total Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
8 | Interface ID | Drops Count |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
12 | Timestamp (High) |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
16 | Timestamp (Low) |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
20 | Captured Packet Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
24 | Original Packet Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
28 / /
/ Packet Data /
/ variable length, padded to 32 bits /
/ /
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
/ /
/ Options (variable) /
/ /
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Block Total Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Args:
schema: Parsed block schema.
header: Parsed PCAP-NG header schema.
Returns:
Parsed packet data.
"""
warn('PCAP-NG: Packet Block has been obsolete! Please use Enhanced Packet Block and/or '
'Simple Packet Block instead.', DeprecatedFormatWarning, stacklevel=stacklevel())
timestamp, timestamp_epoch = self._read_timestamp(schema.timestamp_high, schema.timestamp_low,
interface_id=schema.interface_id)
data = Data_PacketBlock(
type=header.type,
length=schema.length,
section_number=self._sect,
number=self._fnum,
interface_id=schema.interface_id,
drop_count=schema.drop_count,
timestamp=timestamp,
timestamp_epoch=timestamp_epoch,
captured_len=schema.captured_length,
original_len=schema.original_length,
options=self._read_pcapng_options(schema.options),
)
# NOTE: The padding of the packet data to 32 bits, if it is not all
# zeros (#1437).
self._read_padding(schema, data, 'padding_data')
# NOTE: The octets after ``opt_endofopt``, if any (#1437).
self._read_padding(schema, data, 'padding_opts', 'padding_opts', nonzero=False)
# NOTE: A captured length running past the block is kept as declared,
# and the packet data is the octets the block holds (#1405).
if schema.captured_length > len(schema.packet_data):
warn(f'PCAP-NG: [Block {header.type}] captured length {schema.captured_length} runs past '
f'the block, which holds {len(schema.packet_data)} octet(s); kept as captured',
ProtocolWarning, stacklevel=stacklevel())
return self._decode_next_layer(data, self._get_linktype(schema.interface_id),
schema.captured_length) # type: ignore[return-value]
[docs]
def _read_block_short(self, schema: 'Schema_EnhancedPacketBlock | Schema_PacketBlock | Schema_UnknownBlock', *,
header: 'Schema_PCAPNG') -> 'Data_EnhancedPacketBlock | Data_PacketBlock':
"""Read a PCAP-NG EPB or packet block (obsolete) shorter than its fixed fields.
The fixed fields of either block span 32 octets, so a smaller Block
Total Length cannot hold them.
Args:
schema: Parsed block schema, or on the construction path the
:class:`~pcapkit.protocols.schema.misc.pcapng.UnknownBlock` that
:meth:`_make_block_short` built.
header: Parsed PCAP-NG header schema.
Returns:
Parsed packet data, with the fixed fields the block holds and zero
for those it does not, no packet data and no options. The block body
is kept as ``block_raw``, which the block's ``make`` writes back
verbatim, so the block rebuilds at the length it declared
(:issue:`1414`).
"""
if isinstance(schema, Schema_UnknownBlock):
body = schema.body
else:
# the octets unpacked for the fixed fields, which start with the
# Block Total Length and run past the end of the block
octets = bytes(schema)
body = octets[4:schema.length - 8]
warn(f'PCAP-NG: [Block {header.type}] block length {schema.length} is below the '
f'32-octet minimum; kept as captured', ProtocolWarning, stacklevel=stacklevel())
if octets[schema.length - 8:schema.length - 4] != octets[:4]:
warn(f'PCAP-NG: [Block {header.type}] block length mismatch: {schema.length} != '
f'{octets[schema.length - 8:schema.length - 4].hex()}', ProtocolWarning,
stacklevel=stacklevel())
endian = '<' if self._byte == 'little' else '>'
fields = (body + bytes(20))[:20]
drop_count = 0
if header.type == Enum_BlockType.Packet_Block:
interface_id, drop_count, timestamp_high, timestamp_low, captured_len, original_len = \
struct.unpack(f'{endian}HHIIII', fields)
else:
interface_id, timestamp_high, timestamp_low, captured_len, original_len = \
struct.unpack(f'{endian}IIIII', fields)
timestamp, timestamp_epoch = self._read_timestamp(timestamp_high, timestamp_low,
interface_id=interface_id)
kwargs = {
'type': header.type,
'length': schema.length,
'section_number': self._sect,
'number': self._fnum,
'interface_id': interface_id,
'timestamp': timestamp,
'timestamp_epoch': timestamp_epoch,
'captured_len': captured_len,
'original_len': original_len,
'options': self._read_pcapng_options([]),
} # type: dict[str, Any]
data = (Data_PacketBlock(drop_count=drop_count, **kwargs)
if header.type == Enum_BlockType.Packet_Block
else Data_EnhancedPacketBlock(**kwargs)) # type: Data_EnhancedPacketBlock | Data_PacketBlock
data.__update__([('block_raw', body)])
return self._decode_next_layer(data, self._get_linktype(interface_id), 0) # type: ignore[return-value]
[docs]
def _read_pcapng_options(self, options_schema: 'list[Schema_Option]') -> 'Option':
"""Read PCAP-NG options.
Structure of PCAP-NG option:
.. code-block:: text
1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Option Code | Option Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
/ Option Value /
/ variable length, padded to 32 bits /
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
/ /
/ . . . other options . . . /
/ /
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Option Code == opt_endofopt | Option Length == 0 |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Args:
options_schema: Parsed PCAP-NG options.
Returns:
Parsed PCAP-NG options data.
"""
options = OrderedMultiDict() # type: Option
# an area kept as captured holds no options to read
if any(isinstance(schema, bytes) for schema in options_schema):
return options
for schema in options_schema:
type = schema.type
name = self._lookup_registry(self.__option__, _option_key(type))
if isinstance(name, str):
meth_name = f'_read_option_{name}'
meth = cast('OptionParser',
getattr(self, meth_name, self._read_option_unknown))
else:
meth = name[0]
data = meth(schema, options=options)
# NOTE: The padding to 32 bits, if it is not all zeros (#1437).
self._read_padding(schema, data)
# record option data
options.add(type, data)
self._opt[type] += 1
# break when ``opt_endofopt`` is reached
if type == Enum_OptionType.opt_endofopt:
break
return options
[docs]
def _read_option_unknown(self, schema: 'Schema_UnknownOption', *,
options: 'Option') -> 'Data_UnknownOption':
"""Read unknown PCAP-NG option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
option = Data_UnknownOption(
type=schema.type,
length=schema.length,
data=schema.data,
)
return option
[docs]
def _read_option_endofopt(self, schema: 'Schema_EndOfOption', *,
options: 'Option') -> 'Data_EndOfOption':
"""Read PCAP-NG ``opt_endofopt`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [opt_endofopt] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
if schema.length != 0:
raise ProtocolError(f'PCAP-NG: [opt_endofopt] invalid length (expected 0, got {schema.length})')
option = Data_EndOfOption(
type=schema.type,
length=schema.length,
)
return option
[docs]
def _read_option_custom(self, schema: 'Schema_CustomOption', *,
options: 'Option') -> 'Data_CustomOption':
"""Read PCAP-NG ``opt_custom`` option.
Structure of PCAP-NG ``opt_custom`` option:
.. code-block:: text
1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Custom Option Code | Option Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Private Enterprise Number (PEN) |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
/ Custom Data /
/ variable length, padded to 32 bits /
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
Raises:
ProtocolError: If the option is shorter than its 4-octet private
enterprise number, which Option Length includes.
"""
if schema.length < 4:
raise ProtocolError(f'PCAP-NG: [opt_custom] invalid length (expected at least 4, got {schema.length})')
option = Data_CustomOption(
type=schema.type,
length=schema.length,
pen=schema.pen,
data=schema.data,
)
return option
[docs]
def _read_option_if_name(self, schema: 'Schema_IF_NameOption', *,
options: 'Option') -> 'Data_IF_NameOption':
"""Read PCAP-NG ``if_name`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_name] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_name] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
option = Data_IF_NameOption(
type=schema.type,
length=schema.length,
name=schema.name,
)
return option
[docs]
def _read_option_if_description(self, schema: 'Schema_IF_DescriptionOption', *,
options: 'Option') -> 'Data_IF_DescriptionOption':
"""Read PCAP-NG ``if_description`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_description] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_description] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
option = Data_IF_DescriptionOption(
type=schema.type,
length=schema.length,
description=schema.description,
)
return option
[docs]
def _read_option_if_ipv4(self, schema: 'Schema_IF_IPv4AddrOption', *,
options: 'Option') -> 'Data_IF_IPv4AddrOption':
"""Read PCAP-NG ``if_IPv4addr`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_IPv4addr] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if schema.length != 8:
raise ProtocolError(f'PCAP-NG: [if_IPv4addr] invalid length (expected 8, got {schema.length})')
option = Data_IF_IPv4AddrOption(
type=schema.type,
length=schema.length,
interface=schema.interface,
)
return option
[docs]
def _read_option_if_ipv6(self, schema: 'Schema_IF_IPv6AddrOption', *,
options: 'Option') -> 'Data_IF_IPv6AddrOption':
"""Read PCAP-NG ``if_IPv6addr`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_IPv6addr] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if schema.length != 17:
raise ProtocolError(f'PCAP-NG: [if_IPv6addr] invalid length (expected 17, got {schema.length})')
option = Data_IF_IPv6AddrOption(
type=schema.type,
length=schema.length,
interface=schema.interface,
)
return option
[docs]
def _read_option_if_mac(self, schema: 'Schema_IF_MACAddrOption', *,
options: 'Option') -> 'Data_IF_MACAddrOption':
"""Read PCAP-NG ``if_MACaddr`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_MACaddr] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_MACaddr] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
if schema.length != 6:
raise ProtocolError(f'PCAP-NG: [if_MACaddr] invalid length (expected 6, got {schema.length})')
option = Data_IF_MACAddrOption(
type=schema.type,
length=schema.length,
interface=self._read_mac_addr(schema.interface),
)
return option
[docs]
def _read_option_if_eui(self, schema: 'Schema_IF_EUIAddrOption', *,
options: 'Option') -> 'Data_IF_EUIAddrOption':
"""Read PCAP-NG ``if_EUIaddr`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_EUIaddr] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_EUIaddr] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
if schema.length != 8:
raise ProtocolError(f'PCAP-NG: [if_EUIaddr] invalid length (expected 8, got {schema.length})')
option = Data_IF_EUIAddrOption(
type=schema.type,
length=schema.length,
interface=self._read_eui_addr(schema.interface),
)
return option
[docs]
def _read_option_if_speed(self, schema: 'Schema_IF_SpeedOption', *,
options: 'Option') -> 'Data_IF_SpeedOption':
"""Read PCAP-NG ``if_speed`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_speed] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_speed] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
if schema.length != 8:
raise ProtocolError(f'PCAP-NG: [if_speed] invalid length (expected 8, got {schema.length})')
option = Data_IF_SpeedOption(
type=schema.type,
length=schema.length,
speed=schema.speed,
)
return option
[docs]
def _read_option_if_tsresol(self, schema: 'Schema_IF_TSResolOption', *,
options: 'Option') -> 'Data_IF_TSResolOption':
"""Read PCAP-NG ``if_tsresol`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_tsresol] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_tsresol] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
if schema.length != 1:
raise ProtocolError(f'PCAP-NG: [if_tsresol] invalid length (expected 1, got {schema.length})')
option = Data_IF_TSResolOption(
type=schema.type,
length=schema.length,
resolution=schema.resolution,
)
return option
[docs]
def _read_option_if_tzone(self, schema: 'Schema_IF_TZoneOption', *,
options: 'Option') -> 'Data_IF_TZoneOption':
"""Read PCAP-NG ``if_tzone`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_tzone] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_tzone] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
if schema.length != 4:
raise ProtocolError(f'PCAP-NG: [if_tzone] invalid length (expected 4, got {schema.length})')
offset = datetime.timedelta(seconds=schema.tzone)
try:
timezone = datetime.timezone(offset) # type: datetime.timezone | datetime.timedelta
except ValueError:
# a :class:`~datetime.timezone` holds strictly less than 24 hours;
# the signed 32-bit field holds far more, so keep it as captured
warn(f'PCAP-NG: [if_tzone] offset of {schema.tzone} second(s) is not within 24 hours '
f'of UTC; kept as captured', ProtocolWarning, stacklevel=stacklevel())
timezone = offset
option = Data_IF_TZoneOption(
type=schema.type,
length=schema.length,
timezone=timezone,
)
return option
[docs]
def _read_option_if_filter(self, schema: 'Schema_IF_FilterOption', *,
options: 'Option') -> 'Data_IF_FilterOption':
"""Read PCAP-NG ``if_filter`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_filter] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_filter] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
if schema.length < 1:
raise ProtocolError(f'PCAP-NG: [if_filter] invalid length (expected 1+, got {schema.length})')
option = Data_IF_FilterOption(
type=schema.type,
length=schema.length,
code=schema.code,
expression=schema.filter,
)
return option
[docs]
def _read_option_if_os(self, schema: 'Schema_IF_OSOption', *,
options: 'Option') -> 'Data_IF_OSOption':
"""Read PCAP-NG ``if_os`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_os] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_os] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
option = Data_IF_OSOption(
type=schema.type,
length=schema.length,
os=schema.os,
)
return option
[docs]
def _read_option_if_fcslen(self, schema: 'Schema_IF_FCSLenOption', *,
options: 'Option') -> 'Data_IF_FCSLenOption':
"""Read PCAP-NG ``if_fcslen`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_fcslen] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_fcslen] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
if schema.length != 1:
raise ProtocolError(f'PCAP-NG: [if_fcslen] invalid length (expected 1, got {schema.length})')
option = Data_IF_FCSLenOption(
type=schema.type,
length=schema.length,
fcs_length=schema.fcslen,
)
return option
[docs]
def _read_option_if_tsoffset(self, schema: 'Schema_IF_TSOffsetOption', *,
options: 'Option') -> 'Data_IF_TSOffsetOption':
"""Read PCAP-NG ``if_tsoffset`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_tsoffset] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_tsoffset] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
if schema.length != 8:
raise ProtocolError(f'PCAP-NG: [if_tsoffset] invalid length (expected 8, got {schema.length})')
option = Data_IF_TSOffsetOption(
type=schema.type,
length=schema.length,
offset=schema.tsoffset,
)
return option
[docs]
def _read_option_if_hardware(self, schema: 'Schema_IF_HardwareOption', *,
options: 'Option') -> 'Data_IF_HardwareOption':
"""Read PCAP-NG ``if_hardware`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_hardware] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_hardware] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
option = Data_IF_HardwareOption(
type=schema.type,
length=schema.length,
hardware=schema.hardware,
)
return option
[docs]
def _read_option_if_txspeed(self, schema: 'Schema_IF_TxSpeedOption', *,
options: 'Option') -> 'Data_IF_TxSpeedOption':
"""Read PCAP-NG ``if_txspeed`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_txspeed] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_txspeed] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
if schema.length != 8:
raise ProtocolError(f'PCAP-NG: [if_txspeed] invalid length (expected 8, got {schema.length})')
option = Data_IF_TxSpeedOption(
type=schema.type,
length=schema.length,
speed=schema.tx_speed,
)
return option
[docs]
def _read_option_if_rxspeed(self, schema: 'Schema_IF_RxSpeedOption', *,
options: 'Option') -> 'Data_IF_RxSpeedOption':
"""Read PCAP-NG ``if_rxspeed`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_rxspeed] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_rxspeed] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
if schema.length != 8:
raise ProtocolError(f'PCAP-NG: [if_rxspeed] invalid length (expected 8, got {schema.length})')
option = Data_IF_RxSpeedOption(
type=schema.type,
length=schema.length,
speed=schema.rx_speed,
)
return option
[docs]
def _read_option_epb_flags(self, schema: 'Schema_EPB_FlagsOption', *,
options: 'Option') -> 'Data_EPB_FlagsOption':
"""Read PCAP-NG ``epb_flags`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Enhanced_Packet_Block:
raise ProtocolError(f'PCAP-NG: [epb_flags] option must be in Enhanced Packet Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [epb_flags] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
if schema.length != 4:
raise ProtocolError(f'PCAP-NG: [epb_flags] invalid length (expected 8, got {schema.length})')
option = Data_EPB_FlagsOption(
type=schema.type,
length=schema.length,
direction=PacketDirection(schema.flags['direction']),
reception=PacketReception(schema.flags['reception']),
fcs_len=schema.flags['fcs_len'],
checksum_not_ready=bool(schema.flags['checksum_not_ready']),
checksum_valid=bool(schema.flags['checksum_valid']),
tcp_segmentation_offloaded=bool(schema.flags['tcp_segmentation_offloaded']),
reserved=schema.flags['reserved'],
crc_error=bool(schema.flags['crc_error']),
too_long=bool(schema.flags['too_long']),
too_short=bool(schema.flags['too_short']),
gap_error=bool(schema.flags['gap_error']),
unaligned_error=bool(schema.flags['unaligned_error']),
delimiter_error=bool(schema.flags['delimiter_error']),
preamble_error=bool(schema.flags['preamble_error']),
symbol_error=bool(schema.flags['symbol_error']),
)
return option
[docs]
def _read_option_epb_hash(self, schema: 'Schema_EPB_HashOption', *,
options: 'Option') -> 'Data_EPB_HashOption':
"""Read PCAP-NG ``epb_hash`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Enhanced_Packet_Block:
raise ProtocolError(f'PCAP-NG: [epb_hash] option must be in Enhanced Packet Block, '
f'but found in {self._type} block.')
option = Data_EPB_HashOption(
type=schema.type,
length=schema.length,
algorithm=schema.func,
hash=schema.data,
)
return option
[docs]
def _read_option_epb_dropcount(self, schema: 'Schema_EPB_DropCountOption', *,
options: 'Option') -> 'Data_EPB_DropCountOption':
"""Read PCAP-NG ``epb_dropcount`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Enhanced_Packet_Block:
raise ProtocolError(f'PCAP-NG: [epb_dropcount] option must be in Enhanced Packet Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [epb_dropcount] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
if schema.length != 8:
raise ProtocolError(f'PCAP-NG: [epb_dropcount] invalid length (expected 8, got {schema.length})')
option = Data_EPB_DropCountOption(
type=schema.type,
length=schema.length,
drop_count=schema.drop_count,
)
return option
[docs]
def _read_option_epb_packetid(self, schema: 'Schema_EPB_PacketIDOption', *,
options: 'Option') -> 'Data_EPB_PacketIDOption':
"""Read PCAP-NG ``epb_packetid`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Enhanced_Packet_Block:
raise ProtocolError(f'PCAP-NG: [epb_packetid] option must be in Enhanced Packet Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [epb_packetid] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
if schema.length != 8:
raise ProtocolError(f'PCAP-NG: [epb_packetid] invalid length (expected 8, got {schema.length})')
option = Data_EPB_PacketIDOption(
type=schema.type,
length=schema.length,
packet_id=schema.packet_id,
)
return option
[docs]
def _read_option_epb_queue(self, schema: 'Schema_EPB_QueueOption', *,
options: 'Option') -> 'Data_EPB_QueueOption':
"""Read PCAP-NG ``epb_queue`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Enhanced_Packet_Block:
raise ProtocolError(f'PCAP-NG: [epb_queue] option must be in Enhanced Packet Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [epb_queue] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
if schema.length != 4:
raise ProtocolError(f'PCAP-NG: [epb_queue] invalid length (expected 4, got {schema.length})')
option = Data_EPB_QueueOption(
type=schema.type,
length=schema.length,
queue_id=schema.queue_id,
)
return option
[docs]
def _read_option_epb_verdict(self, schema: 'Schema_EPB_VerdictOption', *,
options: 'Option') -> 'Data_EPB_VerdictOption':
"""Read PCAP-NG ``epb_verdict`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Enhanced_Packet_Block:
raise ProtocolError(f'PCAP-NG: [epb_verdict] option must be in Enhanced Packet Block, '
f'but found in {self._type} block.')
if schema.length < 1:
raise ProtocolError(f'PCAP-NG: [epb_verdict] invalid length (expected 1+, got {schema.length})')
option = Data_EPB_VerdictOption(
type=schema.type,
length=schema.length,
verdict=schema.verdict,
value=schema.value,
)
return option
[docs]
def _read_option_ns_dnsname(self, schema: 'Schema_NS_DNSNameOption', *,
options: 'Option') -> 'Data_NS_DNSNameOption':
"""Read PCAP-NG ``ns_dnsname`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Name_Resolution_Block:
raise ProtocolError(f'PCAP-NG: [ns_dnsname] option must be in Name Resolution Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [ns_dnsname] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
option = Data_NS_DNSNameOption(
type=schema.type,
length=schema.length,
name=schema.name,
)
return option
[docs]
def _read_option_ns_dnsipv4(self, schema: 'Schema_NS_DNSIP4AddrOption', *,
options: 'Option') -> 'Data_NS_DNSIP4AddrOption':
"""Read PCAP-NG ``ns_dnsIP4addr`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Name_Resolution_Block:
raise ProtocolError(f'PCAP-NG: [ns_dnsIP4addr] option must be in Name Resolution Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [ns_dnsIP4addr] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
if schema.length != 4:
raise ProtocolError(f'PCAP-NG: [ns_dnsIP4addr] invalid length (expected 4, got {schema.length})')
option = Data_NS_DNSIP4AddrOption(
type=schema.type,
length=schema.length,
ip=schema.ip,
)
return option
[docs]
def _read_option_ns_dnsipv6(self, schema: 'Schema_NS_DNSIP6AddrOption', *,
options: 'Option') -> 'Data_NS_DNSIP6AddrOption':
"""Read PCAP-NG ``ns_dnsIP6addr`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Name_Resolution_Block:
raise ProtocolError(f'PCAP-NG: [ns_dnsIP6addr] option must be in Name Resolution Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [ns_dnsIP6addr] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
if schema.length != 16:
raise ProtocolError(f'PCAP-NG: [ns_dnsIP6addr] invalid length (expected 16, got {schema.length})')
option = Data_NS_DNSIP6AddrOption(
type=schema.type,
length=schema.length,
ip=schema.ip,
)
return option
[docs]
def _read_option_isb_starttime(self, schema: 'Schema_ISB_StartTimeOption', *,
options: 'Option') -> 'Data_ISB_StartTimeOption':
"""Read PCAP-NG ``isb_starttime`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Interface_Statistics_Block:
raise ProtocolError(f'PCAP-NG: [isb_starttime] option must be in Interface Statistics Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_starttime] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
if schema.length != 8:
raise ProtocolError(f'PCAP-NG: [isb_starttime] invalid length (expected 8, got {schema.length})')
timestamp, timestamp_epoch = self._read_timestamp(schema.timestamp_high, schema.timestamp_low,
interface_id=self._isb_interface_id)
option = Data_ISB_StartTimeOption(
type=schema.type,
length=schema.length,
timestamp=timestamp,
timestamp_epoch=timestamp_epoch,
)
return option
[docs]
def _read_option_isb_endtime(self, schema: 'Schema_ISB_EndTimeOption', *,
options: 'Option') -> 'Data_ISB_EndTimeOption':
"""Read PCAP-NG ``isb_endtime`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Interface_Statistics_Block:
raise ProtocolError(f'PCAP-NG: [isb_endtime] option must be in Interface Statistics Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_endtime] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
if schema.length != 8:
raise ProtocolError(f'PCAP-NG: [isb_endtime] invalid length (expected 8, got {schema.length})')
timestamp, timestamp_epoch = self._read_timestamp(schema.timestamp_high, schema.timestamp_low,
interface_id=self._isb_interface_id)
option = Data_ISB_EndTimeOption(
type=schema.type,
length=schema.length,
timestamp=timestamp,
timestamp_epoch=timestamp_epoch,
)
return option
[docs]
def _read_option_isb_ifrecv(self, schema: 'Schema_ISB_IFRecvOption', *,
options: 'Option') -> 'Data_ISB_IFRecvOption':
"""Read PCAP-NG ``isb_ifrecv`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Interface_Statistics_Block:
raise ProtocolError(f'PCAP-NG: [isb_ifrecv] option must be in Interface Statistics Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_ifrecv] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
if schema.length != 8:
raise ProtocolError(f'PCAP-NG: [isb_ifrecv] invalid length (expected 8, got {schema.length})')
option = Data_ISB_IFRecvOption(
type=schema.type,
length=schema.length,
packets=schema.packets,
)
return option
[docs]
def _read_option_isb_ifdrop(self, schema: 'Schema_ISB_IFDropOption', *,
options: 'Option') -> 'Data_ISB_IFDropOption':
"""Read PCAP-NG ``isb_ifdrop`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Interface_Statistics_Block:
raise ProtocolError(f'PCAP-NG: [isb_ifdrop] option must be in Interface Statistics Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_ifdrop] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
if schema.length != 8:
raise ProtocolError(f'PCAP-NG: [isb_ifdrop] invalid length (expected 8, got {schema.length})')
option = Data_ISB_IFDropOption(
type=schema.type,
length=schema.length,
packets=schema.packets,
)
return option
[docs]
def _read_option_isb_filteraccept(self, schema: 'Schema_ISB_FilterAcceptOption', *,
options: 'Option') -> 'Data_ISB_FilterAcceptOption':
"""Read PCAP-NG ``isb_filteraccept`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Interface_Statistics_Block:
raise ProtocolError(f'PCAP-NG: [isb_filteraccept] option must be in Interface Statistics Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_filteraccept] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
if schema.length != 8:
raise ProtocolError(f'PCAP-NG: [isb_filteraccept] invalid length (expected 8, got {schema.length})')
option = Data_ISB_FilterAcceptOption(
type=schema.type,
length=schema.length,
packets=schema.packets,
)
return option
[docs]
def _read_option_isb_osdrop(self, schema: 'Schema_ISB_OSDropOption', *,
options: 'Option') -> 'Data_ISB_OSDropOption':
"""Read PCAP-NG ``isb_osdrop`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Interface_Statistics_Block:
raise ProtocolError(f'PCAP-NG: [isb_osdrop] option must be in Interface Statistics Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_osdrop] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
if schema.length != 8:
raise ProtocolError(f'PCAP-NG: [isb_osdrop] invalid length (expected 8, got {schema.length})')
option = Data_ISB_OSDropOption(
type=schema.type,
length=schema.length,
packets=schema.packets,
)
return option
[docs]
def _read_option_isb_usrdeliv(self, schema: 'Schema_ISB_UsrDelivOption', *,
options: 'Option') -> 'Data_ISB_UsrDelivOption':
"""Read PCAP-NG ``isb_usrdeliv`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Interface_Statistics_Block:
raise ProtocolError(f'PCAP-NG: [isb_usrdeliv] option must be in Interface Statistics Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_usrdeliv] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
if schema.length != 8:
raise ProtocolError(f'PCAP-NG: [isb_usrdeliv] invalid length (expected 8, got {schema.length})')
option = Data_ISB_UsrDelivOption(
type=schema.type,
length=schema.length,
packets=schema.packets,
)
return option
[docs]
def _read_option_pack_flags(self, schema: 'Schema_PACK_FlagsOption', *,
options: 'Option') -> 'Data_PACK_FlagsOption':
"""Read PCAP-NG ``pack_flags`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Packet_Block:
raise ProtocolError(f'PCAP-NG: [pack_flags] option must be in Packet Block, '
f'but found in {self._type} block.')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [pack_flags] option must be only one, '
f'but {self._opt[schema.type] + 1} found.')
if schema.length != 4:
raise ProtocolError(f'PCAP-NG: [pack_flags] invalid length (expected 8, got {schema.length})')
option = Data_PACK_FlagsOption(
type=schema.type,
length=schema.length,
direction=PacketDirection(schema.flags['direction']),
reception=PacketReception(schema.flags['reception']),
fcs_len=schema.flags['fcs_len'],
checksum_not_ready=bool(schema.flags['checksum_not_ready']),
checksum_valid=bool(schema.flags['checksum_valid']),
tcp_segmentation_offloaded=bool(schema.flags['tcp_segmentation_offloaded']),
reserved=schema.flags['reserved'],
crc_error=bool(schema.flags['crc_error']),
too_long=bool(schema.flags['too_long']),
too_short=bool(schema.flags['too_short']),
gap_error=bool(schema.flags['gap_error']),
unaligned_error=bool(schema.flags['unaligned_error']),
delimiter_error=bool(schema.flags['delimiter_error']),
preamble_error=bool(schema.flags['preamble_error']),
symbol_error=bool(schema.flags['symbol_error']),
)
return option
[docs]
def _read_option_pack_hash(self, schema: 'Schema_PACK_HashOption', *,
options: 'Option') -> 'Data_PACK_HashOption':
"""Read PCAP-NG ``pack_hash`` option.
Args:
schema: Parsed option schema.
options: Parsed PCAP-NG options.
Returns:
Parsed option data.
"""
if self._type != Enum_BlockType.Packet_Block:
raise ProtocolError(f'PCAP-NG: [pack_hash] option must be in Packet Block, '
f'but found in {self._type} block.')
option = Data_PACK_HashOption(
type=schema.type,
length=schema.length,
algorithm=schema.func,
hash=schema.data,
)
return option
[docs]
def _read_nrb_records(self, records_schema: 'list[Schema_NameResolutionRecord]') -> 'Record':
"""Read PCAP-NG name resolution records.
Args:
records_schema: Parsed name resolution records.
Returns:
Parsed PCAP-NG name resolution records data.
"""
records = OrderedMultiDict() # type: Record
# an area kept as captured holds no records to read
if any(isinstance(schema, bytes) for schema in records_schema):
return records
for schema in records_schema:
type = schema.type
name = self._lookup_registry(self.__record__, type)
if isinstance(name, str):
meth_name = f'_read_record_{name}'
meth = cast('RecordParser',
getattr(self, meth_name, self._read_record_unknown))
else:
meth = name[0]
data = meth(schema, records=records)
# NOTE: The padding to 32 bits, if it is not all zeros (#1437).
self._read_padding(schema, data)
# record option data
records.add(type, data)
# break when ``nrb_record_end`` is reached
if type == Enum_RecordType.nrb_record_end:
break
return records
[docs]
def _read_record_unknown(self, schema: 'Schema_UnknownRecord', *,
records: 'Record') -> 'Data_UnknownRecord':
"""Read PCAP-NG unknown name resolution records.
Args:
schema: Parsed name resolution record schema.
records: Parsed PCAP-NG records.
Returns:
Parsed name resolution record data.
"""
record = Data_UnknownRecord(
type=schema.type,
length=schema.length,
data=schema.data,
)
return record
[docs]
def _read_record_end(self, schema: 'Schema_EndRecord', *,
records: 'Record') -> 'Data_EndRecord':
"""Read PCAP-NG ``nrb_record_end`` name resolution records.
Args:
schema: Parsed name resolution record schema.
records: Parsed PCAP-NG records.
Returns:
Parsed name resolution record data.
"""
if schema.length != 0:
raise ProtocolError(f'PCAP-NG: [nrb_record_end] invalid length (expected 0, got {schema.length})')
record = Data_EndRecord(
type=schema.type,
length=schema.length,
)
return record
[docs]
def _read_record_ipv4(self, schema: 'Schema_IPv4Record', *,
records: 'Record') -> 'Data_IPv4Record':
"""Read PCAP-NG ``nrb_record_ipv4`` name resolution records.
Args:
schema: Parsed name resolution record schema.
records: Parsed PCAP-NG records.
Returns:
Parsed name resolution record data.
"""
record = Data_IPv4Record(
type=schema.type,
length=schema.length,
ip=schema.ip,
records=tuple(schema.names),
)
return record
[docs]
def _read_record_ipv6(self, schema: 'Schema_IPv6Record', *,
records: 'Record') -> 'Data_IPv6Record':
"""Read PCAP-NG ``nrb_record_ipv6`` name resolution records.
Args:
schema: Parsed name resolution record schema.
records: Parsed PCAP-NG records.
Returns:
Parsed name resolution record data.
"""
record = Data_IPv6Record(
type=schema.type,
length=schema.length,
ip=schema.ip,
records=tuple(schema.names),
)
return record
[docs]
def _read_secrets_unknown(self, schema: 'Schema_UnknownSecrets', *,
block: 'Schema_DecryptionSecretsBlock') -> 'Data_UnknownSecrets':
"""Read PCAP-NG unknown secrets.
Args:
schema: Parsed secret schema.
block: Parsed PCAP-NG decryption secrets block.
"""
secrets = Data_UnknownSecrets(
data=schema.data,
)
return secrets
[docs]
def _read_secrets_tls(self, schema: 'Schema_TLSKeyLog', *,
block: 'Schema_DecryptionSecretsBlock') -> 'Data_TLSKeyLog':
"""Read PCAP-NG TLS key log secrets.
Args:
schema: Parsed secret schema.
block: Parsed PCAP-NG decryption secrets block.
"""
secrets = Data_TLSKeyLog(
# NOTE: each label's entries as an OrderedMultiInfo, which ``Info``
# does not do for a mapping held in a dict (#1484).
entries={label: OrderedMultiInfo(entry) for (label, entry) in schema.entries.items()},
)
# NOTE: the log text is kept as well, since :attr:`entries` holds
# neither its comments nor the order of its lines, and a rebuild writes
# it back verbatim.
secrets.__update__(data=schema.data)
return secrets
[docs]
def _read_secrets_wireguard(self, schema: 'Schema_WireGuardKeyLog', *,
block: 'Schema_DecryptionSecretsBlock') -> 'Data_WireGuardKeyLog':
"""Read PCAP-NG WireGuard key log secrets.
Args:
schema: Parsed secret schema.
block: Parsed PCAP-NG decryption secrets block.
Returns:
Parsed decryption secrets data.
"""
secrets = Data_WireGuardKeyLog(
entries=schema.entries,
)
# NOTE: the log text is kept as well, since :attr:`entries` holds
# neither its comments nor the order of its lines, and a rebuild writes
# it back verbatim.
secrets.__update__(data=schema.data)
return secrets
[docs]
def _read_secrets_zigbee_nwk(self, schema: 'Schema_ZigBeeNWKKey', *,
block: 'Schema_DecryptionSecretsBlock') -> 'Data_ZigBeeNWKKey':
"""Read PCAP-NG ZigBee NWK Key secrets.
Structure of ZigBee NWK Key secrets:
.. code-block:: text
0 1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+---------------------------------------------------------------+
0 | Block Type = 0x0000000A |
+---------------------------------------------------------------+
4 | Block Total Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
8 | Secrets Type = 0x5a4e574b |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
12 | Secrets Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
16 | AES-128 |
| NWK Key |
| (16 octets) |
| (128 bits) |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
32 | PAN ID | padding (0) |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
36 / /
/ Options (variable) /
/ /
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
/ Block Total Length /
+---------------------------------------------------------------+
Args:
schema: Parsed secret schema.
block: Parsed PCAP-NG decryption secrets block.
Returns:
Parsed decryption secrets data.
"""
secrets = Data_ZigBeeNWKKey(
nwk_key=schema.key,
pan_id=schema.panid,
)
# NOTE: the octets as the block holds them are kept as well, since a
# Secrets Length may declare fewer or more than the key's fixed width,
# and a rebuild writes them back verbatim (#1986).
secrets.__update__(data=bytes(schema))
return secrets
[docs]
def _read_secrets_zigbee_aps(self, schema: 'Schema_ZigBeeAPSKey', *,
block: 'Schema_DecryptionSecretsBlock') -> 'Data_ZigBeeAPSKey':
"""Read PCAP-NG ZigBee APS Key secrets.
Structure of ZigBee APS Key secrets:
.. code-block:: text
0 1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+---------------------------------------------------------------+
0 | Block Type = 0x0000000A |
+---------------------------------------------------------------+
4 | Block Total Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
8 | Secrets Type = 0x5a415053 |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
12 | Secrets Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
16 | AES-128 |
| APS Key |
| (16 octets) |
| (128 bits) |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
32 | PAN ID | Low Node Short Address |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
36 | High Node Short Address | padding (0) |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
40 / /
/ Options (variable) /
/ /
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
/ Block Total Length /
+---------------------------------------------------------------+
Args:
schema: Parsed secret schema.
block: Parsed PCAP-NG decryption secrets block.
Returns:
Parsed decryption secrets data.
"""
secrets = Data_ZigBeeAPSKey(
aps_key=schema.key,
pan_id=schema.panid,
short_address=schema.addr_high << 16 | schema.addr_low,
)
# NOTE: kept as for ``_read_secrets_zigbee_nwk`` (#1986).
secrets.__update__(data=bytes(schema))
return secrets
[docs]
def _make_block_unknown(self, block: 'Optional[Data_UnknownBlock]' = None, *,
data: 'bytes' = b'',
**kwargs: 'Any') -> 'Schema_UnknownBlock':
"""Make unknown PCAP-NG block.
Args:
block: Block data model.
data: Unspecified block data.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed block schema.
"""
if block is not None:
data = block.body
# NOTE: The body is padded to 32 bits, and Block Total Length counts it
# together with the 12 octets of block type and both length fields.
data += bytes(-len(data) % 4)
length = len(data) + GENERIC_BLOCK_MINIMUM
return Schema_UnknownBlock(
length=length,
body=data,
length2=length,
)
[docs]
def _make_block_shb(self, block: 'Optional[Data_SectionHeaderBlock]' = None, *,
version: 'tuple[int, int] | VersionInfo' = (1, 0),
major_version: 'Optional[int]' = None,
minor_version: 'Optional[int]' = None,
section_length: 'int' = -1,
options: 'Optional[Option | list[Schema_Option | tuple[Enum_OptionType, dict[str, Any]] | bytes]]' = None,
byteorder: 'Literal["big", "little"]' = sys.byteorder,
**kwargs: 'Any') -> 'Schema_SectionHeaderBlock':
"""Make PCAP-NG section header block (SHB).
Args:
block: Block data model.
version: Version information.
major_version: Major version number.
minor_version: Minor version number.
section_length: Section length.
options: Block options.
byteorder: Byte order of the section; a parsed block keeps its own.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed block schema.
"""
if block is not None:
byteorder = block.byteorder
self._byte = byteorder
if block is not None and hasattr(block, 'block_raw'):
return self._make_block_short(block) # type: ignore[return-value]
if block is not None:
major_version = block.version.major
minor_version = block.version.minor
section_length = block.section_length
options = self._block_area(block, 'options')
else:
if major_version is None:
major_version = version[0]
if minor_version is None:
minor_version = version[1]
if options is not None:
options_value, total_length = self._make_pcapng_options(options, namespace='opt')
else:
options_value, total_length = [], 0
total_length += self._make_options_trailer(block, options_value)
return Schema_SectionHeaderBlock(
length=total_length + BLOCK_MINIMUM[Enum_BlockType.Section_Header_Block],
magic=BYTE_ORDER_MAGIC,
major=major_version,
minor=minor_version,
section_length=section_length,
options=options_value,
length2=total_length + BLOCK_MINIMUM[Enum_BlockType.Section_Header_Block],
)
[docs]
def _make_block_idb(self, block: 'Optional[Data_InterfaceDescriptionBlock]' = None, *,
linktype: 'Enum_LinkType | StdlibEnum | AenumEnum | str | int' = Enum_LinkType.NULL,
linktype_default: 'Optional[int]' = None,
linktype_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long
linktype_reversed: 'bool' = False,
snaplen: 'int' = 0xFFFF_FFFF_FFFF_FFFF,
options: 'Optional[Option | list[Schema_Option | tuple[Enum_OptionType, dict[str, Any]] | bytes]]' = None,
**kwargs: 'Any') -> 'Schema_InterfaceDescriptionBlock':
"""Make PCAP-NG interface description block (IDB).
Args:
block: Block data model.
linktype: Link layer protocol type.
linktype_default: Default value of link layer protocol type.
linktype_namespace: Namespace of link layer protocol type.
linktype_reversed: Reversed flag for link layer protocol type namespace.
snaplen: Snap length.
options: Block options.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed block schema.
"""
if block is not None and hasattr(block, 'block_raw'):
return self._make_block_short(block) # type: ignore[return-value]
if block is not None:
linktype_val = block.linktype
snaplen = block.snaplen
options = self._block_area(block, 'options')
else:
linktype_val = self._make_index(linktype, linktype_default, namespace=linktype_namespace, # type: ignore[assignment]
reversed=linktype_reversed, pack=False)
if options is not None:
options_value, total_length = self._make_pcapng_options(options, namespace='if')
else:
options_value, total_length = [], 0
total_length += self._make_options_trailer(block, options_value)
return self._make_padding(Schema_InterfaceDescriptionBlock(
length=total_length + BLOCK_MINIMUM[Enum_BlockType.Interface_Description_Block],
linktype=linktype_val,
snaplen=snaplen,
options=options_value,
length2=total_length + BLOCK_MINIMUM[Enum_BlockType.Interface_Description_Block],
), block, 'reserved')
[docs]
def _make_block_epb(self, block: 'Optional[Data_EnhancedPacketBlock]' = None, *,
interface_id: 'int' = 0,
timestamp: 'Optional[dt_type | int | float | Decimal]' = None,
captured_len: 'Optional[int]' = None,
original_len: 'Optional[int]' = None,
packet_data: 'bytes | ProtocolBase | Schema' = b'',
options: 'Optional[Option | list[Schema_Option | tuple[Enum_OptionType, dict[str, Any]] | bytes]]' = None,
**kwargs: 'Any') -> 'Schema_EnhancedPacketBlock':
"""Make PCAP-NG enhanced packet block (EPB).
Args:
block: Block data model.
interface_id: Interface ID.
timestamp: Packet timestamp.
captured_len: Captured length.
original_len: Original length.
packet_data: Payload of the block.
options: Block options.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed block schema.
"""
if block is not None and hasattr(block, 'block_raw'):
return self._make_block_short(block) # type: ignore[return-value]
if block is not None:
interface_id = block.interface_id
timestamp = block.timestamp_epoch
captured_len = block.captured_len
original_len = block.original_len
packet_data = self._make_packet_data(block)
options = self._block_area(block, 'options')
timestamp_high, timestamp_low = self._make_timestamp(timestamp, interface_id=interface_id)
if captured_len is None:
if self._ctx is None:
snaplen = 0xFFFF_FFFF_FFFF_FFFF
else:
# a snaplen of zero means no limit
snaplen = self._get_interface(interface_id).snaplen or 0xFFFF_FFFF_FFFF_FFFF
captured_len = min(len(packet_data), snaplen)
if original_len is None:
original_len = len(packet_data)
# NOTE: Packet Data holds exactly Captured Packet Length octets, so a
# packet longer than that -- e.g. than the interface's snaplen -- is cut
# to it rather than written out whole behind a clipped length.
if len(packet_data) > captured_len:
packet_data = bytes(packet_data)[:captured_len]
packet_len = math.ceil(len(packet_data) / 4) * 4
if options is not None:
options_value, total_length = self._make_pcapng_options(options, namespace='epb')
else:
options_value, total_length = [], 0
total_length += self._make_options_trailer(block, options_value)
return self._make_padding(Schema_EnhancedPacketBlock(
length=total_length + BLOCK_MINIMUM[Enum_BlockType.Enhanced_Packet_Block] + packet_len,
interface_id=interface_id,
timestamp_high=timestamp_high,
timestamp_low=timestamp_low,
captured_len=captured_len,
original_len=original_len,
packet_data=packet_data,
options=options_value,
length2=total_length + BLOCK_MINIMUM[Enum_BlockType.Enhanced_Packet_Block] + packet_len,
), block, 'padding_data')
[docs]
def _make_block_spb(self, block: 'Optional[Data_SimplePacketBlock]' = None, *,
original_len: 'Optional[int]' = None,
packet_data: 'bytes | ProtocolBase | Schema' = b'',
**kwargs: 'Any') -> 'Schema_SimplePacketBlock':
"""Make PCAP-NG simple packet block (SPB).
Args:
block: Block data model.
original_len: Original length.
packet_data: Payload of the block.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed block schema.
"""
if block is not None:
original_len = block.original_len
packet_data = self._make_packet_data(block)
# NOTE: The data model's ``captured_len`` already bounds the data,
# including a block that kept octets past the snaplen (#1384).
captured_len = block.captured_len
else:
if original_len is None:
original_len = len(packet_data)
# NOTE: An SPB always belongs to interface 0, and its Packet Data is
# ``min(original_len, snaplen)`` octets long -- the length the parser
# reads back -- so a longer packet is cut to it.
if self._ctx is None or not self._ctx.interfaces:
snaplen = 0xFFFF_FFFF_FFFF_FFFF
else:
snaplen = self._ctx.interfaces[0].snaplen or 0xFFFF_FFFF_FFFF_FFFF
captured_len = min(original_len, snaplen)
if len(packet_data) > captured_len:
packet_data = bytes(packet_data)[:captured_len]
packet_len = math.ceil(len(packet_data) / 4) * 4
return self._make_padding(Schema_SimplePacketBlock(
length=packet_len + BLOCK_MINIMUM[Enum_BlockType.Simple_Packet_Block],
original_len=original_len,
packet_data=packet_data,
length2=packet_len + BLOCK_MINIMUM[Enum_BlockType.Simple_Packet_Block],
), block, 'padding', 'padding_data')
[docs]
def _make_block_short(
self, block: 'Data_EnhancedPacketBlock | Data_PacketBlock | Data_SectionHeaderBlock | Data_InterfaceDescriptionBlock', # pylint: disable=line-too-long
) -> 'Schema_UnknownBlock':
"""Make a PCAP-NG EPB, packet block (obsolete), SHB or IDB kept as captured.
Args:
block: Block data model, whose ``block_raw`` holds the body of a
block shorter than its fixed fields (see :meth:`_read_block_short`
and :meth:`_read_block_short_header`).
Returns:
Constructed block schema, the body written back verbatim.
"""
return self._make_block_unknown(data=block.block_raw) # type: ignore[union-attr]
[docs]
def _make_block_nrb(self, block: 'Optional[Data_NameResolutionBlock]' = None, *,
records: 'Optional[Record | list[Schema_NameResolutionRecord | tuple[Enum_RecordType, dict[str, Any]] | bytes]]' = None,
options: 'Optional[Option | list[Schema_Option | tuple[Enum_OptionType, dict[str, Any]] | bytes]]' = None,
**kwargs: 'Any') -> 'Schema_NameResolutionBlock':
"""Make PCAP-NG name resolution block (NRB).
Args:
block: Block data model.
records: Name resolution records.
options: Block options.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed block schema.
"""
if block is not None:
records = self._block_area(block, 'records')
options = self._block_area(block, 'options')
if records is not None:
records_value, records_length = self._make_nrb_records(records)
else:
records_value, records_length = [], 0
if options is not None:
options_value, options_length = self._make_pcapng_options(options, namespace='ns')
else:
options_value, options_length = [], 0
options_length += self._make_options_trailer(block, options_value)
return Schema_NameResolutionBlock(
length=options_length + records_length + GENERIC_BLOCK_MINIMUM,
records=records_value,
options=options_value,
length2=options_length + records_length + GENERIC_BLOCK_MINIMUM,
)
[docs]
def _make_block_isb(self, block: 'Optional[Data_InterfaceStatisticsBlock]' = None, *,
interface_id: 'int' = 0,
timestamp: 'Optional[dt_type | int | float | Decimal]' = None,
options: 'Optional[Option | list[Schema_Option | tuple[Enum_OptionType, dict[str, Any]] | bytes]]' = None,
**kwargs: 'Any') -> 'Schema_InterfaceStatisticsBlock':
"""Make PCAP-NG interface statistics block (ISB).
Args:
block: Block data model.
interface_id: Interface ID.
timestamp: Block timestamp.
options: Block options.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed block schema.
"""
if block is not None:
interface_id = block.interface_id
timestamp = block.timestamp_epoch
options = self._block_area(block, 'options')
timestamp_high, timestamp_low = self._make_timestamp(timestamp, interface_id=interface_id)
# NOTE: ``isb_starttime`` and ``isb_endtime`` are scaled by the
# resolution of this block's interface, as on the parse path.
self._isb_interface_id = interface_id # pylint: disable=attribute-defined-outside-init
if options is not None:
options_value, total_length = self._make_pcapng_options(options, namespace='isb')
else:
options_value, total_length = [], 0
total_length += self._make_options_trailer(block, options_value)
return Schema_InterfaceStatisticsBlock(
length=total_length + BLOCK_MINIMUM[Enum_BlockType.Interface_Statistics_Block],
interface_id=interface_id,
timestamp_high=timestamp_high,
timestamp_low=timestamp_low,
options=options_value,
length2=total_length + BLOCK_MINIMUM[Enum_BlockType.Interface_Statistics_Block],
)
[docs]
def _make_block_systemd(self, block: 'Optional[Data_SystemdJournalExportBlock]' = None, *,
entries: 'Optional[list[OrderedMultiDict[str, str | bytes]] | bytes]' = None,
**kwargs: 'Any') -> 'Schema_SystemdJournalExportBlock':
"""Make PCAP-NG :manpage:`systemd(1)` journal export block.
Args:
block: Block data model.
entries: :manpage:`systemd(1)` journal export entries.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed block schema.
"""
self._byte = sys.byteorder
if block is not None:
entries = cast('list[OrderedMultiDict[str, str | bytes]] | bytes',
getattr(block, 'entry_raw', block.data))
if entries is None:
data = b''
elif isinstance(entries, bytes):
data = entries
else:
data = Schema_SystemdJournalExportBlock.dump_entries(entries)
# NOTE: The entry MUST be zero-padded to 32 bits, and the Block Total
# Length counts that padding, as the ``entry`` field reads it back.
data += bytes(-len(data) % 4)
return Schema_SystemdJournalExportBlock(
length=len(data) + GENERIC_BLOCK_MINIMUM,
entry=data,
length2=len(data) + GENERIC_BLOCK_MINIMUM,
)
[docs]
def _make_block_dsb(self, block: 'Optional[Data_DecryptionSecretsBlock]' = None, *,
secrets_type: 'Enum_SecretsType | StdlibEnum | AenumEnum | str | int' = Enum_SecretsType.TLS_Key_Log,
secrets_type_default: 'Optional[int]' = None,
secrets_type_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long
secrets_type_reversed: 'bool' = False,
secrets_data: 'Schema_DSBSecrets | Data_DSBSecrets | bytes | dict[str, Any]' = b'',
options: 'Optional[Option | list[Schema_Option | tuple[Enum_OptionType, dict[str, Any]] | bytes]]' = None,
**kwargs: 'Any') -> 'Schema_DecryptionSecretsBlock':
"""Make PCAP-NG decryption secrets block (DSB).
Args:
block: Block data model.
secrets_type: Decryption secrets type.
secrets_type_default: Default value of decryption secrets type.
secrets_type_namespace: Namespace of decryption secrets type.
secrets_type_reversed: Reversed flag for namespace of decryption secrets type.
secrets_data: Decryption secrets data.
options: Block options.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed block schema.
"""
declared = None # type: Optional[int]
if block is not None:
secrets_type_val = block.secrets_type
secrets_data = block.secrets_data
declared = block.secrets_length
options = self._block_area(block, 'options')
else:
secrets_type_val = self._make_index(secrets_type, secrets_type_default, namespace=secrets_type_namespace, # type: ignore[assignment]
reversed=secrets_type_reversed, pack=False)
if isinstance(secrets_data, bytes):
secrets_data_val = secrets_data # type: bytes | Schema_DSBSecrets
elif isinstance(secrets_data, (dict, Data_DSBSecrets)):
name = self._lookup_registry(self.__secrets__, secrets_type_val)
if isinstance(name, str):
meth_name = f'_make_secrets_{name}'
meth = cast('SecretsConstructor',
getattr(self, meth_name, self._make_secrets_unknown))
else:
meth = name[1]
if isinstance(secrets_data, dict):
secrets_data_val = meth(secrets_type_val, **secrets_data)
else:
secrets_data_val = meth(secrets_type_val, secrets_data)
elif isinstance(secrets_data, Schema):
secrets_data_val = secrets_data
else:
raise ProtocolError(f'PCAP-NG: [DSB] secrets {secrets_type_val} invalid format')
# NOTE: Secrets Length excludes the padding to the next 32-bit boundary,
# which ``padding_data`` writes and the block length has to include.
if block is not None and isinstance(secrets_data_val, Schema):
# NOTE: Packed in the context of the block it was parsed from, so
# that a key log the block cut is still read as cut (#1980).
secrets_held = len(secrets_data_val.pack({'__packet__': {
'length': block.length, 'secrets_length': block.secrets_length,
}}))
else:
secrets_held = len(secrets_data_val)
secrets_padded = secrets_held + -secrets_held % 4
# NOTE: A Secrets Length running past the block is kept as declared, as
# an EPB keeps its captured length, so the block rebuilds as captured;
# the block length is still sized from the secrets written (#1980).
# Only a true overrun is kept -- the one the parse side bounds and
# warns on, measured by ``secrets_area`` -- so a length that fits the
# block, or one an edit has outdated, follows the secrets written.
from pcapkit.protocols.schema.misc.pcapng import secrets_area
if block is not None and declared is not None and declared > secrets_area({
'length': block.length, 'secrets_length': declared,
}):
secrets_length = declared
else:
secrets_length = secrets_held
if options is not None:
options_value, total_length = self._make_pcapng_options(options, namespace='dsb')
else:
options_value, total_length = [], 0
total_length += self._make_options_trailer(block, options_value)
return self._make_padding(Schema_DecryptionSecretsBlock(
length=total_length + secrets_padded + BLOCK_MINIMUM[Enum_BlockType.Decryption_Secrets_Block],
secrets_type=secrets_type_val,
secrets_length=secrets_length,
secrets_data=secrets_data_val,
options=options_value,
length2=total_length + secrets_padded + BLOCK_MINIMUM[Enum_BlockType.Decryption_Secrets_Block],
), block, 'padding_data')
[docs]
def _make_block_cb(self, block: 'Optional[Data_CustomBlock]' = None, *,
pen: 'int' = 0,
data: 'Schema | bytes' = b'',
options: 'Optional[Option | list[Schema_Option | tuple[Enum_OptionType, dict[str, Any]] | bytes]]' = None,
**kwargs: 'Any') -> 'Schema_CustomBlock':
"""Make PCAP-NG custom block (CB).
Args:
block: Block data model.
pen: Private enterprise number.
data: Custom data.
options: Block options.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed block schema.
"""
if block is not None:
pen = block.pen
data = block.data
options = cast('Option', getattr(block, 'options', None))
# NOTE: Unlike every other block, a custom block has no ``OptionField``
# to name an option registry: :class:`~pcapkit.protocols.schema.misc.pcapng.CustomBlock`
# spans the custom data, its padding and the options as one opaque blob,
# since only the owner of the private enterprise number knows where the
# custom data ends. Its options are therefore resolved in the generic
# ``opt_*`` space, which is the only one valid in every block.
if options is not None:
options_value, _ = self._make_pcapng_options(options, namespace='opt')
else:
options_value, _ = [], 0
# NOTE: ``-n % 4`` is the octet count needed to reach the next 32-bit
# boundary, and zero when already there. Integer arithmetic throughout,
# unlike the ``math.ceil(n / 4) * 4 - n`` idiom used elsewhere in this
# file, which routes a length through a float.
cb_data = data.pack() if isinstance(data, Schema) else data
cb_data += bytes(-len(cb_data) % 4)
for option in options_value:
cb_data += option.pack() if isinstance(option, Schema) else option
cb_data += bytes(-len(cb_data) % 4)
# NOTE: both custom block types span the same fixed fields, and this
# method makes either, so the one entry stands for both.
minimum = BLOCK_MINIMUM[Enum_BlockType.Custom_Block_that_rewriters_can_copy_into_new_files]
return Schema_CustomBlock(
length=len(cb_data) + minimum,
pen=pen,
data=cb_data,
length2=len(cb_data) + minimum,
)
[docs]
def _make_block_packet(self, block: 'Optional[Data_PacketBlock]' = None, *,
interface_id: 'int' = 0,
drop_count: 'int' = 0,
timestamp: 'Optional[dt_type | int | float | Decimal]' = None,
captured_len: 'Optional[int]' = None,
original_len: 'Optional[int]' = None,
packet_data: 'bytes | ProtocolBase | Schema' = b'',
options: 'Optional[Option | list[Schema_Option | tuple[Enum_OptionType, dict[str, Any]] | bytes]]' = None,
**kwargs: 'Any') -> 'Schema_PacketBlock':
"""Make PCAP-NG packet block (obsolete).
Args:
block: Block data model.
interface_id: Interface ID.
drop_count: Drops count.
timestamp: Packet timestamp.
captured_len: Captured length.
original_len: Original length.
packet_data: Payload of the block.
options: Block options.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed block schema.
"""
warn('PCAP-NG: Packet Block has been obsolete! Please use Enhanced Packet Block and/or '
'Simple Packet Block instead.', DeprecatedFormatWarning, stacklevel=stacklevel())
if block is not None and hasattr(block, 'block_raw'):
return self._make_block_short(block) # type: ignore[return-value]
if block is not None:
interface_id = block.interface_id
drop_count = block.drop_count
timestamp = block.timestamp_epoch
captured_len = block.captured_len
original_len = block.original_len
packet_data = self._make_packet_data(block)
options = self._block_area(block, 'options')
timestamp_high, timestamp_low = self._make_timestamp(timestamp, interface_id=interface_id)
if captured_len is None:
if self._ctx is None:
snaplen = 0xFFFF_FFFF_FFFF_FFFF
else:
# a snaplen of zero means no limit
snaplen = self._get_interface(interface_id).snaplen or 0xFFFF_FFFF_FFFF_FFFF
captured_len = min(len(packet_data), snaplen)
if original_len is None:
original_len = len(packet_data)
# NOTE: Packet Data holds exactly Captured Packet Length octets, so a
# packet longer than that -- e.g. than the interface's snaplen -- is cut
# to it rather than written out whole behind a clipped length.
if len(packet_data) > captured_len:
packet_data = bytes(packet_data)[:captured_len]
packet_len = math.ceil(len(packet_data) / 4) * 4
if options is not None:
options_value, total_length = self._make_pcapng_options(options, namespace='pack')
else:
options_value, total_length = [], 0
total_length += self._make_options_trailer(block, options_value)
return self._make_padding(Schema_PacketBlock(
length=total_length + BLOCK_MINIMUM[Enum_BlockType.Packet_Block] + packet_len,
interface_id=interface_id,
drop_count=drop_count,
timestamp_high=timestamp_high,
timestamp_low=timestamp_low,
captured_length=captured_len,
original_length=original_len,
packet_data=packet_data,
options=options_value,
length2=total_length + BLOCK_MINIMUM[Enum_BlockType.Packet_Block] + packet_len,
), block, 'padding_data')
[docs]
def _make_pcapng_options(self, options: 'Option | list[Schema_Option | tuple[Enum_OptionType, dict[str, Any]] | bytes] | bytes',
namespace: 'str') -> 'tuple[list[Schema_Option | bytes], int]':
"""Make options for PCAP-NG.
Args:
options: PCAP-NG options.
namespace: Namespace of options.
Returns:
Tuple of options and total length of options.
"""
# an option area kept as captured is written back verbatim
if isinstance(options, bytes):
return [options], len(options)
has_endofopt = False
total_length = 0
if isinstance(options, list):
options_list = [] # type: list[Schema_Option | bytes]
for schema in options:
if isinstance(schema, bytes):
code = Enum_OptionType.get(int.from_bytes(schema[0:2], self._byte, signed=False),
namespace=namespace)
if code == Enum_OptionType.opt_endofopt: # moved to the end, written once
has_endofopt = True
continue
data = schema # type: Schema_Option | bytes
data_len = len(data)
elif isinstance(schema, Schema):
code = schema.type
if code == Enum_OptionType.opt_endofopt: # moved to the end, written once
has_endofopt = True
continue
data = schema
data_len = len(schema.pack())
else:
code, args = cast('tuple[Enum_OptionType, dict[str, Any]]', schema)
if code == Enum_OptionType.opt_endofopt: # moved to the end, written once
has_endofopt = True
continue
name = self._lookup_registry(self.__option__, _option_key(code))
if isinstance(name, str):
meth_name = f'_make_option_{name}'
meth = cast('OptionConstructor',
getattr(self, meth_name, self._make_option_unknown))
else:
meth = name[1]
data = self._make_padding(meth(code, **args), args)
data_len = len(data.pack())
options_list.append(data)
total_length += data_len
self._opt[code] += 1
if has_endofopt:
opt_endofopt = self._make_option_endofopt(Enum_OptionType.opt_endofopt)
total_length += len(opt_endofopt.pack())
options_list.append(opt_endofopt)
return options_list, total_length
options_list = []
for code, option in options.items(multi=True):
if code == Enum_OptionType.opt_endofopt: # moved to the end, written once
has_endofopt = True
continue
name = self._lookup_registry(self.__option__, _option_key(code))
if isinstance(name, str):
meth_name = f'_make_option_{name}'
meth = cast('OptionConstructor',
getattr(self, meth_name, self._make_option_unknown))
else:
meth = name[1]
data = self._make_padding(meth(code, option), option)
data_len = len(data.pack())
options_list.append(data)
total_length += data_len
self._opt[code] += 1
if has_endofopt:
opt_endofopt = self._make_option_endofopt(Enum_OptionType.opt_endofopt)
total_length += len(opt_endofopt.pack())
options_list.append(opt_endofopt)
return options_list, total_length
[docs]
def _make_option_unknown(self, type: 'Enum_OptionType', option: 'Optional[Data_UnknownOption]' = None, *,
data: 'bytes' = b'',
**kwargs: 'Any') -> 'Schema_UnknownOption':
"""Make unknown PCAP-NG option.
Args:
type: Option type.
option: Option data model.
data: Unspecified option data.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if option is not None:
data = option.data
length = len(data)
return Schema_UnknownOption(
type=type,
length=length,
data=data,
)
[docs]
def _make_option_endofopt(self, type: 'Enum_OptionType', option: 'Optional[Data_EndOfOption]' = None,
**kwargs: 'Any') -> 'Schema_EndOfOption':
"""Make PCAP-NG ``opt_endofopt`` option.
Args:
type: Option type.
option: Option data model.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [opt_endofopt] option must be only one, '
f'but {self._opt[type] + 1} found.')
return Schema_EndOfOption(
type=type,
length=0,
)
[docs]
def _make_option_custom(self, type: 'Enum_OptionType', option: 'Optional[Data_CustomOption]' = None, *,
pen: 'int' = 0xFFFFFFFF,
data: 'bytes' = b'',
**kwargs: 'Any') -> 'Schema_CustomOption':
"""Make PCAP-NG ``opt_custom`` option.
Args:
type: Option type.
option: Option data model.
pen: Private enterprise number.
data: Custom data.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if option is not None:
pen = option.pen
data = option.data
return Schema_CustomOption(
type=type,
length=len(data) + 4,
pen=pen,
data=data
)
[docs]
def _make_option_if_name(self, type: 'Enum_OptionType', option: 'Optional[Data_IF_NameOption]' = None, *,
name: 'str' = '',
**kwargs: 'Any') -> 'Schema_IF_NameOption':
"""Make PCAP-NG ``if_name`` option.
Args:
type: Option type.
option: Option data model.
name: Interface name.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_name] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_name] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
name = option.name
return Schema_IF_NameOption(
type=type,
length=_utf8_length(name),
name=name,
)
[docs]
def _make_option_if_description(self, type: 'Enum_OptionType', option: 'Optional[Data_IF_DescriptionOption]' = None, *,
description: 'str' = '',
**kwargs: 'Any') -> 'Schema_IF_DescriptionOption':
"""Make PCAP-NG ``if_description`` option.
Args:
type: Option type.
option: Option data model.
description: Interface description.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_description] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_description] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
description = option.description
return Schema_IF_DescriptionOption(
type=type,
length=_utf8_length(description),
description=description,
)
[docs]
def _make_option_if_ipv4(self, type: 'Enum_OptionType', option: 'Optional[Data_IF_IPv4AddrOption]' = None, *,
interface: 'IPv4Interface | str' = '192.168.1.1/255.255.255.0',
**kwargs: 'Any') -> 'Schema_IF_IPv4AddrOption':
"""Make PCAP-NG ``if_IPv4addr`` option.
Args:
type: Option type.
option: Option data model.
interface: IPv4 interface.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_IPv4addr] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if option is not None:
interface = option.interface
return Schema_IF_IPv4AddrOption(
type=type,
length=8,
interface=interface,
)
[docs]
def _make_option_if_ipv6(self, type: 'Enum_OptionType', option: 'Optional[Data_IF_IPv6AddrOption]' = None, *,
interface: 'IPv6Interface | str' = '2001:0db8:85a3:08d3:1319:8a2e:0370:7344/64',
**kwargs: 'Any') -> 'Schema_IF_IPv6AddrOption':
"""Make PCAP-NG ``if_IPv6addr`` option.
Args:
type: Option type.
option: Option data model.
interface: IPv6 interface.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_IPv6addr] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if option is not None:
interface = option.interface
return Schema_IF_IPv6AddrOption(
type=type,
length=17,
interface=interface,
)
[docs]
def _make_option_if_mac(self, type: 'Enum_OptionType', option: 'Optional[Data_IF_MACAddrOption]' = None, *,
interface: 'str | bytes | bytearray' = '00:01:02:03:04:05',
**kwargs: 'Any') -> 'Schema_IF_MACAddrOption':
"""Make PCAP-NG ``if_MACaddr`` option.
Args:
type: Option type.
option: Option data model.
interface: MAC interface address.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_MACaddr] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_MACaddr] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
interface = option.interface
return Schema_IF_MACAddrOption(
type=type,
length=6,
interface=self._make_mac_addr(interface),
)
[docs]
def _make_option_if_eui(self, type: 'Enum_OptionType', option: 'Optional[Data_IF_EUIAddrOption]' = None, *,
interface: 'str | bytes | bytearray' = '02:34:56:FF:FE:78:9A:BC',
**kwargs: 'Any') -> 'Schema_IF_EUIAddrOption':
"""Make PCAP-NG ``if_EUIaddr`` option.
Args:
type: Option type.
option: Option data model.
interface: Hardware EUI address.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_EUIaddr] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_EUIaddr] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
interface = option.interface
return Schema_IF_EUIAddrOption(
type=type,
length=8,
interface=self._make_eui_addr(interface),
)
[docs]
def _make_option_if_speed(self, type: 'Enum_OptionType', option: 'Optional[Data_IF_SpeedOption]' = None, *,
speed: 'int' = 100000000,
**kwargs: 'Any') -> 'Schema_IF_SpeedOption':
"""Make PCAP-NG ``if_speed`` option.
Args:
type: Option type.
option: Option data model.
speed: Interface speed, in bits per second.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_speed] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_speed] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
speed = option.speed
return Schema_IF_SpeedOption(
type=type,
length=8,
speed=speed,
)
[docs]
def _make_option_if_tsresol(self, type: 'Enum_OptionType', option: 'Optional[Data_IF_TSResolOption]' = None, *,
resolution: 'int' = DEFAULT_RESOLUTION,
**kwargs: 'Any') -> 'Schema_IF_TSResolOption':
"""Make PCAP-NG ``if_tsresol`` option.
Args:
type: Option type.
option: Option data model.
resolution: Resolution of timestamps, in units per second.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_tsresol] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_tsresol] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
resolution = option.resolution
test = math.log10(resolution).as_integer_ratio()[1]
if test == 1:
flag = 0
resl = int(math.log10(resolution))
else:
test = math.log2(resolution).as_integer_ratio()[1]
if test == 1:
flag = 1
resl = int(math.log2(resolution))
else:
raise ProtocolError(f'PCAP-NG: [if_tsresol] option resolution must be power of 10 or 2, '
f'but {resolution} found.')
return Schema_IF_TSResolOption(
type=type,
length=1,
tsresol={
'flag': flag,
'resolution': resl,
},
)
[docs]
def _make_option_if_tzone(self, type: 'Enum_OptionType', option: 'Optional[Data_IF_TZoneOption]' = None, *,
tzone: 'timezone | timedelta | int' = 0,
**kwargs: 'Any') -> 'Schema_IF_TZoneOption':
"""Make PCAP-NG ``if_tzone`` option.
Args:
type: Option type.
option: Option data model.
tzone: Timezone offset, in seconds.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_tzone] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_tzone] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
tzone = option.timezone
if isinstance(tzone, int):
tzone_val = tzone
elif isinstance(tzone, datetime.timedelta):
tzone_val = int(tzone.total_seconds())
elif isinstance(tzone, datetime.timezone):
tzone_val = int(tzone.utcoffset(None).total_seconds())
else:
raise ProtocolError(f'PCAP-NG: [if_tzone] option timezone must be int, timedelta or timezone, '
f'but {builtins.type(tzone).__name__} found.')
return Schema_IF_TZoneOption(
type=type,
length=4,
tzone=tzone_val,
)
[docs]
def _make_option_if_filter(self, type: 'Enum_OptionType', option: 'Optional[Data_IF_FilterOption]' = None, *,
filter: 'Enum_FilterType | StdlibEnum | AenumEnum | str | int' = Enum_FilterType(0),
filter_default: 'Optional[int]' = None,
filter_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long
filter_reversed: 'bool' = False,
expression: 'bytes | str' = b'',
**kwargs: 'Any') -> 'Schema_IF_FilterOption':
"""Make PCAP-NG ``if_filter`` option.
Args:
type: Option type.
option: Option data model.
filter: Filter type.
filter_default: Default filter value.
filter_namespace: Filter namespace.
filter_reversed: Whether filter namespace is reversed.
expression: Filter expression.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_filter] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_filter] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
filter_val = option.code
expr_val = option.expression
else:
filter_val = self._make_index(filter, filter_default, namespace=filter_namespace, # type: ignore[assignment]
reversed=filter_reversed, pack=False)
expr_val = expression if isinstance(expression, bytes) else expression.encode()
return Schema_IF_FilterOption(
type=type,
length=1 + len(expr_val),
code=filter_val,
filter=expr_val,
)
[docs]
def _make_option_if_os(self, type: 'Enum_OptionType', option: 'Optional[Data_IF_OSOption]' = None, *,
os: 'str' = platform.platform(),
**kwargs: 'Any') -> 'Schema_IF_OSOption':
"""Make PCAP-NG ``if_os`` option.
Args:
type: Option type.
option: Option data model.
os: Operating system name.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_os] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_os] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
os = option.os
return Schema_IF_OSOption(
type=type,
length=_utf8_length(os),
os=os,
)
[docs]
def _make_option_if_fcslen(self, type: 'Enum_OptionType', option: 'Optional[Data_IF_FCSLenOption]' = None, *,
fcs_length: 'int' = 4,
**kwargs: 'Any') -> 'Schema_IF_FCSLenOption':
"""Make PCAP-NG ``if_fcslen`` option.
Args:
type: Option type.
option: Option data model.
fcs_length: FCS length, in bytes.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_fcslen] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_fcslen] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
fcs_length = option.fcs_length
return Schema_IF_FCSLenOption(
type=type,
length=1,
fcslen=fcs_length,
)
[docs]
def _make_option_if_tsoffset(self, type: 'Enum_OptionType', option: 'Optional[Data_IF_TSOffsetOption]' = None, *,
offset: 'int' = 0,
**kwargs: 'Any') -> 'Schema_IF_TSOffsetOption':
"""Make PCAP-NG ``if_tsoffset`` option.
Args:
type: Option type.
option: Option data model.
offset: Timestamp offset in seconds.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_tsoffset] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_tsoffset] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
offset = option.offset
return Schema_IF_TSOffsetOption(
type=type,
length=8,
tsoffset=offset,
)
[docs]
def _make_option_if_hardware(self, type: 'Enum_OptionType', option: 'Optional[Data_IF_HardwareOption]' = None, *,
hardware: 'str' = platform.processor(),
**kwargs: 'Any') -> 'Schema_IF_HardwareOption':
"""Make PCAP-NG ``if_hardware`` option.
Args:
type: Option type.
option: Option data model.
hardware: Hardware information.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_hardware] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_hardware] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
hardware = option.hardware
return Schema_IF_HardwareOption(
type=type,
length=_utf8_length(hardware),
hardware=hardware,
)
[docs]
def _make_option_if_txspeed(self, type: 'Enum_OptionType', option: 'Optional[Data_IF_TxSpeedOption]' = None, *,
speed: 'int' = 100000000,
**kwargs: 'Any') -> 'Schema_IF_TxSpeedOption':
"""Make PCAP-NG ``if_txspeed`` option.
Args:
type: Option type.
option: Option data model.
speed: Interface transmit speed, in bits per second.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_txspeed] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_txspeed] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
speed = option.speed
return Schema_IF_TxSpeedOption(
type=type,
length=8,
tx_speed=speed,
)
[docs]
def _make_option_if_rxspeed(self, type: 'Enum_OptionType', option: 'Optional[Data_IF_RxSpeedOption]' = None, *,
speed: 'int' = 100000000,
**kwargs: 'Any') -> 'Schema_IF_RxSpeedOption':
"""Make PCAP-NG ``if_rxspeed`` option.
Args:
type: Option type.
option: Option data model.
speed: Interface receive speed, in bits per second.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_rxspeed] option must be in Interface Description Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_rxspeed] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
speed = option.speed
return Schema_IF_RxSpeedOption(
type=type,
length=8,
rx_speed=speed,
)
[docs]
def _make_option_epb_flags(self, type: 'Enum_OptionType', option: 'Optional[Data_EPB_FlagsOption]' = None, *,
direction: 'PacketDirection | StdlibEnum | AenumEnum | str | int' = PacketDirection.UNKNOWN,
direction_default: 'Optional[int]' = None,
direction_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long
direction_reversed: 'bool' = False,
reception: 'PacketReception | StdlibEnum | AenumEnum | str | int' = PacketReception.UNKNOWN,
reception_default: 'Optional[int]' = None,
reception_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long
reception_reversed: 'bool' = False,
fcs_len: 'int' = 0,
checksum_not_ready: 'bool' = False,
checksum_valid: 'bool' = False,
tcp_segmentation_offloaded: 'bool' = False,
reserved: 'int' = 0,
crc_error: 'bool' = False,
too_long: 'bool' = False,
too_short: 'bool' = False,
gap_error: 'bool' = False,
unaligned_error: 'bool' = False,
delimiter_error: 'bool' = False,
preamble_error: 'bool' = False,
symbol_error: 'bool' = False,
**kwargs: 'Any') -> 'Schema_EPB_FlagsOption':
"""Make PCAP-NG ``epb_flags`` option.
Args:
type: Option type.
option: Option data model.
direction: Packet direction.
direction_default: Default value of packet direction.
direction_namespace: Namespace of packet direction.
direction_reversed: Whether to reverse packet direction namespace.
reception: Packet reception.
reception_default: Default value of packet reception.
reception_namespace: Namespace of packet reception.
reception_reversed: Whether to reverse packet reception namespace.
fcs_len: Length of FCS field, in bytes.
checksum_not_ready: Whether the checksum is not ready.
checksum_valid: Whether the checksum is valid.
tcp_segmentation_offloaded: Whether TCP segmentation was offloaded.
reserved: Bits 12 to 23 of the flags word, verbatim.
crc_error: Whether CRC error occurred.
too_long: Whether packet is too long.
too_short: Whether packet is too short.
gap_error: Whether gap error occurred.
unaligned_error: Whether unaligned error occurred.
delimiter_error: Whether delimiter error occurred.
preamble_error: Whether preamble error occurred.
symbol_error: Whether symbol error occurred.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Enhanced_Packet_Block:
raise ProtocolError(f'PCAP-NG: [epb_flags] option must be in Enhanced Packet Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [epb_flags] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
direction_val = option.direction
reception_val = option.reception
fcs_len = option.fcs_len
checksum_not_ready = option.checksum_not_ready
checksum_valid = option.checksum_valid
tcp_segmentation_offloaded = option.tcp_segmentation_offloaded
reserved = option.reserved
crc_error = option.crc_error
too_long = option.too_long
too_short = option.too_short
gap_error = option.gap_error
unaligned_error = option.unaligned_error
delimiter_error = option.delimiter_error
preamble_error = option.preamble_error
symbol_error = option.symbol_error
else:
direction_val = self._make_index(direction, direction_default, namespace=direction_namespace, # type: ignore[assignment]
reversed=direction_reversed, pack=False)
reception_val = self._make_index(reception, reception_default, namespace=reception_namespace, # type: ignore[assignment]
reversed=reception_reversed, pack=False)
return Schema_EPB_FlagsOption(
type=type,
length=4,
flags={
'direction': getattr(direction_val, 'value', direction_val),
'reception': getattr(reception_val, 'value', reception_val),
'fcs_len': fcs_len,
'checksum_not_ready': int(checksum_not_ready),
'checksum_valid': int(checksum_valid),
'tcp_segmentation_offloaded': int(tcp_segmentation_offloaded),
'reserved': reserved,
'crc_error': int(crc_error),
'too_long': int(too_long),
'too_short': int(too_short),
'gap_error': int(gap_error),
'unaligned_error': int(unaligned_error),
'delimiter_error': int(delimiter_error),
'preamble_error': int(preamble_error),
'symbol_error': int(symbol_error),
},
)
[docs]
def _make_option_epb_hash(self, type: 'Enum_OptionType', option: 'Optional[Data_EPB_HashOption]' = None, *,
algorithm: 'Enum_HashAlgorithm | StdlibEnum | AenumEnum | int | str' = Enum_HashAlgorithm.two_s_complement,
algorithm_default: 'Optional[int]' = None,
algorithm_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long
algorithm_reversed: 'bool' = False,
hash: 'bytes' = b'',
**kwargs: 'Any') -> 'Schema_EPB_HashOption':
"""Make PCAP-NG ``epb_hash`` option.
Args:
type: Option type.
option: Option data model.
algorithm: Hash algorithm.
algorithm_default: Default value of hash algorithm.
algorithm_namespace: Namespace of hash algorithm.
algorithm_reversed: Whether to reverse hash algorithm namespace.
hash: Hash value.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Enhanced_Packet_Block:
raise ProtocolError(f'PCAP-NG: [epb_hash] option must be in Enhanced Packet Block, '
f'but found in {self._type} block.')
if option is not None:
algo_val = option.algorithm
hash = option.hash
else:
algo_val = self._make_index(algorithm, algorithm_default, namespace=algorithm_namespace, # type: ignore[assignment]
reversed=algorithm_reversed, pack=False)
return Schema_EPB_HashOption(
type=type,
length=1 + len(hash),
func=algo_val,
data=hash,
)
[docs]
def _make_option_epb_dropcount(self, type: 'Enum_OptionType', option: 'Optional[Data_EPB_DropCountOption]' = None, *,
drop_count: 'int' = 0,
**kwargs: 'Any') -> 'Schema_EPB_DropCountOption':
"""Make PCAP-NG ``epb_dropcount`` option.
Args:
type: Option type.
option: Option data model.
drop_count: Number of dropped packets.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Enhanced_Packet_Block:
raise ProtocolError(f'PCAP-NG: [epb_dropcount] option must be in Enhanced Packet Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [epb_dropcount] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
drop_count = option.drop_count
return Schema_EPB_DropCountOption(
type=type,
length=8,
drop_count=drop_count,
)
[docs]
def _make_option_epb_packetid(self, type: 'Enum_OptionType', option: 'Optional[Data_EPB_PacketIDOption]' = None, *,
packet_id: 'int' = 0,
**kwargs: 'Any') -> 'Schema_EPB_PacketIDOption':
"""Make PCAP-NG ``epb_packetid`` option.
Args:
type: Option type.
option: Option data model.
packet_id: Packet ID.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Enhanced_Packet_Block:
raise ProtocolError(f'PCAP-NG: [epb_packetid] option must be in Enhanced Packet Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [epb_packetid] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
packet_id = option.packet_id
return Schema_EPB_PacketIDOption(
type=type,
length=8,
packet_id=packet_id,
)
[docs]
def _make_option_epb_queue(self, type: 'Enum_OptionType', option: 'Optional[Data_EPB_QueueOption]' = None, *,
queue_id: 'int' = 0,
**kwargs: 'Any') -> 'Schema_EPB_QueueOption':
"""Make PCAP-NG ``epb_queue`` option.
Args:
type: Option type.
option: Option data model.
queue_id: Queue ID.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Enhanced_Packet_Block:
raise ProtocolError(f'PCAP-NG: [epb_queue] option must be in Enhanced Packet Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [epb_queue] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
queue_id = option.queue_id
return Schema_EPB_QueueOption(
type=type,
length=4,
queue_id=queue_id,
)
[docs]
def _make_option_epb_verdict(self, type: 'Enum_OptionType', option: 'Optional[Data_EPB_VerdictOption]' = None, *,
verdict: 'Enum_VerdictType | StdlibEnum | AenumEnum | str | int' = Enum_VerdictType.Hardware,
verdict_default: 'Optional[int]' = None,
verdict_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long
verdict_reversed: 'bool' = False,
value: 'bytes' = b'',
**kwargs: 'Any') -> 'Schema_EPB_VerdictOption':
"""Make PCAP-NG ``epb_verdict`` option.
Args:
type: Option type.
option: Option data model.
verdict: Verdict type.
verdict_default: Default value for verdict.
verdict_namespace: Namespace for verdict.
verdict_reversed: Whether to reverse the namespace.
value: Verdict value.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Enhanced_Packet_Block:
raise ProtocolError(f'PCAP-NG: [epb_verdict] option must be in Enhanced Packet Block, '
f'but found in {self._type} block.')
# NOTE: No only-one guard: the specification allows ``epb_verdict`` to
# appear multiple times, and the parser accepts every occurrence.
if option is not None:
verdict_val = option.verdict
value = option.value
else:
verdict_val = self._make_index(verdict, verdict_default, namespace=verdict_namespace, # type: ignore[assignment]
reversed=verdict_reversed, pack=False)
return Schema_EPB_VerdictOption(
type=type,
length=1 + len(value),
verdict=verdict_val,
value=value,
)
[docs]
def _make_option_ns_dnsname(self, type: 'Enum_OptionType', option: 'Optional[Data_NS_DNSNameOption]' = None, *,
name: 'str' = '',
**kwargs: 'Any') -> 'Schema_NS_DNSNameOption':
"""Make PCAP-NG ``ns_dnsname`` option.
Args:
type: Option type.
option: Option data model.
name: DNS server name.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Name_Resolution_Block:
raise ProtocolError(f'PCAP-NG: [ns_dnsname] option must be in Name Resolution Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [ns_dnsname] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
name = option.name
return Schema_NS_DNSNameOption(
type=type,
length=_utf8_length(name),
name=name,
)
[docs]
def _make_option_ns_dnsipv4(self, type: 'Enum_OptionType', option: 'Optional[Data_NS_DNSIP4AddrOption]' = None, *,
ip: 'str | bytes | IPv4Address | int' = '8.8.8.8',
**kwargs: 'Any') -> 'Schema_NS_DNSIP4AddrOption':
"""Make PCAP-NG ``ns_dnsIP4addr`` option.
Args:
type: Option type.
option: Option data model.
ip: DNS server IPv4 address.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Name_Resolution_Block:
raise ProtocolError(f'PCAP-NG: [ns_dnsIP4addr] option must be in Name Resolution Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [ns_dnsIP4addr] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
ip = option.ip
return Schema_NS_DNSIP4AddrOption(
type=type,
length=4,
ip=ip,
)
[docs]
def _make_option_ns_dnsipv6(self, type: 'Enum_OptionType', option: 'Optional[Data_NS_DNSIP6AddrOption]' = None, *,
ip: 'str | bytes | IPv6Address | int' = '8.8.8.8',
**kwargs: 'Any') -> 'Schema_NS_DNSIP6AddrOption':
"""Make PCAP-NG ``ns_dnsIP6addr`` option.
Args:
type: Option type.
option: Option data model.
ip: DNS server IPv6 address.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Name_Resolution_Block:
raise ProtocolError(f'PCAP-NG: [ns_dnsIP6addr] option must be in Name Resolution Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [ns_dnsIP6addr] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
ip = option.ip
return Schema_NS_DNSIP6AddrOption(
type=type,
length=16,
ip=ip,
)
[docs]
def _make_option_isb_starttime(self, type: 'Enum_OptionType', option: 'Optional[Data_ISB_StartTimeOption]' = None, *,
timestamp: 'Optional[int | float | dt_type | Decimal]' = None,
**kwargs: 'Any') -> 'Schema_ISB_StartTimeOption':
"""Make PCAP-NG ``isb_starttime`` option.
Args:
type: Option type.
option: Option data model.
timestamp: Start time.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Interface_Statistics_Block:
raise ProtocolError(f'PCAP-NG: [isb_starttime] option must be in Interface Statistics Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_starttime] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
timestamp = option.timestamp_epoch
ts_high, ts_low = self._make_timestamp(timestamp, interface_id=self._isb_interface_id)
return Schema_ISB_StartTimeOption(
type=type,
length=8,
timestamp_high=ts_high,
timestamp_low=ts_low,
)
[docs]
def _make_option_isb_endtime(self, type: 'Enum_OptionType', option: 'Optional[Data_ISB_EndTimeOption]' = None, *,
timestamp: 'Optional[int | float | dt_type | Decimal]' = None,
**kwargs: 'Any') -> 'Schema_ISB_EndTimeOption':
"""Make PCAP-NG ``isb_endtime`` option.
Args:
type: Option type.
option: Option data model.
timestamp: End time.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Interface_Statistics_Block:
raise ProtocolError(f'PCAP-NG: [isb_endtime] option must be in Interface Statistics Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_endtime] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
timestamp = option.timestamp_epoch
ts_high, ts_low = self._make_timestamp(timestamp, interface_id=self._isb_interface_id)
return Schema_ISB_EndTimeOption(
type=type,
length=8,
timestamp_high=ts_high,
timestamp_low=ts_low,
)
[docs]
def _make_option_isb_ifrecv(self, type: 'Enum_OptionType', option: 'Optional[Data_ISB_IFRecvOption]' = None, *,
packets: 'int' = 0,
**kwargs: 'Any') -> 'Schema_ISB_IFRecvOption':
"""Make PCAP-NG ``isb_ifrecv`` option.
Args:
type: Option type.
option: Option data model.
packets: Number of received packets.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Interface_Statistics_Block:
raise ProtocolError(f'PCAP-NG: [isb_ifrecv] option must be in Interface Statistics Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_ifrecv] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
packets = option.packets
return Schema_ISB_IFRecvOption(
type=type,
length=8,
packets=packets,
)
[docs]
def _make_option_isb_ifdrop(self, type: 'Enum_OptionType', option: 'Optional[Data_ISB_IFDropOption]' = None, *,
packets: 'int' = 0,
**kwargs: 'Any') -> 'Schema_ISB_IFDropOption':
"""Make PCAP-NG ``isb_ifdrop`` option.
Args:
type: Option type.
option: Option data model.
packets: Number of dropped packets.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Interface_Statistics_Block:
raise ProtocolError(f'PCAP-NG: [isb_ifdrop] option must be in Interface Statistics Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_ifdrop] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
packets = option.packets
return Schema_ISB_IFDropOption(
type=type,
length=8,
packets=packets,
)
[docs]
def _make_option_isb_filteraccept(self, type: 'Enum_OptionType', option: 'Optional[Data_ISB_FilterAcceptOption]' = None, *,
packets: 'int' = 0,
**kwargs: 'Any') -> 'Schema_ISB_FilterAcceptOption':
"""Make PCAP-NG ``isb_filteraccept`` option.
Args:
type: Option type.
option: Option data model.
packets: Number of packets accepted by the filter.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Interface_Statistics_Block:
raise ProtocolError(f'PCAP-NG: [isb_filteraccept] option must be in Interface Statistics Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_filteraccept] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
packets = option.packets
return Schema_ISB_FilterAcceptOption(
type=type,
length=8,
packets=packets,
)
[docs]
def _make_option_isb_osdrop(self, type: 'Enum_OptionType', option: 'Optional[Data_ISB_OSDropOption]' = None, *,
packets: 'int' = 0,
**kwargs: 'Any') -> 'Schema_ISB_OSDropOption':
"""Make PCAP-NG ``isb_osdrop`` option.
Args:
type: Option type.
option: Option data model.
packets: Number of packets dropped by the OS.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Interface_Statistics_Block:
raise ProtocolError(f'PCAP-NG: [isb_osdrop] option must be in Interface Statistics Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_osdrop] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
packets = option.packets
return Schema_ISB_OSDropOption(
type=type,
length=8,
packets=packets,
)
[docs]
def _make_option_isb_usrdeliv(self, type: 'Enum_OptionType', option: 'Optional[Data_ISB_UsrDelivOption]' = None, *,
packets: 'int' = 0,
**kwargs: 'Any') -> 'Schema_ISB_UsrDelivOption':
"""Make PCAP-NG ``isb_usrdeliv`` option.
Args:
type: Option type.
option: Option data model.
packets: Number of packets delivered to the user.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Interface_Statistics_Block:
raise ProtocolError(f'PCAP-NG: [isb_usrdeliv] option must be in Interface Statistics Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_usrdeliv] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
packets = option.packets
return Schema_ISB_UsrDelivOption(
type=type,
length=8,
packets=packets,
)
[docs]
def _make_option_pack_flags(self, type: 'Enum_OptionType', option: 'Optional[Data_PACK_FlagsOption]' = None, *,
direction: 'PacketDirection | StdlibEnum | AenumEnum | str | int' = PacketDirection.UNKNOWN,
direction_default: 'Optional[int]' = None,
direction_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long
direction_reversed: 'bool' = False,
reception: 'PacketReception | StdlibEnum | AenumEnum | str | int' = PacketReception.UNKNOWN,
reception_default: 'Optional[int]' = None,
reception_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long
reception_reversed: 'bool' = False,
fcs_len: 'int' = 0,
checksum_not_ready: 'bool' = False,
checksum_valid: 'bool' = False,
tcp_segmentation_offloaded: 'bool' = False,
reserved: 'int' = 0,
crc_error: 'bool' = False,
too_long: 'bool' = False,
too_short: 'bool' = False,
gap_error: 'bool' = False,
unaligned_error: 'bool' = False,
delimiter_error: 'bool' = False,
preamble_error: 'bool' = False,
symbol_error: 'bool' = False,
**kwargs: 'Any') -> 'Schema_PACK_FlagsOption':
"""Make PCAP-NG ``pack_flags`` option.
Args:
type: Option type.
option: Option data model.
direction: Packet direction.
direction_default: Default value of packet direction.
direction_namespace: Namespace of packet direction.
direction_reversed: Whether to reverse packet direction namespace.
reception: Packet reception.
reception_default: Default value of packet reception.
reception_namespace: Namespace of packet reception.
reception_reversed: Whether to reverse packet reception namespace.
fcs_len: Length of FCS field, in bytes.
checksum_not_ready: Whether the checksum is not ready.
checksum_valid: Whether the checksum is valid.
tcp_segmentation_offloaded: Whether TCP segmentation was offloaded.
reserved: Bits 12 to 23 of the flags word, verbatim.
crc_error: Whether CRC error occurred.
too_long: Whether packet is too long.
too_short: Whether packet is too short.
gap_error: Whether gap error occurred.
unaligned_error: Whether unaligned error occurred.
delimiter_error: Whether delimiter error occurred.
preamble_error: Whether preamble error occurred.
symbol_error: Whether symbol error occurred.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Packet_Block:
raise ProtocolError(f'PCAP-NG: [pack_flags] option must be in Packet Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [pack_flags] option must be only one, '
f'but {self._opt[type] + 1} found.')
if option is not None:
direction_val = option.direction
reception_val = option.reception
fcs_len = option.fcs_len
checksum_not_ready = option.checksum_not_ready
checksum_valid = option.checksum_valid
tcp_segmentation_offloaded = option.tcp_segmentation_offloaded
reserved = option.reserved
crc_error = option.crc_error
too_long = option.too_long
too_short = option.too_short
gap_error = option.gap_error
unaligned_error = option.unaligned_error
delimiter_error = option.delimiter_error
preamble_error = option.preamble_error
symbol_error = option.symbol_error
else:
direction_val = self._make_index(direction, direction_default, namespace=direction_namespace, # type: ignore[assignment]
reversed=direction_reversed, pack=False)
reception_val = self._make_index(reception, reception_default, namespace=reception_namespace, # type: ignore[assignment]
reversed=reception_reversed, pack=False)
return Schema_PACK_FlagsOption(
type=type,
length=4,
flags={
'direction': getattr(direction_val, 'value', direction_val),
'reception': getattr(reception_val, 'value', reception_val),
'fcs_len': fcs_len,
'checksum_not_ready': int(checksum_not_ready),
'checksum_valid': int(checksum_valid),
'tcp_segmentation_offloaded': int(tcp_segmentation_offloaded),
'reserved': reserved,
'crc_error': int(crc_error),
'too_long': int(too_long),
'too_short': int(too_short),
'gap_error': int(gap_error),
'unaligned_error': int(unaligned_error),
'delimiter_error': int(delimiter_error),
'preamble_error': int(preamble_error),
'symbol_error': int(symbol_error),
},
)
[docs]
def _make_option_pack_hash(self, type: 'Enum_OptionType', option: 'Optional[Data_PACK_HashOption]' = None, *,
algorithm: 'Enum_HashAlgorithm | StdlibEnum | AenumEnum | int | str' = Enum_HashAlgorithm.two_s_complement,
algorithm_default: 'Optional[int]' = None,
algorithm_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long
algorithm_reversed: 'bool' = False,
hash: 'bytes' = b'',
**kwargs: 'Any') -> 'Schema_PACK_HashOption':
"""Make PCAP-NG ``pack_hash`` option.
Args:
type: Option type.
option: Option data model.
algorithm: Hash algorithm.
algorithm_default: Default value of hash algorithm.
algorithm_namespace: Namespace of hash algorithm.
algorithm_reversed: Whether to reverse hash algorithm namespace.
hash: Hash value.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
"""
if self._type != Enum_BlockType.Packet_Block:
raise ProtocolError(f'PCAP-NG: [pack_hash] option must be in Packet Block, '
f'but found in {self._type} block.')
if option is not None:
algo_val = option.algorithm
hash = option.hash
else:
algo_val = self._make_index(algorithm, algorithm_default, namespace=algorithm_namespace, # type: ignore[assignment]
reversed=algorithm_reversed, pack=False)
return Schema_PACK_HashOption(
type=type,
length=1 + len(hash),
func=algo_val,
data=hash,
)
[docs]
def _make_nrb_records(self, records: 'Record | list[Schema_NameResolutionRecord | tuple[Enum_RecordType, dict[str, Any]] | bytes] | bytes') -> 'tuple[list[Schema_NameResolutionRecord | bytes], int]':
"""Make name resolution records for PCAP-NG.
Args:
records: PCAP-NG name resolution records.
Returns:
Tuple of name resolution records and total length of the records.
"""
# a record area kept as captured is written back verbatim
if isinstance(records, bytes):
return [records], len(records)
has_record_end = False
total_length = 0
if isinstance(records, list):
records_list = [] # type: list[Schema_NameResolutionRecord | bytes]
for schema in records:
if isinstance(schema, bytes):
code = Enum_RecordType.get(int.from_bytes(schema[0:2], self._byte, signed=False))
if code == Enum_RecordType.nrb_record_end: # moved to the end, written once
has_record_end = True
continue
data = schema # type: Schema_NameResolutionRecord | bytes
data_len = len(data)
elif isinstance(schema, Schema):
code = schema.type
if code == Enum_RecordType.nrb_record_end: # moved to the end, written once
has_record_end = True
continue
data = schema
data_len = len(schema.pack())
else:
code, args = cast('tuple[Enum_RecordType, dict[str, Any]]', schema)
if code == Enum_RecordType.nrb_record_end: # moved to the end, written once
has_record_end = True
continue
name = self._lookup_registry(self.__record__, code)
if isinstance(name, str):
meth_name = f'_make_record_{name}'
meth = cast('RecordConstructor',
getattr(self, meth_name, self._make_record_unknown))
else:
meth = name[1]
data = self._make_padding(meth(code, **args), args)
data_len = len(data.pack())
records_list.append(data)
total_length += data_len
if has_record_end:
nrb_record_end = self._make_record_end(Enum_RecordType.nrb_record_end) # type: ignore[arg-type]
total_length += len(nrb_record_end.pack())
records_list.append(nrb_record_end)
return records_list, total_length
records_list = []
for code, record in records.items(multi=True):
if code == Enum_RecordType.nrb_record_end: # moved to the end, written once
has_record_end = True
continue
name = self._lookup_registry(self.__record__, code)
if isinstance(name, str):
meth_name = f'_make_record_{name}'
meth = cast('RecordConstructor',
getattr(self, meth_name, self._make_record_unknown))
else:
meth = name[1]
data = self._make_padding(meth(code, record), record)
data_len = len(data.pack())
records_list.append(data)
total_length += data_len
if has_record_end:
nrb_record_end = self._make_record_end(Enum_RecordType.nrb_record_end) # type: ignore[arg-type]
total_length += len(nrb_record_end.pack())
records_list.append(nrb_record_end)
return records_list, total_length
[docs]
def _make_record_unknown(self, type: 'Enum_RecordType', record: 'Optional[Data_UnknownRecord]' = None, *,
data: 'bytes' = b'',
**kwargs: 'Any') -> 'Schema_UnknownRecord':
"""Make PCAP-NG unknown name resolution record.
Args:
type: Record type.
record: Record data model.
data: Record data.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed record schema.
"""
if record is not None:
data = record.data
return Schema_UnknownRecord(
type=type,
length=len(data),
data=data,
)
[docs]
def _make_record_end(self, type: 'Enum_RecordType', record: 'Optional[Data_EndRecord]' = None,
**kwargs: 'Any') -> 'Schema_EndRecord':
"""Make PCAP-NG ``nrb_record_end`` name resolution record.
Args:
type: Record type.
record: Record data model.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed record schema.
"""
return Schema_EndRecord(
type=type,
length=0,
)
[docs]
def _make_record_ipv4(self, type: 'Enum_RecordType', record: 'Optional[Data_IPv4Record]' = None, *,
ip: 'IPv4Address | str | bytes | int' = '127.0.0.1',
names: 'Optional[list[str]]' = None,
**kwargs: 'Any') -> 'Schema_IPv4Record':
"""Make PCAP-NG ``nrb_record_ipv4`` name resolution record.
Args:
type: Record type.
record: Record data model.
ip: IPv4 address.
names: Host names.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed record schema.
"""
if record is not None:
ip = record.ip
names = cast('list[str]', record.records)
if names is None:
names = []
records = _join_names(names)
return Schema_IPv4Record(
type=type,
length=4 + _utf8_length(records),
ip=ip,
resol=records,
)
[docs]
def _make_record_ipv6(self, type: 'Enum_RecordType', record: 'Optional[Data_IPv6Record]' = None, *,
ip: 'IPv6Address | str | bytes | int' = '127.0.0.1',
names: 'Optional[list[str]]' = None,
**kwargs: 'Any') -> 'Schema_IPv6Record':
"""Make PCAP-NG ``nrb_record_ipv6`` name resolution record.
Args:
type: Record type.
record: Record data model.
ip: IPv6 address.
names: Host names.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed record schema.
"""
if record is not None:
ip = record.ip
names = cast('list[str]', record.records)
if names is None:
names = []
records = _join_names(names)
return Schema_IPv6Record(
type=type,
length=16 + _utf8_length(records),
ip=ip,
resol=records,
)
[docs]
def _make_secrets_unknown(self, type: 'Enum_SecretsType', secrets: 'Optional[Data_UnknownSecrets]' = None, *,
data: 'bytes' = b'',
**kwargs: 'Any') -> 'Schema_UnknownSecrets':
"""Make PCAP-NG unknown secrets.
Args:
type: Secrets type.
secrets: Secrets data model.
data: Secrets data.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed secrets schema.
"""
if secrets is not None:
data = secrets.data
return Schema_UnknownSecrets(
data=data,
)
[docs]
def _make_secrets_tls(self, type: 'Enum_SecretsType', secrets: 'Optional[Data_TLSKeyLog]' = None, *,
entries: 'Optional[dict[Enum_TLSKeyLabel, OrderedMultiDict[bytes, bytes]]]' = None,
**kwargs: 'Any') -> 'Schema_TLSKeyLog':
"""Make PCAP-NG TLS Key Log secrets.
Args:
type: Secrets type.
secrets: Secrets data model. The key log text it was parsed from,
if any, is written back verbatim; otherwise its entries are.
On a parsed model the text wins, so an edit to its
``entries`` alone is silently ignored.
entries: TLS Key Log entries.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed secrets schema.
"""
from pcapkit import __version__
if secrets is not None:
text = secrets.get('data')
if isinstance(text, str):
return Schema_TLSKeyLog(
data=text,
)
# NOTE: each an OrderedMultiInfo, which is an OrderedMultiDict.
entries = cast('dict[Enum_TLSKeyLabel, OrderedMultiDict[bytes, bytes]]', secrets.entries)
if entries is None:
entries = {}
data = [f'# generated by PyPCAPKit v{__version__}\n'] # type: list[str]
for label, entry in entries.items():
for k, v in entry.items(multi=True):
data.append(f'{label.name} {k.hex()} {v.hex()}\n')
return Schema_TLSKeyLog(
data=''.join(data),
)
[docs]
def _make_secrets_wireguard(self, type: 'Enum_SecretsType', secrets: 'Optional[Data_WireGuardKeyLog]' = None, *,
entries: 'Optional[OrderedMultiDict[WireGuardKeyLabel, bytes]]' = None,
**kwargs: 'Any') -> 'Schema_WireGuardKeyLog':
"""Make PCAP-NG WireGuard secrets.
Args:
type: Secrets type.
secrets: Secrets data model. The key log text it was parsed from,
if any, is written back verbatim; otherwise its entries are.
On a parsed model the text wins, so an edit to its
``entries`` alone is silently ignored.
entries: WireGuard Key Log entries.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed secrets schema.
"""
from pcapkit import __version__
if secrets is not None:
text = secrets.get('data')
if isinstance(text, str):
return Schema_WireGuardKeyLog(
data=text,
)
entries = secrets.entries
if entries is None:
entries = OrderedMultiDict()
data = [f'# generated by PyPCAPKit v{__version__}\n'] # type: list[str]
for label, value in entries.items(multi=True):
data.append(f'{label.name} = {base64.b64encode(value).decode()}\n')
return Schema_WireGuardKeyLog(
data=''.join(data),
)
[docs]
def _make_secrets_zigbee_nwk(self, type: 'Enum_SecretsType', secrets: 'Optional[Data_ZigBeeNWKKey]' = None, *,
nwk_key: 'bytes' = b'',
pan_id: 'int' = 0,
**kwargs: 'Any') -> 'Schema_ZigBeeNWKKey':
"""Make PCAP-NG ZigBee NWK Key secrets.
Args:
type: Secrets type.
secrets: Secrets data model. The Secrets Data octets it was parsed
from, if any, are written back verbatim, so that a key shorter
or longer than its fixed width rebuilds as captured
(:issue:`1986`); otherwise its fields are. On a parsed model
the octets win, so an edit to its fields alone is silently
ignored.
nwk_key: NWK key.
pan_id: PAN ID.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed secrets schema.
"""
if secrets is not None:
held = secrets.get('data')
if isinstance(held, bytes):
return zigbee_key_held(Schema_ZigBeeNWKKey, held)
nwk_key = secrets.nwk_key
pan_id = secrets.pan_id
return Schema_ZigBeeNWKKey(
key=nwk_key,
panid=pan_id,
)
[docs]
def _make_secrets_zigbee_aps(self, type: 'Enum_SecretsType', secrets: 'Optional[Data_ZigBeeAPSKey]' = None, *,
aps_key: 'bytes' = b'',
pan_id: 'int' = 0,
short_address: 'int' = 0,
**kwargs: 'Any') -> 'Schema_ZigBeeAPSKey':
"""Make PCAP-NG ZigBee APS Key secrets.
Args:
type: Secrets type.
secrets: Secrets data model. The Secrets Data octets it was parsed
from, if any, are written back verbatim, as for
:meth:`_make_secrets_zigbee_nwk`; otherwise its fields are.
aps_key: APS key.
pan_id: PAN ID.
short_address: Short address.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed secrets schema.
"""
if secrets is not None:
held = secrets.get('data')
if isinstance(held, bytes):
return zigbee_key_held(Schema_ZigBeeAPSKey, held)
aps_key = secrets.aps_key
pan_id = secrets.pan_id
short_address = secrets.short_address
return Schema_ZigBeeAPSKey(
key=aps_key,
panid=pan_id,
addr_high=short_address >> 16,
addr_low=short_address & 0xFFFF,
)