# -*- coding: utf-8 -*-
# mypy: disable-error-code=dict-item
"""PCAP-NG File Format
=========================
.. module:: pcapkit.protocols.misc.pcapng.protocol
:mod:`pcapkit.protocols.misc.pcapng.protocol` contains
:class:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG`,
which implements an extractor for PCAP-NG file format [*]_, and the
:class:`~pcapkit.protocols.misc.pcapng.protocol.PacketDirection`,
:class:`~pcapkit.protocols.misc.pcapng.protocol.PacketReception` and
:class:`~pcapkit.protocols.misc.pcapng.protocol.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 struct
import sys
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.schema.byteorder import packet_byteorder
from pcapkit.corekit.schema.padding import pad_length
from pcapkit.corekit.version import VersionInfo
from pcapkit.protocols.data import Packet as Data_Packet
from pcapkit.protocols.misc.pcap.header.schema import RESOLUTION
from pcapkit.protocols.misc.pcapng.data import PCAPNG as Data_PCAPNG
from pcapkit.protocols.misc.pcapng.data import CommentOption as Data_CommentOption
from pcapkit.protocols.misc.pcapng.data import CustomBlock as Data_CustomBlock
from pcapkit.protocols.misc.pcapng.data import CustomOption as Data_CustomOption
from pcapkit.protocols.misc.pcapng.data import DecryptionSecretsBlock as Data_DecryptionSecretsBlock
from pcapkit.protocols.misc.pcapng.data import DSBSecrets as Data_DSBSecrets
from pcapkit.protocols.misc.pcapng.data import EndOfOption as Data_EndOfOption
from pcapkit.protocols.misc.pcapng.data import EndRecord as Data_EndRecord
from pcapkit.protocols.misc.pcapng.data import EnhancedPacketBlock as Data_EnhancedPacketBlock
from pcapkit.protocols.misc.pcapng.data import EPB_DropCountOption as Data_EPB_DropCountOption
from pcapkit.protocols.misc.pcapng.data import EPB_FlagsOption as Data_EPB_FlagsOption
from pcapkit.protocols.misc.pcapng.data import EPB_HashOption as Data_EPB_HashOption
from pcapkit.protocols.misc.pcapng.data import EPB_PacketIDOption as Data_EPB_PacketIDOption
from pcapkit.protocols.misc.pcapng.data import EPB_QueueOption as Data_EPB_QueueOption
from pcapkit.protocols.misc.pcapng.data import EPB_VerdictOption as Data_EPB_VerdictOption
from pcapkit.protocols.misc.pcapng.data import IF_DescriptionOption as Data_IF_DescriptionOption
from pcapkit.protocols.misc.pcapng.data import IF_EUIAddrOption as Data_IF_EUIAddrOption
from pcapkit.protocols.misc.pcapng.data import IF_FCSLenOption as Data_IF_FCSLenOption
from pcapkit.protocols.misc.pcapng.data import IF_FilterOption as Data_IF_FilterOption
from pcapkit.protocols.misc.pcapng.data import IF_HardwareOption as Data_IF_HardwareOption
from pcapkit.protocols.misc.pcapng.data import IF_IPv4AddrOption as Data_IF_IPv4AddrOption
from pcapkit.protocols.misc.pcapng.data import IF_IPv6AddrOption as Data_IF_IPv6AddrOption
from pcapkit.protocols.misc.pcapng.data import IF_MACAddrOption as Data_IF_MACAddrOption
from pcapkit.protocols.misc.pcapng.data import IF_NameOption as Data_IF_NameOption
from pcapkit.protocols.misc.pcapng.data import IF_OSOption as Data_IF_OSOption
from pcapkit.protocols.misc.pcapng.data import IF_RxSpeedOption as Data_IF_RxSpeedOption
from pcapkit.protocols.misc.pcapng.data import IF_SpeedOption as Data_IF_SpeedOption
from pcapkit.protocols.misc.pcapng.data import IF_TSOffsetOption as Data_IF_TSOffsetOption
from pcapkit.protocols.misc.pcapng.data import IF_TSResolOption as Data_IF_TSResolOption
from pcapkit.protocols.misc.pcapng.data import IF_TxSpeedOption as Data_IF_TxSpeedOption
from pcapkit.protocols.misc.pcapng.data import IF_TZoneOption as Data_IF_TZoneOption
from pcapkit.protocols.misc.pcapng.data import \
InterfaceDescriptionBlock as Data_InterfaceDescriptionBlock
from pcapkit.protocols.misc.pcapng.data import \
InterfaceStatisticsBlock as Data_InterfaceStatisticsBlock
from pcapkit.protocols.misc.pcapng.data import IPv4Record as Data_IPv4Record
from pcapkit.protocols.misc.pcapng.data import IPv6Record as Data_IPv6Record
from pcapkit.protocols.misc.pcapng.data import ISB_EndTimeOption as Data_ISB_EndTimeOption
from pcapkit.protocols.misc.pcapng.data import ISB_FilterAcceptOption as Data_ISB_FilterAcceptOption
from pcapkit.protocols.misc.pcapng.data import ISB_IFDropOption as Data_ISB_IFDropOption
from pcapkit.protocols.misc.pcapng.data import ISB_IFRecvOption as Data_ISB_IFRecvOption
from pcapkit.protocols.misc.pcapng.data import ISB_OSDropOption as Data_ISB_OSDropOption
from pcapkit.protocols.misc.pcapng.data import ISB_StartTimeOption as Data_ISB_StartTimeOption
from pcapkit.protocols.misc.pcapng.data import ISB_UsrDelivOption as Data_ISB_UsrDelivOption
from pcapkit.protocols.misc.pcapng.data import NameResolutionBlock as Data_NameResolutionBlock
from pcapkit.protocols.misc.pcapng.data import NameResolutionRecord as Data_NameResolutionRecord
from pcapkit.protocols.misc.pcapng.data import NS_DNSIP4AddrOption as Data_NS_DNSIP4AddrOption
from pcapkit.protocols.misc.pcapng.data import NS_DNSIP6AddrOption as Data_NS_DNSIP6AddrOption
from pcapkit.protocols.misc.pcapng.data import NS_DNSNameOption as Data_NS_DNSNameOption
from pcapkit.protocols.misc.pcapng.data import Option as Data_Option
from pcapkit.protocols.misc.pcapng.data import PACK_FlagsOption as Data_PACK_FlagsOption
from pcapkit.protocols.misc.pcapng.data import PACK_HashOption as Data_PACK_HashOption
from pcapkit.protocols.misc.pcapng.data import PacketBlock as Data_PacketBlock
from pcapkit.protocols.misc.pcapng.data import SectionHeaderBlock as Data_SectionHeaderBlock
from pcapkit.protocols.misc.pcapng.data import SimplePacketBlock as Data_SimplePacketBlock
from pcapkit.protocols.misc.pcapng.data import \
SystemdJournalExportBlock as Data_SystemdJournalExportBlock
from pcapkit.protocols.misc.pcapng.data import TLSKeyLog as Data_TLSKeyLog
from pcapkit.protocols.misc.pcapng.data import TLSKeyLogEntries as Data_TLSKeyLogEntries
from pcapkit.protocols.misc.pcapng.data import UnknownBlock as Data_UnknownBlock
from pcapkit.protocols.misc.pcapng.data import UnknownOption as Data_UnknownOption
from pcapkit.protocols.misc.pcapng.data import UnknownRecord as Data_UnknownRecord
from pcapkit.protocols.misc.pcapng.data import UnknownSecrets as Data_UnknownSecrets
from pcapkit.protocols.misc.pcapng.data import WireGuardKeyLog as Data_WireGuardKeyLog
from pcapkit.protocols.misc.pcapng.data import ZigBeeAPSKey as Data_ZigBeeAPSKey
from pcapkit.protocols.misc.pcapng.data import ZigBeeNWKKey as Data_ZigBeeNWKKey
from pcapkit.protocols.misc.pcapng.schema import (BLOCK_MINIMUM, BYTE_ORDER_MAGIC,
DEFAULT_RESOLUTION, GENERIC_BLOCK_MINIMUM)
from pcapkit.protocols.misc.pcapng.schema import PCAPNG as Schema_PCAPNG
from pcapkit.protocols.misc.pcapng.schema import BlockType as Schema_BlockType
from pcapkit.protocols.misc.pcapng.schema import CommentOption as Schema_CommentOption
from pcapkit.protocols.misc.pcapng.schema import CustomBlock as Schema_CustomBlock
from pcapkit.protocols.misc.pcapng.schema import CustomOption as Schema_CustomOption
from pcapkit.protocols.misc.pcapng.schema import \
DecryptionSecretsBlock as Schema_DecryptionSecretsBlock
from pcapkit.protocols.misc.pcapng.schema import DSBSecrets as Schema_DSBSecrets
from pcapkit.protocols.misc.pcapng.schema import EndOfOption as Schema_EndOfOption
from pcapkit.protocols.misc.pcapng.schema import EndRecord as Schema_EndRecord
from pcapkit.protocols.misc.pcapng.schema import EnhancedPacketBlock as Schema_EnhancedPacketBlock
from pcapkit.protocols.misc.pcapng.schema import EPB_DropCountOption as Schema_EPB_DropCountOption
from pcapkit.protocols.misc.pcapng.schema import EPB_FlagsOption as Schema_EPB_FlagsOption
from pcapkit.protocols.misc.pcapng.schema import EPB_HashOption as Schema_EPB_HashOption
from pcapkit.protocols.misc.pcapng.schema import EPB_PacketIDOption as Schema_EPB_PacketIDOption
from pcapkit.protocols.misc.pcapng.schema import EPB_QueueOption as Schema_EPB_QueueOption
from pcapkit.protocols.misc.pcapng.schema import EPB_VerdictOption as Schema_EPB_VerdictOption
from pcapkit.protocols.misc.pcapng.schema import IF_DescriptionOption as Schema_IF_DescriptionOption
from pcapkit.protocols.misc.pcapng.schema import IF_EUIAddrOption as Schema_IF_EUIAddrOption
from pcapkit.protocols.misc.pcapng.schema import IF_FCSLenOption as Schema_IF_FCSLenOption
from pcapkit.protocols.misc.pcapng.schema import IF_FilterOption as Schema_IF_FilterOption
from pcapkit.protocols.misc.pcapng.schema import IF_HardwareOption as Schema_IF_HardwareOption
from pcapkit.protocols.misc.pcapng.schema import IF_IPv4AddrOption as Schema_IF_IPv4AddrOption
from pcapkit.protocols.misc.pcapng.schema import IF_IPv6AddrOption as Schema_IF_IPv6AddrOption
from pcapkit.protocols.misc.pcapng.schema import IF_MACAddrOption as Schema_IF_MACAddrOption
from pcapkit.protocols.misc.pcapng.schema import IF_NameOption as Schema_IF_NameOption
from pcapkit.protocols.misc.pcapng.schema import IF_OSOption as Schema_IF_OSOption
from pcapkit.protocols.misc.pcapng.schema import IF_RxSpeedOption as Schema_IF_RxSpeedOption
from pcapkit.protocols.misc.pcapng.schema import IF_SpeedOption as Schema_IF_SpeedOption
from pcapkit.protocols.misc.pcapng.schema import IF_TSOffsetOption as Schema_IF_TSOffsetOption
from pcapkit.protocols.misc.pcapng.schema import IF_TSResolOption as Schema_IF_TSResolOption
from pcapkit.protocols.misc.pcapng.schema import IF_TxSpeedOption as Schema_IF_TxSpeedOption
from pcapkit.protocols.misc.pcapng.schema import IF_TZoneOption as Schema_IF_TZoneOption
from pcapkit.protocols.misc.pcapng.schema import \
InterfaceDescriptionBlock as Schema_InterfaceDescriptionBlock
from pcapkit.protocols.misc.pcapng.schema import \
InterfaceStatisticsBlock as Schema_InterfaceStatisticsBlock
from pcapkit.protocols.misc.pcapng.schema import IPv4Record as Schema_IPv4Record
from pcapkit.protocols.misc.pcapng.schema import IPv6Record as Schema_IPv6Record
from pcapkit.protocols.misc.pcapng.schema import ISB_EndTimeOption as Schema_ISB_EndTimeOption
from pcapkit.protocols.misc.pcapng.schema import \
ISB_FilterAcceptOption as Schema_ISB_FilterAcceptOption
from pcapkit.protocols.misc.pcapng.schema import ISB_IFDropOption as Schema_ISB_IFDropOption
from pcapkit.protocols.misc.pcapng.schema import ISB_IFRecvOption as Schema_ISB_IFRecvOption
from pcapkit.protocols.misc.pcapng.schema import ISB_OSDropOption as Schema_ISB_OSDropOption
from pcapkit.protocols.misc.pcapng.schema import ISB_StartTimeOption as Schema_ISB_StartTimeOption
from pcapkit.protocols.misc.pcapng.schema import ISB_UsrDelivOption as Schema_ISB_UsrDelivOption
from pcapkit.protocols.misc.pcapng.schema import NameResolutionBlock as Schema_NameResolutionBlock
from pcapkit.protocols.misc.pcapng.schema import NameResolutionRecord as Schema_NameResolutionRecord
from pcapkit.protocols.misc.pcapng.schema import NS_DNSIP4AddrOption as Schema_NS_DNSIP4AddrOption
from pcapkit.protocols.misc.pcapng.schema import NS_DNSIP6AddrOption as Schema_NS_DNSIP6AddrOption
from pcapkit.protocols.misc.pcapng.schema import NS_DNSNameOption as Schema_NS_DNSNameOption
from pcapkit.protocols.misc.pcapng.schema import PACK_FlagsOption as Schema_PACK_FlagsOption
from pcapkit.protocols.misc.pcapng.schema import PACK_HashOption as Schema_PACK_HashOption
from pcapkit.protocols.misc.pcapng.schema import PacketBlock as Schema_PacketBlock
from pcapkit.protocols.misc.pcapng.schema import SectionHeaderBlock as Schema_SectionHeaderBlock
from pcapkit.protocols.misc.pcapng.schema import SimplePacketBlock as Schema_SimplePacketBlock
from pcapkit.protocols.misc.pcapng.schema import \
SystemdJournalExportBlock as Schema_SystemdJournalExportBlock
from pcapkit.protocols.misc.pcapng.schema import TLSKeyLog as Schema_TLSKeyLog
from pcapkit.protocols.misc.pcapng.schema import UnknownBlock as Schema_UnknownBlock
from pcapkit.protocols.misc.pcapng.schema import UnknownOption as Schema_UnknownOption
from pcapkit.protocols.misc.pcapng.schema import UnknownRecord as Schema_UnknownRecord
from pcapkit.protocols.misc.pcapng.schema import UnknownSecrets as Schema_UnknownSecrets
from pcapkit.protocols.misc.pcapng.schema import WireGuardKeyLog as Schema_WireGuardKeyLog
from pcapkit.protocols.misc.pcapng.schema import ZigBeeAPSKey as Schema_ZigBeeAPSKey
from pcapkit.protocols.misc.pcapng.schema import ZigBeeNWKKey as Schema_ZigBeeNWKKey
from pcapkit.protocols.misc.pcapng.schema import zigbee_key_held
from pcapkit.protocols.protocol import ProtocolBase
from pcapkit.protocols.schema import Schema
from pcapkit.utilities.compat import StrEnum, localcontext
from pcapkit.utilities.contextvars import frame_length
from pcapkit.utilities.datetime import fromtimestamp, totimestamp
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 ipaddress import IPv4Address, IPv4Interface, IPv6Address, IPv6Interface
from typing import (IO, Any, Callable, ClassVar, Counter, DefaultDict, Mapping, Optional, Tuple,
Type, TypeVar, Union)
from mypy_extensions import DefaultArg, KwArg, NamedArg
from typing_extensions import Literal
from pcapkit.corekit.enum import EnumRegistry
from pcapkit.foundation.engines.pcapng import Context
from pcapkit.protocols.misc.pcapng.schema import Option as Schema_Option
from pcapkit.protocols.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')
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
def _block_type_name(block_type: 'Enum_BlockType | int') -> 'str':
"""Return the name of a block type, for an option's placement message.
Args:
block_type: The block type to name, a member or a plain :obj:`int`
code.
Returns:
The block type's name, e.g. ``Enhanced_Packet_Block``; ``Block`` and
the code, e.g. ``Block 256``, for a code no member holds.
Note:
``block_type`` is looked up rather than having its :attr:`~enum.Enum.name`
read directly, because :attr:`PCAPNG._type <PCAPNG._type>` is not
always a member on the make path (:issue:`2301`).
:meth:`PCAPNG.make <pcapkit.protocols.misc.pcapng.protocol.PCAPNG.make>` assigns
what :meth:`~pcapkit.protocols.protocol.Protocol._make_index` resolves
the code to: its member, or the plain :obj:`int` for a code in range
that no member holds (:issue:`2594`).
"""
try:
return Enum_BlockType(block_type).name
except ValueError:
return f'Block {int(block_type)}'
[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 section 4.2 makes the TLS key labels the IANA registry ``Enum_TLSKeyLabel``,
# not re-exported here (#1744); draft-ietf-opsawg-pcapng-06 closes ``WireGuardKeyLabel``.
[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.protocol.PCAPNG.__proto__>`
attribute:
.. list-table::
:header-rows: 1
* - Index
- Protocol
* - :attr:`pcapkit.const.reg.linktype.LinkType.NULL`
- :class:`pcapkit.protocols.link.loopback.protocol.Loopback`
* - :attr:`pcapkit.const.reg.linktype.LinkType.ETHERNET`
- :class:`pcapkit.protocols.link.ethernet.protocol.Ethernet`
* - :attr:`pcapkit.const.reg.linktype.LinkType.IEEE802_11`
- :class:`pcapkit.protocols.link.ieee802_11.protocol.IEEE802_11`
* - :attr:`pcapkit.const.reg.linktype.LinkType.FRELAY`
- :class:`pcapkit.protocols.link.frad.protocol.FRAD`
* - :attr:`pcapkit.const.reg.linktype.LinkType.LOOP`
- :class:`pcapkit.protocols.link.loopback.protocol.Loopback`
* - :attr:`pcapkit.const.reg.linktype.LinkType.LINUX_SLL`
- :class:`pcapkit.protocols.link.linux_sll1.protocol.LinuxSLL1`
* - :attr:`pcapkit.const.reg.linktype.LinkType.IEEE802_11_RADIOTAP`
- :class:`pcapkit.protocols.link.radiotap.protocol.Radiotap`
* - :attr:`pcapkit.const.reg.linktype.LinkType.IPV4`
- :class:`pcapkit.protocols.internet.ipv4.protocol.IPv4`
* - :attr:`pcapkit.const.reg.linktype.LinkType.IPV6`
- :class:`pcapkit.protocols.internet.ipv6.protocol.IPv6`
* - :attr:`pcapkit.const.reg.linktype.LinkType.LINUX_SLL2`
- :class:`pcapkit.protocols.link.linux_sll2.protocol.LinuxSLL2`
: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.protocol.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.protocol.PCAPNG._read_block_shb`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_block_shb`
* - :attr:`~pcapkit.const.pcapng.block_type.BlockType.Interface_Description_Block`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_block_idb`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_block_idb`
* - :attr:`~pcapkit.const.pcapng.block_type.BlockType.Enhanced_Packet_Block`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_block_epb`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_block_epb`
* - :attr:`~pcapkit.const.pcapng.block_type.BlockType.Simple_Packet_Block`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_block_spb`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_block_spb`
* - :attr:`~pcapkit.const.pcapng.block_type.BlockType.Name_Resolution_Block`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_block_nrb`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_block_nrb`
* - :attr:`~pcapkit.const.pcapng.block_type.BlockType.Interface_Statistics_Block`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_block_isb`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_block_isb`
* - :attr:`~pcapkit.const.pcapng.block_type.BlockType.systemd_Journal_Export_Block`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_block_systemd`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_block_systemd`
* - :attr:`~pcapkit.const.pcapng.block_type.BlockType.Decryption_Secrets_Block`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_block_dsb`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.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.protocol.PCAPNG._read_block_cb`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.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.protocol.PCAPNG._read_block_cb`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_block_cb`
* - :attr:`~pcapkit.const.pcapng.block_type.BlockType.Packet_Block`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_block_packet`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.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.protocol.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.protocol.PCAPNG._read_option_endofopt`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_endofopt`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.opt_comment`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_comment`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.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.protocol.PCAPNG._read_option_custom`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_custom`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_name`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_if_name`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_if_name`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_description`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_if_description`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_if_description`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_IPv4addr`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_if_ipv4`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_if_ipv4`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_IPv6addr`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_if_ipv6`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_if_ipv6`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_MACaddr`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_if_mac`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_if_mac`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_EUIaddr`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_if_eui`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_if_eui`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_speed`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_if_speed`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_if_speed`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_tsresol`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_if_tsresol`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_if_tsresol`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_tzone`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_if_tzone`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_if_tzone`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_filter`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_if_filter`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_if_filter`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_os`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_if_os`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_if_os`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_fcslen`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_if_fcslen`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_if_fcslen`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_tsoffset`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_if_tsoffset`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_if_tsoffset`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_hardware`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_if_hardware`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_if_hardware`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_txspeed`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_if_txspeed`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_if_txspeed`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.if_rxspeed`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_if_rxspeed`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_if_rxspeed`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.epb_flags`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_epb_flags`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_epb_flags`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.epb_hash`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_epb_hash`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_epb_hash`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.epb_dropcount`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_epb_dropcount`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_epb_dropcount`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.epb_packetid`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_epb_packetid`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_epb_packetid`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.epb_queue`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_epb_queue`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_epb_queue`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.epb_verdict`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_epb_verdict`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_epb_verdict`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.ns_dnsname`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_ns_dnsname`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_ns_dnsname`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.ns_dnsIP4addr`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_ns_dnsipv4`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_ns_dnsipv4`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.ns_dnsIP6addr`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_ns_dnsipv6`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_ns_dnsipv6`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.isb_starttime`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_isb_starttime`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_isb_starttime`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.isb_endtime`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_isb_endtime`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_isb_endtime`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.isb_ifrecv`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_isb_ifrecv`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_isb_ifrecv`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.isb_ifdrop`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_isb_ifdrop`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_isb_ifdrop`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.isb_filteraccept`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_isb_filteraccept`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_isb_filteraccept`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.isb_osdrop`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_isb_osdrop`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_isb_osdrop`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.isb_usrdeliv`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_isb_usrdeliv`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_isb_usrdeliv`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.pack_flags`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_pack_flags`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_option_pack_flags`
* - :attr:`~pcapkit.const.pcapng.option_type.OptionType.pack_hash`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_option_pack_hash`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.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.protocol.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.protocol.PCAPNG._read_record_end`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_record_end`
* - :attr:`~pcapkit.const.pcapng.record_type.RecordType.nrb_record_ipv4`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_record_ipv4`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_record_ipv4`
* - :attr:`~pcapkit.const.pcapng.record_type.RecordType.nrb_record_ipv6`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_record_ipv6`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.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.protocol.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.protocol.PCAPNG._read_secrets_tls`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_secrets_tls`
* - :attr:`~pcapkit.const.pcapng.secrets_type.SecretsType.WireGuard_Key_Log`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_secrets_wireguard`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_secrets_wireguard`
* - :attr:`~pcapkit.const.pcapng.secrets_type.SecretsType.ZigBee_NWK_Key`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_secrets_zigbee_nwk`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_secrets_zigbee_nwk`
* - :attr:`~pcapkit.const.pcapng.secrets_type.SecretsType.ZigBee_APS_Key`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_secrets_zigbee_aps`
- :meth:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG._make_secrets_zigbee_aps`
.. versionchanged:: 1.5.0
``context`` returns the caller supplied
:class:`~pcapkit.corekit.context.ContextRegistry` inherited from
:class:`~pcapkit.protocols.protocol.ProtocolBase`, on every block. The
block's section ``Context``, which ``context`` returned through 1.4.x,
is :attr:`section_context` (:issue:`2172`).
.. versionchanged:: 1.6.0
* Each enumeration field of :meth:`make` and of the block and option
makers, and the code of each option and the type of each name
resolution record given to a block maker, takes a member or an
:class:`int` code, resolved with the enumeration's ``get``, an option
code in the block's namespace; a name, a :obj:`bool` for a field
wider than one bit, or a code the field cannot carry raises
:exc:`~pcapkit.utilities.exceptions.ProtocolError`, rather than
resolving against an enumeration inferred from the caller's frame,
writing a different value or raising another exception
(:issue:`2594`).
* An option or a name resolution record given to a block maker as raw
octets too short to hold its 16-bit code raises
:exc:`~pcapkit.utilities.exceptions.ProtocolError`, rather than
reading empty octets as code 0 and writing an end-of-options or
end-of-records marker for them (:issue:`2723`).
.. versionremoved:: 1.6.0
The ``type_default``, ``type_namespace`` and ``type_reversed``
arguments of :meth:`make`, and the ``_default``, ``_namespace`` and
``_reversed`` arguments that went with each enumeration field of the
block and option makers -- ``linktype``, ``secrets_type``, ``filter``,
``direction``, ``reception``, ``algorithm`` and ``verdict``; resolve a
name with :meth:`~pcapkit.protocols.protocol.Protocol.resolve_index`
first (:issue:`2594`).
"""
if TYPE_CHECKING:
#: Section context of the block; see :attr:`section_context`.
_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.misc.pcapng.schema.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.ProtocolBase` subclass.
__proto__ = collections.defaultdict(
lambda: ModuleDescriptor('pcapkit.protocols.misc.raw.protocol', 'Raw'),
{
Enum_LinkType.NULL: ModuleDescriptor('pcapkit.protocols.link', 'Loopback'),
Enum_LinkType.ETHERNET: ModuleDescriptor('pcapkit.protocols.link', 'Ethernet'),
Enum_LinkType.IEEE802_11: ModuleDescriptor('pcapkit.protocols.link', 'IEEE802_11'),
Enum_LinkType.FRELAY: ModuleDescriptor('pcapkit.protocols.link', 'FRAD'),
Enum_LinkType.LOOP: ModuleDescriptor('pcapkit.protocols.link', 'Loopback'),
Enum_LinkType.LINUX_SLL: ModuleDescriptor('pcapkit.protocols.link', 'LinuxSLL1'),
Enum_LinkType.IEEE802_11_RADIOTAP: ModuleDescriptor('pcapkit.protocols.link', 'Radiotap'),
Enum_LinkType.IPV4: ModuleDescriptor('pcapkit.protocols.internet', 'IPv4'),
Enum_LinkType.IPV6: ModuleDescriptor('pcapkit.protocols.internet', 'IPv6'),
Enum_LinkType.LINUX_SLL2: ModuleDescriptor('pcapkit.protocols.link', 'LinuxSLL2'),
},
) # type: DefaultDict[Enum_LinkType | int, ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]]
#: What :attr:`__proto__` is keyed by -- the enum a key must be a member of,
#: and whether a bare :class:`int` is a valid key too. Declared by the class
#: *owning* the registry, so
#: :func:`~pcapkit.foundation.registry.protocols.protocol.register_protocol_code`
#: checks a ``code=`` key as
#: :func:`~pcapkit.foundation.registry.protocols.misc.register_pcapng` does;
#: :class:`~pcapkit.protocols.misc.pcap.frame.protocol.Frame`, whose registry
#: is keyed by the same enum, declares its own (:issue:`2268`).
__proto_key__: 'ClassVar[tuple[Type[EnumRegistry], bool]]' = (Enum_LinkType, False)
#: 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.Interface_Description_Block: 'idb',
Enum_BlockType.Packet_Block: 'packet',
Enum_BlockType.Simple_Packet_Block: 'spb',
Enum_BlockType.Name_Resolution_Block: 'nrb',
Enum_BlockType.Interface_Statistics_Block: 'isb',
Enum_BlockType.Enhanced_Packet_Block: 'epb',
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.Section_Header_Block: 'shb',
Enum_BlockType.Custom_Block_that_rewriters_should_not_copy_into_new_files: 'cb',
},
) # 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_APS_Key: 'zigbee_aps',
Enum_SecretsType.ZigBee_NWK_Key: 'zigbee_nwk',
},
) # 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:`Protocol.length <pcapkit.protocols.protocol.Protocol.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: Not cached, unlike the inherited one, which reads the stream: this only
# walks filled buffers, and a cache would hand a second ``unpack`` stale octets.
@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:`Protocol.packet <pcapkit.protocols.protocol.Protocol.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.__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 block (its Block Type), then the block's up to the payload. The block
# is found by identity, so a renamed field cannot fold the body into the header.
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 section_context(self) -> 'Context':
"""Section context of current PCAP-NG block.
Raises:
UnsupportedCall: If the block holds no section context: a block
given ``ctx=None``, or a Section Header Block, which starts a
section rather than reading one, unless the
:class:`~pcapkit.foundation.engines.pcapng.PCAPNG` engine read
it and assigned it the section it opens.
.. versionadded:: 1.5.0
Renamed from ``context``, with no alias. ``context`` is the one
inherited from :class:`~pcapkit.protocols.protocol.ProtocolBase`,
the caller supplied parsing context of
:mod:`pcapkit.corekit.context` (:issue:`2172`).
"""
if self._ctx is None:
raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute 'section_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 resolution is finer than microseconds."""
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:
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:
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:
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.ProtocolBase` subclass
Raises:
pcapkit.utilities.exceptions.RegistryError: If ``protocol`` is not a
class, or not a :class:`~pcapkit.protocols.protocol.ProtocolBase` 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.misc.register_linktype`
writing to both cannot make either warn about the other. See
:meth:`Protocol.register
<pcapkit.protocols.protocol.Protocol.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: Name of the ``_read_block_${name}`` and ``_make_block_${name}``
methods, or a ``(parser, constructor)`` pair of callables; see
:attr:`__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: Name of the ``_read_option_${name}`` and
``_make_option_${name}`` methods, or a ``(parser, constructor)``
pair of callables; see :attr:`__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: Name of the ``_read_record_${name}`` and
``_make_record_${name}`` methods, or a ``(parser, constructor)``
pair of callables; see :attr:`__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: Name of the ``_read_secrets_${name}`` and
``_make_secrets_${name}`` methods, or a ``(parser, constructor)``
pair of callables; see :attr:`__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:
The special keyword argument ``__packet__`` passes the global
packet data to underlying methods, for when the current instance
does not have it.
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:
The special keyword argument ``__packet__`` passes the global
packet data to underlying methods, for when the current instance
does not have it.
A block handed to :meth:`make` as raw :obj:`bytes` is written back
verbatim past the Block Type, so the schema :meth:`make` returns
holds the octets rather than the block's fields. Those octets are
parsed here, off the buffer :meth:`pack` has just filled, which is
what a reader of the file would do with them (:issue:`2354`).
"""
header = cast('Optional[Schema_PCAPNG]', self.__header__)
if header is None or isinstance(header.block, bytes):
packet = kwargs.get('__packet__', {}) # packet data
# NOTE: A tail too short for any block is refused on construction too:
# a caller's raw octets are no more trustworthy than a file's.
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 interface 0's snaplen; 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 means no limit too; 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: Read through :attr:`packet`, so the payload ``ProtocolBase.__init__``
# injects is the value computed here, not 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 cut block read as zeros, so a value its
# parser refuses is an artefact of the cut and is kept as captured.
if not _read or self._captured(_seek_set) >= schema.block.length:
raise
block = self._read_block_cut(schema, _seek_set, error)
# NOTE: A malformed option or record area arrives as one :obj:`bytes` item and parses
# as none; kept as ``options_raw`` or ``records_raw``, ``make`` writes it back (#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: A copy of the block, 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: Bounded at :data:`sys.maxsize`, past which the read raises :exc:`OverflowError`
# rather than coming back short as the truncation handling below expects (#1975).
bound = min(schema.block.length, sys.maxsize)
self._data = self._read_fileng(bound)
# NOTE: On a cut capture Block Total Length runs past the end, and the next block would
# read a negative length and fail the extraction (#431), so the octets read mark its end.
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: Zero-filled fields and cut entries cannot rebuild the block at its declared
# length, so :meth:`__post_init__` re-parses the ``__truncated_raw__`` kept (#1470).
block.__update__([('__truncated_raw__', self._data)])
# NOTE: A whole read at the bound stopped short of the block's end, so the stream skips
# the rest, to where Block Total Length puts the next block (#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 | int' = Enum_BlockType.Simple_Packet_Block,
block: 'bytes | Data_PCAPNG | Schema_BlockType | dict[str, Any]' = b'',
**kwargs: 'Any') -> 'Schema_PCAPNG':
"""Make PCAP-NG block data.
Args:
type: Block type.
block: Block data.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed packet data.
Raises:
ProtocolError: If ``type`` is a name, a :obj:`bool`, or a member
or code the 32-bit Block Type field cannot carry; or if
``block`` is none of the forms this maker reads -- raw
:obj:`bytes`, a :obj:`dict` of block arguments, a
:class:`~pcapkit.protocols.misc.pcapng.data.PCAPNG` data model
or a :class:`~pcapkit.protocols.schema.Schema`.
"""
type_val = self._make_index(type, Enum_BlockType, bits=32, name='block type', alias='PCAP-NG')
self._type = type_val
# NOTE: Settle the byte order before the block maker reads a raw option's or record's type
# code in it (#2353); an SHB overrides it. ``or {}``, as ``__packet__`` may be ``None``.
if self._ctx is not None:
self._byte = self._ctx.section.byteorder
else:
self._byte = packet_byteorder(kwargs.get('__packet__') or {})
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 block: must be bytes, a dict, '
f'a PCAPNG data model or a Schema, not {block!r}')
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:
The first block has ``num`` ``0`` and ``ctx`` :obj:`None`, marking it
as 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: Section 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 captured octets, parsed
# again as in :meth:`read`, rather than at its declared length (#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 re-parsed on this instance, counting each option
# afresh against its "only one" guard, so the make pass's count must not carry over.
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]
@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,
}
[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.misc.pcapng.schema.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 _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]
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 _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_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_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())
# NOTE: The layers below are told both lengths, as a PCAP frame tells them (#2653).
token = frame_length.set((data.captured_len, data.original_len))
try:
return self._decode_next_layer(data, self._get_linktype(schema.interface_id),
schema.captured_length) # type: ignore[return-value]
finally:
frame_length.reset(token)
[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: Packet data past interface 0's snaplen is kept (#1384) but reported,
# as the specification says the snaplen determines the length.
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 + pad_length(data.original_len, 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())
# NOTE: The layers below are told both lengths, as a PCAP frame tells them (#2653).
token = frame_length.set((data.captured_len, data.original_len))
try:
return self._decode_next_layer(data, self._get_linktype(0), data.captured_len) # type: ignore[return-value]
finally:
frame_length.reset(token)
[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_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())
# NOTE: The layers below are told both lengths, as a PCAP frame tells them (#2653).
token = frame_length.set((data.captured_len, data.original_len))
try:
return self._decode_next_layer(data, self._get_linktype(schema.interface_id),
schema.captured_len) # type: ignore[return-value]
finally:
frame_length.reset(token)
[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 arrives as one :obj:`bytes` item and parses as none;
# kept as ``entry_raw``, the block's ``make`` writes it 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_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_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'not {self._opt[schema.type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_name] option must be only one, '
f'not {self._opt[schema.type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_description] option must be only one, '
f'not {self._opt[schema.type] + 1}')
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'not {_block_type_name(self._type)}')
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'not {_block_type_name(self._type)}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_MACaddr] option must be only one, '
f'not {self._opt[schema.type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_EUIaddr] option must be only one, '
f'not {self._opt[schema.type] + 1}')
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_mac_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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_speed] option must be only one, '
f'not {self._opt[schema.type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_tsresol] option must be only one, '
f'not {self._opt[schema.type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_tzone] option must be only one, '
f'not {self._opt[schema.type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_filter] option must be only one, '
f'not {self._opt[schema.type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_os] option must be only one, '
f'not {self._opt[schema.type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_fcslen] option must be only one, '
f'not {self._opt[schema.type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_tsoffset] option must be only one, '
f'not {self._opt[schema.type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_hardware] option must be only one, '
f'not {self._opt[schema.type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_txspeed] option must be only one, '
f'not {self._opt[schema.type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [if_rxspeed] option must be only one, '
f'not {self._opt[schema.type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [epb_flags] option must be only one, '
f'not {self._opt[schema.type] + 1}')
if schema.length != 4:
raise ProtocolError(f'PCAP-NG: [epb_flags] invalid length (expected 4, 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'not {_block_type_name(self._type)}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [epb_dropcount] option must be only one, '
f'not {self._opt[schema.type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [epb_packetid] option must be only one, '
f'not {self._opt[schema.type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [epb_queue] option must be only one, '
f'not {self._opt[schema.type] + 1}')
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'not {_block_type_name(self._type)}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [ns_dnsname] option must be only one, '
f'not {self._opt[schema.type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [ns_dnsIP4addr] option must be only one, '
f'not {self._opt[schema.type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [ns_dnsIP6addr] option must be only one, '
f'not {self._opt[schema.type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_starttime] option must be only one, '
f'not {self._opt[schema.type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_endtime] option must be only one, '
f'not {self._opt[schema.type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_ifrecv] option must be only one, '
f'not {self._opt[schema.type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_ifdrop] option must be only one, '
f'not {self._opt[schema.type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_filteraccept] option must be only one, '
f'not {self._opt[schema.type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_osdrop] option must be only one, '
f'not {self._opt[schema.type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_usrdeliv] option must be only one, '
f'not {self._opt[schema.type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[schema.type] > 0:
raise ProtocolError(f'PCAP-NG: [pack_flags] option must be only one, '
f'not {self._opt[schema.type] + 1}')
if schema.length != 4:
raise ProtocolError(f'PCAP-NG: [pack_flags] invalid length (expected 4, 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'not {_block_type_name(self._type)}')
option = Data_PACK_HashOption(
type=schema.type,
length=schema.length,
algorithm=schema.func,
hash=schema.data,
)
return option
[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.
Returns:
Parsed decryption secrets data.
"""
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.
Returns:
Parsed decryption secrets data.
"""
secrets = Data_TLSKeyLog(
# NOTE: each label's entries become an OrderedMultiInfo as the model
# stores them (#1484, #2726).
entries=Data_TLSKeyLogEntries(schema.entries),
)
# NOTE: the log text is kept too, as :attr:`entries` loses its comments and
# line order; 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 too, as :attr:`entries` loses its comments and
# line order; a rebuild writes it back verbatim.
secrets.__update__(data=schema.data)
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 _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 are kept too, as Secrets Length may differ from the key's
# fixed width; a rebuild writes them back verbatim (#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(pad_length(len(data), 4))
length = len(data) + GENERIC_BLOCK_MINIMUM
return Schema_UnknownBlock(
length=length,
body=data,
length2=length,
)
[docs]
def _make_block_idb(self, block: 'Optional[Data_InterfaceDescriptionBlock]' = None, *,
linktype: 'Enum_LinkType | int' = Enum_LinkType.NULL,
snaplen: 'int' = 0,
options: 'Optional[Option | list[Schema_Option | tuple[Enum_OptionType | int, dict[str, Any]] | bytes]]' = None, # pylint: disable=line-too-long
**kwargs: 'Any') -> 'Schema_InterfaceDescriptionBlock':
"""Make PCAP-NG interface description block (IDB).
Args:
block: Block data model.
linktype: Link layer protocol type.
snaplen: Snap length, an unsigned 32-bit value; 0, the default,
means no limit.
options: Block options.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed block schema.
Raises:
ProtocolError: If ``linktype`` is a name, a :obj:`bool`, or a
member or code the 16-bit LinkType field cannot carry; or if
``snaplen`` is not an :obj:`int` (a :obj:`bool` included), or
is outside ``0`` to ``0xFFFFFFFF``, which the 32-bit SnapLen
field would truncate (:issue:`2278`).
.. versionchanged:: 1.5.0
``snaplen``'s default changed from ``0xFFFF_FFFF_FFFF_FFFF`` -- which
the 32-bit field silently truncated to ``0xFFFFFFFF`` -- to ``0``,
and a value outside the field's range now raises instead of being
truncated (:issue:`2278`).
.. versionchanged:: 1.6.0
``snaplen`` is checked by
:meth:`~pcapkit.protocols.protocol.Protocol._make_uint`, so a
:obj:`bool` or a non-integer raises
:exc:`~pcapkit.utilities.exceptions.ProtocolError` too
(:issue:`2614`).
"""
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, Enum_LinkType, bits=16, name='[IDB] linktype',
alias='PCAP-NG')
snaplen = self._make_uint(snaplen, bits=32, name='[IDB] snaplen', alias='PCAP-NG')
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)
schema = 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')
# NOTE: The octets after ``opt_endofopt``, if any (#1437, #2776).
return self._make_padding(schema, block, 'padding', 'padding_opts')
[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 | int, dict[str, Any]] | bytes]]' = None, # pylint: disable=line-too-long
**kwargs: 'Any') -> 'Schema_PacketBlock':
"""Make PCAP-NG packet block (obsolete).
Args:
block: Block data model.
interface_id: Interface ID.
drop_count: Drops count, ``0xFFFF`` for "not available".
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.
Raises:
ProtocolError: If ``drop_count`` is not an :obj:`int` (a
:obj:`bool` included), or is outside ``0`` to ``0xFFFF``.
.. versionchanged:: 1.6.0
``drop_count`` is checked by
:meth:`~pcapkit.protocols.protocol.Protocol._make_uint`, so a value
the 16-bit field cannot hold, ``-1`` included, or a :obj:`bool`
raises :exc:`~pcapkit.utilities.exceptions.ProtocolError` rather
than being written masked (:issue:`2614`).
"""
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')
drop_count = self._make_uint(drop_count, bits=16, name='[PB] drop_count', alias='PCAP-NG')
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 longer
# packet 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 = len(packet_data) + pad_length(len(packet_data), 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)
schema = 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')
# NOTE: The octets after ``opt_endofopt``, if any (#1437, #2776).
return self._make_padding(schema, block, 'padding_opts')
[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 belongs to interface 0, its Packet Data ``min(original_len,
# snaplen)`` octets long as the parser reads it, so a longer packet is cut.
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 = len(packet_data) + pad_length(len(packet_data), 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_nrb(self, block: 'Optional[Data_NameResolutionBlock]' = None, *,
records: 'Optional[Record | list[Schema_NameResolutionRecord | tuple[Enum_RecordType | int, dict[str, Any]] | bytes]]' = None, # pylint: disable=line-too-long
options: 'Optional[Option | list[Schema_Option | tuple[Enum_OptionType | int, dict[str, Any]] | bytes]]' = None, # pylint: disable=line-too-long
**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)
# NOTE: The octets after ``opt_endofopt``, if any (#1437, #2776).
return self._make_padding(Schema_NameResolutionBlock(
length=options_length + records_length + GENERIC_BLOCK_MINIMUM,
records=records_value,
options=options_value,
length2=options_length + records_length + GENERIC_BLOCK_MINIMUM,
), block, 'padding', 'padding_opts')
[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 | int, dict[str, Any]] | bytes]]' = None, # pylint: disable=line-too-long
**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)
# NOTE: The octets after ``opt_endofopt``, if any (#1437, #2776).
return self._make_padding(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],
), block, 'padding', 'padding_opts')
[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 | int, dict[str, Any]] | bytes]]' = None, # pylint: disable=line-too-long
**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 longer
# packet 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 = len(packet_data) + pad_length(len(packet_data), 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)
schema = 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')
# NOTE: The octets after ``opt_endofopt``, if any (#1437, #2776).
return self._make_padding(schema, block, 'padding_opts')
[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(pad_length(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 | int' = Enum_SecretsType.TLS_Key_Log,
secrets_data: 'Schema_DSBSecrets | Data_DSBSecrets | bytes | dict[str, Any]' = b'',
options: 'Optional[Option | list[Schema_Option | tuple[Enum_OptionType | int, dict[str, Any]] | bytes]]' = None, # pylint: disable=line-too-long
**kwargs: 'Any') -> 'Schema_DecryptionSecretsBlock':
"""Make PCAP-NG decryption secrets block (DSB).
Args:
block: Block data model.
secrets_type: Decryption secrets type.
secrets_data: Decryption secrets data. Raw :obj:`bytes` are read as
the Secrets Data of a parsed block are, through the schema
registered for ``secrets_type`` (:issue:`2273`).
options: Block options.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed block schema.
Raises:
ProtocolError: If ``secrets_type`` is a name, a :obj:`bool`, or a
member or code the 32-bit Secrets Type field cannot carry; or
if ``secrets_data`` is none of the forms this maker reads --
raw :obj:`bytes`, a :obj:`dict` of secrets arguments, a
:class:`~pcapkit.protocols.misc.pcapng.data.DSBSecrets` data
model or a :class:`~pcapkit.protocols.schema.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, Enum_SecretsType, bits=32,
name='[DSB] secrets_type', alias='PCAP-NG')
if isinstance(secrets_data, bytes):
from pcapkit.protocols.misc.pcapng.schema import dsb_secrets_selector
# NOTE: Read through the secrets schema the parse side selects, as :meth:`read`
# consumes a schema rather than raw octets; it packs back to ``secrets_data`` (#2273).
secrets_data_val = cast('Schema_DSBSecrets', dsb_secrets_selector({
'secrets_type': secrets_type_val,
'secrets_length': len(secrets_data),
'length': len(secrets_data) + BLOCK_MINIMUM[Enum_BlockType.Decryption_Secrets_Block],
}).unpack(secrets_data, {}))
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] invalid secrets_data: must be bytes, a dict, '
f'a DSBSecrets data model or a Schema, not {secrets_data!r}')
# 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 + pad_length(secrets_held, 4)
# NOTE: Only a true overrun, past ``secrets_area`` as the parse side bounds it, keeps its
# declared Secrets Length so the block rebuilds as captured; others follow the secrets (#1980).
from pcapkit.protocols.misc.pcapng.schema 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)
schema = 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')
# NOTE: The octets after ``opt_endofopt``, if any (#1437, #2776).
return self._make_padding(schema, block, 'padding_opts')
[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 | int, dict[str, Any]] | bytes]]' = None, # pylint: disable=line-too-long
**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: A custom block spans its data, padding and options as one blob, with no
# ``OptionField``, so its options resolve in the generic ``opt_*`` space.
if options is not None:
options_value, _ = self._make_pcapng_options(options, namespace='opt')
else:
options_value, _ = [], 0
cb_data = data.pack() if isinstance(data, Schema) else data
cb_data += bytes(pad_length(len(cb_data), 4))
for option in options_value:
cb_data += option.pack() if isinstance(option, Schema) else option
cb_data += bytes(pad_length(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_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 | int, dict[str, Any]] | bytes]]' = None, # pylint: disable=line-too-long
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, ``-1`` (or ``0xFFFFFFFFFFFFFFFF``)
for "not specified".
options: Block options.
byteorder: Byte order of the section; a parsed block keeps its own.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed block schema.
Raises:
ProtocolError: If ``section_length`` is not an :obj:`int` (a
:obj:`bool` included), or is outside ``-2 ** 63`` to
``2 ** 64 - 1``.
.. versionchanged:: 1.6.0
``section_length`` is checked by
:meth:`~pcapkit.protocols.protocol.Protocol._make_sint`, so a value
outside the 64-bit field in either spelling, a :obj:`bool` or a
non-integer raises
:exc:`~pcapkit.utilities.exceptions.ProtocolError` rather than being
written masked or failing to pack; ``-1`` and
``0xFFFFFFFFFFFFFFFF`` still write the same octets
(:issue:`2614`).
"""
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]
# NOTE: the specification spells "not specified" both -1 and 0xFFFFFFFFFFFFFFFF, so the
# signed field also takes its unsigned spelling, packing both alike; wider is refused (#2617).
self._make_sint(section_length, bits=64, name='[SHB] section_length', alias='PCAP-NG', allow_unsigned=True)
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)
# NOTE: The octets after ``opt_endofopt``, if any (#1437, #2776).
return self._make_padding(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],
), block, 'padding', 'padding_opts')
[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'not {self._opt[type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_name] option must be only one, '
f'not {self._opt[type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_description] option must be only one, '
f'not {self._opt[type] + 1}')
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'not {_block_type_name(self._type)}')
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'not {_block_type_name(self._type)}')
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'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_MACaddr] option must be only one, '
f'not {self._opt[type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_EUIaddr] option must be only one, '
f'not {self._opt[type] + 1}')
if option is not None:
interface = option.interface
return Schema_IF_EUIAddrOption(
type=type,
length=8,
interface=self._make_mac_addr(interface, size=8, name='EUI address'),
)
[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'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_speed] option must be only one, '
f'not {self._opt[type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_tsresol] option must be only one, '
f'not {self._opt[type] + 1}')
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'not {resolution}')
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 from UTC, as a :class:`~datetime.timezone`,
a :class:`~datetime.timedelta` or an :obj:`int` of seconds.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
Raises:
ProtocolError: If ``tzone`` is none of those types, or is a
:obj:`bool`, or its seconds do not fit the signed 32-bit field.
.. versionchanged:: 1.6.0
An offset the signed 32-bit field cannot hold, or a :obj:`bool`,
raises :exc:`~pcapkit.utilities.exceptions.ProtocolError` rather
than being written truncated (:issue:`2614`).
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_tzone] option must be in Interface Description Block, '
f'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_tzone] option must be only one, '
f'not {self._opt[type] + 1}')
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'not {builtins.type(tzone).__name__}')
# NOTE: checked rather than masked, see ``IF_TZoneOption.tzone``; the
# signed value is written, as the field reads it back.
self._make_sint(tzone_val, bits=32, name='[if_tzone] tzone', alias='PCAP-NG')
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 | int' = Enum_FilterType(0),
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.
expression: Filter expression.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
Raises:
ProtocolError: If ``filter`` is a name, a :obj:`bool`, or a member
or code the 8-bit filter type field cannot carry.
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_filter] option must be in Interface Description Block, '
f'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_filter] option must be only one, '
f'not {self._opt[type] + 1}')
if option is not None:
filter_val = option.code
expr_val = option.expression
else:
filter_val = self._make_index(filter, Enum_FilterType, bits=8, name='[if_filter] filter',
alias='PCAP-NG')
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'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_os] option must be only one, '
f'not {self._opt[type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_fcslen] option must be only one, '
f'not {self._opt[type] + 1}')
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.
Raises:
ProtocolError: If ``offset`` is not an :obj:`int`, is a
:obj:`bool`, or does not fit the signed 64-bit field.
.. versionchanged:: 1.6.0
An offset the signed 64-bit field cannot hold, a :obj:`bool` or a
non-integer raises :exc:`~pcapkit.utilities.exceptions.ProtocolError`
rather than being written truncated, written as zero (``None``) or
failing to pack (:issue:`2614`).
"""
if self._type != Enum_BlockType.Interface_Description_Block:
raise ProtocolError(f'PCAP-NG: [if_tsoffset] option must be in Interface Description Block, '
f'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_tsoffset] option must be only one, '
f'not {self._opt[type] + 1}')
if option is not None:
offset = option.offset
# NOTE: checked rather than masked, see ``IF_TSOffsetOption.tsoffset``.
self._make_sint(offset, bits=64, name='[if_tsoffset] offset', alias='PCAP-NG')
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'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_hardware] option must be only one, '
f'not {self._opt[type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_txspeed] option must be only one, '
f'not {self._opt[type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [if_rxspeed] option must be only one, '
f'not {self._opt[type] + 1}')
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 | int' = PacketDirection.UNKNOWN,
reception: 'PacketReception | int' = PacketReception.UNKNOWN,
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.
reception: Packet reception.
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.
Raises:
ProtocolError: If ``direction`` or ``reception`` is a name, a
:obj:`bool`, or a member or code its 2-bit or 3-bit sub-field
cannot carry.
"""
if self._type != Enum_BlockType.Enhanced_Packet_Block:
raise ProtocolError(f'PCAP-NG: [epb_flags] option must be in Enhanced Packet Block, '
f'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [epb_flags] option must be only one, '
f'not {self._opt[type] + 1}')
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, PacketDirection, bits=2,
name='[epb_flags] direction', alias='PCAP-NG')
reception_val = self._make_index(reception, PacketReception, bits=3,
name='[epb_flags] reception', alias='PCAP-NG')
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 | int' = Enum_HashAlgorithm.two_s_complement,
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.
hash: Hash value.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
Raises:
ProtocolError: If ``algorithm`` is a name, a :obj:`bool`, or a
member or code the 8-bit hash algorithm field cannot carry.
"""
if self._type != Enum_BlockType.Enhanced_Packet_Block:
raise ProtocolError(f'PCAP-NG: [epb_hash] option must be in Enhanced Packet Block, '
f'not {_block_type_name(self._type)}')
if option is not None:
algo_val = option.algorithm
hash = option.hash
else:
algo_val = self._make_index(algorithm, Enum_HashAlgorithm, bits=8,
name='[epb_hash] algorithm', alias='PCAP-NG')
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'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [epb_dropcount] option must be only one, '
f'not {self._opt[type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [epb_packetid] option must be only one, '
f'not {self._opt[type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [epb_queue] option must be only one, '
f'not {self._opt[type] + 1}')
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 | int' = Enum_VerdictType.Hardware,
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.
value: Verdict value.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
Raises:
ProtocolError: If ``verdict`` is a name, a :obj:`bool`, or a
member or code the 8-bit verdict type field cannot carry.
"""
if self._type != Enum_BlockType.Enhanced_Packet_Block:
raise ProtocolError(f'PCAP-NG: [epb_verdict] option must be in Enhanced Packet Block, '
f'not {_block_type_name(self._type)}')
# 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, Enum_VerdictType, bits=8,
name='[epb_verdict] verdict', alias='PCAP-NG')
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'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [ns_dnsname] option must be only one, '
f'not {self._opt[type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [ns_dnsIP4addr] option must be only one, '
f'not {self._opt[type] + 1}')
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' = '2001:4860:4860::8888',
**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'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [ns_dnsIP6addr] option must be only one, '
f'not {self._opt[type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_starttime] option must be only one, '
f'not {self._opt[type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_endtime] option must be only one, '
f'not {self._opt[type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_ifrecv] option must be only one, '
f'not {self._opt[type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_ifdrop] option must be only one, '
f'not {self._opt[type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_filteraccept] option must be only one, '
f'not {self._opt[type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_osdrop] option must be only one, '
f'not {self._opt[type] + 1}')
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'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [isb_usrdeliv] option must be only one, '
f'not {self._opt[type] + 1}')
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 | int' = PacketDirection.UNKNOWN,
reception: 'PacketReception | int' = PacketReception.UNKNOWN,
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.
reception: Packet reception.
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.
Raises:
ProtocolError: If ``direction`` or ``reception`` is a name, a
:obj:`bool`, or a member or code its 2-bit or 3-bit sub-field
cannot carry.
"""
if self._type != Enum_BlockType.Packet_Block:
raise ProtocolError(f'PCAP-NG: [pack_flags] option must be in Packet Block, '
f'not {_block_type_name(self._type)}')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [pack_flags] option must be only one, '
f'not {self._opt[type] + 1}')
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, PacketDirection, bits=2,
name='[pack_flags] direction', alias='PCAP-NG')
reception_val = self._make_index(reception, PacketReception, bits=3,
name='[pack_flags] reception', alias='PCAP-NG')
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 | int' = Enum_HashAlgorithm.two_s_complement,
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.
hash: Hash value.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed option schema.
Raises:
ProtocolError: If ``algorithm`` is a name, a :obj:`bool`, or a
member or code the 8-bit hash algorithm field cannot carry.
"""
if self._type != Enum_BlockType.Packet_Block:
raise ProtocolError(f'PCAP-NG: [pack_hash] option must be in Packet Block, '
f'not {_block_type_name(self._type)}')
if option is not None:
algo_val = option.algorithm
hash = option.hash
else:
algo_val = self._make_index(algorithm, Enum_HashAlgorithm, bits=8,
name='[pack_hash] algorithm', alias='PCAP-NG')
return Schema_PACK_HashOption(
type=type,
length=1 + len(hash),
func=algo_val,
data=hash,
)
[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' = '::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[Mapping[Enum_TLSKeyLabel, OrderedMultiDict[bytes, bytes]]]' = None,
data: 'Optional[str]' = 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.
data: TLS key log text, written back verbatim in place of
``entries``, as the text of ``secrets`` is.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed secrets schema.
.. versionchanged:: 1.6.0
Takes the key log text as ``data``, so that the :obj:`dict` of a
parsed model rebuilds the log as captured, as the model itself does,
rather than as its entries under a generated comment
(:issue:`2698`).
"""
from pcapkit import __version__
if secrets is not None:
data = secrets.get('data')
if isinstance(data, str):
return Schema_TLSKeyLog(
data=data,
)
if secrets is not None:
# NOTE: each an OrderedMultiInfo, which is an OrderedMultiDict.
entries = cast('Mapping[Enum_TLSKeyLabel, OrderedMultiDict[bytes, bytes]]', secrets.entries)
if entries is None:
entries = {}
lines = [f'# generated by PyPCAPKit v{__version__}\n'] # type: list[str]
for label, entry in entries.items():
for k, v in entry.items(multi=True):
lines.append(f'{label.name} {k.hex()} {v.hex()}\n')
return Schema_TLSKeyLog(
data=''.join(lines),
)
[docs]
def _make_secrets_wireguard(self, type: 'Enum_SecretsType', secrets: 'Optional[Data_WireGuardKeyLog]' = None, *,
entries: 'Optional[OrderedMultiDict[WireGuardKeyLabel, bytes]]' = None,
data: 'Optional[str]' = 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.
data: WireGuard key log text, written back verbatim in place of
``entries``, as the text of ``secrets`` is.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed secrets schema.
.. versionchanged:: 1.6.0
Takes the key log text as ``data``, as
:meth:`_make_secrets_tls` does (:issue:`2698`).
"""
from pcapkit import __version__
if secrets is not None:
data = secrets.get('data')
if isinstance(data, str):
return Schema_WireGuardKeyLog(
data=data,
)
if secrets is not None:
entries = secrets.entries
if entries is None:
entries = OrderedMultiDict()
lines = [f'# generated by PyPCAPKit v{__version__}\n'] # type: list[str]
for label, value in entries.items(multi=True):
lines.append(f'{label.name} = {base64.b64encode(value).decode()}\n')
return Schema_WireGuardKeyLog(
data=''.join(lines),
)
[docs]
def _make_secrets_zigbee_aps(self, type: 'Enum_SecretsType', secrets: 'Optional[Data_ZigBeeAPSKey]' = None, *,
aps_key: 'bytes' = bytes(16),
pan_id: 'int' = 0,
short_address: 'int' = 0,
data: 'Optional[bytes]' = None,
**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, 16 octets.
pan_id: PAN ID.
short_address: Short address.
data: Secrets Data octets, written back verbatim in place of the
fields, as the octets of ``secrets`` are.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed secrets schema.
Raises:
ProtocolError: If ``aps_key`` is not 16 octets.
.. versionchanged:: 1.6.0
* Takes the Secrets Data octets as ``data``, as
:meth:`_make_secrets_zigbee_nwk` does (:issue:`2698`).
* ``aps_key`` of any other width, empty included, raises
:exc:`~pcapkit.utilities.exceptions.ProtocolError` rather than
being padded or cut to the 16-octet key, and defaults to 16 zero
octets where it defaulted to empty (:issue:`2676`).
"""
if secrets is not None:
data = secrets.get('data')
if isinstance(data, bytes):
return zigbee_key_held(Schema_ZigBeeAPSKey, data)
if secrets is not None:
aps_key = secrets.aps_key
pan_id = secrets.pan_id
short_address = secrets.short_address
return Schema_ZigBeeAPSKey(
key=self._make_octets(aps_key, size=16, name='[ZigBee APS Key] key', alias='PCAP-NG'),
panid=pan_id,
addr_high=short_address >> 16,
addr_low=short_address & 0xFFFF,
)
[docs]
def _make_secrets_zigbee_nwk(self, type: 'Enum_SecretsType', secrets: 'Optional[Data_ZigBeeNWKKey]' = None, *,
nwk_key: 'bytes' = bytes(16),
pan_id: 'int' = 0,
data: 'Optional[bytes]' = None,
**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, 16 octets.
pan_id: PAN ID.
data: Secrets Data octets, written back verbatim in place of
``nwk_key`` and ``pan_id``, as the octets of ``secrets`` are.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed secrets schema.
Raises:
ProtocolError: If ``nwk_key`` is not 16 octets.
.. versionchanged:: 1.6.0
* Takes the Secrets Data octets as ``data``, so that the :obj:`dict`
of a parsed model rebuilds the key as captured, as the model itself
does, rather than as its zero-filled fields (:issue:`2698`).
* ``nwk_key`` of any other width, empty included, raises
:exc:`~pcapkit.utilities.exceptions.ProtocolError` rather than
being padded or cut to the 16-octet key, and defaults to 16 zero
octets where it defaulted to empty (:issue:`2676`).
"""
if secrets is not None:
data = secrets.get('data')
if isinstance(data, bytes):
return zigbee_key_held(Schema_ZigBeeNWKKey, data)
if secrets is not None:
nwk_key = secrets.nwk_key
pan_id = secrets.pan_id
return Schema_ZigBeeNWKKey(
key=self._make_octets(nwk_key, size=16, name='[ZigBee NWK Key] key', alias='PCAP-NG'),
panid=pan_id,
)
[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.misc.pcapng.schema.PCAPNG.type>` have been padded out
of nothing ``__length__`` is negative -- which
:func:`~pcapkit.protocols.misc.pcapng.schema.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 both of :meth:`unpack`'s paths. On the parsing path the
tail is what the file has left to read; on the construction path it is
the octets of a block handed to :meth:`make` as raw :obj:`bytes`, which
:meth:`unpack` parses off the buffer :meth:`pack` has just filled
(:issue:`2354`) -- a caller's octets are no more trustworthy than a
file's. Octets too short for this floor are therefore refused here
rather than by :meth:`read`'s own Block Total Length check, which sees
only what gets past the floor and so names an impossible *declared*
length, never a tail too short to declare one.
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)
[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.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:
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:
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 zone, by default in both branches: the EPB timestamp is
# an offset from the UTC epoch (draft-ietf-opsawg-pcapng-02 section 4.3).
if self._ctx is None:
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: The UTC epoch, with no ``tzone.utcoffset`` added: the timestamp is
# zone-free (draft section 4.3), and section 4.2 says ``if_tzone`` SHOULD NOT be used.
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`` sets only the zone the instant renders in, never the instant. Not the
# stdlib's ``fromtimestamp``, whose C-library range on Windows is 1970 to 3000 (#2099).
ts_datetime = fromtimestamp(ts_ratio[0] / ts_ratio[1], tzone)
# an epoch past what ``datetime`` or a 64-bit ``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)
@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]') -> 'int':
"""Count the octets a parsed block held past its ``opt_endofopt``.
Args:
block: Block data model, if any.
Returns:
Number of octets, which the block's length takes in.
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 :meth:`_make_padding` writes them back into the padding field after
the options, verbatim (:issue:`1437`). As an item of the options, they
would read as an area kept as captured, and a block rebuilt from its
``info`` would lose its options and ``padding_opts`` (:issue:`2776`).
.. versionchanged:: 1.6.0
Takes no ``options``, and no longer appends the octets to them
(:issue:`2776`).
"""
trailer = block.get('padding_opts') if block is not None else None
if not isinstance(trailer, (bytes, bytearray)) or not trailer:
return 0
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.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.
Which clock a defaulted ``timestamp`` is read to follows the
resolution of the interface ``interface_id`` names, not
:attr:`self.nanosecond <nanosecond>`. That property resolves its
interface through the parsed data model, which a block being made
has not got, so reading it here would make a ``make`` that names
no ``timestamp`` -- of an Enhanced Packet Block, an obsolete
Packet Block or an Interface Statistics Block -- raise
:exc:`AttributeError` on ``_info`` (:issue:`2362`).
.. versionchanged:: 1.5.0
A defaulted ``timestamp`` no longer reads
:attr:`self.nanosecond <nanosecond>`, so it no longer requires a
parsed data model (:issue:`2362`).
"""
with localcontext(prec=64):
resolution = self._get_resolution(interface_id)
if timestamp is None:
if resolution > RESOLUTION[False]:
timestamp = decimal.Decimal(time.time_ns()) / 1_000_000_000
else:
timestamp = decimal.Decimal(time.time())
else:
if isinstance(timestamp, datetime.datetime):
timestamp = totimestamp(timestamp)
timestamp = decimal.Decimal(timestamp)
ts_info = int(
(timestamp - self._get_offset(interface_id)) * resolution
)
return (ts_info >> 32) & 0xFFFF_FFFF, ts_info & 0xFFFF_FFFF
[docs]
def _make_mac_addr(self, addr: 'str | bytes | bytearray', *, size: 'Optional[int]' = 6,
name: 'str' = 'MAC address', alias: 'Optional[str]' = None) -> 'bytes':
"""Make a MAC address, or an EUI-64 one with ``size=8``.
Args:
addr: MAC or EUI address.
size: Octets in the address, as for :meth:`Protocol._make_mac_addr
<pcapkit.protocols.protocol.Protocol._make_mac_addr>`.
name: Name of the address, for the error message.
alias: Prefix of the error message, ``'PCAP-NG'`` by default.
Returns:
The address packed as ``size`` octets, 6 by default.
Raises:
ProtocolError: If ``addr`` is not a :class:`str`, :class:`bytes`
or :class:`bytearray` (a :obj:`bool` included), or is not a
``:``- or ``-``-separated *hex* encoded address of ``size``
octets.
"""
prefix = 'PCAP-NG' if alias is None else alias
return super()._make_mac_addr(addr, size=size, name=name, alias=prefix)
[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:
``dict_`` gains ``__next_type__`` and ``__next_name__``, the next
layer's protocol type and name, which :meth:`Info.to_dict
<pcapkit.corekit.infoclass.Info.to_dict>` carries and the dumpers
leave out, and ``protocols``, 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.protocol.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.protocol.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_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_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 name resolution record data
records.add(type, data)
# break when ``nrb_record_end`` is reached
if type == Enum_RecordType.nrb_record_end:
break
return records
[docs]
def _make_pcapng_options(self, options: 'Option | list[Schema_Option | tuple[Enum_OptionType | int, dict[str, Any]] | bytes] | bytes', # pylint: disable=line-too-long
namespace: 'str') -> 'tuple[list[Schema_Option | bytes], int]':
"""Make options for PCAP-NG.
Args:
options: PCAP-NG options. An option's code is a member, kept as
given, or an :class:`int`, resolved in ``namespace``.
namespace: Namespace of options.
Returns:
Tuple of options and total length of options.
Raises:
ProtocolError: If an option's code is a name, a :obj:`bool`, or a
code the 16-bit option code field cannot carry
(:issue:`2594`), or if an option given as raw octets is
shorter than that field (:issue:`2723`).
"""
# 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):
# NOTE: Octets short of the code field would read as code 0
# and be taken for an opt_endofopt (#2723).
if len(schema) < 2:
raise ProtocolError(f'PCAP-NG: invalid option: {schema!r} holds no whole option code')
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:
kind, args = cast('tuple[Enum_OptionType | int, dict[str, Any]]', schema)
# NOTE: a member is a str, which ``_make_index`` refuses, and a code is
# resolved in the block's namespace, as a raw option's is.
code = kind if isinstance(kind, Enum_OptionType) else Enum_OptionType.get(
self._make_uint(kind, bits=16, name='option code', alias='PCAP-NG'), namespace=namespace)
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 key, option in options.items(multi=True):
code = key if isinstance(key, Enum_OptionType) else Enum_OptionType.get(
self._make_uint(key, bits=16, name='option code', alias='PCAP-NG'), namespace=namespace)
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_nrb_records(self, records: 'Record | list[Schema_NameResolutionRecord | tuple[Enum_RecordType | int, dict[str, Any]] | bytes] | bytes') -> 'tuple[list[Schema_NameResolutionRecord | bytes], int]': # pylint: disable=line-too-long
"""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.
Raises:
ProtocolError: If a record's type is a name, a :obj:`bool`, or a
code the 16-bit record type field cannot carry
(:issue:`2594`), or if a record given as raw octets is shorter
than that field (:issue:`2723`).
"""
# 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):
# NOTE: Octets short of the type field would read as type 0
# and be taken for an nrb_record_end (#2723).
if len(schema) < 2:
raise ProtocolError(f'PCAP-NG: invalid record: {schema!r} holds no whole record type')
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:
kind, args = cast('tuple[Enum_RecordType | int, dict[str, Any]]', schema)
code = self._make_index(kind, Enum_RecordType, bits=16, name='record type', alias='PCAP-NG')
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 key, record in records.items(multi=True):
code = self._make_index(key, Enum_RecordType, bits=16, name='record type', alias='PCAP-NG')
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