# -*- coding: utf-8 -*-
# mypy: disable-error-code=dict-item
"""Root Protocol
===================
.. module:: pcapkit.protocols.protocol
:mod:`pcapkit.protocols.protocol` contains
:class:`~pcapkit.protocols.protocol.Protocol`, the abstract base class for a
new protocol stack, and :class:`~pcapkit.protocols.protocol.ProtocolBase`,
from which the built-in protocol families derive, with its metaclass
:class:`~pcapkit.protocols.protocol.ProtocolMeta`.
.. versionchanged:: 1.5.0
Constructing a protocol with a keyword that none of its signatures
declares raises :exc:`~pcapkit.utilities.exceptions.UnsupportedCall`,
where it was discarded (:issue:`617`). A next layer captured shorter than
its header is kept as :class:`~pcapkit.protocols.misc.raw.protocol.Raw`, where it
was parsed with zero-filled fields (:issue:`1170`).
"""
import abc
import collections
import collections.abc
import contextlib
import copy
import difflib
import enum
import functools
import importlib
import inspect
import io
import os
import shutil
import string
import struct
import textwrap
import types
import urllib.parse
from typing import TYPE_CHECKING, Any, Generic, Optional, Type, TypeVar, cast, overload
import aenum
from pcapkit.corekit.context import ContextRegistry
from pcapkit.corekit.decorators import beholder, seekset
from pcapkit.corekit.fields.misc import ConditionalField, SchemaField, SwitchField
from pcapkit.corekit.infoclass import Info
from pcapkit.corekit.module import ModuleDescriptor
from pcapkit.corekit.multidict import MultiDict
from pcapkit.corekit.protochain import ProtoChain
from pcapkit.corekit.sentinels import ABSENT
from pcapkit.protocols.data import Data
from pcapkit.protocols.data import Packet as Data_Packet
from pcapkit.protocols.schema import Schema
from pcapkit.utilities import config
from pcapkit.utilities.chardet import detect
from pcapkit.utilities.compat import cached_property
from pcapkit.utilities.contextvars import beholding, frame_depth, rebuilding
from pcapkit.utilities.exceptions import (BaseError, ProtocolError, ProtocolNotFound,
ProtocolNotImplemented, RegistryError, StructError,
UnsupportedCall, stacklevel)
from pcapkit.utilities.inspect import isabstract
from pcapkit.utilities.warnings import (ProtocolWarning, RegistryWarning, UnknownFieldWarning,
hold_warnings, warn)
if TYPE_CHECKING:
from enum import IntEnum as StdlibEnum
from types import FrameType
from typing import IO, Any, Callable, DefaultDict, Iterable, Optional, Type
from aenum import IntEnum as AenumEnum
from typing_extensions import Literal, Self
from pcapkit.corekit.context import ProtocolContext
from pcapkit.corekit.enum import EnumLookup
__all__ = ['Protocol', 'ProtocolBase']
_PT = TypeVar('_PT', bound='Data')
_ST = TypeVar('_ST', bound='Schema')
_CTX = TypeVar('_CTX', bound='ProtocolContext')
_VT = TypeVar('_VT')
_ET = TypeVar('_ET', bound='EnumLookup')
# readable characters' order list
readable = [ord(char) for char in filter(lambda char: not char.isspace(), string.printable)]
def _passes_raw(value: 'Any', nested: 'bool' = False) -> 'bool':
"""Tell if ``make()`` keywords may carry a raw :obj:`bytes` item.
Args:
value: keyword arguments of ``make()``, or a value within them
nested: if ``value`` sits inside a container
Returns:
:data:`True` if :obj:`bytes` sits anywhere inside a :obj:`list`,
:obj:`tuple`, :obj:`dict` or data model among ``value``, or a
:class:`~pcapkit.protocols.schema.Schema` does anywhere. A
top-level :obj:`bytes` keyword (e.g. a payload) does not count. A data
model is looked into as a :obj:`dict` is, since a maker builds from its
fields as given, and a multi-mapping through every value of each key
(:issue:`2725`).
"""
if isinstance(value, bytes):
return nested
if isinstance(value, Schema): # a ready-made item, which ``make()`` may pack into raw octets
return True
if isinstance(value, (Info, MultiDict)):
return any(_passes_raw(item, nested) for key, item in value.items(multi=True) if key != '__packet__')
if isinstance(value, dict):
return any(_passes_raw(item, nested) for key, item in value.items() if key != '__packet__')
if isinstance(value, (list, tuple)):
return any(_passes_raw(item, True) for item in value)
return False
@functools.lru_cache(maxsize=None)
def _schema_slots(schema: 'Type[Schema]') -> 'tuple[str, ...]':
"""Names of the fields of ``schema`` that hold a schema.
Args:
schema: schema class
Returns:
Names of the :class:`~pcapkit.corekit.fields.misc.SchemaField` and
:class:`~pcapkit.corekit.fields.misc.SwitchField` fields, including
those wrapped in a :class:`~pcapkit.corekit.fields.misc.ConditionalField`.
"""
slots = []
for name, field in schema.__fields__.items():
while isinstance(field, ConditionalField):
field = field.field
if isinstance(field, (SchemaField, SwitchField)):
slots.append(name)
return tuple(slots)
def _holds_raw(value: 'Any') -> 'bool':
"""Tell if a made schema may hold raw :obj:`bytes` in place of a schema.
Args:
value: made schema, or a value within it
Returns:
:data:`True` if ``value`` holds :obj:`bytes` as a list item, or as
the value of a field that holds a schema
(:class:`~pcapkit.corekit.fields.misc.SchemaField` or
:class:`~pcapkit.corekit.fields.misc.SwitchField`).
"""
if isinstance(value, (list, tuple)):
return any(isinstance(item, bytes) or _holds_raw(item) for item in value)
if isinstance(value, Schema):
for name, field in value.__fields__.items():
item = value.get(name)
while isinstance(field, ConditionalField):
field = field.field
if isinstance(item, bytes) and isinstance(field, (SchemaField, SwitchField)):
return True
if _holds_raw(item):
return True
return False
def _raw_where_parsed(made: 'Any', parsed: 'Any') -> 'bool':
"""Tell if ``made`` holds raw :obj:`bytes` where ``parsed`` holds a schema.
Args:
made: made schema, or a value within it
parsed: the same position in the schema parsed from its packed octets
Returns:
:data:`True` if, at some position both share, ``made`` holds
:obj:`bytes` and ``parsed`` a :class:`~pcapkit.protocols.schema.Schema`.
"""
if isinstance(made, bytes):
return isinstance(parsed, Schema)
if isinstance(made, (list, tuple)) and isinstance(parsed, (list, tuple)):
return any(_raw_where_parsed(item, other) for item, other in zip(made, parsed))
if isinstance(made, Schema) and isinstance(parsed, Schema):
return any(_raw_where_parsed(made.get(name), parsed.get(name))
for name in made.__fields__ if name in parsed.__fields__)
return False
#: Construction keywords that name no field.
#: :meth:`Protocol.__init__ <pcapkit.protocols.protocol.Protocol.__init__>` or the
#: schema layer consumes them, not :meth:`make <pcapkit.protocols.protocol.Protocol.make>`.
#: No signature declares them, so :func:`_declared_keywords` takes them from here.
#:
#: ``packet`` is injected by the library: :meth:`Protocol.__init__
#: <pcapkit.protocols.protocol.Protocol.__init__>` puts
#: ``packet=self.packet.payload`` into every parsed ``_info``, which the default
#: :meth:`Protocol._make_data <pcapkit.protocols.protocol.Protocol._make_data>`
#: (``data.to_dict()``) carries into the keywords :meth:`Protocol.from_data
#: <pcapkit.protocols.protocol.Protocol.from_data>` rebuilds from. Refusing it
#: would break ``from_data`` for every protocol whose ``make`` lacks it, starting
#: with :class:`~pcapkit.protocols.misc.no_payload.protocol.NoPayload`. Where it
#: is a real field, as the packet *type* of :meth:`HIP.make
#: <pcapkit.protocols.internet.hip.protocol.HIP.make>`, listing it here only
#: stops it being refused.
OUT_OF_BAND_KEYWORDS = frozenset({'_layer', '_protocol', '__context__',
'__packet__', 'packet'})
#: Cache for :func:`_declared_keywords`, keyed by protocol class. Signatures do
#: not change after the class is created and the walk below is
#: :math:`O(\\text{MRO} \\times \\text{methods})`, so it runs once per class
#: rather than once per constructed packet.
_DECLARED_KEYWORDS = {} # type: dict[type, Optional[frozenset[str]]]
#: Methods a construction keyword may be for.
#: :meth:`Protocol.__post_init__ <pcapkit.protocols.protocol.Protocol.__post_init__>`
#: passes the same ``**kwargs`` to :meth:`pack <pcapkit.protocols.protocol.Protocol.pack>`
#: (and so ``make``) *and* :meth:`unpack <pcapkit.protocols.protocol.Protocol.unpack>`
#: (and so ``read``).
_KEYWORD_CONSUMERS = ('make', 'read', 'pack', 'unpack', '__post_init__', '__init__')
[docs]
def _declared_keywords(cls: 'type') -> 'Optional[frozenset[str]]':
"""Collect every keyword the protocol ``cls`` declares a parameter for.
Args:
cls: Protocol class to inspect.
Returns:
Names of every keyword-acceptable parameter declared by any of
:data:`_KEYWORD_CONSUMERS` anywhere in the MRO of ``cls``, plus every
entry of :attr:`Protocol.__keywords__
<pcapkit.protocols.protocol.Protocol.__keywords__>` found there,
plus :data:`OUT_OF_BAND_KEYWORDS`. :obj:`None` if ``cls`` *itself* sets
``__keywords__`` to :obj:`None`, meaning its keywords cannot be enumerated
and are not to be checked -- inherited :obj:`None` does not count, as
:attr:`Protocol.__keywords__
<pcapkit.protocols.protocol.Protocol.__keywords__>` explains.
The union is wider than the signature of ``cls.make``:
:meth:`Protocol.__post_init__
<pcapkit.protocols.protocol.Protocol.__post_init__>` hands one ``**kwargs``
to both the construction and the parse, so a keyword ``read`` declares
travels through ``make`` too, as ``extension`` of :meth:`HIP.read
<pcapkit.protocols.internet.hip.protocol.HIP.read>` does.
The walk covers the whole MRO, not the most derived override of each method,
so a subclass forwarding to its parent keeps the parent's keywords. The
exception is a class that sets :attr:`Protocol.__own_make_keywords__
<pcapkit.protocols.protocol.Protocol.__own_make_keywords__>`: its ``make``
counts only while no subclass overrides it (:issue:`2286`).
"""
try:
return _DECLARED_KEYWORDS[cls]
except KeyError:
pass
unchecked = False
names = set(OUT_OF_BAND_KEYWORDS)
# NOTE: The class whose ``make`` is the one ``cls`` resolves to; any other
# ``make`` in the MRO has been overridden.
make_owner = next((klass for klass in cls.__mro__ if 'make' in klass.__dict__), None)
for klass in cls.__mro__:
# NOTE: ``inspect.signature`` misses a keyword read out of ``**kwargs``,
# so each class lists its own in ``__keywords__``, unioned down the MRO.
keywords = klass.__dict__.get('__keywords__', ABSENT)
if keywords is None:
# NOTE: Not inherited: HTTP forwards to HTTPv1 or HTTPv2, which each
# declare their keywords in full and so stay checked.
if klass is cls:
unchecked = True
elif keywords is not ABSENT:
names.update(keywords)
for method in _KEYWORD_CONSUMERS:
# NOTE: From ``__dict__``, not ``getattr``, so each class in the MRO
# contributes its own definition, not the most derived one.
func = klass.__dict__.get(method)
if func is None:
continue
# NOTE: Read only while it is the ``make`` that ``cls`` uses; from
# ``__dict__`` so the marker is not inherited.
if (method == 'make' and klass is not make_owner
and klass.__dict__.get('__own_make_keywords__', False)):
continue
try:
signature = inspect.signature(func)
except (TypeError, ValueError): # pragma: no cover
# NOTE: Skipped, not fatal; the worst case is a false rejection.
continue
for name, param in signature.parameters.items():
if name in ('self', 'cls'):
continue
if param.kind in (inspect.Parameter.POSITIONAL_OR_KEYWORD,
inspect.Parameter.KEYWORD_ONLY):
names.add(name)
declared = None if unchecked else frozenset(names)
_DECLARED_KEYWORDS[cls] = declared
return declared
def _check_construction_keywords(cls: 'type', kwargs: 'dict[str, Any]',
strict: 'bool' = True) -> 'None':
"""Reject construction keywords that the protocol ``cls`` declares nowhere.
Args:
cls: Protocol class being constructed.
kwargs: Construction keywords to check, the out-of-band ones left out.
:meth:`Protocol.__init__
<pcapkit.protocols.protocol.Protocol.__init__>` passes those
remaining after it has consumed them; :meth:`Protocol.from_data
<pcapkit.protocols.protocol.Protocol.from_data>` passes its
caller's own, less ``_layer``, ``_protocol`` and ``__context__``,
before :meth:`Protocol._make_data
<pcapkit.protocols.protocol.Protocol._make_data>` adds to them.
strict: Whether an unexpected keyword is an error. :data:`True` for a
caller's own construction; :data:`False` when the keywords were
generated by :meth:`Protocol._make_data
<pcapkit.protocols.protocol.Protocol._make_data>` rather than
written by anybody -- see :meth:`Protocol.from_data
<pcapkit.protocols.protocol.Protocol.from_data>`.
Raises:
UnsupportedCall: If ``strict`` and any keyword is not among those
:func:`_declared_keywords` collects for ``cls``.
Warns:
UnknownFieldWarning: The same finding when not ``strict``.
"""
declared = _declared_keywords(cls)
if declared is None:
return
unexpected = sorted(key for key in kwargs if key not in declared)
if not unexpected:
return
# NOTE: The check catches misspellings, so suggest the closest declared
# name, e.g. ``seq_no`` for ``seq``.
report = [] # type: list[str]
for key in unexpected:
suggestions = difflib.get_close_matches(key, declared, n=1)
report.append(f'{key!r} (did you mean {suggestions[0]!r}?)' if suggestions else repr(key))
listed = ', '.join(report)
if strict:
raise UnsupportedCall(f'{cls.__name__}: unexpected keyword(s): {listed}')
# NOTE: A warning, not an error: the caller cannot fix keys ``_make_data`` generated, and
# raising would break ``from_data``; the default ``stacklevel`` blames that call.
warn(f'{cls.__name__}._make_data returned keyword(s) that no signature of '
f'{cls.__name__} declares, so they are discarded: {listed}',
UnknownFieldWarning)
def _short_of_hint(klass: 'Type[ProtocolBase]', captured: 'int') -> 'bool':
"""Tell whether ``captured`` octets may be too few for a ``klass`` header.
Args:
klass: Next layer protocol class.
captured: Octets captured for it.
Returns:
Whether a next layer of ``klass`` parsed from ``captured`` octets may be
replaced with :class:`~pcapkit.protocols.misc.raw.protocol.Raw` by
:meth:`Protocol._parse_next_layer
<pcapkit.protocols.protocol.Protocol._parse_next_layer>`, and so has
its warnings held.
The hint is read from the class, before there is an instance to ask, which
is enough for every protocol shipped: each returns a constant. A hint that
needs an instance raises here instead, and the layer counts as short, which
only means its warnings are held until the parse is known to be kept.
"""
try:
hint = klass.__length_hint__(cast('ProtocolBase', None))
except Exception: # pylint: disable=broad-except
return True
return hint is not None and captured < hint
def _sibling_model(cls: 'Type[ProtocolBase]', role: 'str') -> 'Any':
"""The ``role`` class for a protocol class defined without one (:issue:`2534`).
Args:
cls: The protocol class being defined.
role: ``'schema'`` or ``'data'``, the module to look in.
Returns:
The class named after ``cls`` in the ``role`` module of the package whose
``protocol`` module defines ``cls``, if there is one; otherwise
:class:`~pcapkit.protocols.misc.raw.schema.Raw` or
:class:`~pcapkit.protocols.misc.raw.data.Raw`.
"""
package, _, name = cls.__module__.rpartition('.')
if name == 'protocol' and package != __name__.rpartition('.')[0]:
sibling = f'{package}.{role}'
try:
module = importlib.import_module(sibling)
except ModuleNotFoundError as exc:
if exc.name != sibling:
raise
else:
found = getattr(module, cls.__name__, None)
if isinstance(found, type):
return found
# NOTE: imported here rather than at the top: Raw's package is under
# pcapkit.protocols.misc, whose modules import this one (#2534).
if role == 'schema':
from pcapkit.protocols.misc.raw.schema import \
Raw as Schema_Raw # pylint: disable=import-outside-toplevel
return Schema_Raw
from pcapkit.protocols.misc.raw.data import \
Raw as Data_Raw # pylint: disable=import-outside-toplevel
return Data_Raw
[docs]
class ProtocolBase(Generic[_PT, _ST], metaclass=ProtocolMeta):
"""Abstract base class for all protocol families.
The built-in protocol families -- the layer base classes
:class:`~pcapkit.protocols.link.link.Link`,
:class:`~pcapkit.protocols.internet.internet.Internet`,
:class:`~pcapkit.protocols.transport.transport.Transport` and
:class:`~pcapkit.protocols.application.application.Application`, and the
miscellaneous protocols in :mod:`pcapkit.protocols.misc`, PCAP and PCAP-NG
included -- derive from this class directly. Its metaclass is
:class:`ProtocolMeta`.
Note:
This class is for internal use only. For customisation, subclass
:class:`Protocol` to create a new protocol stack, or the matching
layer base class above for a new protocol in an existing layer.
"""
if TYPE_CHECKING:
#: Parsed packet data.
_info: '_PT'
#: Raw packet data.
_data: 'bytes'
#: Source packet stream.
_file: 'IO[bytes]'
#: Next layer protocol instance.
_next: 'ProtocolBase'
#: Protocol chain instance.
_protos: 'ProtoChain'
#: If a header set before :meth:`unpack` may hold raw items.
_make_raw: 'bool'
# Internal data storage for cached properties.
__cached__: 'dict[str, Any]'
#: Protocol packet data definition.
__data__: 'Type[_PT]'
#: Protocol header schema definition.
__schema__: 'Type[_ST]'
#: Protocol header schema instance.
__header__: '_ST'
##########################################################################
# Defaults.
##########################################################################
#: Layer of protocol, can be one of ``Link``, ``Internet``, ``Transport``
#: and ``Application``. For example, the layer of
#: :class:`~pcapkit.protocols.link.ethernet.protocol.Ethernet` is ``Link``. However,
#: certain protocols are not in any layer, such as
#: :class:`~pcapkit.protocols.misc.raw.protocol.Raw`, and thus its layer is :obj:`None`.
__layer__: 'Optional[Literal["Link", "Internet", "Transport", "Application"]]' = None
#: 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:`ProtocolBase` subclass.
__proto__: 'DefaultDict[int, ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]]' = collections.defaultdict(
lambda: ModuleDescriptor('pcapkit.protocols.misc.raw.protocol', 'Raw'),
)
#: Read-only copy of :attr:`__proto__` as the class defining it shipped it,
#: taken by :meth:`__init_subclass__` before anything can register into it.
#: :meth:`_next_layer_entry` reads it to tell a descriptor handed back from
#: a new one without resolving it (:issue:`1504`).
_shipped_proto: 'types.MappingProxyType[Any, ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]]' = \
types.MappingProxyType({})
#: Protocol registry by name, holding each protocol class under its
#: upper-cased class name, ``'IPV4'`` for :class:`~pcapkit.protocols.internet.ipv4.protocol.IPv4`.
#: Each built-in protocol is a :class:`~pcapkit.corekit.module.ModuleDescriptor`
#: here, so the map names every one before any is imported, and
#: :meth:`get_protocol` imports a class only when it resolves its entry
#: (:issue:`2763`). :meth:`Protocol.__init_subclass__` adds each concrete
#: :class:`Protocol` subclass as it is defined, :meth:`register_protocol`
#: replaces an entry, and :attr:`ProtocolMeta.registry` is a read-only view.
#:
#: .. versionadded:: 1.6.0
__registry__: 'dict[str, ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]]' = {
# Link Layer
'ARP': ModuleDescriptor('pcapkit.protocols.link.arp.protocol', 'ARP'),
'C_TAG': ModuleDescriptor('pcapkit.protocols.link.c_tag.protocol', 'C_Tag'),
'ETHERNET': ModuleDescriptor('pcapkit.protocols.link.ethernet.protocol', 'Ethernet'),
'FRAD': ModuleDescriptor('pcapkit.protocols.link.frad.protocol', 'FRAD'),
'GRE': ModuleDescriptor('pcapkit.protocols.link.gre.protocol', 'GRE'),
'IEEE802_11': ModuleDescriptor('pcapkit.protocols.link.ieee802_11.protocol', 'IEEE802_11'),
'INARP': ModuleDescriptor('pcapkit.protocols.link.inarp.protocol', 'InARP'),
'L2TP': ModuleDescriptor('pcapkit.protocols.link.l2tp.protocol', 'L2TP'),
'L2TPV2': ModuleDescriptor('pcapkit.protocols.link.l2tpv2.protocol', 'L2TPv2'),
'LINUXSLL': ModuleDescriptor('pcapkit.protocols.link.linux_sll.protocol', 'LinuxSLL'),
'LINUXSLL1': ModuleDescriptor('pcapkit.protocols.link.linux_sll1.protocol', 'LinuxSLL1'),
'LINUXSLL2': ModuleDescriptor('pcapkit.protocols.link.linux_sll2.protocol', 'LinuxSLL2'),
'LOOPBACK': ModuleDescriptor('pcapkit.protocols.link.loopback.protocol', 'Loopback'),
'NETLINK': ModuleDescriptor('pcapkit.protocols.link.netlink.protocol', 'Netlink'),
'RADIOTAP': ModuleDescriptor('pcapkit.protocols.link.radiotap.protocol', 'Radiotap'),
'S_TAG': ModuleDescriptor('pcapkit.protocols.link.s_tag.protocol', 'S_Tag'),
'VLAN': ModuleDescriptor('pcapkit.protocols.link.vlan.protocol', 'VLAN'),
# Internet Layer
'AH': ModuleDescriptor('pcapkit.protocols.internet.ah.protocol', 'AH'),
'ESP': ModuleDescriptor('pcapkit.protocols.internet.esp.protocol', 'ESP'),
'HIP': ModuleDescriptor('pcapkit.protocols.internet.hip.protocol', 'HIP'),
'HOPOPT': ModuleDescriptor('pcapkit.protocols.internet.hopopt.protocol', 'HOPOPT'),
'ICMP': ModuleDescriptor('pcapkit.protocols.internet.icmp.protocol', 'ICMP'),
'ICMPV4': ModuleDescriptor('pcapkit.protocols.internet.icmpv4.protocol', 'ICMPv4'),
'ICMPV6': ModuleDescriptor('pcapkit.protocols.internet.icmpv6.protocol', 'ICMPv6'),
'IGMP': ModuleDescriptor('pcapkit.protocols.internet.igmp.protocol', 'IGMP'),
'IGMPV1': ModuleDescriptor('pcapkit.protocols.internet.igmpv1.protocol', 'IGMPv1'),
'IGMPV2': ModuleDescriptor('pcapkit.protocols.internet.igmpv2.protocol', 'IGMPv2'),
'IGMPV3': ModuleDescriptor('pcapkit.protocols.internet.igmpv3.protocol', 'IGMPv3'),
'IP': ModuleDescriptor('pcapkit.protocols.internet.ip.protocol', 'IP'),
'IPSEC': ModuleDescriptor('pcapkit.protocols.internet.ipsec.protocol', 'IPsec'),
'IPV4': ModuleDescriptor('pcapkit.protocols.internet.ipv4.protocol', 'IPv4'),
'IPV6': ModuleDescriptor('pcapkit.protocols.internet.ipv6.protocol', 'IPv6'),
'IPV6_EXT': ModuleDescriptor('pcapkit.protocols.internet.ipv6_ext.protocol', 'IPv6_Ext'),
'IPV6_FRAG': ModuleDescriptor('pcapkit.protocols.internet.ipv6_frag.protocol', 'IPv6_Frag'),
'IPV6_OPTS': ModuleDescriptor('pcapkit.protocols.internet.ipv6_opts.protocol', 'IPv6_Opts'),
'IPV6_ROUTE': ModuleDescriptor('pcapkit.protocols.internet.ipv6_route.protocol', 'IPv6_Route'),
'IPX': ModuleDescriptor('pcapkit.protocols.internet.ipx.protocol', 'IPX'),
'MH': ModuleDescriptor('pcapkit.protocols.internet.mh.protocol', 'MH'),
# Transport Layer
'DCCP': ModuleDescriptor('pcapkit.protocols.transport.dccp.protocol', 'DCCP'),
'SCTP': ModuleDescriptor('pcapkit.protocols.transport.sctp.protocol', 'SCTP'),
'TCP': ModuleDescriptor('pcapkit.protocols.transport.tcp.protocol', 'TCP'),
'UDP': ModuleDescriptor('pcapkit.protocols.transport.udp.protocol', 'UDP'),
# Application Layer
'DRARP': ModuleDescriptor('pcapkit.protocols.application.drarp.protocol', 'DRARP'),
'DTLS': ModuleDescriptor('pcapkit.protocols.application.dtls.protocol', 'DTLS'),
'FTP': ModuleDescriptor('pcapkit.protocols.application.ftp.protocol', 'FTP'),
'FTP_DATA': ModuleDescriptor('pcapkit.protocols.application.ftp_data.protocol', 'FTP_DATA'),
'HTTP': ModuleDescriptor('pcapkit.protocols.application.http.protocol', 'HTTP'),
'HTTPV1': ModuleDescriptor('pcapkit.protocols.application.httpv1.protocol', 'HTTPv1'),
'HTTPV2': ModuleDescriptor('pcapkit.protocols.application.httpv2.protocol', 'HTTPv2'),
'NGAP': ModuleDescriptor('pcapkit.protocols.application.ngap.protocol', 'NGAP'),
'OSPF': ModuleDescriptor('pcapkit.protocols.application.ospf.protocol', 'OSPF'),
'RARP': ModuleDescriptor('pcapkit.protocols.application.rarp.protocol', 'RARP'),
'SSL': ModuleDescriptor('pcapkit.protocols.application.ssl.protocol', 'SSL'),
'SSLV2': ModuleDescriptor('pcapkit.protocols.application.sslv2.protocol', 'SSLv2'),
'SSLV3': ModuleDescriptor('pcapkit.protocols.application.sslv3.protocol', 'SSLv3'),
'TLS': ModuleDescriptor('pcapkit.protocols.application.tls.protocol', 'TLS'),
# Miscellaneous
'FRAME': ModuleDescriptor('pcapkit.protocols.misc.pcap.frame.protocol', 'Frame'),
'HEADER': ModuleDescriptor('pcapkit.protocols.misc.pcap.header.protocol', 'Header'),
'IP_PSEUDO': ModuleDescriptor('pcapkit.protocols.misc.checksum.ip_pseudo.protocol', 'IP_Pseudo'),
'NOPAYLOAD': ModuleDescriptor('pcapkit.protocols.misc.no_payload.protocol', 'NoPayload'),
'PCAPNG': ModuleDescriptor('pcapkit.protocols.misc.pcapng.protocol', 'PCAPNG'),
'RAW': ModuleDescriptor('pcapkit.protocols.misc.raw.protocol', 'Raw'),
}
#: Read-only copy of :attr:`__registry__` as shipped, which
#: :meth:`register_protocol` reads to put a built-in's descriptor back
#: rather than its class, as :attr:`_shipped_proto` serves
#: :meth:`register` (:issue:`1504`).
_shipped_registry: 'types.MappingProxyType[str, ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]]' = \
types.MappingProxyType(dict(__registry__))
#: Construction keywords this protocol takes out of ``**kwargs`` by name
#: rather than declaring as a parameter, as :meth:`IPv6_Ext.read
#: <pcapkit.protocols.internet.ipv6_ext.protocol.IPv6_Ext.read>` does with
#: ``alias``. :func:`_declared_keywords` reads signatures, which miss such a
#: name, so the construction check of :meth:`__init__` accepts only those
#: listed here, unioned over the MRO. Prefer a declared parameter, which also
#: documents the keyword (:issue:`617`).
#:
#: :obj:`None` skips the check, for a *dispatcher* whose real signature is
#: chosen at call time: :meth:`HTTP.make
#: <pcapkit.protocols.application.http.protocol.HTTP.make>` declares only
#: ``version`` and forwards the rest to :meth:`HTTPv1.make
#: <pcapkit.protocols.application.httpv1.protocol.HTTPv1.make>` or
#: :meth:`HTTPv2.make <pcapkit.protocols.application.httpv2.protocol.HTTPv2.make>`.
#: A protocol that skips the check discards an unknown keyword silently.
#: Unlike a set, :obj:`None` is **not** inherited, so ``HTTPv1`` and
#: ``HTTPv2``, which declare their keywords in full, stay checked.
__keywords__: 'Optional[frozenset[str]]' = frozenset()
#: Whether the parameters of this class's own :meth:`make` are accepted only
#: while that ``make`` is the one in use. By default
#: :func:`_declared_keywords` takes every ``make`` in the MRO. A base whose
#: subclasses override ``make`` *without* forwarding sets this, so its
#: keywords are refused for them rather than accepted and dropped, as
#: :class:`~pcapkit.protocols.internet.ipv6_ext.protocol.IPv6_Ext` does for
#: ``len`` and ``data`` (:issue:`2286`). Not inherited, and
#: :attr:`__keywords__` and the other construction methods are unaffected.
__own_make_keywords__: 'bool' = False
#: Whether :meth:`from_data` is rebuilding this instance from a parsed data
#: model, rather than from keywords somebody wrote. It decides only whether
#: the construction keyword check of :meth:`__init__` raises or warns, and
#: is set for that call alone; every other code path sees :data:`False`.
__reconstructing__: 'bool' = False
#: The ``__short_read__`` record of the data model :meth:`from_data`
#: rebuilds this instance from. :meth:`pack` cuts the header it makes back
#: to the octets captured, and :meth:`__post_init__` has :meth:`unpack`
#: parse them at the width ``__short_width__`` records, so the rebuild is
#: parsed as its capture was (:issue:`2703`, :issue:`2761`). Set for that
#: call alone, like :attr:`__reconstructing__`; otherwise :data:`None`.
__short_read__: 'Optional[tuple[str, int]]' = None
#: Caller supplied parsing context; see :mod:`pcapkit.corekit.context`.
#: :meth:`self.__init__ <Protocol.__init__>` sets a
#: :class:`~pcapkit.corekit.context.ContextRegistry`; an instance built
#: without ``__init__`` reads the class-level :data:`None`.
_exctx: 'Optional[ContextRegistry]' = None
#: Whether :meth:`_parse_next_layer` kept this layer as
#: :class:`~pcapkit.protocols.misc.raw.protocol.Raw` unparsed, being past
#: :data:`~pcapkit.utilities.config.FRAME_LAYER_LIMIT`. Such a layer carries
#: an ``error`` as a failed parse does; this tells the two apart, for what
#: follows the limit is not to be read on (:issue:`1610`).
_past_layer_limit: 'bool' = False
##########################################################################
# Properties.
##########################################################################
# name of current protocol
@property
@abc.abstractmethod
def name(self) -> 'str':
"""Name of current protocol."""
# acronym of current protocol
@property
def alias(self) -> 'str':
"""Acronym of current protocol."""
return self.__class__.__name__
# key name for the data model
@property
def info_name(self) -> 'str':
"""Key the enclosing layer's :term:`data model` holds :attr:`info` under."""
return self.__class__.__name__.lower()
# data model of current instance
@property
def info(self) -> '_PT':
"""Parsed :term:`data model` of current instance."""
return self._info
# binary packet data if current instance
@property
def data(self) -> 'bytes':
"""Binary packet data of current instance."""
return self._data
# header length of current protocol
@property
@abc.abstractmethod
def length(self) -> 'int':
"""Header length of current protocol."""
# payload of current instance
@property
def payload(self) -> 'ProtocolBase':
"""Payload of current instance."""
return self._next
# name of next layer protocol
@property
def protocol(self) -> 'Optional[str]':
"""Name of next layer protocol (if any)."""
with contextlib.suppress(IndexError):
return self._protos[0]
return None
# protocol chain of current instance
@property
def protochain(self) -> 'ProtoChain':
"""Protocol chain of current instance.
Note:
:class:`~pcapkit.protocols.misc.no_payload.protocol.NoPayload` has no chain and
narrows this to :obj:`None`; :meth:`_decode_next_layer` passes it
on as ``basis``, which
:class:`~pcapkit.corekit.protochain.ProtoChain` accepts.
"""
return self._protos
# packet data
[docs]
@cached_property
def packet(self) -> 'Data_Packet':
"""Packet data of the protocol.
Note:
The split takes :attr:`self.length <length>` as the length of the
octets *preceding* the payload, and the payload as running to the
end of the buffer: a header followed by its payload.
A protocol laid out otherwise overrides this, or its header eats the
payload and the payload is ``b''``.
:class:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG` does: its
:attr:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG.length` is the
wire's Block Total Length, and the captured octets sit ahead of the
option list and the trailing length field. The payload matters
because :meth:`Protocol.__init__` injects it into every parsed
``_info``, which the dumpers write out.
"""
try:
return self._read_packet(header=self.length)
except UnsupportedCall:
return Data_Packet(
header=b'',
payload=self._read_packet(),
)
# header schema
[docs]
@cached_property
def schema(self) -> '_ST':
"""Header :term:`schema` instance of the protocol."""
return self.__header__
# caller supplied parsing context
@property
def context(self) -> 'ContextRegistry':
"""Caller supplied parsing context.
See Also:
:mod:`pcapkit.corekit.context` for what this channel is for, and
:meth:`self._get_context
<pcapkit.protocols.protocol.Protocol._get_context>` for how a
protocol implementation reaches its own entry.
"""
return ContextRegistry.make(self._exctx)
##########################################################################
# Methods.
##########################################################################
[docs]
@classmethod
def id(cls) -> 'tuple[str, ...]':
"""Index ID of the protocol.
Returns:
By default, it returns the name of the protocol. In certain cases,
the method may return multiple values.
See Also:
:meth:`pcapkit.protocols.protocol.Protocol.__getitem__`
"""
return (cls.__name__,)
[docs]
@abc.abstractmethod
def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> '_PT':
"""Read (parse) packet data.
Args:
length: Length of packet data.
**kwargs: Arbitrary keyword arguments.
Returns:
Parsed packet data.
"""
[docs]
@abc.abstractmethod
def make(self, **kwargs: 'Any') -> '_ST':
"""Make (construct) packet data.
Args:
**kwargs: Arbitrary keyword arguments.
Returns:
Curated protocol schema data.
Note:
The ``**kwargs`` absorbs the keywords :meth:`Protocol.__post_init__
<pcapkit.protocols.protocol.Protocol.__post_init__>` hands to the
parse as well as the construction, so an implementation need not
declare every keyword it is called with. A caller's keyword that no
signature declares still raises
:exc:`~pcapkit.utilities.exceptions.UnsupportedCall` from
:meth:`Protocol.__init__
<pcapkit.protocols.protocol.Protocol.__init__>` (:issue:`617`).
Warning:
**Calling this method directly is not checked**, and discards an
undeclared keyword silently. The check lives in
:meth:`Protocol.__init__
<pcapkit.protocols.protocol.Protocol.__init__>`, the one place the
keywords converge, so it covers ``SomeProtocol(...)`` and its
:meth:`pack`, but not ``object.__new__(cls).make(**kwargs)``, as
:meth:`HTTP.make <pcapkit.protocols.application.http.protocol.HTTP.make>`
calls its versioned implementation. Construct through the
constructor to get the check.
"""
[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.
The octets are built from ``kwargs`` alone, through :meth:`make`;
the instance's state is not read, so a bare
``SomeProtocol(...).pack()`` returns a default header, not the
instance's octets, which :attr:`data` holds.
Nor is that state written. Once the instance is parsed or
constructed, the header is made on a shallow copy (see
:meth:`_packing_target`), so :attr:`schema`, :attr:`info` and
anything :meth:`make` sets, such as
:attr:`TCP.connection <pcapkit.protocols.transport.tcp.protocol.TCP.connection>`,
still describe that packet. Only the call :meth:`__post_init__`
makes to construct the instance keeps the header, for
:meth:`unpack` to read back.
When :meth:`from_data` rebuilds a header the capture cut short,
that call cuts it back to the octets captured, through
``__short_read__`` and :meth:`Schema._cut_short_read
<pcapkit.protocols.schema.Schema._cut_short_read>`
(:issue:`2703`). :meth:`__post_init__` then drops that header, so
:meth:`unpack` parses the octets back and the rebuild holds the
values its parse read, not those :meth:`make` filled in past the
cut (:issue:`2761`).
.. versionchanged:: 1.5.0
Calling it no longer replaces the instance's
:attr:`__header__ <Protocol.__header__>`, which left :attr:`schema`
disagreeing with :attr:`info` (:issue:`1769`).
"""
target = self._packing_target()
header = target.__header__ = target.make(**kwargs)
#: bool: If the made header may hold raw items (see :meth:`unpack`).
target._make_raw = _passes_raw(kwargs) or any( # pylint: disable=protected-access
isinstance(header.get(name), bytes) for name in _schema_slots(type(header)))
packet = kwargs.get('__packet__', {}) # packet data
data = header.pack(packet)
if target.__short_read__ is None:
return data
# NOTE: Cut back before ``unpack`` reads and verifies it, at the width
# ``from_data`` passes on (#2703, #2761).
header.__dict__['__short_read__'] = target.__short_read__
header._cut_short_read() # pylint: disable=protected-access
return bytes(header)
[docs]
def unpack(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> '_PT':
"""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 :attr:`self.__header__ <Protocol.__header__>` already set, by
:meth:`from_schema` or by the :meth:`pack` call of
:meth:`__post_init__`, is kept, unless it holds raw :obj:`bytes`
where parsing its packed octets yields a schema -- e.g. a raw
option given to ``make()``, directly or inside a data model -- which
:meth:`read` cannot consume; it is then replaced by the parsed one
(:issue:`1312`).
A header :meth:`from_data` rebuilds from a short read is parsed
from the octets captured, its warnings discarded, and given its
data model's ``__short_read__`` before :meth:`read` runs, as the
record of a cut those octets may not show (:issue:`2761`).
Raises:
ProtocolError: If the made header holds raw items and its packed
octets do not parse, either as a schema or by :meth:`read`.
The message carries the original error's, which is also
chained as ``__cause__``.
.. versionchanged:: 1.6.0
A raw item that ``make()`` takes from inside a data model, such as
the one :meth:`from_data` passes on, is replaced too, rather than
reaching :meth:`read` (:issue:`2725`).
"""
packet = kwargs.get('__packet__', {}) # packet data
if cast('Optional[_ST]', self.__header__) is None and self.__short_read__ is not None:
# NOTE: A rebuilt short read is parsed as quietly as its made header was read,
# and keeps the record of a cut its octets may not show (#2711, #2761).
with hold_warnings() as held:
held.discard = True
header = cast('_ST', self.__schema__.unpack(self._file, length, packet)) # type: ignore[call-arg,misc]
header.__dict__['__short_read__'] = self.__short_read__
self.__header__ = header
elif cast('Optional[_ST]', self.__header__) is None:
self.__header__ = cast('_ST', self.__schema__.unpack(self._file, length, packet)) # type: ignore[call-arg,misc]
elif self._make_raw and _holds_raw(self.__header__):
# NOTE: A schema may accept octets :meth:`read` rejects, e.g. an option
# overrunning the option area, so the read is covered too (#1394).
try:
parsed = cast('_ST', type(self.__header__).unpack(io.BytesIO(self._data), length, packet)) # type: ignore[call-arg,misc]
if _raw_where_parsed(self.__header__, parsed):
self.__header__ = parsed
return self.read(length, **kwargs)
except (BaseError, struct.error) as error: # pylint: disable=no-member
raise ProtocolError(f'{type(self).__name__}: malformed raw item given to make(): {error}') from error
return self.read(length, **kwargs)
[docs]
@staticmethod
def decode(byte: bytes, *, encoding: 'Optional[str]' = None,
errors: 'Literal["strict", "ignore", "replace"]' = 'strict') -> 'str':
"""Decode :obj:`bytes` into :obj:`str`.
Should decoding fail using ``encoding``, the method will try again decoding
the :obj:`bytes` as ``'unicode_escape'`` with ``'replace'`` for error handling.
See Also:
The method is a wrapping function for :meth:`bytes.decode`.
Args:
byte: Source bytestring.
encoding: The encoding with which to decode the :obj:`bytes`.
If not provided, :mod:`pcapkit` will first try detecting its encoding
using |chardet|_. The fallback encoding is **UTF-8**.
errors: The error handling scheme to use for the handling of decoding errors.
The default is ``'strict'`` meaning that decoding errors raise a
:exc:`UnicodeDecodeError`. Other possible values are ``'ignore'`` and ``'replace'``
as well as any other name registered with :func:`codecs.register_error` that
can handle :exc:`UnicodeDecodeError`.
"""
charset = encoding or detect(byte)
try:
return byte.decode(charset, errors=errors)
except UnicodeError:
return byte.decode('unicode_escape', errors='replace')
[docs]
@staticmethod
def unquote(url: str, *, encoding: 'str' = 'utf-8',
errors: 'Literal["strict", "ignore", "replace"]' = 'replace') -> 'str':
"""Unquote URLs into readable format.
Should decoding fail, the method will try again replacing ``'%'`` with ``'\\x'`` then
decoding the ``url`` as ``'unicode_escape'`` with ``'replace'`` for error handling.
See Also:
This method is a wrapper function for :func:`urllib.parse.unquote`.
Args:
url: URL string.
encoding: The encoding with which to decode the :obj:`bytes`.
errors: The error handling scheme to use for the handling of decoding errors.
The default is ``'replace'``, which substitutes a replacement marker for
each undecodable sequence. Other possible values are ``'strict'``, under
which decoding errors raise a :exc:`UnicodeDecodeError`, and ``'ignore'``,
as well as any other name registered with :func:`codecs.register_error` that
can handle :exc:`UnicodeDecodeError`.
"""
try:
return urllib.parse.unquote(url, encoding=encoding, errors=errors)
except UnicodeError:
return url.replace('%', r'\x').encode().decode('unicode_escape', errors='replace')
[docs]
@staticmethod
def resolve_index(name: 'str | int | StdlibEnum | AenumEnum', default: 'Optional[int]' = None, *,
namespace: 'dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]',
reversed: 'bool' = False) -> 'int | StdlibEnum | AenumEnum': # pylint: disable=redefined-builtin
"""Resolve a name for an enumeration field of :meth:`make`.
A field of :meth:`make` that holds an enumeration takes a member or an
:class:`int` code, and refuses a name (:issue:`2594`). This resolves a
name against a ``namespace`` given explicitly, so that its result can
be passed on::
>>> from pcapkit.const.reg.ethertype import EtherType
>>> from pcapkit.protocols.link.ethernet.protocol import Ethernet
>>> code = Ethernet.resolve_index('Address_Resolution_Protocol',
... namespace=EtherType)
>>> Ethernet(type=code).info.type
<EtherType.Address_Resolution_Protocol: 2054>
Args:
name: a member name; a member or an :class:`int` is returned as
it is, for :meth:`make` to resolve
default: the code to return when ``name`` is not in ``namespace``
namespace: the enumeration ``name`` names a member of, or a
:obj:`dict` of codes and names
reversed: if ``namespace`` is a :obj:`dict` of names to codes,
rather than of codes to names
Returns:
The member ``name`` names, for an enumeration ``namespace``; the
code it names, for a :obj:`dict`; else ``default``.
Raises:
ProtocolNotImplemented: If ``name`` is a :obj:`str` that is not in
``namespace`` and ``default`` is :data:`None`.
.. versionadded:: 1.6.0
Takes over the ``<field>_default``, ``<field>_namespace`` and
``<field>_reversed`` arguments of :meth:`make` (:issue:`2594`).
"""
if not isinstance(name, str):
return name
try:
if isinstance(namespace, type) and issubclass(namespace, (enum.Enum, aenum.Enum)):
return namespace[name]
if isinstance(namespace, dict):
if reversed:
return cast('dict[str, int]', namespace)[name]
return {value: key for key, value in cast('dict[int, str]', namespace).items()}[name]
raise KeyError(name) # caught below, so a non-mapping misses as an unknown name does
except KeyError as error:
if default is None:
raise ProtocolNotImplemented(f'{name!r} is not a name in {namespace!r}') from error
return default
[docs]
@staticmethod
def expand_comp(value: 'str | ProtocolBase | Type[ProtocolBase]') -> 'tuple':
"""Expand protocol class to protocol name.
The method is used to expand protocol class to protocol name, in the
following manner:
1. If ``value`` is a protocol instance, the method will return the
protocol class, and the protocol names in upper case obtained from
:meth:`Protocol.id <pcapkit.protocols.protocol.Protocol.id>`.
2. If ``value`` is a protocol class, the method will return the
protocol class itself, and the protocol names in upper case
obtained from :meth:`Protocol.id <pcapkit.protocols.protocol.Protocol.id>`.
3. If ``value`` is :obj:`str`, the method will attempt to search for
the existing registered protocol class with
:meth:`ProtocolBase.get_protocol` and follow **step 2**; otherwise,
return the value in upper case alone.
Args:
value: Protocol class or name.
"""
if isinstance(value, type) and issubclass(value, ProtocolBase):
comp = (value, *(name.upper() for name in value.id()))
elif isinstance(value, ProtocolBase):
comp = (type(value), *(name.upper() for name in value.id()))
elif (proto := ProtocolBase.get_protocol(value)) is not None:
comp = (proto, *(name.upper() for name in proto.id()))
else:
comp = (value.upper(),)
return comp
[docs]
@classmethod
def analyze(cls, proto: 'int', payload: 'bytes', **kwargs: 'Any') -> 'ProtocolBase':
"""Analyse packet payload.
Args:
proto: Protocol registry number.
payload: Packet payload.
**kwargs: Arbitrary keyword arguments.
Returns:
Parsed payload as a :class:`~pcapkit.protocols.protocol.ProtocolBase`
instance.
"""
protocol = cls._lookup_next_layer(cls.__proto__, proto)
payload_io = io.BytesIO(payload)
try:
report = protocol(payload_io, len(payload), **kwargs) # type: ignore[abstract]
except Exception as exc:
if isinstance(exc, StructError) and exc.eof: # pylint: disable=no-member
from pcapkit.protocols.misc.no_payload.protocol import NoPayload as protocol # pylint: disable=import-outside-toplevel # isort: skip
else:
from pcapkit.protocols.misc.raw.protocol import Raw as protocol # pylint: disable=import-outside-toplevel # isort: skip
# error = traceback.format_exc(limit=1).strip().rsplit(os.linesep, maxsplit=1)[-1]
# log error
#logger.error(str(exc), exc_info=exc, stack_info=DEVMODE, stacklevel=stacklevel())
report = protocol(payload_io, len(payload), **kwargs) # type: ignore[abstract]
return report
[docs]
@classmethod
def register(cls, code: 'int', protocol: 'ModuleDescriptor | Type[ProtocolBase]') -> 'None':
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
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 ``code`` holds a
different entry, naming the displaced one and its replacement.
An unresolved :class:`~pcapkit.corekit.module.ModuleDescriptor`
incumbent counts as different from the class it names.
Note:
As for :func:`register_protocol
<pcapkit.foundation.registry.protocols.protocol.register_protocol>`,
handing back the entry stored under ``code`` -- the same class
object, or an equal
:class:`~pcapkit.corekit.module.ModuleDescriptor` -- is a silent
no-op that keeps the stored entry (:issue:`1364`). The comparison
resolves nothing, so a stored descriptor stays deferred.
A descriptor equal to the one ``code`` shipped with restores that
shipped object, unresolved, so undoing an override leaves the entry
as shipped (:issue:`1504`); see ``_next_layer_entry()``, which
every layer's ``register`` shares.
"""
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_protocol(cls, protocol: 'Type[ProtocolBase]') -> 'None':
"""Register a protocol class in :attr:`__registry__`, replacing what its name holds.
The key is ``protocol.__name__.upper()``. Defining a concrete
:class:`Protocol` subclass registers it already, but keeps an entry its
name holds, so this is how to replace one, or to register a class that
derives :class:`ProtocolBase` alone.
Arguments:
protocol: :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 the name holds a
different class, naming it and its replacement, and their ids
where the two read the same. A
:class:`~pcapkit.corekit.module.ModuleDescriptor` entry is
resolved to compare, so registering the class it names is silent
and leaves the descriptor in place.
Note:
A built-in protocol is stored as the descriptor it shipped as, per
:attr:`_shipped_registry`, rather than as its class, so undoing an
override leaves the entry as shipped, to be imported on lookup.
.. versionadded:: 1.6.0
"""
if not isinstance(protocol, type):
raise RegistryError(f'protocol must be a class, not {protocol!r}')
if not issubclass(protocol, ProtocolBase):
raise RegistryError(f'protocol must be a Protocol subclass, not {protocol!r}')
name = protocol.__name__.upper()
incumbent = ProtocolBase.get_protocol(name)
if incumbent is protocol:
return
if incumbent is not None:
incumbent_repr, protocol_repr = repr(incumbent), repr(protocol)
if incumbent_repr == protocol_repr:
# NOTE: #710: distinct classes can share a repr(), e.g. a factory's
# fresh class of one name per call; id() tells them apart.
incumbent_repr = f'{incumbent_repr} (id={id(incumbent):#x})'
protocol_repr = f'{protocol_repr} (id={id(protocol):#x})'
warn(f'protocol {name} already registered, overwriting {incumbent_repr} '
f'with {protocol_repr}', RegistryWarning)
shipped = ProtocolBase._shipped_registry.get(name)
if isinstance(shipped, ModuleDescriptor) and \
(shipped.module, shipped.name) == (protocol.__module__, protocol.__name__):
ProtocolBase.__registry__[name] = shipped
else:
ProtocolBase.__registry__[name] = protocol
@overload # pragma: no cover
@classmethod
def get_protocol(cls, name: 'str') -> 'Optional[Type[ProtocolBase]]': ...
@overload # pragma: no cover
@classmethod
def get_protocol(cls, name: 'str', default: '_VT') -> 'Type[ProtocolBase] | _VT': ...
[docs]
@classmethod
def get_protocol(cls, name: 'str', default: 'Any' = None) -> 'Any':
"""Look a protocol class up in :attr:`__registry__` by name.
Arguments:
name: protocol name, in any case: ``'ipv4'`` finds
:class:`~pcapkit.protocols.internet.ipv4.protocol.IPv4`
default: what to return where no entry has the name
Returns:
The class registered under ``name.upper()``, imported where its entry
is a :class:`~pcapkit.corekit.module.ModuleDescriptor`, or
``default``. The entry is left as it is, so a descriptor is resolved
again on each lookup and follows a reloaded module.
Raises:
pcapkit.utilities.exceptions.ProtocolError: If a descriptor names a
class its module does not define.
.. versionadded:: 1.6.0
"""
entry = ProtocolBase.__registry__.get(name.upper())
if entry is None:
return default
if isinstance(entry, ModuleDescriptor):
return entry.klass
return entry
[docs]
@classmethod
def from_schema(cls, schema: '_ST | dict[str, Any]') -> 'Self':
"""Create protocol instance from schema.
Args:
schema: Protocol schema.
Returns:
Protocol instance.
"""
if not isinstance(schema, Schema):
schema = cast('_ST', cls.__schema__.from_dict(schema))
self = cls.__new__(cls)
self.__header__ = schema
# initialize protocol instance
self.__init__(bytes(schema), len(schema)) # type: ignore[misc]
return self
[docs]
@classmethod
def from_data(cls, data: '_PT | dict[str, Any]', **kwargs: 'Any') -> 'Self':
"""Create protocol instance from data.
``data`` may be the :obj:`dict` that :meth:`Info.to_dict
<pcapkit.corekit.infoclass.Info.to_dict>` returns, which carries the
metadata the rebuild reads: ``__next_name__`` and ``__next_type__``,
which the payload is dispatched on, ``__short_read__`` and
``__short_width__``, and, for
:class:`~pcapkit.protocols.internet.ipv6.protocol.IPv6`, the extension
header chain under ``__exthdr__``. A hand-written :obj:`dict` without
them rebuilds no payload (nor IPv6 extension header) and cuts no short
read back; nothing is guessed from its other keys (see
:meth:`_make_payload`, :issue:`2003`).
A header the capture cut short, as ``__short_read__`` records, is cut
back and read at the width ``__short_width__`` records, so the rebuild
parses the octets captured as the capture did (:issue:`2703`): it holds
the values its parse read, not those :meth:`make` fills in past the cut
(:issue:`2761`), and the ``Raw`` next layer that parse left
(:issue:`2717`). Inside a payload :meth:`_make_payload` rebuilds, which
gives only its octets, the header made is read instead.
Args:
data: Protocol data.
**kwargs: Construction keywords that ``data`` does not carry, such as
the ``num`` and ``header`` of
:class:`~pcapkit.protocols.misc.pcap.frame.protocol.Frame`, or the
``num``, ``sct`` and ``ctx`` of
:class:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG`. They take
precedence over the keywords :meth:`_make_data` returns.
Returns:
Protocol instance.
Raises:
UnsupportedCall: If any of ``kwargs`` matches no parameter of the
protocol, as for a direct construction.
"""
if not isinstance(data, Data):
data = cast('_PT', cls.__data__.from_dict(data))
# NOTE: The caller's own keywords are checked strictly, as for a direct
# construction, less the out-of-band ones ``__init__`` consumes.
_check_construction_keywords(cls, {
key: val for key, val in kwargs.items()
if key not in ('_layer', '_protocol', '__context__')
})
self = cls.__new__(cls)
kwargs = {**self._make_data(data), **kwargs}
# NOTE: The rest came from ``_make_data``, so ``__init__`` warns, not raises.
# Both flags last the call alone; ``pack`` cuts back by ``__short_read__`` (#2703).
self.__reconstructing__ = True
self.__short_read__ = cast('Optional[tuple[str, int]]', data.get('__short_read__'))
# NOTE: A cut header is read at the width its parse read it at, as recorded, so
# the rebuild holds the values and the next layer its parse did (#2717, #2761).
width = data.get('__short_width__')
if self.__short_read__ is not None and width is not None:
kwargs.setdefault('length', width)
try:
# initialize protocol instance
self.__init__(**kwargs) # type: ignore[misc]
finally:
del self.__reconstructing__
del self.__short_read__
return self
##########################################################################
# Data models.
##########################################################################
def __new__(cls, *args: 'Any', **kwargs: 'Any') -> 'Self': # pylint: disable=unused-argument
self = super().__new__(cls)
# NOTE: Assign this attribute after ``__new__`` to avoid shared memory
# reference between instances.
self.__cached__ = {}
self.__header__ = None # type: ignore[assignment]
#: bool: If a header set before :meth:`unpack` may hold raw items;
#: :meth:`pack` narrows it down for a made one.
self._make_raw = True
return self
@overload # pragma: no cover
def __init__(self, file: 'IO[bytes] | bytes', length: 'Optional[int]' = ..., **kwargs: 'Any') -> 'None': ...
@overload # pragma: no cover
def __init__(self, **kwargs: 'Any') -> 'None': ...
[docs]
def __init__(self, file: 'Optional[IO[bytes] | bytes]' = None, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'None':
"""Initialisation.
Args:
file: Source packet stream.
length: Length of packet data.
_layer (str): Parse packet until ``_layer``
(:attr:`self._exlayer <pcapkit.protocols.protocol.Protocol._exlayer>`).
While parsing, the un-prefixed ``layer`` is accepted as well --
see the note below.
_protocol: Parse packet until ``_protocol``
(:attr:`self._exproto <pcapkit.protocols.protocol.Protocol._exproto>`),
given as a protocol name (:obj:`str`), a protocol class or a
protocol instance.
While parsing, the un-prefixed ``protocol`` is accepted as well --
see the note below.
packet: Packet context of the enclosing layer, a :obj:`dict` as
handed over by
:meth:`self._import_next_layer
<pcapkit.protocols.protocol.Protocol._import_next_layer>`.
While parsing, it is republished as ``__packet__`` so that
:meth:`self.unpack <Protocol.unpack>` -- and through it the
schema -- can see it; see the note below.
__context__: Caller supplied parsing context (:attr:`self._exctx <pcapkit.protocols.protocol.Protocol._exctx>`;
see :mod:`pcapkit.corekit.context`): a
:class:`~pcapkit.corekit.context.ContextRegistry`, a
:class:`~pcapkit.corekit.context.ProtocolContext`, a mapping of
names to :class:`~pcapkit.corekit.context.ProtocolContext`, or an
iterable of :class:`~pcapkit.corekit.context.ProtocolContext`.
It is consumed here rather
than being forwarded to :meth:`self.read <Protocol.read>`, and is
propagated to nested layers by
:meth:`self._import_next_layer
<pcapkit.protocols.protocol.Protocol._import_next_layer>`.
**kwargs: Arbitrary keyword arguments.
Raises:
UnsupportedCall: When constructing (``file`` is :obj:`None`), if a
keyword names no parameter of this protocol's :meth:`make`,
:meth:`read`, :meth:`pack`, :meth:`unpack`,
:meth:`__post_init__` or :meth:`__init__`, anywhere in the MRO
-- except a ``make`` whose class sets
:attr:`__own_make_keywords__`, which counts only while it is
the ``make`` of the protocol being constructed -- and is not
listed in :attr:`__keywords__`. Parsing (``file`` is given) is
unaffected.
Note:
The ``_layer``, ``_protocol`` and ``__context__`` keywords are
*out-of-band*: they configure the parse rather than describe the
packet, and are consumed here, the one point every producer passes
through, so a spelling normalised here is normalised for the
engines, every :meth:`_import_next_layer
<pcapkit.protocols.protocol.Protocol._import_next_layer>` and any
third-party protocol alike.
``packet`` is read but not consumed: it stays in the keywords
forwarded to :meth:`__post_init__`, and while parsing a copy is
republished beside it as ``__packet__``.
"""
#logger.debug('%s(file, %s, **%s)', type(self).__name__, length, kwargs)
# Parsing rather than constructing, as in ``__post_init__``. ``layer``,
# ``protocol`` and ``packet`` are out-of-band only then; else ``make()`` fields.
parsing = file is not None
#: int: File pointer.
self._seekset = io.SEEK_SET # type: int
#: Parse packet until such layer.
self._exlayer = kwargs.pop('_layer', None) # type: Optional[str]
#: Parse packet until such protocol.
self._exproto = kwargs.pop('_protocol', None) # type: Optional[str | ProtocolBase | Type[ProtocolBase]]
# NOTE: The engines, every ``_import_next_layer`` and ``pcapkit.extract`` spell the
# limits ``layer``/``protocol``, so both are accepted; the prefixed one wins (GH-356).
if parsing:
layer = kwargs.pop('layer', None)
protocol = kwargs.pop('protocol', None)
if self._exlayer is None:
self._exlayer = layer
if self._exproto is None:
self._exproto = protocol
# NOTE: ``Extractor.__init__`` passes ``'none'``/``'null'`` for an omitted
# ``layer``/``protocol`` (``extract`` too, for ``layer``), meaning no limit.
if isinstance(self._exlayer, str) and self._exlayer.lower() == 'none':
self._exlayer = None
if isinstance(self._exproto, str) and self._exproto.lower() == 'null':
self._exproto = None
#: pcapkit.corekit.context.ContextRegistry: Caller supplied parsing context.
# NOTE: A registry is adopted as-is rather than copied by ``make()``: the
# contexts are shared across layers regardless.
__context__ = kwargs.pop('__context__', None)
self._exctx = (__context__ if isinstance(__context__, ContextRegistry)
else ContextRegistry.make(__context__)) # type: ContextRegistry
#: bool: Whether to stop parsing the next layer of protocol.
self._sigterm = self._check_term_threshold()
# NOTE: ``packet=`` from ``_import_next_layer`` reaches the schema as ``__packet__`` (GH-382),
# copied since ``Schema.unpack`` writes into it and IPv6 extension headers share one dict.
if parsing and '__packet__' not in kwargs and isinstance(kwargs.get('packet'), dict):
kwargs['__packet__'] = dict(kwargs['packet'])
# NOTE: Construction only: every ``make`` ends in ``**kwargs``, so a misspelt field would
# keep its default silently (#617). Parse keywords go unchecked: dropping one changes no octet.
if not parsing:
_check_construction_keywords(
type(self), kwargs, strict=not self.__reconstructing__)
# post-init customisations
self.__post_init__(file, length, **kwargs) # type: ignore[arg-type]
# inject packet payload into the data model
self._info.__update__(packet=self.packet.payload)
@overload # pragma: no cover
def __post_init__(self, file: 'IO[bytes] | bytes', length: 'Optional[int]' = ..., **kwargs: 'Any') -> 'None': ...
@overload # pragma: no cover
def __post_init__(self, **kwargs: 'Any') -> 'None': ...
[docs]
def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None,
length: 'Optional[int]' = None, **kwargs: 'Any') -> 'None':
"""Post initialisation hook.
Args:
file: Source packet stream.
length: Length of packet data.
**kwargs: Arbitrary keyword arguments.
See Also:
For construction arguments, please refer to
:meth:`self.make <pcapkit.protocols.protocol.Protocol.make>`.
Note:
Inside a payload :meth:`_make_payload` rebuilds, the parse's
warnings are dropped, since the enclosing layer parses the same
octets again and reports them (:issue:`2711`). A parse that raises
reports its warnings, unless
:func:`~pcapkit.corekit.decorators.beholder` replaces the layer.
"""
if file is None:
_data = self.pack(**kwargs)
# NOTE: A short read's header is rebuilt from what was captured, at ``from_data``'s width,
# not from ``make``'s fill past the cut (#2703, #2761); a rebuilt payload keeps its octets.
if self.__short_read__ is not None and not rebuilding.get():
self.__header__ = None # type: ignore[assignment]
else:
_data = file if isinstance(file, bytes) else file.read(length) # type: ignore[arg-type]
#: bytes: Raw packet data.
self._data = _data
#: io.BytesIO: Source packet stream.
self._file = io.BytesIO(self._data)
if not rebuilding.get():
#: pcapkit.protocols.data.Data: Parsed packet data.
self._info = self.unpack(length, **kwargs)
return
# NOTE: The enclosing layer parses these octets again and reports what
# they carry; as in ``_parse_next_layer``, a raise keeps them (#2711).
with hold_warnings() as held:
try:
self._info = self.unpack(length, **kwargs)
except Exception:
held.discard = beholding.get()
raise
held.discard = True
[docs]
def __init_subclass__(cls, /, schema: 'Optional[Type[_ST]]' = None,
data: 'Optional[Type[_PT]]' = None,
code: 'Any' = None,
*args: 'Any', **kwargs: 'Any') -> 'None':
"""Initialisation for subclasses.
Args:
schema: Schema class.
data: Data class.
code: Next-layer dispatch registration key(s). :data:`None` (the
default) skips registration entirely -- see below.
*args: Arbitrary positional arguments.
**kwargs: Arbitrary keyword arguments.
Raises:
UnsupportedCall: If any unrecognised class keyword is given.
Sets :attr:`self.__schema__ <pcapkit.protocols.protocol.Protocol.__schema__>`
and :attr:`self.__data__ <pcapkit.protocols.protocol.Protocol.__data__>`
on the subclass and, if ``code`` is given, registers it into the
next-layer dispatch :term:`registry` (or registries) ``code`` names,
e.g. :attr:`Link.__proto__ <pcapkit.protocols.link.link.Link.__proto__>`.
Notes:
An omitted ``schema`` or ``data`` is the class of the subclass's
own name in the ``schema`` or ``data`` module beside the
``protocol`` module defining it, as
:class:`~pcapkit.protocols.link.l2tp.protocol.L2TP` finds
:class:`~pcapkit.protocols.link.l2tp.schema.L2TP`
(:issue:`2534`), else
:class:`~pcapkit.protocols.misc.raw.schema.Raw` or
:class:`~pcapkit.protocols.misc.raw.data.Raw`.
Dispatch registration is **opt-in**, like the ``engine=``/``protocol=``/
``fmt=`` keywords of :class:`~pcapkit.foundation.engines.engine.Engine`,
:class:`~pcapkit.foundation.reassembly.reassembly.Reassembly`,
:class:`~pcapkit.foundation.traceflow.traceflow.TraceFlow` and
:class:`~pcapkit.dumpkit.common.Dumper`. Every built-in protocol omits
``code``: the built-in dispatch tables (e.g. :attr:`Link.__proto__
<pcapkit.protocols.link.link.Link.__proto__>`) are literal assignments
in each layer module. A class defined without ``code`` can be
registered later through the owning class's :meth:`register` or the
matching ``register_*`` helper in
:mod:`pcapkit.foundation.registry.protocols`.
``code`` accepts:
* a single enum member, whose *type* determines the destination
registry (or registries) -- e.g. any
:class:`~pcapkit.const.reg.ethertype.EtherType` member always means
:class:`~pcapkit.protocols.link.link.Link`, and any
:class:`~pcapkit.const.reg.linktype.LinkType` member means *both*
:class:`~pcapkit.protocols.misc.pcap.frame.protocol.Frame` **and**
:class:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG`;
* a :class:`dict` mapping a destination class to a key, for a key
that cannot name its own destination -- a raw :class:`int` port,
for instance, is ambiguous between
:class:`~pcapkit.protocols.transport.tcp.protocol.TCP` and
:class:`~pcapkit.protocols.transport.udp.protocol.UDP`, and *must* use this
form;
* an iterable mixing either of the above, to register the same class
into several registries from a single declaration -- e.g. a
:class:`~pcapkit.protocols.link.l2tp.protocol.L2TP` subclass reachable both
by its IP protocol number and by a UDP port.
The mapping form is accepted even where the key's type could be
inferred. An enum member whose type names no known destination raises
:exc:`~pcapkit.utilities.exceptions.RegistryError` rather than guessing.
See Also:
:func:`pcapkit.foundation.registry.protocols.protocol.register_protocol_code`
implements the resolution described above.
"""
if args or kwargs:
unexpected = ', '.join([*map(repr, args), *sorted(kwargs)])
raise UnsupportedCall(f'{cls.__name__}: unexpected class keyword(s): {unexpected}')
super().__init_subclass__()
if schema is None:
schema = cast('Type[_ST]', _sibling_model(cls, 'schema'))
if data is None:
data = cast('Type[_PT]', _sibling_model(cls, 'data'))
cls.__schema__ = schema
cls.__data__ = data
# NOTE: Copy a class's own ``__proto__`` before anything registers into it, so
# ``_next_layer_entry`` restores a shipped descriptor unresolved (#1504).
if '__proto__' in cls.__dict__:
table = cls.__dict__['__proto__']
cls._shipped_proto = types.MappingProxyType(
dict(table) if isinstance(table, collections.abc.Mapping) else {})
if code is not None:
from pcapkit.foundation.registry.protocols import \
register_protocol_code # pylint: disable=import-outside-toplevel
register_protocol_code(cls, code)
[docs]
def __repr__(self) -> 'str':
"""Returns representation of parsed protocol data.
Example:
>>> protocol
<Frame frame=<Frame ..., ethernet=Ethernet(...), protocols='Ethernet:IPv6:Raw', packet=b'...'>>
"""
if (cached := self.__cached__.get('__repr__')) is not None:
return cached
# cache and return
repr_ = f'<{self.alias} {self.info_name}={self._info!r}>'
self.__cached__['__repr__'] = repr_
return repr_
[docs]
def __str__(self) -> 'str':
"""Returns formatted hex representation of source data stream.
Example:
>>> protocol
<Frame frame=<Frame ..., ethernet=Ethernet(...), protocols='Ethernet:IPv6:Raw', packet=b'...'>>
>>> print(protocol)
00 00 00 00 00 00 00 a6 87 f9 27 93 16 ee fe 80 00 00 00 ..........'........
00 00 00 1c cd 7c 77 ba c7 46 b7 87 00 0e aa 00 00 00 00 .....|w..F.........
fe 80 00 00 00 00 00 00 1c cd 7c 77 ba c7 46 b7 01 01 a4 ..........|w..F....
5e 60 d9 6b 97 ^`.k.
"""
if (cached := self.__cached__.get('__str__')) is not None:
return cached
hexbuf = ' '.join(textwrap.wrap(self._data.hex(), 2))
strbuf = ''.join(chr(char) if char in readable else '.' for char in self._data)
number = shutil.get_terminal_size().columns // 4 - 1
length = number * 3
hexlst = textwrap.wrap(hexbuf, length)
strlst = list(iter(functools.partial(io.StringIO(strbuf).read, number), ''))
# cache and return
str_ = os.linesep.join(map(lambda x: f'{x[0].ljust(length)} {x[1]}', zip(hexlst, strlst)))
self.__cached__['__str__'] = str_
return str_
def __bytes__(self) -> 'bytes':
"""Returns source data stream in :obj:`bytes`."""
return self._data
def __len__(self) -> 'int':
"""Total length of corresponding protocol."""
if (cached := self.__cached__.get('__len__')) is not None:
return cached
# cache and return
len_ = len(self._data)
self.__cached__['__len__'] = len_
return len_
[docs]
def __length_hint__(self) -> 'Optional[int]':
"""Return an estimated length for the object."""
def __iter__(self) -> 'IO[bytes]':
"""Iterate through :attr:`self._data <pcapkit.protocols.protocol.Protocol._data>`."""
return io.BytesIO(self._data)
[docs]
def __getitem__(self, key: 'str | Protocol | Type[Protocol]') -> 'ProtocolBase':
"""Subscription (``getitem``) support.
* If ``key`` is a :class:`~pcapkit.protocols.protocol.Protocol` object,
the method will fetch its indexes (:meth:`self.id <pcapkit.protocols.protocol.Protocol.id>`).
* Later, search the packet's chain of protocols with the calculated ``key``.
* If no matches, then raises :exc:`~pcapkit.utilities.exceptions.ProtocolNotFound`.
Args:
key: Indexing key.
Returns:
The sub-packet from the current packet of indexed protocol.
Raises:
ProtocolNotFound: If ``key`` is not in the current packet.
See Also:
The method calls
:meth:`self.expand_comp <pcapkit.protocols.protocol.Protocol.expand_comp>`
to handle the ``key`` and expand it for robust searching.
"""
comp = self.expand_comp(key)
# if it's itself
test_comp = (type(self), *(name.upper() for name in self.id()))
for test in comp:
if test in test_comp:
return self
# then check recursively
from pcapkit.protocols.misc.no_payload.protocol import \
NoPayload # pylint: disable=import-outside-toplevel
payload = self._next
while not isinstance(payload, NoPayload):
test_comp = (type(payload), *(name.upper() for name in payload.id()))
for test in comp:
if test in test_comp:
return payload
payload = payload.payload
raise ProtocolNotFound(key)
[docs]
def __contains__(self, name: 'str | Protocol | Type[Protocol]') -> 'bool':
"""Returns if certain protocol is in the instance.
Args:
name: Name to search
See Also:
The method calls
:meth:`self.expand_comp <pcapkit.protocols.protocol.Protocol.expand_comp>`
to handle the ``name`` and expand it for robust searching.
"""
comp = self.expand_comp(name)
# if it's itself
test_comp = (type(self), *(name.upper() for name in self.id()))
for test in comp:
if test in test_comp:
return True
# then check recursively
from pcapkit.protocols.misc.no_payload.protocol import \
NoPayload # pylint: disable=import-outside-toplevel
payload = self._next
while not isinstance(payload, NoPayload):
test_comp = (type(payload), *(name.upper() for name in payload.id()))
for test in comp:
if test in test_comp:
return True
payload = payload.payload
return False
[docs]
@classmethod
@abc.abstractmethod
def __index__(cls) -> 'StdlibEnum | AenumEnum':
"""Numeral registry index of the protocol."""
@classmethod
def __eq__(cls, other: 'object') -> 'bool':
"""Returns if ``other`` is of the same protocol as the current object.
Args:
other: Comparison against the object.
"""
if isinstance(other, type) and issubclass(other, ProtocolBase):
return cls is other
if isinstance(other, ProtocolBase):
return cls.id() == other.id()
if isinstance(other, str):
test_comp = cls.expand_comp(cls)
return other.upper() in test_comp
return False
def __hash__(self) -> 'int':
"""Return the hash value for :attr:`self._data <pcapkit.protocols.protocol.Protocol._data>`."""
return hash(self._data)
##########################################################################
# Utilities.
##########################################################################
[docs]
@classmethod
def _make_data(cls, data: 'Data') -> 'dict[str, Any]':
"""Create key-value pairs from ``data`` for protocol construction.
Args:
data: protocol data
Returns:
Key-value pairs for protocol construction.
"""
return data.to_dict()
[docs]
def _packing_target(self) -> 'Self':
"""Return the instance that :meth:`pack` makes its header on.
Returns:
The instance itself while it is being constructed, i.e. before
:meth:`__post_init__` has set its :attr:`info`, since its
:meth:`unpack` reads the made header back; a shallow copy of it
otherwise, so that neither that header nor an attribute that
:meth:`make` sets is left on the instance (:issue:`1769`).
Note:
A :meth:`pack` override that does not call this one has to build
on this target too, as :meth:`Frame.pack
<pcapkit.protocols.misc.pcap.frame.protocol.Frame.pack>` and
:meth:`PCAPNG.pack <pcapkit.protocols.misc.pcapng.protocol.PCAPNG.pack>` do.
The copy is shallow, so a :meth:`make` that mutates a container
in place still reaches the instance's.
"""
# NOTE: ``_info`` is the last thing ``__post_init__`` sets, from
# ``unpack``; until then the instance holds no packet to keep.
if '_info' not in vars(self):
return self
return copy.copy(self)
@classmethod
def _next_layer_entry(cls, code: 'Any', protocol: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]'
) -> 'Optional[ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]]':
"""Work out what a next-layer ``register`` stores under ``code``.
:meth:`register` and every override of it, such as
:meth:`Link.register <pcapkit.protocols.link.link.Link.register>`,
share this; each still warns and stores for itself.
Arguments:
code: registry key
protocol: module descriptor or a :class:`ProtocolBase` subclass
Returns:
:obj:`None` if ``protocol`` is the entry stored under ``code``
already -- the same object, or a descriptor equal to a stored one
(:issue:`1364`). Otherwise the entry to store: if ``protocol`` is a
descriptor equal to the one ``code`` shipped with, per
:attr:`_shipped_proto`, the shipped object itself, unresolved;
else ``protocol`` resolved to its class.
Raises:
pcapkit.utilities.exceptions.RegistryError: If ``protocol`` is not a
class, or not a :class:`ProtocolBase` subclass.
Note:
The shipped descriptor is matched before it is resolved, so an
override undone by handing back the shipped entry leaves the
registry as it shipped: the entry is still a descriptor, nothing is
imported, and the class is looked up when a frame dispatches to it
(:issue:`1504`).
"""
incumbent = cls.__proto__.get(code)
if incumbent is not None and (incumbent is protocol or (
isinstance(incumbent, ModuleDescriptor) and isinstance(protocol, ModuleDescriptor)
and incumbent == protocol)):
return None
shipped = cls._shipped_proto.get(code)
if (isinstance(protocol, ModuleDescriptor) and isinstance(shipped, ModuleDescriptor)
and protocol == shipped):
return shipped
if isinstance(protocol, ModuleDescriptor):
protocol = protocol.klass
if not isinstance(protocol, type):
raise RegistryError(f'protocol must be a class, not {protocol!r}')
if not issubclass(protocol, ProtocolBase):
raise RegistryError(f'protocol must be a Protocol subclass, not {protocol!r}')
return protocol
[docs]
def _get_context(self, cls: 'Optional[Type[_CTX]]' = None) -> 'Optional[_CTX]':
"""Get the caller supplied context for this protocol, if any.
The lookup is keyed on :meth:`self.id
<pcapkit.protocols.protocol.Protocol.id>`, so a
protocol finds its own context without knowing how the caller spelled
the registry.
Args:
cls: Expected context class; when given, a context registered
under this protocol's name but of another type is ignored
rather than returned for the implementation to trip over.
Returns:
The matching context, or :data:`None` when the caller supplied
none.
See Also:
:mod:`pcapkit.corekit.context`
"""
registry = self._exctx
if registry is None:
return None
return registry.match(self.id(), cls)
[docs]
def _get_payload(self) -> 'bytes':
"""Get payload from :attr:`self.__header__ <Protocol.__header__>`.
Returns:
Payload of :attr:`self.__header__ <Protocol.__header__>` as :obj:`bytes`.
See Also:
This is a wrapper function for :meth:`pcapkit.protocols.schema.Schema.get_payload`.
"""
return self.__header__.get_payload()
def _read_protos(self, size: int) -> 'Optional[StdlibEnum | AenumEnum]': # pylint: disable=unused-argument
"""Read next layer protocol type.
* If *succeed*, returns the enum of next layer protocol.
* If *fail*, returns :obj:`None`.
Arguments:
size: buffer size
"""
def _read_fileng(self, *args: 'Any', **kwargs: 'Any') -> 'bytes':
"""Read file buffer (:attr:`self._file <pcapkit.protocols.protocol.Protocol._file>`).
This method wraps the :meth:`file.read <io.BytesIO.read>` call.
Args:
*args: arbitrary positional arguments
**kwargs: arbitrary keyword arguments
Returns:
bytes: Data read from file buffer.
"""
return self._file.read(*args, **kwargs)
[docs]
def _read_unpack(self, size: 'int' = 1, *, signed: 'bool' = False,
lilendian: 'bool' = False, quiet: 'bool' = False) -> 'int':
"""Read bytes and unpack for integers.
Arguments:
size: buffer size
signed: signed flag
lilendian: little-endian flag
quiet: quiet (no exception) flag
Returns:
Unpacked data upon success
Raises:
StructError: If the buffer is empty, whatever ``quiet`` says, or if
fewer than ``size`` octets were read, without ``quiet``.
"""
endian = '<' if lilendian else '>'
if size == 8: # unpack to 8-byte integer (long long)
kind = 'q' if signed else 'Q'
elif size == 4: # unpack to 4-byte integer (int / long)
kind = 'i' if signed else 'I'
elif size == 2: # unpack to 2-byte integer (short)
kind = 'h' if signed else 'H'
elif size == 1: # unpack to 1-byte integer (char)
kind = 'b' if signed else 'B'
else: # no struct format, see int.from_bytes below
kind = None
mem = self._file.read(size)
if not mem:
raise StructError('unpack: empty buffer', quiet=True, eof=True)
if kind is None:
# A short read fails here as :func:`struct.unpack` makes it fail
# for the sizes that have a format code (#2279).
if len(mem) < size and not quiet:
raise StructError(f'{self.__class__.__name__}: unpack failed')
end = 'little' if lilendian else 'big' # type: Literal['little', 'big']
buf = int.from_bytes(mem, end, signed=signed)
else:
fmt = f'{endian}{kind}'
try:
buf = struct.unpack(fmt, mem)[0] # pylint: disable=no-member
except struct.error as error: # pylint: disable=no-member
if quiet:
end = 'little' if lilendian else 'big'
buf = int.from_bytes(mem, end, signed=signed)
return buf
raise StructError(f'{self.__class__.__name__}: unpack failed') from error
return buf
def _read_binary(self, size: 'int' = 1) -> 'str':
"""Read bytes and convert into binaries.
Arguments:
size: buffer size
Returns:
Binary bits (``0``/``1``).
Raises:
StructError: If the buffer is empty, with ``eof`` set as by
:meth:`_read_unpack`, or if fewer than ``size`` octets were read.
"""
# ``max`` makes a non-positive size read nothing rather than the
# rest of the stream.
mem = self._file.read(max(size, 0))
if len(mem) < size:
if not mem:
raise StructError('unpack: empty buffer', quiet=True, eof=True)
raise StructError(f'{self.__class__.__name__}: unpack failed')
return ''.join(bin(byte)[2:].zfill(8) for byte in mem)
[docs]
def _read_octets(self, data: 'bytes', pos: 'int' = 0, *, size: 'int', what: 'str') -> 'bytes':
"""Read exactly ``size`` octets out of ``data`` at ``pos``.
A plain slice comes back short past the end of ``data``, which would
let a truncated field parse rather than fail.
Args:
data: buffer being read
pos: offset of the octets in ``data``
size: octets to read
what: name of the field, for the error message
Returns:
The ``size`` octets at ``pos``.
Raises:
ProtocolError: If fewer than ``size`` octets remain from ``pos``.
The message names :attr:`alias`.
.. versionadded:: 1.6.0
"""
octets = data[pos:pos + size]
if len(octets) != size:
raise ProtocolError(f'{self.alias}: invalid format: {what} runs past the end of the buffer')
return octets
[docs]
def _read_uint(self, data: 'bytes', pos: 'int' = 0, *, size: 'int', what: 'str',
lilendian: 'bool' = False) -> 'int':
"""Read a ``size``-octet unsigned integer out of ``data`` at ``pos``.
:meth:`_read_unpack` reads the file; this reads a buffer already in
hand. The inverse is :meth:`_make_pack`, not :meth:`_make_uint`, which
checks a value and packs nothing.
Args:
data: buffer being read
pos: offset of the integer in ``data``
size: octets in the integer
what: name of the field, for the error message
lilendian: little-endian flag
Returns:
The integer.
Raises:
ProtocolError: If fewer than ``size`` octets remain from ``pos``,
as :meth:`_read_octets` raises it.
.. versionadded:: 1.6.0
"""
octets = self._read_octets(data, pos, size=size, what=what)
return int.from_bytes(octets, 'little' if lilendian else 'big')
[docs]
@staticmethod
def _read_sint(value: 'int', bits: 'int') -> 'int':
"""Read ``value``, an unsigned ``bits``-bit field, as a two's-complement one.
It takes the unsigned integer a field has already read, such as a
:class:`~pcapkit.corekit.fields.strings.BitField` sub-field, rather
than a buffer.
Args:
value: the unsigned value read from the field
bits: width of the field, in bits
Returns:
The signed value the same bits denote.
.. versionadded:: 1.6.0
"""
return value - (1 << bits) if value & (1 << (bits - 1)) else value
[docs]
def _read_vector(self, data: 'bytes', pos: 'int' = 0, *, size: 'int' = 1,
what: 'str') -> 'tuple[bytes, int]':
"""Read a length-prefixed vector out of ``data``.
The vector is a ``size``-octet unsigned big-endian length, then that
many octets, as the variable-length vectors of TLS's presentation
language are (:rfc:`9846#section-3.4`).
Args:
data: buffer being read
pos: offset of the length prefix in ``data``
size: octets in the length prefix
what: name of the vector, for the error message
Returns:
The vector's octets, and the offset just past them.
Raises:
ProtocolError: If the length prefix, or the octets it counts, run
past the end of ``data``. The message names :attr:`alias`.
.. versionadded:: 1.6.0
"""
start = pos + size
end = start + self._read_uint(data, pos, size=size, what=f'{what} length')
return self._read_octets(data, start, size=end - start, what=what), end
[docs]
def _read_whole_vector(self, data: 'bytes', *, size: 'int' = 1, what: 'str') -> 'bytes':
"""Read the one length-prefixed vector that ``data`` holds.
The vector is read as :meth:`_read_vector` reads it from offset zero,
and has to end where ``data`` does.
Args:
data: buffer holding the vector and nothing else
size: octets in the length prefix
what: name of the vector, for the error message
Returns:
The vector's octets.
Raises:
ProtocolError: If the vector runs past the end of ``data``, or
octets follow it. The message names :attr:`alias`.
.. versionadded:: 1.6.0
"""
octets, end = self._read_vector(data, 0, size=size, what=what)
if end != len(data):
raise ProtocolError(f'{self.alias}: invalid format: {what} carries excess octets')
return octets
[docs]
def _read_uints(self, data: 'bytes', *, size: 'int' = 1, what: 'str') -> 'tuple[int, ...]':
"""Read ``data`` as a run of ``size``-octet unsigned big-endian integers.
Args:
data: buffer holding the integers and nothing else
size: octets in each integer
what: name of the list, for the error message
Returns:
The integers, in order.
Raises:
ProtocolError: If the length of ``data`` is not a multiple of
``size``, whose trailing octets would otherwise be padded out
when the list is made again. The message names :attr:`alias`.
.. versionadded:: 1.6.0
"""
if len(data) % size:
raise ProtocolError(f'{self.alias}: invalid format: {what} length {len(data)} '
f'is not a multiple of {size}')
return tuple(int.from_bytes(data[index:index + size], 'big') for index in range(0, len(data), size))
[docs]
def _read_mac_addr(self, addr: 'bytes') -> 'str':
"""Read a MAC address, an EUI-64 one included.
Args:
addr: the address octets
Returns:
Colon (``:``) separated *hex* encoded address.
.. versionadded:: 1.6.0
"""
return addr.hex(':')
def _read_text(self, data: 'bytes') -> 'tuple[str, Optional[str]]':
"""Decode text, recording how to encode it back.
The text is decoded by :meth:`decode`. Where the result does not
encode back to ``data`` as UTF-8 -- which ``make()`` uses by default --
the detected charset is returned with it, or ``'latin-1'`` (decoding
again) if that does not reproduce ``data`` either, so that the original
octets survive a rebuild.
Args:
data: raw text
Returns:
The decoded text and the charset to encode it with, the latter
:obj:`None` for UTF-8.
.. versionadded:: 1.6.0
"""
value = self.decode(data)
for charset in (None, detect(data)):
try:
if value.encode(charset or 'utf-8') == data:
return value, charset
except (LookupError, UnicodeError):
pass
return data.decode('latin-1'), 'latin-1'
@overload # pragma: no cover
def _read_packet(self, length: 'Optional[int]' = ..., *, header: 'None' = ...) -> 'bytes': ...
@overload # pragma: no cover
def _read_packet(self, *, header: 'int', payload: 'Optional[int]' = ..., discard: 'Literal[True]') -> 'bytes': ...
@overload # pragma: no cover
def _read_packet(self, *, header: 'int', payload: 'Optional[int]' = ..., discard: 'Literal[False]' = ...) -> 'Data_Packet': ... # pylint: disable=line-too-long
[docs]
@seekset # type: ignore[misc]
def _read_packet(self, length: 'Optional[int]' = None, *, header: 'Optional[int]' = None,
payload: 'Optional[int]' = None, discard: bool = False) -> 'bytes | Data_Packet':
"""Read raw packet data.
Arguments:
length: length of the packet
header: length of the packet header
payload: length of the packet payload
discard: flag if discard header data
* If ``header`` is omitted, returns the whole packet data in :obj:`bytes`.
* If ``discard`` is set as :data:`True`, returns the packet body (in
:obj:`bytes`) only.
* Otherwise, returns the header and payload data as
:class:`~pcapkit.protocols.data.Packet` object.
"""
if header is not None:
data_header = self._read_fileng(header)
data_payload = self._read_fileng(payload)
if discard:
return data_payload
return Data_Packet(
header=data_header,
payload=data_payload
)
return self._read_fileng(length)
[docs]
@classmethod
def _make_pack(cls, integer: 'int', *, size: 'int' = 1,
signed: 'bool' = False, lilendian: 'bool' = False) -> 'bytes':
"""Pack integers to bytes.
Arguments:
integer: integer to be packed
size: buffer size
signed: signed flag
lilendian: little-endian flag
Returns:
Packed data upon success.
Raises:
StructError: If failed to pack the integer.
"""
endian = '<' if lilendian else '>'
if size == 8: # pack from 8-byte integer (long long)
kind = 'q' if signed else 'Q'
elif size == 4: # pack from 4-byte integer (int / long)
kind = 'i' if signed else 'I'
elif size == 2: # pack from 2-byte integer (short)
kind = 'h' if signed else 'H'
elif size == 1: # pack from 1-byte integer (char)
kind = 'b' if signed else 'B'
else: # no struct format, see int.to_bytes below
kind = None
if kind is None:
end = 'little' if lilendian else 'big' # type: Literal['little', 'big']
# ``int.to_bytes`` raises these where ``struct.pack`` raises
# ``struct.error`` for the other sizes (#2291).
try:
buf = integer.to_bytes(size, end, signed=signed)
except (AttributeError, OverflowError, TypeError, ValueError) as error:
raise StructError(f'{cls.__name__}: pack failed') from error
else:
try:
fmt = f'{endian}{kind}'
buf = struct.pack(fmt, integer) # pylint: disable=no-member
except struct.error as error: # pylint: disable=no-member
raise StructError(f'{cls.__name__}: pack failed') from error
return buf
[docs]
def _make_octets(self, value: 'Any', *, size: 'int', name: 'str',
alias: 'Optional[str]' = None) -> 'bytes':
"""Check that ``value`` is exactly ``size`` octets.
The octets :meth:`_read_octets` reads are written as they are, so
their size is all there is to check.
Args:
value: the value given for the field
size: octets in the field
name: name of the field, for the error message
alias: prefix of the error message, for a caller whose
:attr:`alias` raises before ``read()`` sets it; :attr:`alias`
by default
Returns:
``value`` as :class:`bytes`.
Raises:
ProtocolError: If ``value`` is not :class:`bytes` or
:class:`bytearray`, or is not ``size`` octets long. The
message names ``alias``.
.. versionadded:: 1.6.0
"""
if not isinstance(value, (bytes, bytearray)) or len(value) != size:
unit = 'octet' if size == 1 else 'octets'
raise ProtocolError(f'{self.alias if alias is None else alias}: invalid {name}: '
f'must be {size} {unit}, not {value!r}')
return bytes(value)
[docs]
@classmethod
def _fill_short_read(cls, value: 'Any', *, size: 'int', model: 'Any', field: 'str',
offset: 'int' = 0) -> 'Any':
"""Zero-fill the octets of ``value`` that a capture cut short.
A model parsed from a capture cut short records, in its
``__short_read__``, the field of :attr:`__schema__` the capture ended
inside and the octets of it present. A padding field keeps only those
octets, where the other fields read the rest as zeros, so a value
read at or past the cut comes back short. Zero-filled, it rebuilds as
the rest of the header does, and the rebuild is cut back to what was
captured. A value of any other width is handed back for
:meth:`_make_octets` to refuse.
Args:
value: the value taken from ``model``
size: octets in the value's field
model: the data model ``value`` was taken from
field: the field of :attr:`__schema__` that holds ``value``
offset: where ``value`` starts in ``field``
Returns:
``value`` zero-filled to ``size`` where it is exactly the octets
the cut kept of it; otherwise ``value`` itself.
.. versionadded:: 1.6.0
"""
if not isinstance(value, (bytes, bytearray)) or len(value) >= size:
return value
get = getattr(model, 'get', None)
record = get('__short_read__') if callable(get) else None
if record is None:
return value
name, count = record
fields = list(cls.__schema__.__fields__)
if name == field:
kept = min(max(count - offset, 0), size)
elif name in fields and field in fields and fields.index(name) < fields.index(field):
kept = 0
else:
return value
if len(value) == kept:
return bytes(value).ljust(size, b'\x00')
return value
[docs]
def _make_uint(self, value: 'Any', *, bits: 'int', name: 'str', alias: 'Optional[str]' = None) -> 'int':
"""Check that ``value`` is an unsigned integer that fits in ``bits`` bits.
A :obj:`bool` passes only for a one-bit field, so that a flag given
for a wider field is refused rather than read as ``0`` or ``1``.
Args:
value: the value given for the field
bits: width of the field, in bits
name: name of the field, for the error message
alias: prefix of the error message, for a caller whose
:attr:`alias` raises before ``read()`` sets it; :attr:`alias`
by default
Returns:
``value`` as a plain :class:`int`.
Raises:
ProtocolError: If ``value`` is not an :class:`int`, is a
:obj:`bool` for a field wider than one bit, is negative, or
does not fit in ``bits`` bits. The message names ``alias``.
.. versionadded:: 1.6.0
"""
if (not isinstance(value, int) or (isinstance(value, bool) and bits != 1)
or not 0 <= value < (1 << bits)):
raise ProtocolError(f'{self.alias if alias is None else alias}: invalid {name}: {value!r}')
return int(value)
[docs]
def _make_sint(self, value: 'Any', *, bits: 'int', name: 'str', alias: 'Optional[str]' = None,
allow_unsigned: 'bool' = False) -> 'int':
"""Check that ``value`` is a signed integer that fits in ``bits`` bits.
The inverse of :meth:`_read_sint`. A :obj:`bool` is refused whatever
``bits`` is, since no signed field holds a flag.
Args:
value: the value given for the field
bits: width of the field, in bits
name: name of the field, for the error message
alias: prefix of the error message, for a caller whose
:attr:`alias` raises before ``read()`` sets it or is not the
prefix its messages carry; :attr:`alias` by default
allow_unsigned: also accept the unsigned spelling of the field's
``bits`` bits, i.e. up to ``2 ** bits - 1``, for a field whose
specification spells one value both ways, as a sentinel
written ``-1`` and ``0xFF...F``; both denote the same bits
Returns:
The unsigned integer whose ``bits`` bits denote ``value`` in
two's complement.
Raises:
ProtocolError: If ``value`` is not an :class:`int`, is a
:obj:`bool`, or does not fit in ``bits`` bits as two's
complement (or, with ``allow_unsigned``, as an unsigned
integer either). The message names ``alias``.
.. versionadded:: 1.6.0
"""
bound = 1 << (bits - 1)
limit = 2 * bound if allow_unsigned else bound
if not isinstance(value, int) or isinstance(value, bool) or not -bound <= value < limit:
raise ProtocolError(f'{self.alias if alias is None else alias}: invalid {name}: {value!r}')
return value & ((1 << bits) - 1)
[docs]
@staticmethod
def _uint_size(value: 'int', *, minimum: 'int' = 0) -> 'int':
"""Count the octets a non-negative integer needs.
``value.bit_length()`` is rounded up to whole octets: ``// 8`` alone
floors, and counts one octet too few for a width that is not a
multiple of eight.
Args:
value: the integer, not negative
minimum: the fewest octets to count, for a field that cannot be
empty although ``0`` needs none
Returns:
The octets ``value`` needs, and at least ``minimum``.
.. versionadded:: 1.6.0
"""
return max(minimum, (value.bit_length() + 7) // 8)
[docs]
@classmethod
def _make_vector(cls, octets: 'bytes', *, size: 'int' = 1) -> 'bytes':
"""Write ``octets`` as a length-prefixed vector.
The inverse of :meth:`_read_vector`: a ``size``-octet unsigned
big-endian length, packed by :meth:`_make_pack`, then ``octets``.
Args:
octets: the vector's octets
size: octets in the length prefix
Returns:
The length prefix followed by ``octets``.
Raises:
StructError: If the length of ``octets`` does not fit in ``size``
octets.
.. versionadded:: 1.6.0
"""
return cls._make_pack(len(octets), size=size) + octets
[docs]
@classmethod
def _make_uints(cls, values: 'Iterable[int]', *, size: 'int' = 1) -> 'bytes':
"""Write ``values`` as a run of ``size``-octet unsigned big-endian integers.
The inverse of :meth:`_read_uints`, each value packed by
:meth:`_make_pack`.
Args:
values: the integers, or anything :class:`int` converts to one,
such as an enumeration member
size: octets in each integer
Returns:
The packed integers, concatenated in order.
Raises:
StructError: If a value is negative, or does not fit in ``size``
octets.
.. versionadded:: 1.6.0
"""
return b''.join(cls._make_pack(int(value), size=size) for value in values)
[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``.
The inverse of :meth:`_read_mac_addr`.
Args:
addr: ``:``- or ``-``-separated *hex* encoded address
size: octets in the address, or :data:`None` for any number of
them, none included
name: name of the address, for the error message
alias: prefix of the error message, for a caller whose
:attr:`alias` raises before ``read()`` sets it or is not the
prefix its messages carry; :attr:`alias` by default
Returns:
The address octets.
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. The message names ``alias``.
.. versionadded:: 1.6.0
"""
prefix = self.alias if alias is None else alias
if not isinstance(addr, (str, bytes, bytearray)):
raise ProtocolError(f'{prefix}: invalid {name}: must be a str, bytes or '
f'bytearray, not {addr!r}')
text = addr.encode() if isinstance(addr, str) else bytes(addr)
octets = text.replace(b'-', b':').split(b':') if text else []
if ((size is None or len(octets) == size)
and all(len(octet) == 2 and all(chr(digit) in string.hexdigits for digit in octet)
for octet in octets)):
return bytes.fromhex(b''.join(octets).decode())
raise ProtocolError(f'{prefix}: invalid {name}: {addr!r}')
[docs]
def _make_index(self, value: 'Any', namespace: 'Type[_ET]', *, bits: 'int', name: 'str',
alias: 'Optional[str]' = None) -> '_ET':
"""Resolve the code given for an enumeration field of ``make()``.
The field takes a member of ``namespace`` or an :class:`int` code,
and either resolves with ``namespace``'s
:meth:`~pcapkit.corekit.enum.EnumLookup.get`. A name is refused rather
than looked up, so a caller holding one resolves it first, with
:meth:`resolve_index` (:issue:`2594`). The code is checked as
:meth:`_make_uint` checks an integer, so a :obj:`bool` passes only for
a one-bit field (:issue:`2617`): the field's width decides what is
refused, not the enumeration's range.
Args:
value: the value given for the field
namespace: the enumeration the field holds
bits: width of the field, in bits
name: name of the field, for the error message
alias: prefix of the error message, for a caller whose
:attr:`alias` raises before ``read()`` sets it; :attr:`alias`
by default
Returns:
What ``namespace``'s :meth:`~pcapkit.corekit.enum.EnumLookup.get`
resolves ``value`` to. For most registries that includes an
unassigned code, as the ``Unassigned`` or ``Reserved`` member their
``_missing_`` builds. Where ``get`` raises :exc:`ValueError` for a
code the field carries, as a closed enumeration does, the code
itself comes back as an :class:`int`, so that the field is rebuilt
as captured.
Raises:
ProtocolError: If ``value`` is not an :class:`int` (a name among
them), is a :obj:`bool` for a field wider than one bit, is
negative, or does not fit in ``bits`` bits. The message names
``alias``.
.. versionchanged:: 1.6.0
Takes the enumeration and the width of the field, in place of the
``default``, ``namespace``, ``reversed`` and ``pack`` arguments,
refuses a code the field cannot carry, and refuses a name rather
than resolving it against an enumeration inferred from the calling
method's frame (:issue:`2594`).
"""
code = self._make_uint(value, bits=bits, name=name, alias=alias)
try:
return namespace.get(code)
except ValueError:
return cast('_ET', code)
[docs]
@classmethod
def _make_payload(cls, data: 'Data') -> 'ProtocolBase':
"""Create payload from ``data`` for protocol construction.
The payload's type and name are ``__next_type__`` and
``__next_name__``. If either is missing or :data:`None`, or ``data``
holds nothing under ``__next_name__``, the payload is a
:class:`~pcapkit.protocols.misc.no_payload.protocol.NoPayload`, as for
a protocol with no next layer; otherwise :meth:`Protocol.from_data
<pcapkit.protocols.protocol.Protocol.from_data>` builds it.
:meth:`Info.to_dict <pcapkit.corekit.infoclass.Info.to_dict>` carries
both keys, so ``data`` rebuilt from such a :obj:`dict` dispatches as
the info did. A hand-written :obj:`dict` must carry them too, as it
must ``__short_read__`` and, for IPv6, ``__exthdr__``: the payload is
never guessed from the other keys of ``data`` (:issue:`2003`).
The payload is built one layer further into the frame, as
:meth:`_parse_next_layer` counts it, so a rebuild stops at
:data:`~pcapkit.utilities.config.FRAME_LAYER_LIMIT` where the parse did
(:issue:`1610`). The payload's parse reports nothing, the limit
included: this protocol's own parse goes over the same octets, so the
rebuilt frame reports each warning, and each error a next-layer fallback
absorbs, once, as the parsed one did (:issue:`2711`).
Args:
data: protocol data
Returns:
Payload for protocol construction.
"""
proto = cast('Optional[Type[Protocol]]', data.get('__next_type__'))
name = cast('Optional[str]', data.get('__next_name__'))
if (proto is None or not (isinstance(proto, type) and issubclass(proto, ProtocolBase))
or name is None or name not in data):
from pcapkit.protocols.misc.no_payload.protocol import \
NoPayload # pylint: disable=import-outside-toplevel
return NoPayload()
token, rebuilt = frame_depth.set(frame_depth.get() + 1), rebuilding.set(True)
try:
return proto.from_data(data[name])
finally:
rebuilding.reset(rebuilt)
frame_depth.reset(token)
[docs]
@staticmethod
def _lookup_registry(registry: 'DefaultDict[Any, _VT]', code: 'Any') -> '_VT':
"""Look up a dispatch :term:`registry` entry without recording a miss.
Arguments:
registry: dispatch registry to read, i.e. :attr:`self.__proto__
<pcapkit.protocols.protocol.Protocol.__proto__>` or one of the
per-protocol ``__option__`` / ``__chunk__`` / ``__block__``
family. Passed in rather than read from the class, so that a
caller reaching the registry through an instance keeps doing
so.
code: registry key to look up, i.e. the wire code being dispatched on
Returns:
The entry registered for ``code``, or the fallback ``registry``
declares when ``code`` is not registered.
Important:
Each of these registries is a :class:`collections.defaultdict` on a
*class* attribute, shared process-wide, so ``registry[code]`` would
insert every code it misses: one packet with an unrecognised code
would grow the registry permanently. The entry buys nothing, costs a
spurious "already registered" warning on the next genuine
``register`` of that code, and makes "is this code registered?"
depend on what has been parsed, so the fallback is read from the
default factory directly.
"""
if code in registry:
return registry[code]
return cast('Callable[[], _VT]', registry.default_factory)()
[docs]
@staticmethod
def _lookup_next_layer(registry: 'DefaultDict[int, ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]]',
proto: 'int') -> 'Type[ProtocolBase]':
"""Look up the protocol class registered for a next layer code.
Arguments:
registry: next layer protocol registry, i.e. :attr:`self.__proto__
<pcapkit.protocols.protocol.Protocol.__proto__>`. Passed in
rather than read from the class, so that a caller reaching the
registry through an instance keeps doing so.
proto: next layer protocol index
Returns:
The class registered for ``proto``, or the fallback ``registry``
declares -- normally :class:`~pcapkit.protocols.misc.raw.protocol.Raw` -- when
``proto`` is not registered.
Important:
The lookup is :meth:`self._lookup_registry
<pcapkit.protocols.protocol.Protocol._lookup_registry>`, so a miss
does not grow the shared registry. The entry found, registered or
the fallback, may be a
:class:`~pcapkit.corekit.module.ModuleDescriptor`, resolved to the
class it names.
Nothing is written back. A hit memoised under ``proto`` would keep
a class :func:`importlib.reload` makes stale, and make parsing
mutate a registry the whole process shares (:issue:`1559`); so
would a memo in the default factory or in a cache beside the
registry (:issue:`421`, :issue:`425`, :issue:`555`). Each lookup
resolves its descriptor again, kept cheap by
:attr:`ModuleDescriptor.klass
<pcapkit.corekit.module.ModuleDescriptor.klass>` reading
:data:`sys.modules` rather than re-entering
:func:`importlib.import_module` (:issue:`574`).
"""
protocol = ProtocolBase._lookup_registry(registry, proto)
if isinstance(protocol, ModuleDescriptor):
return protocol.klass
return protocol
[docs]
def _decode_next_layer(self, dict_: '_PT', proto: 'int', length: 'Optional[int]' = None, *,
packet: 'Optional[dict[str, Any]]' = None) -> '_PT':
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 <Protocol.unpack>`)
Returns:
Current protocol with next layer extracted.
Notes:
The method adds a ``__next_type__`` key to ``dict_`` to store the
next layer protocol type, and a ``__next_name__`` key to store the
next layer protocol name. :meth:`Info.to_dict
<pcapkit.corekit.infoclass.Info.to_dict>` carries both, so that
:meth:`_make_payload` dispatches a :obj:`dict` rebuild on them,
while the dumpers and ``repr`` leave them out (:issue:`2003`).
"""
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,
'__next_type__': type(next_),
'__next_name__': layer,
})
self._next = next_ # pylint: disable=attribute-defined-outside-init
self._protos = ProtoChain(self.__class__, self.alias, basis=chain) # pylint: disable=attribute-defined-outside-init
return dict_
[docs]
@beholder
def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *,
packet: 'Optional[dict[str, Any]]' = None) -> 'ProtocolBase':
"""Import next layer extractor.
Arguments:
proto: next layer protocol index
length: valid (*non-padding*) length
packet: packet info (passed from :meth:`self.unpack <Protocol.unpack>`)
Returns:
Instance of next layer.
"""
if TYPE_CHECKING:
protocol: 'Type[ProtocolBase]'
file_ = self._get_payload()
if length is None:
length = len(file_)
empty = self._empty_next_layer(file_, length)
if empty is not None:
protocol = empty
elif self._sigterm:
from pcapkit.protocols.misc.raw.protocol import Raw as protocol # isort: skip # pylint: disable=import-outside-toplevel
else:
protocol = self._lookup_next_layer(self.__proto__, proto)
return self._parse_next_layer(protocol, file_, length, alias=proto, packet=packet,
layer=self._exlayer, protocol=self._exproto,
__context__=self._exctx)
def _empty_next_layer(self, file_: 'bytes', length: 'int') -> 'Optional[Type[ProtocolBase]]':
r"""The class a next layer with nothing to parse is kept as.
Arguments:
file\_: payload of the current layer
length: valid (*non-padding*) length
Returns:
:data:`None` if octets were captured past the header and it leaves
its next layer a length, which is then parsed. Otherwise
:class:`~pcapkit.protocols.misc.raw.protocol.Raw` if no octets were
captured past a header leaving its next layer a negative length, and
:class:`~pcapkit.protocols.misc.no_payload.protocol.NoPayload` if
the header leaves none, or no octets past it.
Note:
A header leaves a negative length when it declares more than it was
given: one the capture cut short, read at the octets captured, or an
IPv6 extension header overrunning the payload (:issue:`2477`). Such a
next layer is kept raw, not dispatched on a code the cut may have
read as zeros (:issue:`1170`). :meth:`from_data` reads the rebuild of
a cut header at the width its parse read it at, which leaves it the
same length, so the rebuild keeps ``Raw`` where its data model did
(:issue:`2717`, :issue:`2761`).
.. versionadded:: 1.6.0
"""
if file_ and length != 0:
return None
if not file_ and length < 0:
from pcapkit.protocols.misc.raw.protocol import Raw # isort: skip # pylint: disable=import-outside-toplevel
return Raw
from pcapkit.protocols.misc.no_payload.protocol import NoPayload # isort: skip # pylint: disable=import-outside-toplevel
return NoPayload
[docs]
@staticmethod
def _parse_next_layer(klass: 'Type[ProtocolBase]', file_: 'bytes', length: 'int',
**kwargs: 'Any') -> 'ProtocolBase':
r"""Parse the next layer, or keep it as raw octets if its header was not captured.
Arguments:
klass: next layer protocol class
file\_: payload of the current layer
length: valid (*non-padding*) length
**kwargs: keyword arguments for the next layer's constructor
Returns:
Instance of ``klass``, or of :class:`~pcapkit.protocols.misc.raw.protocol.Raw`
when fewer octets were captured than its header needs, or when it is
past :data:`~pcapkit.utilities.config.FRAME_LAYER_LIMIT`.
Note:
:meth:`Schema.unpack <pcapkit.protocols.schema.Schema.unpack>`
zero-fills a field the stream ends inside, so a header parsed from
too few octets is filled with values that were never on the wire, and
:meth:`from_data <Protocol.from_data>` rebuilds them (:issue:`1170`).
Such a payload is kept as :class:`~pcapkit.protocols.misc.raw.protocol.Raw`
instead, as an unparseable one is.
The captured octets are the first ``length`` of ``file_``, or all of
it when it holds fewer, as a truncated capture does. They are short
of the header when there are fewer than the next layer's
:meth:`__length_hint__
<pcapkit.protocols.protocol.Protocol.__length_hint__>`, unless the
header it parsed, :attr:`length
<pcapkit.protocols.protocol.Protocol.length>` octets long, fits
in them anyway: for a variable-length header such as
:class:`~pcapkit.protocols.link.l2tpv2.protocol.L2TPv2` the hint is the longest
form rather than the shortest. A next layer that reports no hint is
kept as parsed.
The warnings of a parse that may be replaced are held until that is
known, then:
* **replaced as short** -- dropped, since they describe a header the
result does not have: ``packet length < 0`` twice for an Ethernet
frame of 10 octets (:issue:`1580`);
* **kept** -- reported as the parse ends, inside the guard that
catches them under an ``error`` filter; a layer kept with its
header cut short still warns;
* **raised** -- dropped when
:func:`~pcapkit.corekit.decorators.beholder`, which wraps
:meth:`_import_next_layer`, replaces the layer, whose ``error``
reports the failure (:issue:`2308`); otherwise reported as the
error leaves.
Under ``beholder`` a layer is held however much was captured, since
one that raises is replaced at any length (:issue:`2516`), so a kept
layer's warnings follow those of the layers it carries. Outside it,
a layer captured at least as long as its hint cannot be replaced,
and is parsed holding nothing, even inside one that holds.
A layer past :data:`~pcapkit.utilities.config.FRAME_LAYER_LIMIT` in its frame is not parsed:
it and the rest of the frame are kept as
:class:`~pcapkit.protocols.misc.raw.protocol.Raw`, marked
:attr:`_past_layer_limit`, with one
:exc:`~pcapkit.utilities.warnings.ProtocolWarning` (:issue:`1610`).
Warns:
ProtocolWarning: If the next layer is past :data:`~pcapkit.utilities.config.FRAME_LAYER_LIMIT`.
"""
depth = frame_depth.get() + 1
limit = config.FRAME_LAYER_LIMIT
if limit != -1 and depth > limit:
from pcapkit.protocols.misc.no_payload.protocol import NoPayload # isort: skip # pylint: disable=import-outside-toplevel
from pcapkit.protocols.misc.raw.protocol import Raw # isort: skip # pylint: disable=import-outside-toplevel
# NOTE: Nothing is lost if the next layer is raw already, or empty.
if klass is not Raw and klass is not NoPayload:
message = (f'the frame reached the limit of {limit} layers; the rest of it, '
f'from {klass.__name__} on, is left undissected')
# NOTE: Inside a payload being rebuilt, the layer it is rebuilt
# for warns of the same octets once its own parse gets here.
if not rebuilding.get():
warn(message, ProtocolWarning, stacklevel=stacklevel())
raw = Raw(file_, length, **{**kwargs, 'error': message})
raw._past_layer_limit = True # pylint: disable=protected-access
return raw
captured = min(length, len(file_))
token = frame_depth.set(depth)
try:
# NOTE: Under ``beholder`` a layer that raises is replaced at any length,
# so its warnings are held however much was captured (#2516).
with hold_warnings(beholding.get() or _short_of_hint(klass, captured)) as held:
try:
next_ = klass(file_, length, **kwargs) # type: ignore[abstract]
except Exception:
# NOTE: Under ``beholder`` the layer is replaced and its warnings
# go with it; outside it, they leave with the error (#2308).
held.discard = beholding.get()
raise
length_hint = getattr(next_, '__length_hint__', None)
hint = None if length_hint is None else length_hint()
held.discard = hint is not None and captured < hint and captured < next_.length
finally:
frame_depth.reset(token)
if held.discard:
from pcapkit.protocols.misc.raw.protocol import Raw # isort: skip # pylint: disable=import-outside-toplevel
next_ = Raw(file_, length, **kwargs)
return next_
def _check_term_threshold(self) -> bool:
"""Check if reached termination threshold."""
if self._exlayer is None or (layer := self.__layer__) is None:
layer_match = False
else:
layer_match = layer.upper() == self._exlayer.upper()
if self._exproto is None:
protocol_match = False
else:
protocol_match = False
comp_test = [name.upper() for name in self.id()]
for test in self.expand_comp(self._exproto):
if test in comp_test:
protocol_match = True
break
return layer_match or protocol_match
def _read_qs_rate(self, code: 'int') -> 'int':
"""Read a Quick-Start rate code as a rate in kbps.
:rfc:`4782#section-3.1` encodes the Rate Request, and the Rate Report
formatted exactly as it, as a four-bit ``N`` that stands for
``K * 2 ** N`` bps with ``K`` set to 40,000, and ``N = 0`` for a rate of
zero. The IPv6 option keeps the IPv4 format (:rfc:`4782#section-3.2`),
and the TCP Quick-Start Response formats its Rate Request the same way
(:rfc:`4782#section-4.2`).
Args:
code: the four-bit rate field, ``N``
Returns:
The rate in kbps: ``0`` for ``N = 0``, else ``40 * 2 ** N``.
.. versionadded:: 1.6.0
"""
return 40 * 2 ** code if code else 0
def _make_qs_rate(self, rate: 'Any') -> 'int':
"""Make the Quick-Start rate code for a rate in kbps.
The inverse of :meth:`_read_qs_rate`. A rate between two codes is
rounded down to the lower one, so the option never carries more than
the rate given.
Args:
rate: the rate, in kbps
Returns:
The four-bit rate field, ``N``.
Raises:
ProtocolError: If ``rate`` is not an :class:`int`, is a
:obj:`bool`, or is one the field cannot carry: negative, above
zero but below the 80 kbps of ``N = 1``, or at or above the
``40 * 2 ** 16`` kbps that needs a fifth bit. The message names
:attr:`alias`.
.. versionadded:: 1.6.0
"""
if (not isinstance(rate, int) or isinstance(rate, bool)
or (rate and not 80 <= rate < 40 * 2 ** 16)):
raise ProtocolError(f'{self.alias}: invalid Quick-Start rate: {rate!r}')
return (rate // 40).bit_length() - 1 if rate else 0
[docs]
class Protocol(ProtocolBase, Generic[_PT, _ST]):
"""User-facing abstract base class for third-party protocols.
Inherit it directly only to create a new protocol stack; a new protocol in
an existing layer subclasses that layer's base class instead; see
:doc:`/ext`. The built-in protocol families do not derive from this class
but from :class:`ProtocolBase`, whose metaclass is :class:`ProtocolMeta`.
.. versionchanged:: 1.3.0
The built-in protocol families derive from :class:`ProtocolBase`, so a
built-in protocol is not an instance of this class, where every one
was; and a subclass is registered in ``pcapkit.protocols.__proto__``
when it is defined.
"""
[docs]
def __init_subclass__(cls, /, schema: 'Optional[Type[_ST]]' = None,
data: 'Optional[Type[_PT]]' = None,
code: 'Any' = None,
*args: 'Any', **kwargs: 'Any') -> 'None':
"""Initialisation for subclasses.
Args:
schema: Schema class.
data: Data class.
code: Next-layer dispatch registration key(s). :data:`None` (the
default) skips registration entirely. See
:meth:`ProtocolBase.__init_subclass__` for the accepted
shapes and the enum-type inference rule.
*args: Arbitrary positional arguments.
**kwargs: Arbitrary keyword arguments.
Sets :attr:`self.__schema__ <pcapkit.protocols.protocol.Protocol.__schema__>`
and :attr:`self.__data__ <pcapkit.protocols.protocol.Protocol.__data__>`
on the subclass through :meth:`ProtocolBase.__init_subclass__`, which
also finds an omitted one.
Once that has accepted ``code``, a concrete subclass is also named in
the by-name registry, :attr:`ProtocolBase.__registry__
<pcapkit.protocols.protocol.ProtocolBase.__registry__>`, under its
upper-cased class name; unlike ``code``'s dispatch registration, this
is not opt-in. An abstract subclass, as
:func:`~pcapkit.utilities.inspect.isabstract` reports it, is not named,
nor is a class deriving :class:`ProtocolBase` alone, as every built-in
protocol does. A name holding another entry keeps it, with a
:exc:`~pcapkit.utilities.warnings.RegistryWarning`; only
:meth:`ProtocolBase.register_protocol
<pcapkit.protocols.protocol.ProtocolBase.register_protocol>` replaces
one. The :class:`~pcapkit.corekit.module.ModuleDescriptor` naming the
class's own module and name is no clash, and stays.
See Also:
For more information on the registry, please refer to
:meth:`ProtocolBase.register_protocol <pcapkit.protocols.protocol.ProtocolBase.register_protocol>`.
.. versionchanged:: 1.6.0
An abstract subclass is not registered, and a name already holding
an entry keeps it, where every subclass was registered over it
(:issue:`2763`).
"""
super().__init_subclass__(schema, data, code, *args, **kwargs)
# NOTE: #2763: registered after ``code`` is accepted, never over a held name or this class's
# own descriptor. #2784: not inspect.isabstract, whose __flags__ read GraalPy freezes early.
if not isabstract(cls):
name = cls.__name__.upper()
incumbent = ProtocolBase.__registry__.setdefault(name, cls)
if incumbent is not cls and not (isinstance(incumbent, ModuleDescriptor)
and incumbent == (cls.__module__, cls.__name__)):
incumbent_repr, protocol_repr = repr(incumbent), repr(cls)
if incumbent_repr == protocol_repr: # #710: tell same-named classes apart
incumbent_repr = f'{incumbent_repr} (id={id(incumbent):#x})'
protocol_repr = f'{protocol_repr} (id={id(cls):#x})'
warn(f'protocol {name} already registered, keeping {incumbent_repr} '
f'over {protocol_repr}', RegistryWarning)