# -*- coding: utf-8 -*-
"""Protocol Registries
=========================
.. module:: pcapkit.foundation.registry.protocols
This module provides the protocol registries for :mod:`pcapkit`.
.. versionchanged:: 1.5.0
``NULL``, the default of every ``class_`` argument here, is now the
:data:`pcapkit.corekit.sentinels.NULL` sentinel; in 1.4.x it was the
string ``'(null)'`` (:issue:`833`).
"""
import collections.abc
import enum
import threading
from typing import TYPE_CHECKING, cast, overload
import aenum
from pcapkit.const.reg.apptype import AppType as Enum_AppType
from pcapkit.const.reg.apptype import TransportProtocol
from pcapkit.const.reg.ethertype import EtherType as Enum_EtherType
from pcapkit.const.reg.linktype import LinkType as Enum_LinkType
from pcapkit.const.reg.transtype import TransType as Enum_TransType
from pcapkit.const.sctp.payload_protocol_identifier import \
PayloadProtocolIdentifier as Enum_PayloadProtocolIdentifier
from pcapkit.corekit.module import ModuleDescriptor
from pcapkit.corekit.sentinels import NULL
from pcapkit.protocols import __proto__ as protocol_registry
from pcapkit.protocols.application.httpv2 import HTTP as HTTPv2
from pcapkit.protocols.internet.hip import HIP
from pcapkit.protocols.internet.hopopt import HOPOPT
from pcapkit.protocols.internet.internet import Internet
from pcapkit.protocols.internet.ipv4 import IPv4
from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts
from pcapkit.protocols.internet.ipv6_route import IPv6_Route
from pcapkit.protocols.internet.mh import MH
from pcapkit.protocols.link.link import Link
from pcapkit.protocols.misc.pcap.frame import Frame
from pcapkit.protocols.misc.pcapng import PCAPNG
from pcapkit.protocols.protocol import ProtocolBase
from pcapkit.protocols.schema.application.httpv2 import FrameType as Schema_HTTP_FrameType
from pcapkit.protocols.schema.internet.hip import Parameter as Schema_HIP_Parameter
from pcapkit.protocols.schema.internet.hopopt import Option as Schema_HOPOPT_Option
from pcapkit.protocols.schema.internet.ipv4 import Option as Schema_IPv4_Option
from pcapkit.protocols.schema.internet.ipv6_opts import Option as Schema_IPv6_Opts_Option
from pcapkit.protocols.schema.internet.ipv6_route import \
RoutingType as Schema_IPv6_Route_RoutingType
from pcapkit.protocols.schema.internet.mh import CGAExtension as Schema_MH_CGAExtension
from pcapkit.protocols.schema.internet.mh import Option as Schema_MH_Option
from pcapkit.protocols.schema.internet.mh import Packet as Schema_MH_Packet
from pcapkit.protocols.schema.misc.pcapng import BlockType as Schema_PCAPNG_BlockType
from pcapkit.protocols.schema.misc.pcapng import DSBSecrets as Schema_PCAPNG_DSBSecrets
from pcapkit.protocols.schema.misc.pcapng import \
NameResolutionRecord as Schema_PCAPNG_NameResolutionRecord
from pcapkit.protocols.schema.misc.pcapng import Option as Schema_PCAPNG_Option
from pcapkit.protocols.schema.transport.tcp import MPTCP as Schema_TCP_MPTCP
from pcapkit.protocols.schema.transport.tcp import Option as Schema_TCP_Option
from pcapkit.protocols.transport.sctp import SCTP
from pcapkit.protocols.transport.tcp import TCP
from pcapkit.protocols.transport.udp import UDP
from pcapkit.utilities.exceptions import RegistryError
from pcapkit.utilities.logging import get_logger
from pcapkit.utilities.warnings import RegistryWarning, warn
if TYPE_CHECKING:
from typing import Any, Iterator, Optional, Type
from pcapkit.const.hip.parameter import Parameter as HIP_Parameter
from pcapkit.const.http.frame import Frame as HTTP_Frame
from pcapkit.const.ipv4.option_number import OptionNumber as IPv4_OptionNumber
from pcapkit.const.ipv6.option import Option as IPv6_Option
from pcapkit.const.ipv6.routing import Routing as IPv6_Routing
from pcapkit.const.mh.cga_extension import CGAExtension as MH_CGAExtension
from pcapkit.const.mh.option import Option as MH_Option
from pcapkit.const.mh.packet import Packet as MH_Packet
from pcapkit.const.pcapng.block_type import BlockType as PCAPNG_BlockType
from pcapkit.const.pcapng.option_type import OptionType as PCAPNG_OptionType
from pcapkit.const.pcapng.record_type import RecordType as PCAPNG_RecordType
from pcapkit.const.pcapng.secrets_type import SecretsType as PCAPNG_SecretsType
from pcapkit.const.reg.ethertype import EtherType
from pcapkit.const.reg.linktype import LinkType
from pcapkit.const.reg.transtype import TransType
from pcapkit.const.sctp.payload_protocol_identifier import \
PayloadProtocolIdentifier as SCTP_PayloadProtocolIdentifier
from pcapkit.const.tcp.mp_tcp_option import MPTCPOption as TCP_MPTCPOption
from pcapkit.const.tcp.option import Option as TCP_Option
from pcapkit.corekit.sentinels import NullType
from pcapkit.protocols.application.httpv2 import FrameConstructor as HTTP_FrameConstructor
from pcapkit.protocols.application.httpv2 import FrameParser as HTTP_FrameParser
from pcapkit.protocols.internet.hip import ParameterConstructor as HIP_ParameterConstructor
from pcapkit.protocols.internet.hip import ParameterParser as HIP_ParameterParser
from pcapkit.protocols.internet.hopopt import OptionConstructor as HOPOPT_OptionConstructor
from pcapkit.protocols.internet.hopopt import OptionParser as HOPOPT_OptionParser
from pcapkit.protocols.internet.ipv4 import OptionConstructor as IPv4_OptionConstructor
from pcapkit.protocols.internet.ipv4 import OptionParser as IPv4_OptionParser
from pcapkit.protocols.internet.ipv6_opts import \
OptionConstructor as IPv6_Opts_OptionConstructor
from pcapkit.protocols.internet.ipv6_opts import OptionParser as IPv6_Opts_OptionParser
from pcapkit.protocols.internet.ipv6_route import TypeConstructor as IPv6_Route_TypeConstructor
from pcapkit.protocols.internet.ipv6_route import TypeParser as IPv6_Route_TypeParser
from pcapkit.protocols.internet.mh import ExtensionConstructor as MH_ExtensionConstructor
from pcapkit.protocols.internet.mh import ExtensionParser as MH_ExtensionParser
from pcapkit.protocols.internet.mh import OptionConstructor as MH_OptionConstructor
from pcapkit.protocols.internet.mh import OptionParser as MH_OptionParser
from pcapkit.protocols.internet.mh import PacketConstructor as MH_PacketConstructor
from pcapkit.protocols.internet.mh import PacketParser as MH_PacketParser
from pcapkit.protocols.misc.pcapng import BlockConstructor as PCAPNG_BlockConstructor
from pcapkit.protocols.misc.pcapng import BlockParser as PCAPNG_BlockParser
from pcapkit.protocols.misc.pcapng import OptionConstructor as PCAPNG_OptionConstructor
from pcapkit.protocols.misc.pcapng import OptionParser as PCAPNG_OptionParser
from pcapkit.protocols.misc.pcapng import RecordConstructor as PCAPNG_RecordConstructor
from pcapkit.protocols.misc.pcapng import RecordParser as PCAPNG_RecordParser
from pcapkit.protocols.misc.pcapng import SecretsConstructor as PCAPNG_SecretsConstructor
from pcapkit.protocols.misc.pcapng import SecretsParser as PCAPNG_SecretsParser
from pcapkit.protocols.transport.tcp import MPOptionConstructor as TCP_MPOptionConstructor
from pcapkit.protocols.transport.tcp import MPOptionParser as TCP_MPOptionParser
from pcapkit.protocols.transport.tcp import OptionConstructor as TCP_OptionConstructor
from pcapkit.protocols.transport.tcp import OptionParser as TCP_OptionParser
__all__ = [
'register_protocol',
'register_protocol_code',
'register_linktype',
'register_pcap', 'register_pcapng',
'register_ethertype',
'register_transtype',
'register_ipv4_option', 'register_hip_parameter', 'register_hopopt_option',
'register_ipv6_opts_option', 'register_ipv6_route_routing',
'register_mh_message', 'register_mh_option', 'register_mh_extension',
'register_apptype',
'register_tcp', 'register_udp', 'register_sctp',
'register_tcp_option', 'register_tcp_mp_option',
'register_http_frame',
'register_pcapng_block', 'register_pcapng_option', 'register_pcapng_secrets',
'register_pcapng_record',
]
#: logging.Logger: Module-level logger, a child of the package-wide
#: :data:`pcapkit.utilities.logging.logger`.
logger = get_logger(__name__)
# NOTE: pcapkit.protocols.__proto__
[docs]
def register_protocol(protocol: 'Type[ProtocolBase]') -> 'None':
"""Register a protocol class.
The protocol class must be a subclass of
:class:`~pcapkit.protocols.protocol.ProtocolBase`, and will be registered to
the :data:`pcapkit.protocols.__proto__` registry.
The registry is keyed on ``protocol.__name__.upper()``, which is **not**
unique across this package: several dispatchable protocol classes are
named ``HTTP`` -- the generic base
:class:`pcapkit.protocols.application.http.HTTP` and the version
implementations :class:`pcapkit.protocols.application.httpv1.HTTP` and
:class:`pcapkit.protocols.application.httpv2.HTTP` -- so they all compete
for the single key ``'HTTP'``. Registering one of them replaces whichever
was there, and the replacement is observable through every reader of the
registry, e.g. :meth:`ProtocolBase.expand_comp
<pcapkit.protocols.protocol.ProtocolBase.expand_comp>`, which resolves a
bare protocol name through it. Per :issue:`675` the overwrite is reported,
matching :meth:`ProtocolBase.register
<pcapkit.protocols.protocol.ProtocolBase.register>` and the other
overwrite-warning registries.
The guard reads "key present **and** incumbent is a different class" --
presence alone is not enough. This registry's key is *derived* from the
class rather than supplied by a caller, and this function is the funnel
every wrapper registrar calls -- :func:`register_tcp`, :func:`register_udp`,
:func:`register_apptype`, :func:`register_linktype` and the rest all end by
calling it with their class, and registering their code back undoes that
call's write (:issue:`1505`). So registering one class under two codes, a
supported and documented thing to do, reaches this function twice with the
same class and nothing at stake; a presence-only guard would warn about an
overwrite that overwrote nothing. Warning on the harmless case is not free:
it teaches a caller to filter
:exc:`~pcapkit.utilities.warnings.RegistryWarning` wholesale, and that
filter would then hide the ``HTTP`` collision this warning exists to
surface. The sibling ``register`` methods across the package, each keyed on a
caller-supplied ``code``, apply the same identity criterion (:issue:`718`).
Making the key itself unique would resolve the collision rather than
merely reporting it, but it is a registry-format change that the bare-name
readers outside this module cannot absorb on their own -- the registry is a
documented public attribute, and its readers look a bare name up and then
degrade *silently* on a miss rather than raising, so a re-keying would not
announce itself either. :meth:`ProtocolBase.expand_comp
<pcapkit.protocols.protocol.ProtocolBase.expand_comp>` falls back to the
name as a plain string, and the ``protocol`` properties on
:class:`~pcapkit.foundation.reassembly.reassembly.ReassemblyMeta` and
:class:`~pcapkit.foundation.traceflow.traceflow.TraceFlowMeta` fall back to
:class:`~pcapkit.protocols.misc.raw.Raw`. The registry redesign of
:issue:`514` left this key unchanged and recorded the shared namespace as a
risk it did not fix, so the collision is reported here rather than
resolved.
Args:
protocol: Protocol class.
Raises:
pcapkit.utilities.exceptions.RegistryError: If ``protocol`` is not a
:class:`~pcapkit.protocols.protocol.ProtocolBase` subclass.
Warns:
pcapkit.utilities.warnings.RegistryWarning: If the registry already
holds a *different* class under this protocol's name, i.e. the
registration displaced another protocol class. Re-registering the
same class under the same name is silent.
"""
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 = protocol_registry.get(name)
if incumbent is not None and incumbent is not protocol:
incumbent_repr, protocol_repr = repr(incumbent), repr(protocol)
if incumbent_repr == protocol_repr:
# #710: two *distinct* objects whose repr() -- for an ordinary
# class, its module plus qualname -- happens to coincide, e.g. a
# factory that builds a fresh closure-local class of the same
# name on every call. Appending __module__/__qualname__ would not
# help here: that is exactly what the coinciding repr() already
# renders, so both sides would still print identically. id() is
# the fallback that actually differs, so it is only added in this
# branch -- the common case, two genuinely different classes,
# keeps the plain repr() and stays free of the extra noise.
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)
protocol_registry[name] = protocol
logger.debug('registered protocol: %s', protocol.__name__)
#: Sentinel for a name :data:`pcapkit.protocols.__proto__` did not hold.
_MISSING = object()
class _NameLease:
"""A :data:`pcapkit.protocols.__proto__` entry a code registration put there.
Each code registrar below ends by naming its class in the by-name registry,
and :mod:`pcapkit.foundation.registry` has no ``unregister``: a code is
undone by registering the entry it displaced (:issue:`1363`). Without a
record of that by-name write, handing a code back restored the code and left
the name behind (:issue:`1505`). A lease records what the write displaced and
which codes still dispatch to the class, so the last of them to be handed
back puts the displaced entry back, or removes the name if it was new.
A lease is one write, not one class: the same class can hold two leases on
a name -- e.g. a built-in registered at one port over a custom class
shadowing it at another -- and each undoes only its own write.
"""
__slots__ = ('name', 'protocol', 'displaced', 'codes')
def __init__(self, name: 'str', protocol: 'Type[ProtocolBase]', displaced: 'Any') -> 'None':
#: The key written in :data:`pcapkit.protocols.__proto__`.
self.name = name
#: The class written under it.
self.protocol = protocol
#: The entry it displaced, or :data:`_MISSING` for a new name.
self.displaced = displaced
#: ``(destination, code)`` registrations still holding the lease.
self.codes = set() # type: set[tuple[type, Any]]
#: Name -> its leases, oldest first. Each lease displaced the one below it,
#: unless the name was written by other means in between; the last is the one
#: a lookup sees, unless the name has been written since.
_NAME_LEASES = {} # type: dict[str, list[_NameLease]]
#: ``(destination, code)`` -> the lease its registration holds.
_CODE_LEASES = {} # type: dict[tuple[type, Any], _NameLease]
#: Serialises the lease tables above, so concurrent registrars cannot tear them.
_LEASE_LOCK = threading.RLock()
def _fold_lease(stack: 'list[_NameLease]', index: 'int') -> 'None':
"""Retire ``stack[index]``, a lease now displacing its own class.
That happens once the lease between it and an earlier write of the same
class is handed back, e.g. a custom ``IPv4`` released from under the
built-in ``IPv4`` registered over it. Undoing it would change nothing, so
its codes move to the lease below when that is the same class, and
otherwise -- the class was named by other means -- hold nothing.
Args:
stack: The name's leases, oldest first.
index: Position of the lease to retire.
"""
lease = stack.pop(index)
below = stack[index - 1] if index > 0 else None
if below is not None and below.protocol is not lease.protocol:
below = None
for target in lease.codes:
if _CODE_LEASES.get(target) is not lease:
continue
if below is None:
del _CODE_LEASES[target]
else:
below.codes.add(target)
_CODE_LEASES[target] = below
def _release_name(target: 'tuple[type, Any]') -> 'None':
"""Hand back the lease ``target``'s registration held.
Once no code holds the lease, it leaves its name's stack. On top, it puts
the name back to the entry it displaced, or removes a new name -- unless the
name was written by other means since, in which case that write stands.
Buried, it hands what it displaced to the lease above, if that lease
displaced it.
Args:
target: The ``(destination, code)`` being registered over.
"""
lease = _CODE_LEASES.pop(target, None)
if lease is None:
return
lease.codes.discard(target)
if lease.codes:
return
stack = _NAME_LEASES.get(lease.name, [])
index = next((i for i, item in enumerate(stack) if item is lease), None)
if index is None:
return
del stack[index]
if index < len(stack):
above = stack[index]
if above.displaced is lease.protocol:
above.displaced = lease.displaced
if above.displaced is above.protocol:
_fold_lease(stack, index)
elif protocol_registry.get(lease.name) is lease.protocol:
if lease.displaced is _MISSING:
protocol_registry.pop(lease.name, None)
else:
protocol_registry[lease.name] = lease.displaced
logger.debug('released protocol: %s', lease.protocol.__name__)
if not stack:
_NAME_LEASES.pop(lease.name, None)
def _register_name(protocol: 'Type[ProtocolBase]', *targets: 'tuple[type, Any]') -> 'None':
"""Name ``protocol`` in :data:`pcapkit.protocols.__proto__` for ``targets``.
The tail of every code registrar: :func:`register_protocol`, preceded by
releasing whatever each target code held before, and followed by a lease
when the write changed the name, so that registering the code back undoes
it (:issue:`1505`). A name already showing ``protocol`` -- shipped, defined
by a :class:`~pcapkit.protocols.protocol.Protocol` subclass, or registered
directly -- is not leased, so no restore takes it away; one showing it
through a lease extends that lease to ``targets``.
Args:
protocol: Protocol class just registered -- and so validated -- at ``targets``.
*targets: The ``(destination, code)`` pairs it was registered at.
"""
name = protocol.__name__.upper()
with _LEASE_LOCK:
stack = _NAME_LEASES.get(name)
if stack and stack[-1].protocol is protocol and protocol_registry.get(name) is protocol:
current = stack[-1] # type: Optional[_NameLease]
else:
current = None
for target in targets:
if current is None or _CODE_LEASES.get(target) is not current:
_release_name(target)
incumbent = protocol_registry.get(name, _MISSING)
register_protocol(protocol)
if protocol_registry.get(name) is not protocol:
return
stack = _NAME_LEASES.get(name)
if incumbent is not protocol:
lease = _NameLease(name, protocol, incumbent)
_NAME_LEASES.setdefault(name, []).append(lease)
elif stack and stack[-1].protocol is protocol:
lease = stack[-1]
else:
return
lease.codes.update(targets)
for target in targets:
_CODE_LEASES[target] = lease
#: Destination classes for each enum type a ``code=`` key can carry.
#:
#: Maps an enum type to the class(es) owning the
#: :attr:`ProtocolBase.__proto__ <pcapkit.protocols.protocol.ProtocolBase.__proto__>`
#: dispatch registry keyed by that enum type -- the "registry-of-registries" that
#: lets ``code=`` infer a destination from a key's own type, per :issue:`514`. It
#: is the same targeting :func:`register_ethertype`, :func:`register_transtype`,
#: :func:`register_linktype` and :func:`register_sctp` hard-code by hand, in one
#: table so :func:`register_protocol_code` can consult it. ``LinkType`` naming
#: more than one class is deliberate, not ambiguous -- :func:`register_linktype`
#: already fans out to each of them.
_CODE_DESTINATIONS: 'dict[type, tuple[Type[ProtocolBase], ...]]' = {
Enum_EtherType: (Link,),
Enum_TransType: (Internet,),
Enum_PayloadProtocolIdentifier: (SCTP,),
Enum_LinkType: (Frame, PCAPNG),
}
def _iter_code_targets(code: 'Any') -> 'Iterator[tuple[Type[ProtocolBase], Any]]':
"""Flatten a ``code=`` argument into ``(destination, key)`` pairs.
Args:
code: See :func:`register_protocol_code`.
Yields:
One ``(destination, key)`` pair per registration the caller asked
for -- possibly several, e.g. a single :class:`~pcapkit.const.reg.\
linktype.LinkType` member yields both :class:`Frame` and :class:`PCAPNG`.
Raises:
RegistryError: If a value has no destination it can name or infer;
see :func:`register_protocol_code`.
"""
if isinstance(code, dict):
yield from code.items()
return
if isinstance(code, (enum.Enum, aenum.Enum)):
destinations = _CODE_DESTINATIONS.get(type(code))
if destinations is None:
raise RegistryError(
f'no destination registry is known for enum type {type(code).__name__!r} '
f'(key {code!r}); pass an explicit destination, e.g. code={{SomeClass: {code!r}}}')
for destination in destinations:
yield destination, code
return
if isinstance(code, (str, bytes)):
raise RegistryError(f'code must be an enum member, a {{destination: key}} mapping, or an '
f'iterable thereof, not {code!r}')
if isinstance(code, collections.abc.Iterable):
for item in code:
yield from _iter_code_targets(item)
return
raise RegistryError(f'raw key {code!r} has no destination registry it can infer; pass an '
f'explicit destination, e.g. code={{TCP: {code!r}}}')
[docs]
def register_protocol_code(protocol: 'Type[ProtocolBase]', code: 'Any') -> 'None':
r"""Register ``protocol`` into the next-layer dispatch registry (or
registries) named by ``code``.
This is what backs the ``code`` class keyword of
:class:`~pcapkit.protocols.protocol.ProtocolBase` subclasses; it can also
be called directly to register a class that declined at class-definition
time.
``code`` is normalised into a flat sequence of targets, where each item
is either:
* an enum member, whose *type* is looked up in a small table mapping
enum type to the class(es) owning the matching ``__proto__`` --
:class:`~pcapkit.const.reg.ethertype.EtherType` to
:class:`~pcapkit.protocols.link.link.Link`,
:class:`~pcapkit.const.reg.transtype.TransType` to
:class:`~pcapkit.protocols.internet.internet.Internet`,
:class:`~pcapkit.const.sctp.payload_protocol_identifier.PayloadProtocolIdentifier`
to :class:`~pcapkit.protocols.transport.sctp.SCTP`, and
:class:`~pcapkit.const.reg.linktype.LinkType` to *both*
:class:`~pcapkit.protocols.misc.pcap.frame.Frame` **and**
:class:`~pcapkit.protocols.misc.pcapng.PCAPNG`; or
* a :class:`dict` mapping an explicit destination class to a key -- the
only form accepted for a bare :class:`int`, since e.g. a port number
does not say by itself whether it means
:class:`~pcapkit.protocols.transport.tcp.TCP` or
:class:`~pcapkit.protocols.transport.udp.UDP`;
or an iterable of either, to register ``protocol`` into several
registries from one call -- e.g. a
:class:`~pcapkit.protocols.link.l2tp.L2TP` subclass reachable both by its
IP protocol number and by a UDP port:
.. code-block:: python
register_protocol_code(L2TPv3, [TransType.L2TP, {UDP: 1701}])
The explicit mapping form is accepted for any key, even one whose type
could be inferred -- being more explicit than required is never an
error.
Note:
That example names ``L2TPv3``, which this package does not implement,
rather than :class:`~pcapkit.protocols.link.l2tpv2.L2TPv2`. It is
v3 that is genuinely reachable both ways: :rfc:`3931` §4.1.1 puts it
directly over IP on protocol 115 and §4.1.2 puts it over UDP on port
1701. :class:`L2TPv2 <pcapkit.protocols.link.l2tpv2.L2TPv2>` answers on
port 1701 only, and registering *it* at ``TransType.L2TP`` would point
the :rfc:`2661` parser at a v3-over-IP header -- see
:class:`~pcapkit.protocols.link.l2tp.L2TP` and GitHub issue :issue:`548` for
what that produced when measured.
Args:
protocol: Protocol class to register.
code: Registration key(s); see above.
Raises:
RegistryError: If a bare :class:`int` (or any other non-enum,
non-mapping value) is given without an explicit destination, or if
an enum member's type names no known destination registry --
inference refuses rather than guesses.
"""
for destination, key in _iter_code_targets(code):
destination.register(key, protocol)
logger.debug('registered %s into %s.__proto__: %s', protocol.__name__,
destination.__name__, key)
###############################################################################
# Top-Level Registries
###############################################################################
@overload
def register_linktype(code: 'LinkType', module: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]') -> 'None': ...
@overload
def register_linktype(code: 'LinkType', module: 'str', class_: 'str') -> 'None': ...
[docs]
def register_linktype(code: 'LinkType', module: 'str | ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]',
class_: 'str | NullType' = NULL) -> 'None':
r"""Register a new protocol class.
Notes:
The fully qualified class name of the new protocol class
should be ``{module}.{class_}``.
The function will register the given protocol class to the
following registries:
- :data:`pcapkit.protocols.misc.pcap.frame.Frame.__proto__`
- :data:`pcapkit.protocols.misc.pcapng.PCAPNG.__proto__`
Arguments:
code: protocol code as in :class:`~pcapkit.const.reg.linktype.LinkType`
module: module name or module descriptor or a
:class:`~pcapkit.protocols.protocol.ProtocolBase` subclass
class\_: class name
See Also:
* :func:`pcapkit.foundation.registry.protocols.register_pcap`
* :func:`pcapkit.foundation.registry.protocols.register_pcapng`
"""
if isinstance(module, str):
module = cast('ModuleDescriptor[ProtocolBase]', ModuleDescriptor(module, class_))
Frame.register(code, module)
PCAPNG.register(code, module)
logger.debug('registered linktype protocol: %s', code.name)
# register protocol to protocol registry
if isinstance(module, ModuleDescriptor):
module = module.klass
_register_name(module, (Frame, code), (PCAPNG, code))
@overload
def register_pcap(code: 'LinkType', module: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]') -> 'None': ...
@overload
def register_pcap(code: 'LinkType', module: 'str', class_: 'str') -> 'None': ...
# NOTE: pcapkit.protocols.misc.pcap.frame.Frame.__proto__
[docs]
def register_pcap(code: 'LinkType', module: 'str | ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]',
class_: 'str | NullType' = NULL) -> 'None':
r"""Register a new protocol class.
Notes:
The fully qualified class name of the new protocol class
should be ``{module}.{class_}``.
The function will register the given protocol class to the
:data:`pcapkit.protocols.misc.pcap.frame.Frame.__proto__` registry.
Arguments:
code: protocol code as in :class:`~pcapkit.const.reg.linktype.LinkType`
module: module name or module descriptor or a
:class:`~pcapkit.protocols.protocol.ProtocolBase` subclass
class\_: class name
"""
if isinstance(module, str):
module = cast('ModuleDescriptor[ProtocolBase]', ModuleDescriptor(module, class_))
Frame.register(code, module)
logger.debug('registered PCAP linktype protocol: %s', code.name)
# register protocol to protocol registry
if isinstance(module, ModuleDescriptor):
module = module.klass
_register_name(module, (Frame, code))
@overload
def register_pcapng(code: 'LinkType', module: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]') -> 'None': ...
@overload
def register_pcapng(code: 'LinkType', module: 'str', class_: 'str') -> 'None': ...
# NOTE: pcapkit.protocols.misc.pcapng.PCAPNG.__proto__
[docs]
def register_pcapng(code: 'LinkType', module: 'str | ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]',
class_: 'str | NullType' = NULL) -> 'None':
r"""Register a new protocol class.
Notes:
The fully qualified class name of the new protocol class
should be ``{module}.{class_}``.
The function will register the given protocol class to the
:data:`pcapkit.protocols.misc.pcapng.PCAPNG.__proto__` registry.
Arguments:
code: protocol code as in :class:`~pcapkit.const.reg.linktype.LinkType`
module: module name or module descriptor or a
:class:`~pcapkit.protocols.protocol.ProtocolBase` subclass
class\_: class name
"""
if isinstance(module, str):
module = cast('ModuleDescriptor[ProtocolBase]', ModuleDescriptor(module, class_))
PCAPNG.register(code, module)
logger.debug('registered PCAP-NG linktype protocol: %s', code.name)
# register protocol to protocol registry
if isinstance(module, ModuleDescriptor):
module = module.klass
_register_name(module, (PCAPNG, code))
###############################################################################
# Link Layer Registries
###############################################################################
@overload
def register_ethertype(code: 'EtherType', module: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]') -> 'None': ...
@overload
def register_ethertype(code: 'EtherType', module: 'str', class_: 'str') -> 'None': ...
# NOTE: pcapkit.protocols.link.link.Link.__proto__
[docs]
def register_ethertype(code: 'EtherType', module: 'str | ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]',
class_: 'str | NullType' = NULL) -> 'None':
r"""Register a new protocol class.
Notes:
The fully qualified class name of the new protocol class
should be ``{module}.{class_}``.
The function will register the given protocol class to the
:data:`pcapkit.protocols.link.link.Link.__proto__` registry.
Arguments:
code: protocol code as in :class:`~pcapkit.const.reg.ethertype.EtherType`
module: module name or module descriptor or a
:class:`~pcapkit.protocols.protocol.ProtocolBase` subclass
class\_: class name
"""
if isinstance(module, str):
module = cast('ModuleDescriptor[ProtocolBase]', ModuleDescriptor(module, class_))
Link.register(code, module)
logger.debug('registered ethertype protocol: %s', code.name)
# register protocol to protocol registry
if isinstance(module, ModuleDescriptor):
module = module.klass
_register_name(module, (Link, code))
###############################################################################
# Internet Layer Registries
###############################################################################
@overload
def register_transtype(code: 'TransType', module: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]') -> 'None': ...
@overload
def register_transtype(code: 'TransType', module: 'str', class_: 'str') -> 'None': ...
# NOTE: pcapkit.protocols.internet.internet.Internet.__proto__
[docs]
def register_transtype(code: 'TransType', module: 'str | ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]',
class_: 'str | NullType' = NULL) -> 'None':
r"""Register a new protocol class.
Notes:
The fully qualified class name of the new protocol class
should be ``{module}.{class_}``.
The function will register the given protocol class to the
:data:`pcapkit.protocols.internet.internet.Internet.__proto__` registry.
Arguments:
code: protocol code as in :class:`~pcapkit.const.reg.transtype.TransType`
module: module name or module descriptor or a
:class:`~pcapkit.protocols.protocol.ProtocolBase` subclass
class\_: class name
"""
if isinstance(module, str):
module = cast('ModuleDescriptor[ProtocolBase]', ModuleDescriptor(module, class_))
Internet.register(code, module)
logger.debug('registered transtype protocol: %s', code.name)
# register protocol to protocol registry
if isinstance(module, ModuleDescriptor):
module = module.klass
_register_name(module, (Internet, code))
# NOTE: pcapkit.protocols.internet.ipv4.IPv4.__option__
[docs]
def register_ipv4_option(code: 'IPv4_OptionNumber', meth: 'str | tuple[IPv4_OptionParser, IPv4_OptionConstructor]', *,
schema: 'Optional[Type[Schema_IPv4_Option]]' = None) -> 'None':
"""Register an option parser.
The function will register the given option parser to the
:data:`pcapkit.protocols.internet.ipv4.IPv4.__option__` registry.
Args:
code: :class:`IPv4 <pcapkit.protocols.internet.ipv4.IPv4>` option code as
in :class:`~pcapkit.const.ipv4.option_number.OptionNumber`.
meth: Method name or callable to parse and/or construct the option.
schema: :class:`~pcapkit.protocols.schema.schema.Schema` class for the option.
It should be a subclass of :class:`pcapkit.protocols.schema.internet.ipv4.Option`.
"""
if isinstance(meth, str) and not hasattr(IPv4, f'_read_opt_{meth}'):
raise RegistryError('method must be a valid IPv4 option parser function')
IPv4.register_option(code, meth)
if schema is not None:
Schema_IPv4_Option.register(code, schema)
logger.debug('registered IPv4 option parser: %s', code.name)
# NOTE: pcapkit.protocols.internet.hip.HIP.__parameter__
[docs]
def register_hip_parameter(code: 'HIP_Parameter', meth: 'str | tuple[HIP_ParameterParser, HIP_ParameterConstructor]', *,
schema: 'Optional[Type[Schema_HIP_Parameter]]' = None) -> 'None':
"""Register a parameter parser.
The function will register the given parameter parser to the
:data:`pcapkit.protocols.internet.hip.HIP.__parameter__` registry.
Args:
code: :class:`~pcapkit.protocols.internet.hip.HIP` parameter code as
in :class:`~pcapkit.const.hip.parameter.Parameter`.
meth: Method name or callable to parse and/or construct the parameter.
schema: :class:`~pcapkit.protocols.schema.schema.Schema` class for the parameter.
It should be a subclass of :class:`pcapkit.protocols.schema.internet.hip.Parameter`.
"""
if isinstance(meth, str) and not hasattr(HIP, f'_read_param_{meth}'):
raise RegistryError('method must be a valid HIP parameter parser function')
HIP.register_parameter(code, meth)
if schema is not None:
Schema_HIP_Parameter.register(code, schema)
logger.debug('registered HIP parameter parser: %s', code.name)
# NOTE: pcapkit.protocols.internet.hopopt.HOPOPT.__option__
[docs]
def register_hopopt_option(code: 'IPv6_Option', meth: 'str | tuple[HOPOPT_OptionParser, HOPOPT_OptionConstructor]', *,
schema: 'Optional[Type[Schema_HOPOPT_Option]]' = None) -> 'None':
"""Register an option parser.
The function will register the given option parser to the
:data:`pcapkit.protocols.internet.hopopt.HOPOPT.__option__` registry.
Args:
code: :class:`~pcapkit.protocols.internet.hopopt.HOPOPT` option code as
in :class:`~pcapkit.const.ipv6.option.Option`.
meth: Method name or callable to parse and/or construct the option.
schema: :class:`~pcapkit.protocols.schema.schema.Schema` class for the option.
It should be a subclass of :class:`pcapkit.protocols.schema.internet.hopopt.Option`.
"""
if isinstance(meth, str) and not hasattr(HOPOPT, f'_read_opt_{meth}'):
raise RegistryError('method must be a valid HOPOPT option parser function')
HOPOPT.register_option(code, meth)
if schema is not None:
Schema_HOPOPT_Option.register(code, schema)
logger.debug('registered HOPOPT option parser: %s', code.name)
# NOTE: pcapkit.protocols.internet.ipv6_opts.IPv6_Opts.__option__
[docs]
def register_ipv6_opts_option(code: 'IPv6_Option', meth: 'str | tuple[IPv6_Opts_OptionParser, IPv6_Opts_OptionConstructor]', *,
schema: 'Optional[Type[Schema_IPv6_Opts_Option]]' = None) -> 'None':
"""Register an option parser.
The function will register the given option parser to the
:data:`pcapkit.protocols.internet.ipv6_opts.IPv6_Opts.__option__` registry.
Args:
code: :class:`IPv6-Opts <pcapkit.protocols.internet.ipv6_opts.IPv6_Opts>`
option code as in :class:`~pcapkit.const.ipv6.option.Option`.
meth: Method name or callable to parse and/or construct the option.
schema: :class:`~pcapkit.protocols.schema.schema.Schema` class for the option.
It should be a subclass of :class:`pcapkit.protocols.schema.internet.ipv6_opts.Option`.
"""
if isinstance(meth, str) and not hasattr(IPv6_Opts, f'_read_opt_{meth}'):
raise RegistryError('method must be a valid IPv6-Opts option parser function')
IPv6_Opts.register_option(code, meth)
if schema is not None:
Schema_IPv6_Opts_Option.register(code, schema)
logger.debug('registered IPv6-Opts option parser: %s', code.name)
# NOTE: pcapkit.protocols.internet.ipv6_route.IPv6_Route.__routing__
[docs]
def register_ipv6_route_routing(code: 'IPv6_Routing', meth: 'str | tuple[IPv6_Route_TypeParser, IPv6_Route_TypeConstructor]', *,
schema: 'Optional[Type[Schema_IPv6_Route_RoutingType]]' = None) -> 'None':
"""Register a routing data parser.
The function will register the given routing data parser to the
:data:`pcapkit.protocols.internet.ipv6_route.IPv6_Route.__routing__` registry.
Args:
code: :class:`IPv6-Route <pcapkit.protocols.internet.ipv6_route.IPv6_Route>`
data type code as in :class:`~pcapkit.const.ipv6.routing.Routing`.
meth: Method name or callable to parse and/or construct the data.
schema: :class:`~pcapkit.protocols.schema.schema.Schema` class for the routing data.
It should be a subclass of :class:`pcapkit.protocols.schema.internet.ipv6_route.RoutingType`.
"""
if isinstance(meth, str) and not hasattr(IPv6_Route, f'_read_data_type_{meth}'):
raise RegistryError('method must be a valid IPv6-Route routing data parser function')
IPv6_Route.register_routing(code, meth)
if schema is not None:
Schema_IPv6_Route_RoutingType.register(code, schema)
logger.debug('registered IPv6-Route routing data parser: %s', code.name)
# NOTE: pcapkit.protocols.internet.mh.MH.__message__
[docs]
def register_mh_message(code: 'MH_Packet', meth: 'str | tuple[MH_PacketParser, MH_PacketConstructor]', *,
schema: 'Optional[Type[Schema_MH_Packet]]' = None) -> 'None':
"""Register an MH message type parser.
The function will register the given message type parser to the
:data:`pcapkit.protocols.internet.mh.MH.__message__` registry.
Args:
code: :class:`~pcapkit.protocols.internet.mh.MH`
message type code as in :class:`~pcapkit.const.mh.packet.Packet`.
meth: Method name or callable to parse and/or construct the data.
schema: :class:`~pcapkit.protocols.schema.schema.Schema` class for the message type.
It should be a subclass of :class:`pcapkit.protocols.schema.internet.mh.Packet`.
"""
if isinstance(meth, str) and not hasattr(MH, f'_read_msg_{meth}'):
raise RegistryError('method must be a valid MH message type parser function')
MH.register_message(code, meth)
if schema is not None:
Schema_MH_Packet.register(code, schema)
logger.debug('registered MH message type parser: %s', code.name)
# NOTE: pcapkit.protocols.internet.mh.MH.__option__
[docs]
def register_mh_option(code: 'MH_Option', meth: 'str | tuple[MH_OptionParser, MH_OptionConstructor]', *,
schema: 'Optional[Type[Schema_MH_Option]]' = None) -> 'None':
"""Register an MH option parser.
The function will register the given option parser to the
:data:`pcapkit.protocols.internet.mh.MH.__option__` registry.
Args:
code: :class:`~pcapkit.protocols.internet.mh.MH`
option code as in :class:`~pcapkit.const.mh.option.Option`.
meth: Method name or callable to parse and/or construct the data.
schema: :class:`~pcapkit.protocols.schema.schema.Schema` class for the option.
It should be a subclass of :class:`pcapkit.protocols.schema.internet.mh.Option`.
"""
if isinstance(meth, str) and not hasattr(MH, f'_read_opt_{meth}'):
raise RegistryError('method must be a valid MH option parser function')
MH.register_option(code, meth)
if schema is not None:
Schema_MH_Option.register(code, schema)
logger.debug('registered MH option parser: %s', code.name)
# NOTE: pcapkit.protocols.internet.mh.MH.__extension__
[docs]
def register_mh_extension(code: 'MH_CGAExtension', meth: 'str | tuple[MH_ExtensionParser, MH_ExtensionConstructor]', *,
schema: 'Optional[Type[Schema_MH_CGAExtension]]' = None) -> 'None':
"""Register a CGA extension parser.
The function will register the given CGA extension parser to the
:data:`pcapkit.protocols.internet.mh.MH.__extension__` registry.
Args:
code: :class:`~pcapkit.protocols.internet.mh.MH`
CGA extension code as in :class:`~pcapkit.const.mh.cga_extension.CGAExtension`.
meth: Method name or callable to parse and/or construct the data.
schema: :class:`~pcapkit.protocols.schema.schema.Schema` class for the CGA extension.
It should be a subclass of :class:`pcapkit.protocols.schema.internet.mh.CGAExtension`.
.. versionchanged:: 1.5.0
``schema`` is keyword-only, as in the other ``register_*`` functions
that take one (:issue:`516`).
"""
if isinstance(meth, str) and not hasattr(MH, f'_read_ext_{meth}'):
raise RegistryError('method must be a valid MH CGA extension function')
MH.register_extension(code, meth)
if schema is not None:
Schema_MH_CGAExtension.register(code, schema)
logger.debug('registered MH CGA extension: %s', code.name)
###############################################################################
# Transport Layer Registries
###############################################################################
@overload
def register_apptype(code: 'int', module: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]', *transport: 'TransportProtocol | str') -> 'None': ...
@overload
def register_apptype(code: 'Enum_AppType', module: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]', *transport: 'TransportProtocol | str') -> 'None': ...
@overload
def register_apptype(code: 'int', module: 'str', class_: 'str', *transport: 'TransportProtocol | str') -> 'None': ...
@overload
def register_apptype(code: 'Enum_AppType', module: 'str', class_: 'str', *transport: 'TransportProtocol | str') -> 'None': ...
[docs]
def register_apptype(code: 'int | Enum_AppType', module: 'str | ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]',
class_: 'str | TransportProtocol | NullType' = NULL,
*transport: 'TransportProtocol | str') -> 'None':
r"""Register a new protocol class.
Notes:
The fully qualified class name of the new protocol class
should be ``{module}.{class_}``.
The function will register the given protocol class to the
:data:`pcapkit.protocols.transport.tcp.TCP.__proto__` and/or
:data:`pcapkit.protocols.transport.udp.UDP.__proto__` registry.
Arguments:
code: port number
module: module name or module descriptor or a
:class:`~pcapkit.protocols.protocol.ProtocolBase` subclass
class\_: class name, meaningful only when ``module`` is a :class:`str`.
Positional, at the same position as the sibling ``register_*``
functions -- but unlike them, a third positional argument here is
not always ``class_``: which parameter it binds to is decided by
``type(module)`` alone, never by what the value looks like.
When ``module`` is **not** a :class:`str`, whatever is given here
is treated as the first ``*transport`` element instead --
prepended, so argument order is preserved -- and ``class_`` itself
stays unset; ``register_apptype(80, Unit, TransportProtocol.udp)``
therefore reaches ``transport`` rather than being silently
swallowed. This is not a heuristic on the value: with a
:class:`str` ``module``, a :class:`str` third argument is
**always** a class name -- ``register_apptype(80, 'a.b', 'tcp')``
registers a class literally named ``'tcp'`` -- and naming a
transport alongside a :class:`str` module takes the fourth
position onward instead.
*transport: transport protocols to register ``code`` under, named **one
at a time**, each as a :class:`~pcapkit.const.reg.apptype.apptype.TransportProtocol`
member or as that member's ``name`` (case-insensitively, e.g. ``'tcp'``,
``'TCP'`` or ``'Tcp'`` alike); the two forms are coerced to the same
member and behave identically,
including which error they raise. Where none is given and ``code`` is
an :class:`~pcapkit.const.reg.apptype.apptype.AppType` member, the
member's own :attr:`~pcapkit.const.reg.apptype.apptype.AppType.proto`
is used, which names exactly the registry the member lives in. A bare
:class:`int` carries no such default and so requires at least one.
Raises:
pcapkit.utilities.exceptions.RegistryError: If no transport protocol is
given for a bare :class:`int`; if one of those given (member or name)
names no port-keyed registry; or if a ``str`` matches no
:class:`~pcapkit.const.reg.apptype.apptype.TransportProtocol` member's
``name``. A composite such as ``tcp | udp``, or its string form
``'tcp|udp'``, names two and is refused for the same reason: one call
registers under one transport protocol.
pcapkit.utilities.exceptions.ProtocolError: Raised lazily, from
:attr:`ModuleDescriptor.klass <pcapkit.corekit.module.ModuleDescriptor.klass>`,
when ``module`` is a :class:`str` and either ``class_`` names no
attribute of it, or ``class_`` was never given at all -- the
latter distinguished from the former rather than reported as a
missing attribute named ``'(null)'``. See GitHub issues :issue:`832` and
:issue:`833`.
Important:
:class:`~pcapkit.protocols.transport.sctp.SCTP` is deliberately **not**
one of the registries this writes to, so naming ``sctp`` here is an error
rather than a no-op. Its
:data:`~pcapkit.protocols.transport.sctp.SCTP.__proto__` registry is
keyed by the DATA chunk's payload protocol identifier rather than by
port number, so writing a port number into it would dispatch on a
number from the wrong registry. Use
:func:`pcapkit.foundation.registry.protocols.register_sctp` instead.
See Also:
* :func:`pcapkit.foundation.registry.protocols.register_tcp`
* :func:`pcapkit.foundation.registry.protocols.register_udp`
* :func:`pcapkit.foundation.registry.protocols.register_sctp`
.. versionremoved:: 1.3.1
``register_port``, which this replaces with no alias.
.. versionchanged:: 1.5.0
The keyword-only ``proto`` is replaced by ``*transport``, one transport
protocol per argument, with no alias (:issue:`806`).
"""
# NOTE: ``class_`` sits at the sibling position -- positional, ahead of
# ``*transport`` -- per maintainer ruling, so a third positional that is
# really a transport would be swallowed as a class name. That is
# disambiguated here by ``type(module)`` alone, never by what ``class_``
# itself looks like. A ``str`` module means
# ``class_`` really is a class name, so it is left untouched -- including
# when it happens to be a string like ``'tcp'``, which is not sniffed for
# looking like a transport. Any other module type means the third
# positional was never a class name to begin with, so it is moved to the
# front of ``transport`` -- prepended, so argument order survives -- and
# ``class_`` is reset to ``NULL`` so the ``ModuleDescriptor`` branch below
# never sees it.
if not isinstance(module, str) and class_ is not NULL:
transport = (cast('TransportProtocol | str', class_),) + transport
class_ = NULL
# NOTE: a ``str`` transport is coerced to its ``TransportProtocol`` member
# up front, before any registry work, so every downstream use -- the
# ``code.proto`` default and the registries lookup below -- sees members
# only. Resolution is by the member's own ``name``, case-insensitively --
# a ruling given in review of the work for #806 -- via ``__members__``
# directly rather than ``TransportProtocol[name]``: this function's
# contract is :exc:`~pcapkit.utilities.exceptions.RegistryError` for
# anything unrecognised, composite-spelled or not, and ``__getitem__``
# raises a bare :exc:`KeyError` on a miss instead of that -- both
# ``TransportProtocol['tcp|udp']`` and ``TransportProtocol['bogus']`` do,
# because ``TransportProtocol`` is not an ``IntFlag``, under which the
# first would silently compose into the value ``3`` (#808).
# ``__members__.get(...)`` lets this function
# raise its own exception on a miss instead of letting ``__getitem__``'s
# propagate, and lowercasing does not turn ``'tcp|udp'`` into a member
# name either way. Anything that is neither a ``str`` nor a
# ``TransportProtocol`` member is rejected here too, rather than falling
# through to the registries lookup below: a bare ``int`` still hashes
# and compares equal to its matching member -- ``TransportProtocol``
# being an ``IntEnum`` rather than an ``IntFlag`` changes neither -- so
# ``registries.get(1)`` resolves to ``TCP`` just as
# ``registries.get(TransportProtocol.tcp)`` does, and would otherwise
# register the port before the ``proto.name`` access below it
# raised -- leaving the port written into ``TCP.__proto__`` with
# ``register_protocol`` never called.
coerced: 'list[TransportProtocol]' = []
for proto in transport:
if isinstance(proto, str):
# NOTE: ``proto`` itself -- not the lowercased lookup key -- is
# what the error message formats with ``!r``, so a mismatched
# name such as ``'TCP '`` still shows its own casing and its
# trailing space rather than the lowercased form.
member = TransportProtocol.__members__.get(proto.lower())
if member is None:
raise RegistryError(f'unknown transport protocol: {proto!r}')
proto = cast('TransportProtocol', member)
elif not isinstance(proto, TransportProtocol):
raise RegistryError(f'unknown transport protocol: {proto!r}')
coerced.append(proto)
transport = tuple(coerced)
# NOTE: the member's own ``proto`` is the single transport protocol of the
# registry it lives in -- GitHub issue #806 -- so it names one destination
# rather than fanning out across every transport IANA gave the service. A
# fan-out would make ``register_apptype(TCP.http, Dummy)`` displace the UDP
# handler for port 80 as well, which no caller naming the TCP member asked
# for.
if not transport:
if not isinstance(code, Enum_AppType):
raise RegistryError(f'no transport protocol given for port {code}; name each '
'transport protocol to register under')
transport = (code.proto,)
if isinstance(code, Enum_AppType):
code = code.port
if isinstance(module, str):
module = cast('ModuleDescriptor[ProtocolBase]', ModuleDescriptor(module, class_))
registries = cast('dict[TransportProtocol, Type[ProtocolBase]]', {
TransportProtocol.tcp: TCP,
TransportProtocol.udp: UDP,
})
targets = [] # type: list[tuple[type, Any]]
for proto in transport:
# NOTE: a composite is not a key here, so ``tcp | udp`` is rejected
# rather than quietly registering under both. Naming them one at a time
# is what the varargs are for.
cls = registries.get(proto)
if cls is None:
raise RegistryError(f'unknown transport protocol: {proto.name}')
cls.register(code, module)
targets.append((cls, code))
logger.debug('registered %s port: %s', proto.name, code)
# register protocol to protocol registry
if isinstance(module, ModuleDescriptor):
module = module.klass
_register_name(module, *targets)
@overload
def register_tcp(code: 'int | Enum_AppType', module: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]') -> 'None': ...
@overload
def register_tcp(code: 'int | Enum_AppType', module: 'str', class_: 'str') -> 'None': ...
# NOTE: pcapkit.protocols.transport.tcp.TCP.__proto__
[docs]
def register_tcp(code: 'int | Enum_AppType', module: 'str | ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]',
class_: 'str | NullType' = NULL) -> 'None':
r"""Register a new protocol class.
Notes:
The fully qualified class name of the new protocol class
should be ``{module}.{class_}``.
The function will register the given protocol class to the
:data:`pcapkit.protocols.transport.tcp.TCP.__proto__` registry.
Arguments:
code: port number
module: module name or module descriptor or a
:class:`~pcapkit.protocols.protocol.ProtocolBase` subclass
class\_: class name
"""
if isinstance(code, Enum_AppType):
code = code.port
if isinstance(module, str):
module = cast('ModuleDescriptor[ProtocolBase]', ModuleDescriptor(module, class_))
TCP.register(code, module)
logger.debug('registered TCP port: %s', code)
# register protocol to protocol registry
if isinstance(module, ModuleDescriptor):
module = module.klass
_register_name(module, (TCP, code))
# NOTE: pcapkit.protocols.transport.tcp.TCP.__option__
[docs]
def register_tcp_option(code: 'TCP_Option', meth: 'str | tuple[TCP_OptionParser, TCP_OptionConstructor]', *,
schema: 'Optional[Type[Schema_TCP_Option]]' = None) -> 'None':
"""Register an option parser.
The function will register the given option parser to the
:data:`pcapkit.protocols.transport.tcp.TCP.__option__` registry.
Args:
code: :class:`~pcapkit.protocols.transport.tcp.TCP` option code as in
:class:`~pcapkit.const.tcp.option.Option`.
meth: Method name or callable to parse and/or construct the option.
schema: :class:`~pcapkit.protocols.schema.schema.Schema` class for the option.
It should be a subclass of :class:`pcapkit.protocols.schema.transport.tcp.Option`.
"""
if isinstance(meth, str) and not hasattr(TCP, f'_read_mode_{meth}'):
raise RegistryError('method must be a TCP option parser function')
TCP.register_option(code, meth)
if schema is not None:
Schema_TCP_Option.register(code, schema)
logger.debug('registered TCP option parser: %s', code.name)
# NOTE: pcapkit.protocols.transport.tcp.TCP.__mp_option__
[docs]
def register_tcp_mp_option(code: 'TCP_MPTCPOption', meth: 'str | tuple[TCP_MPOptionParser, TCP_MPOptionConstructor]', *,
schema: 'Optional[Type[Schema_TCP_MPTCP]]' = None) -> 'None':
"""Register an MPTCP option parser.
The function will register the given option parser to the
:data:`pcapkit.protocols.transport.tcp.TCP.__mp_option__` registry.
Args:
code: Multipath :class:`~pcapkit.protocols.transport.tcp.TCP` option code as in
:class:`~pcapkit.const.tcp.mp_tcp_option.MPTCPOption`.
meth: Method name or callable to parse and/or construct the option.
schema: :class:`~pcapkit.protocols.schema.schema.Schema` class for the option.
It should be a subclass of :class:`pcapkit.protocols.schema.transport.tcp.MPTCP`.
"""
if isinstance(meth, str) and not hasattr(TCP, f'_read_mptcp_{meth}'):
raise RegistryError('method must be a MPTCP option parser function')
TCP.register_mp_option(code, meth)
if schema is not None:
Schema_TCP_MPTCP.register(code, schema)
logger.debug('registered MPTCP option parser: %s', code.name)
@overload
def register_udp(code: 'int | Enum_AppType', module: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]') -> 'None': ...
@overload
def register_udp(code: 'int | Enum_AppType', module: 'str', class_: 'str') -> 'None': ...
# NOTE: pcapkit.protocols.transport.udp.UDP.__proto__
[docs]
def register_udp(code: 'int | Enum_AppType', module: 'str | ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]',
class_: 'str | NullType' = NULL) -> 'None':
r"""Register a new protocol class.
Notes:
The fully qualified class name of the new protocol class
should be ``{module}.{class_}``.
The function will register the given protocol class to the
:data:`pcapkit.protocols.transport.udp.UDP.__proto__` registry.
Arguments:
code: port number
module: module name or module descriptor or a
:class:`~pcapkit.protocols.protocol.ProtocolBase` subclass
class\_: class name
"""
if isinstance(code, Enum_AppType):
code = code.port
if isinstance(module, str):
module = cast('ModuleDescriptor[ProtocolBase]', ModuleDescriptor(module, class_))
UDP.register(code, module)
logger.debug('registered UDP port: %s', code)
# register protocol to protocol registry
if isinstance(module, ModuleDescriptor):
module = module.klass
_register_name(module, (UDP, code))
@overload
def register_sctp(code: 'int | SCTP_PayloadProtocolIdentifier', module: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]') -> 'None': ...
@overload
def register_sctp(code: 'int | SCTP_PayloadProtocolIdentifier', module: 'str', class_: 'str') -> 'None': ...
# NOTE: pcapkit.protocols.transport.sctp.SCTP.__proto__
[docs]
def register_sctp(code: 'int | SCTP_PayloadProtocolIdentifier', module: 'str | ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]',
class_: 'str | NullType' = NULL) -> 'None':
r"""Register a new protocol class.
Notes:
The fully qualified class name of the new protocol class
should be ``{module}.{class_}``.
The function will register the given protocol class to the
:data:`pcapkit.protocols.transport.sctp.SCTP.__proto__` registry.
Arguments:
code: payload protocol identifier (PPID), as in
:class:`~pcapkit.const.sctp.payload_protocol_identifier.PayloadProtocolIdentifier`
module: module name or module descriptor or a
:class:`~pcapkit.protocols.protocol.ProtocolBase` subclass
class\_: class name
Important:
Unlike :func:`register_tcp` and :func:`register_udp`, ``code`` is a
*payload protocol identifier* taken from the DATA chunk, **not** a port
number: SCTP names its upper layer per DATA chunk rather than per
association. See :rfc:`9260#section-3.3.1`.
"""
if isinstance(module, str):
module = cast('ModuleDescriptor[ProtocolBase]', ModuleDescriptor(module, class_))
SCTP.register(code, module)
logger.debug('registered SCTP payload protocol identifier: %s', code)
# register protocol to protocol registry
if isinstance(module, ModuleDescriptor):
module = module.klass
_register_name(module, (SCTP, code))
###############################################################################
# Application Layer Registries
###############################################################################
# NOTE: pcapkit.protocols.application.httpv2.HTTPv2.__frame__
[docs]
def register_http_frame(code: 'HTTP_Frame', meth: 'str | tuple[HTTP_FrameParser, HTTP_FrameConstructor]', *,
schema: 'Optional[Type[Schema_HTTP_FrameType]]' = None) -> 'None':
"""Register a frame parser.
The function will register the given frame parser to the
:data:`pcapkit.protocols.application.httpv2.HTTP.__frame__` registry.
Args:
code: :class:`HTTP/2 <pcapkit.protocols.application.httpv2.HTTP>` frame type
code as in :class:`~pcapkit.const.http.frame.Frame`.
meth: Method name or callable to parse and/or construct the frame.
schema: :class:`~pcapkit.protocols.schema.schema.Schema` class for the frame.
It should be a subclass of :class:`pcapkit.protocols.schema.application.httpv2.FrameType`.
"""
if isinstance(meth, str) and not hasattr(HTTPv2, f'_read_http_{meth}'):
raise RegistryError('method must be a frame parser function')
HTTPv2.register_frame(code, meth)
if schema is not None:
Schema_HTTP_FrameType.register(code, schema)
logger.debug('registered HTTP/2 frame parser: %s', code.name)
###############################################################################
# Miscellaneous Protocol Registries
###############################################################################
# NOTE: pcapkit.protocols.misc.pcapng.PCAPNG.__block__
[docs]
def register_pcapng_block(code: 'PCAPNG_BlockType', meth: 'str | tuple[PCAPNG_BlockParser, PCAPNG_BlockConstructor]', *,
schema: 'Optional[Type[Schema_PCAPNG_BlockType]]' = None) -> 'None':
"""Register a block parser.
The function will register the given block parser to the
:data:`pcapkit.protocols.misc.pcapng.PCAPNG.__block__` registry.
Args:
code: :class:`PCAPNG <pcapkit.protocols.misc.pcapng.PCAPNG>` block type
code as in :class:`~pcapkit.const.pcapng.block_type.BlockType`.
meth: Method name or callable to parse and/or construct the block.
schema: :class:`~pcapkit.protocols.schema.schema.Schema` class for the block.
It should be a subclass of :class:`pcapkit.protocols.schema.misc.pcapng.BlockType`.
"""
if isinstance(meth, str) and not hasattr(PCAPNG, f'_read_block_{meth}'):
raise RegistryError('method must be a block parser function')
PCAPNG.register_block(code, meth)
if schema is not None:
Schema_PCAPNG_BlockType.register(code, schema)
logger.debug('registered PCAP-NG block parser: %s', code.name)
# NOTE: pcapkit.protocols.misc.pcapng.PCAPNG.__option__
[docs]
def register_pcapng_option(code: 'PCAPNG_OptionType', meth: 'str | tuple[PCAPNG_OptionParser, PCAPNG_OptionConstructor]', *,
schema: 'Optional[Type[Schema_PCAPNG_Option]]' = None) -> 'None':
"""Register an option parser.
The function will register the given option parser to the
:data:`pcapkit.protocols.misc.pcapng.PCAPNG.__option__` registry.
Args:
code: :class:`PCAPNG <pcapkit.protocols.misc.pcapng.PCAPNG>` option type
code as in :class:`~pcapkit.const.pcapng.option_type.OptionType`.
meth: Method name or callable to parse and/or construct the option.
schema: :class:`~pcapkit.protocols.schema.schema.Schema` class for the option.
It should be a subclass of :class:`pcapkit.protocols.schema.misc.pcapng.Option`.
"""
if isinstance(meth, str) and not hasattr(PCAPNG, f'_read_option_{meth}'):
raise RegistryError('method must be a option parser function')
PCAPNG.register_option(code, meth)
if schema is not None:
Schema_PCAPNG_Option.register(code, schema)
logger.debug('registered PCAP-NG option parser: %s', code.name)
# NOTE: pcapkit.protocols.misc.pcapng.PCAPNG.__record__
[docs]
def register_pcapng_record(code: 'PCAPNG_RecordType', meth: 'str | tuple[PCAPNG_RecordParser, PCAPNG_RecordConstructor]', *,
schema: 'Optional[Type[Schema_PCAPNG_NameResolutionRecord]]' = None) -> 'None':
"""Register a name resolution record parser.
The function will register the given name resolution record parser to the
:data:`pcapkit.protocols.misc.pcapng.PCAPNG.__record__` registry.
Args:
code: :class:`PCAPNG <pcapkit.protocols.misc.pcapng.PCAPNG>` name
resolution record type code as in :class:`~pcapkit.const.pcapng.record_type.RecordType`.
meth: Method name or callable to parse and/or construct the name
resolution record.
schema: :class:`~pcapkit.protocols.schema.schema.Schema` class for the name resolution record.
It should be a subclass of :class:`pcapkit.protocols.schema.misc.pcapng.NameResolutionRecord`.
"""
if isinstance(meth, str) and not hasattr(PCAPNG, f'_read_record_{meth}'):
raise RegistryError('method must be a name resolution record parser function')
PCAPNG.register_record(code, meth)
if schema is not None:
Schema_PCAPNG_NameResolutionRecord.register(code, schema)
logger.debug('registered PCAP-NG name resolution record parser: %s', code.name)
# NOTE: pcapkit.protocols.misc.pcapng.PCAPNG.__secrets__
[docs]
def register_pcapng_secrets(code: 'PCAPNG_SecretsType', meth: 'str | tuple[PCAPNG_SecretsParser, PCAPNG_SecretsConstructor]', *,
schema: 'Optional[Type[Schema_PCAPNG_DSBSecrets]]' = None) -> 'None':
"""Register a decryption secrets parser.
The function will register the given decryption secrets parser to the
:data:`pcapkit.protocols.misc.pcapng.PCAPNG.__secrets__` registry.
Args:
code: :class:`PCAPNG <pcapkit.protocols.misc.pcapng.PCAPNG>` decryption
secrets type code as in :class:`~pcapkit.const.pcapng.secrets_type.SecretsType`.
meth: Method name or callable to parse and/or construct the decryption secrets.
schema: :class:`~pcapkit.protocols.schema.schema.Schema` class for the decryption secrets.
It should be a subclass of :class:`pcapkit.protocols.schema.misc.pcapng.DSBSecrets`.
"""
if isinstance(meth, str) and not hasattr(PCAPNG, f'_read_secrets_{meth}'):
raise RegistryError('method must be a decryption secrets parser function')
PCAPNG.register_secrets(code, meth)
if schema is not None:
Schema_PCAPNG_DSBSecrets.register(code, schema)
logger.debug('registered PCAP-NG decryption secrets parser: %s', code.name)