diff --git a/docs/source/pcapkit/foundation/registry.rst b/docs/source/pcapkit/foundation/registry.rst index ac7cd1c2fc..a002428874 100644 --- a/docs/source/pcapkit/foundation/registry.rst +++ b/docs/source/pcapkit/foundation/registry.rst @@ -85,6 +85,12 @@ Link Layer Registries .. autofunction:: pcapkit.foundation.registry.protocols.register_loopback +.. autofunction:: pcapkit.foundation.registry.protocols.register_arphrd + +.. autofunction:: pcapkit.foundation.registry.protocols.register_frad + +.. autofunction:: pcapkit.foundation.registry.protocols.register_netlink + Internet Layer Registries ~~~~~~~~~~~~~~~~~~~~~~~~~ diff --git a/docs/source/pcapkit/protocols/link/frad.rst b/docs/source/pcapkit/protocols/link/frad.rst index efd9ea2cd4..5c28183e4e 100644 --- a/docs/source/pcapkit/protocols/link/frad.rst +++ b/docs/source/pcapkit/protocols/link/frad.rst @@ -38,6 +38,7 @@ for SNAP), a 3-octet OUI and a 2-octet PID follow, kept as ``frad.ext_nlpid``, .. automethod:: read .. automethod:: make + .. automethod:: register .. automethod:: __length_hint__ .. automethod:: __index__ diff --git a/docs/source/pcapkit/protocols/link/linux_sll.rst b/docs/source/pcapkit/protocols/link/linux_sll.rst index c7651bc1f7..e49f07dbd9 100644 --- a/docs/source/pcapkit/protocols/link/linux_sll.rst +++ b/docs/source/pcapkit/protocols/link/linux_sll.rst @@ -26,9 +26,15 @@ validation. It has no registry index, so its ``__index__`` raises. :show-inheritance: .. automethod:: id + .. automethod:: register .. automethod:: __index__ .. automethod:: _import_next_layer + .. autoattribute:: __proto__ + :no-value: + .. autoattribute:: __proto_key__ + :no-value: + Auxiliary Functions ------------------- @@ -38,14 +44,13 @@ Auxiliary Functions Auxiliary Constants ------------------- -.. autodata:: pcapkit.protocols.link.linux_sll._ARPHRD_NOT_ETHERTYPE .. autodata:: pcapkit.protocols.link.linux_sll._ETHERTYPE_FIELD :no-value: Hardware Types -------------- -The ``arphrd_type`` values in ``_ARPHRD_NOT_ETHERTYPE`` are named by the enumeration below. +The ``arphrd_type`` keys of ``LinuxSLL.__proto__`` are named by the enumeration below. .. autoclass:: pcapkit.const.linux.arphrd.ARPHRD :members: diff --git a/docs/source/pcapkit/protocols/link/netlink.rst b/docs/source/pcapkit/protocols/link/netlink.rst index 0544718aa3..7873ec60e5 100644 --- a/docs/source/pcapkit/protocols/link/netlink.rst +++ b/docs/source/pcapkit/protocols/link/netlink.rst @@ -43,6 +43,7 @@ format) and is guessed by :func:`nlmsg_byteorder`; either way it is kept as .. automethod:: read .. automethod:: make + .. automethod:: register .. automethod:: __length_hint__ .. automethod:: __index__ diff --git a/pcapkit/all.py b/pcapkit/all.py index 66546887c0..cc6453af7f 100644 --- a/pcapkit/all.py +++ b/pcapkit/all.py @@ -94,6 +94,7 @@ 'register_pcap', 'register_pcapng', 'register_ethertype', 'register_loopback', + 'register_arphrd', 'register_frad', 'register_netlink', 'register_transtype', 'register_ipx', 'register_icmp_type', diff --git a/pcapkit/foundation/__init__.py b/pcapkit/foundation/__init__.py index 562580683f..517b4bd2bf 100644 --- a/pcapkit/foundation/__init__.py +++ b/pcapkit/foundation/__init__.py @@ -36,6 +36,7 @@ 'register_pcap', 'register_pcapng', 'register_ethertype', 'register_loopback', + 'register_arphrd', 'register_frad', 'register_netlink', 'register_transtype', 'register_ipx', 'register_icmp_type', diff --git a/pcapkit/foundation/registry/__init__.py b/pcapkit/foundation/registry/__init__.py index 845a09782c..e5d6eac8eb 100644 --- a/pcapkit/foundation/registry/__init__.py +++ b/pcapkit/foundation/registry/__init__.py @@ -32,6 +32,7 @@ 'register_ethertype', 'register_loopback', + 'register_arphrd', 'register_frad', 'register_netlink', 'register_transtype', 'register_ipx', diff --git a/pcapkit/foundation/registry/protocols.py b/pcapkit/foundation/registry/protocols.py index 71fbae9ba9..344ab8c248 100644 --- a/pcapkit/foundation/registry/protocols.py +++ b/pcapkit/foundation/registry/protocols.py @@ -28,6 +28,7 @@ from pcapkit.const.ipv6.option import Option as IPv6_Option from pcapkit.const.ipv6.routing import Routing as IPv6_Routing from pcapkit.const.ipx.packet import Packet as IPX_Packet +from pcapkit.const.linux.arphrd import ARPHRD as Enum_ARPHRD 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 @@ -61,8 +62,11 @@ from pcapkit.protocols.internet.ipv6_route import IPv6_Route from pcapkit.protocols.internet.ipx import IPX from pcapkit.protocols.internet.mh import MH +from pcapkit.protocols.link.frad import FRAD from pcapkit.protocols.link.link import Link +from pcapkit.protocols.link.linux_sll import LinuxSLL from pcapkit.protocols.link.loopback import Loopback +from pcapkit.protocols.link.netlink import Netlink from pcapkit.protocols.misc.pcap.frame import Frame from pcapkit.protocols.misc.pcapng import PCAPNG from pcapkit.protocols.protocol import ProtocolBase @@ -149,6 +153,7 @@ 'register_ethertype', 'register_loopback', + 'register_arphrd', 'register_frad', 'register_netlink', 'register_transtype', 'register_ipx', @@ -888,6 +893,136 @@ def register_loopback(code: 'int', module: 'str | ModuleDescriptor[ProtocolBase] _register_name(module, (Loopback, code)) +@overload +def register_arphrd(code: 'Enum_ARPHRD', module: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]') -> 'None': ... +@overload +def register_arphrd(code: 'Enum_ARPHRD', module: 'str', class_: 'str') -> 'None': ... + + +# NOTE: pcapkit.protocols.link.linux_sll.LinuxSLL.__proto__ +def register_arphrd(code: 'Enum_ARPHRD', 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.linux_sll.LinuxSLL.__proto__` registry, so + that a Linux "cooked" capture on that ``arphrd_type`` dispatches to it + instead of reading ``protocol_type`` as an EtherType. + + Arguments: + code: ARP hardware type as in :class:`~pcapkit.const.linux.arphrd.ARPHRD` + module: module name or module descriptor or a + :class:`~pcapkit.protocols.protocol.ProtocolBase` subclass + class\_: class name + + Raises: + RegistryError: If ``code`` is not a :class:`~pcapkit.const.linux.arphrd.ARPHRD` + member, e.g. a member of another enum, or ``module`` + is not a :class:`~pcapkit.protocols.protocol.ProtocolBase` + subclass. + + """ + _check_code(code, Enum_ARPHRD, LinuxSLL) # type: ignore[type-abstract] + if isinstance(module, str): + module = cast('ModuleDescriptor[ProtocolBase]', ModuleDescriptor(module, class_)) + + LinuxSLL.register(code, module) + logger.debug('registered ARPHRD protocol: %s', code.name) + + # register protocol to protocol registry + if isinstance(module, ModuleDescriptor): + module = module.klass + _register_name(module, (LinuxSLL, code)) + + +@overload +def register_frad(code: 'int', module: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]') -> 'None': ... +@overload +def register_frad(code: 'int', module: 'str', class_: 'str') -> 'None': ... + + +# NOTE: pcapkit.protocols.link.frad.FRAD.__proto__ +def register_frad(code: 'int', 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.frad.FRAD.__proto__` registry. + + Arguments: + code: NLPID, after resolving the ``0x00`` pad ahead of a SNAP header + module: module name or module descriptor or a + :class:`~pcapkit.protocols.protocol.ProtocolBase` subclass + class\_: class name + + Raises: + RegistryError: If ``code`` is not an :class:`int` (a :class:`bool` is + not accepted, even though it is one), or is a member of an enum. + + """ + _check_code(code, None, FRAD) + if isinstance(module, str): + module = cast('ModuleDescriptor[ProtocolBase]', ModuleDescriptor(module, class_)) + + FRAD.register(code, module) + logger.debug('registered FRAD NLPID: %s', code) + + # register protocol to protocol registry + if isinstance(module, ModuleDescriptor): + module = module.klass + _register_name(module, (FRAD, code)) + + +@overload +def register_netlink(code: 'int', module: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]') -> 'None': ... +@overload +def register_netlink(code: 'int', module: 'str', class_: 'str') -> 'None': ... + + +# NOTE: pcapkit.protocols.link.netlink.Netlink.__proto__ +def register_netlink(code: 'int', 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.netlink.Netlink.__proto__` registry. + + Arguments: + code: ``nlmsg_type`` of the dissected message + module: module name or module descriptor or a + :class:`~pcapkit.protocols.protocol.ProtocolBase` subclass + class\_: class name + + Raises: + RegistryError: If ``code`` is not an :class:`int` (a :class:`bool` is + not accepted, even though it is one), or is a member of an enum. + + """ + _check_code(code, None, Netlink) + if isinstance(module, str): + module = cast('ModuleDescriptor[ProtocolBase]', ModuleDescriptor(module, class_)) + + Netlink.register(code, module) + logger.debug('registered Netlink message type: %s', code) + + # register protocol to protocol registry + if isinstance(module, ModuleDescriptor): + module = module.klass + _register_name(module, (Netlink, code)) + + ############################################################################### # Internet Layer Registries ############################################################################### diff --git a/pcapkit/protocols/link/frad.py b/pcapkit/protocols/link/frad.py index b3f0707044..9e943da988 100644 --- a/pcapkit/protocols/link/frad.py +++ b/pcapkit/protocols/link/frad.py @@ -11,7 +11,7 @@ is ``ARPHRD_FRAD`` -- :class:`~pcapkit.protocols.link.linux_sll.LinuxSLL` dispatches here instead of falling back to :class:`~pcapkit.protocols.misc.raw.Raw` for that one case; see -:data:`~pcapkit.protocols.link.linux_sll._ARPHRD_NOT_ETHERTYPE`. Its +:attr:`LinuxSLL.__proto__ `. Its structure is the Q.922 LAPF address onward, with no trailing FCS: a 2-octet ``frad.addr``, a 1-octet ``frad.control`` (``0x03``, UI, for ordinary data traffic), then a 1-octet ``frad.nlpid`` -- or, when SNAP-encapsulated, the @@ -29,15 +29,17 @@ import collections from typing import TYPE_CHECKING +from pcapkit.const.linux.arphrd import ARPHRD as Enum_ARPHRD from pcapkit.corekit.decorators import beholder from pcapkit.corekit.module import ModuleDescriptor from pcapkit.protocols.data.link.frad import FRAD as Data_FRAD from pcapkit.protocols.link.link import Link from pcapkit.protocols.schema.link.frad import FRAD as Schema_FRAD -from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall +from pcapkit.utilities.exceptions import ProtocolError +from pcapkit.utilities.warnings import RegistryWarning, warn if TYPE_CHECKING: - from typing import Any, DefaultDict, NoReturn, Optional, Type + from typing import Any, DefaultDict, Optional, Type from typing_extensions import Literal @@ -286,6 +288,46 @@ def make(self, payload=payload, ) + @classmethod + def register(cls, code: 'int', protocol: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]') -> 'None': + r"""Register a new protocol class for an NLPID. + + Notes: + The fully qualified class name of the new protocol class + should be ``{protocol.module}.{protocol.name}``. + + Arguments: + code: NLPID, after resolving the ``0x00`` pad ahead of a SNAP header + protocol: module descriptor or a + :class:`~pcapkit.protocols.protocol.ProtocolBase` subclass + + Important: + This overrides :meth:`Link.register + ` because + :attr:`self.__proto__ ` is keyed by NLPID rather + than by EtherType. + + Raises: + pcapkit.utilities.exceptions.RegistryError: If ``protocol`` is not a + class, or not a :class:`~pcapkit.protocols.protocol.ProtocolBase` subclass. + + Warns: + pcapkit.utilities.warnings.RegistryWarning: If this NLPID is already + registered and the incumbent differs from the replacement; see + :meth:`Protocol.register ` + for the guard, and for the shipped descriptor a restore puts + back (:issue:`1504`). + + """ + incumbent = cls.__proto__.get(code) + entry = cls._next_layer_entry(code, protocol) + if entry is None: + return + if incumbent is not None and incumbent is not entry: + warn(f'NLPID {code} already registered, overwriting ' + f'{incumbent!r} with {entry!r}', RegistryWarning) + cls.__proto__[code] = entry + ########################################################################## # Data models. ########################################################################## @@ -303,17 +345,17 @@ def __length_hint__(self) -> 'Literal[10]': return 10 @classmethod - def __index__(cls) -> 'NoReturn': # pylint: disable=invalid-index-returned + def __index__(cls) -> 'Enum_ARPHRD': # pylint: disable=invalid-index-returned """Numeral registry index of the protocol. - Raises: - UnsupportedCall: This protocol is reached only through - :class:`~pcapkit.protocols.link.linux_sll.LinuxSLL`'s - ``arphrd_type`` dispatch, not through a link-layer header - type of its own. + Returns: + Numeral registry index of the protocol in Linux ARP hardware + types, the ``arphrd_type`` that + :attr:`LinuxSLL.__proto__ ` + dispatches here. """ - raise UnsupportedCall(f'{cls.__name__!r} object cannot be interpreted as an integer') + return Enum_ARPHRD.FRAD # type: ignore[return-value] ########################################################################## # Utilities. diff --git a/pcapkit/protocols/link/gre.py b/pcapkit/protocols/link/gre.py index b54b559ee0..3e05581127 100644 --- a/pcapkit/protocols/link/gre.py +++ b/pcapkit/protocols/link/gre.py @@ -12,7 +12,7 @@ is ``ARPHRD_IPGRE`` or ``ARPHRD_IP6GRE`` -- both land on this one class: :class:`~pcapkit.protocols.link.linux_sll.LinuxSLL` dispatches here instead of falling back to :class:`~pcapkit.protocols.misc.raw.Raw` for either case; -see :data:`~pcapkit.protocols.link.linux_sll._ARPHRD_NOT_ETHERTYPE`. The +see :attr:`LinuxSLL.__proto__ `. The "IP"/"IP6" distinction in the two ``arphrd_type`` names is about which IP version the tunnel carries on its *outer* layer -- i.e. which ``arphrd_type`` fired -- not a different GRE header shape; the GRE framing itself is @@ -64,15 +64,16 @@ """ from typing import TYPE_CHECKING +from pcapkit.const.linux.arphrd import ARPHRD as Enum_ARPHRD from pcapkit.const.reg.ethertype import EtherType as Enum_EtherType from pcapkit.protocols.data.link.gre import GRE as Data_GRE from pcapkit.protocols.link.link import Link from pcapkit.protocols.schema.link.gre import GRE as Schema_GRE -from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall +from pcapkit.utilities.exceptions import ProtocolError if TYPE_CHECKING: from enum import IntEnum as StdlibEnum - from typing import Any, NoReturn, Optional, Type + from typing import Any, Optional, Type from aenum import IntEnum as AenumEnum from typing_extensions import Literal @@ -315,17 +316,18 @@ def __length_hint__(self) -> 'Literal[16]': return 16 @classmethod - def __index__(cls) -> 'NoReturn': # pylint: disable=invalid-index-returned + def __index__(cls) -> 'Enum_ARPHRD': # pylint: disable=invalid-index-returned """Numeral registry index of the protocol. - Raises: - UnsupportedCall: This protocol is reached only through - :class:`~pcapkit.protocols.link.linux_sll.LinuxSLL`'s - ``arphrd_type`` dispatch, not through a link-layer header - type of its own. + Returns: + Numeral registry index of the protocol in Linux ARP hardware + types: :attr:`~pcapkit.const.linux.arphrd.ARPHRD.IPGRE`, which + :attr:`LinuxSLL.__proto__ ` + dispatches here alongside + :attr:`~pcapkit.const.linux.arphrd.ARPHRD.IP6GRE`. """ - raise UnsupportedCall(f'{cls.__name__!r} object cannot be interpreted as an integer') + return Enum_ARPHRD.IPGRE # type: ignore[return-value] ########################################################################## # Utilities. diff --git a/pcapkit/protocols/link/l2tpv2.py b/pcapkit/protocols/link/l2tpv2.py index bcdf6095f7..2b5c8c7134 100644 --- a/pcapkit/protocols/link/l2tpv2.py +++ b/pcapkit/protocols/link/l2tpv2.py @@ -398,8 +398,9 @@ def __index__(cls) -> 'NoReturn': # pylint: disable=invalid-index-returned L2TPv2 is reached through **UDP port 1701**, and a port is not an ``__index__`` value anywhere in this package -- every non-raising ``__index__`` returns a :class:`~pcapkit.const.reg.transtype.TransType`, - :class:`~pcapkit.const.reg.ethertype.EtherType` or - :class:`~pcapkit.const.reg.linktype.LinkType`, bar the PCAP frame and + :class:`~pcapkit.const.reg.ethertype.EtherType`, + :class:`~pcapkit.const.reg.linktype.LinkType` or + :class:`~pcapkit.const.linux.arphrd.ARPHRD`, bar the PCAP frame and PCAP-NG block instances that return their own index number, and :meth:`Application.__index__ ` diff --git a/pcapkit/protocols/link/linux_sll.py b/pcapkit/protocols/link/linux_sll.py index 5d46619458..b2f7875492 100644 --- a/pcapkit/protocols/link/linux_sll.py +++ b/pcapkit/protocols/link/linux_sll.py @@ -10,21 +10,25 @@ :class:`~pcapkit.protocols.link.linux_sll2.LinuxSLL2`. """ +import collections from typing import TYPE_CHECKING, Generic -from pcapkit.const.linux.arphrd import ARPHRD +from pcapkit.const.linux.arphrd import ARPHRD as Enum_ARPHRD from pcapkit.const.reg.ethertype import EtherType as Enum_EtherType from pcapkit.corekit.decorators import beholder from pcapkit.corekit.fields.numbers import EnumField +from pcapkit.corekit.module import ModuleDescriptor from pcapkit.protocols.link.link import Link from pcapkit.protocols.protocol import _PT, _ST from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall +from pcapkit.utilities.warnings import RegistryWarning, warn if TYPE_CHECKING: - from typing import Any, NoReturn, Optional + from typing import Any, ClassVar, DefaultDict, NoReturn, Optional, Type from typing_extensions import Literal + from pcapkit.corekit.enum import EnumRegistry from pcapkit.protocols.protocol import ProtocolBase __all__ = ['LinuxSLL'] @@ -33,33 +37,24 @@ #: :class:`~pcapkit.protocols.schema.link.ethernet.Ethernet` resolves its ``type``. _ETHERTYPE_FIELD = EnumField(length=2, namespace=Enum_EtherType) -#: ``arphrd_type`` values whose ``protocol_type`` is not an EtherType, per -#: tcpdump.org's LINUX_SLL and LINUX_SLL2 pages. -_ARPHRD_NOT_ETHERTYPE = frozenset({ - ARPHRD.FRAD, - ARPHRD.IPGRE, - ARPHRD.IEEE80211_RADIOTAP, - ARPHRD.IP6GRE, - ARPHRD.NETLINK, -}) - -def _resolve_next_layer_proto(arphrd_type: 'int', +def _resolve_next_layer_proto(registry: 'DefaultDict[int, Any]', arphrd_type: 'int', protocol_type: 'int') -> 'Enum_EtherType | int': """Resolve the next-layer dispatch key of either version. Args: + registry: The ``arphrd_type`` registry, i.e. :attr:`LinuxSLL.__proto__ + `. arphrd_type: ARP hardware type of the interface the packet was seen on. protocol_type: Protocol (internet layer), as parsed off the wire. Returns: - ``protocol_type`` itself if ``arphrd_type`` is in - :data:`_ARPHRD_NOT_ETHERTYPE`; otherwise the value - :class:`~pcapkit.protocols.link.ethernet.Ethernet` stores for the same - ``type``. + ``protocol_type`` itself if ``arphrd_type`` is a key of ``registry``; + otherwise the value :class:`~pcapkit.protocols.link.ethernet.Ethernet` + stores for the same ``type``. """ - if int(arphrd_type) in _ARPHRD_NOT_ETHERTYPE: + if int(arphrd_type) in registry: return protocol_type return _ETHERTYPE_FIELD.post_process(protocol_type, {}) @@ -103,8 +98,55 @@ class LinuxSLL(Link[_PT, _ST], Generic[_PT, _ST]): # pylint: disable=abstract-m next-layer dispatch they share. See :ref:`linux-sll-labels` for the labels that select a version. + This class supports parsing of the following protocols, which are + registered in the :attr:`self.__proto__ + ` attribute, keyed by + ``arphrd_type``: + + .. list-table:: + :header-rows: 1 + + * - Index + - Protocol + * - :attr:`~pcapkit.const.linux.arphrd.ARPHRD.FRAD` + - :class:`pcapkit.protocols.link.frad.FRAD` + * - :attr:`~pcapkit.const.linux.arphrd.ARPHRD.IPGRE` + - :class:`pcapkit.protocols.link.gre.GRE` + * - :attr:`~pcapkit.const.linux.arphrd.ARPHRD.IEEE80211_RADIOTAP` + - :class:`pcapkit.protocols.link.radiotap.Radiotap` + * - :attr:`~pcapkit.const.linux.arphrd.ARPHRD.IP6GRE` + - :class:`pcapkit.protocols.link.gre.GRE` + * - :attr:`~pcapkit.const.linux.arphrd.ARPHRD.NETLINK` + - :class:`pcapkit.protocols.link.netlink.Netlink` + + Every other ``arphrd_type`` reads ``protocol_type`` as an EtherType and + dispatches it through :attr:`Link.__proto__ + `, as an Ethernet frame does. + """ + ########################################################################## + # Defaults. + ########################################################################## + + #: Protocol index mapping for decoding next layer, keyed by the + #: ``arphrd_type`` values whose ``protocol_type`` is not an EtherType, per + #: tcpdump.org's LINUX_SLL and LINUX_SLL2 pages. + __proto__ = collections.defaultdict( + lambda: ModuleDescriptor('pcapkit.protocols.misc.raw', 'Raw'), + { + Enum_ARPHRD.FRAD: ModuleDescriptor('pcapkit.protocols.link.frad', 'FRAD'), + Enum_ARPHRD.IPGRE: ModuleDescriptor('pcapkit.protocols.link.gre', 'GRE'), + Enum_ARPHRD.IEEE80211_RADIOTAP: ModuleDescriptor('pcapkit.protocols.link.radiotap', 'Radiotap'), + Enum_ARPHRD.IP6GRE: ModuleDescriptor('pcapkit.protocols.link.gre', 'GRE'), + Enum_ARPHRD.NETLINK: ModuleDescriptor('pcapkit.protocols.link.netlink', 'Netlink'), + }, + ) # type: DefaultDict[int | Enum_ARPHRD, ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]] + + #: What :attr:`__proto__` is keyed by, declared as + #: :attr:`Link.__proto_key__ ` is. + __proto_key__: 'ClassVar[tuple[Type[EnumRegistry], bool]]' = (Enum_ARPHRD, False) + ########################################################################## # Methods. ########################################################################## @@ -119,6 +161,48 @@ def id(cls) -> 'tuple[Literal["LinuxSLL1"], Literal["LinuxSLL2"]]': """ return ('LinuxSLL1', 'LinuxSLL2') + @classmethod + def register(cls, code: 'Enum_ARPHRD', protocol: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]') -> 'None': # type: ignore[override] + r"""Register a new protocol class for an ``arphrd_type``. + + Notes: + The fully qualified class name of the new protocol class + should be ``{protocol.module}.{protocol.name}``. + + Arguments: + code: ARP hardware type as in :class:`~pcapkit.const.linux.arphrd.ARPHRD` + protocol: module descriptor or a + :class:`~pcapkit.protocols.protocol.ProtocolBase` subclass + + Important: + This overrides :meth:`Link.register + ` because + :attr:`self.__proto__ ` is keyed by + ``arphrd_type`` rather than by EtherType. A registered + ``arphrd_type`` dispatches through it, and its ``protocol_type`` + is not read as an EtherType. + + Raises: + pcapkit.utilities.exceptions.RegistryError: If ``protocol`` is not a + class, or not a :class:`~pcapkit.protocols.protocol.ProtocolBase` subclass. + + Warns: + pcapkit.utilities.warnings.RegistryWarning: If this ``arphrd_type`` + is already registered and the incumbent differs from the + replacement; see :meth:`Protocol.register + ` for the guard, + and for the shipped descriptor a restore puts back (:issue:`1504`). + + """ + incumbent = cls.__proto__.get(code) + entry = cls._next_layer_entry(code, protocol) + if entry is None: + return + if incumbent is not None and incumbent is not entry: + warn(f'ARP hardware type {code} already registered, overwriting ' + f'{incumbent!r} with {entry!r}', RegistryWarning) + cls.__proto__[code] = entry + ########################################################################## # Data models. ########################################################################## @@ -150,53 +234,43 @@ def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *, # `) Returns: - Instance of next layer: - :class:`~pcapkit.protocols.link.netlink.Netlink` when ``arphrd_type`` - is :attr:`ARPHRD.NETLINK `; - :class:`~pcapkit.protocols.link.frad.FRAD` when ``arphrd_type`` is - :attr:`ARPHRD.FRAD `; - :class:`~pcapkit.protocols.link.gre.GRE` when ``arphrd_type`` is - :attr:`ARPHRD.IPGRE ` or - :attr:`ARPHRD.IP6GRE `; - :class:`~pcapkit.protocols.link.radiotap.Radiotap` when ``arphrd_type`` - is :attr:`ARPHRD.IEEE80211_RADIOTAP - `; - otherwise the dissector :attr:`Link.__proto__ - ` holds for ``proto``, - an EtherType. + Instance of next layer: the class :attr:`self.__proto__ + ` registers for ``arphrd_type`` when it has a key + for it, and otherwise the class :attr:`Link.__proto__ + ` registers for ``proto`` + as an EtherType. A :class:`~pcapkit.protocols.misc.raw.Raw` fallback + records ``proto`` either way. + + Note: + :attr:`self.__proto__ ` holds the + ``arphrd_type`` keys, so the EtherType path names + :attr:`Link.__proto__ ` + itself rather than reaching it through ``super()``, as + :meth:`FRAD._import_next_layer + ` does for a + SNAP EtherType. """ file_ = self._get_payload() if length is None: length = len(file_) - - if not min(length, len(file_)): - return super()._import_next_layer(proto, length, packet=packet) # type: ignore[misc,call-arg] - arphrd_type = int(self.__header__.arphrd_type) # type: ignore[attr-defined] - if arphrd_type == ARPHRD.NETLINK: - from pcapkit.protocols.link.netlink import Netlink # pylint: disable=import-outside-toplevel # isort: skip - return self._parse_next_layer(Netlink, file_, length, alias=proto, packet=packet, - layer=self._exlayer, protocol=self._exproto, - __context__=self._exctx) - if arphrd_type == ARPHRD.FRAD: - from pcapkit.protocols.link.frad import FRAD # pylint: disable=import-outside-toplevel - return self._parse_next_layer(FRAD, file_, length, alias=proto, packet=packet, - layer=self._exlayer, protocol=self._exproto, - __context__=self._exctx) - - if arphrd_type in (ARPHRD.IPGRE, ARPHRD.IP6GRE): - from pcapkit.protocols.link.gre import GRE # pylint: disable=import-outside-toplevel - return self._parse_next_layer(GRE, file_, length, alias=proto, packet=packet, - layer=self._exlayer, protocol=self._exproto, - __context__=self._exctx) - - if arphrd_type == ARPHRD.IEEE80211_RADIOTAP: - from pcapkit.protocols.link.radiotap import \ - Radiotap # pylint: disable=import-outside-toplevel # isort: skip - return self._parse_next_layer(Radiotap, file_, length, alias=proto, packet=packet, - layer=self._exlayer, protocol=self._exproto, - __context__=self._exctx) - - return super()._import_next_layer(proto, length, packet=packet) # type: ignore[misc,call-arg] + if TYPE_CHECKING: + protocol: 'Type[ProtocolBase]' + if not min(length, len(file_)): + from pcapkit.protocols.misc.null import \ + NoPayload as protocol # isort: skip # pylint: disable=import-outside-toplevel + # NOTE: ahead of ``_sigterm``, so a registered ``arphrd_type`` is parsed + # even where parsing stops at this layer (#2510). + elif arphrd_type in self.__proto__: + protocol = self._lookup_next_layer(self.__proto__, arphrd_type) + elif self._sigterm: + from pcapkit.protocols.misc.raw import \ + Raw as protocol # isort: skip # pylint: disable=import-outside-toplevel + else: + protocol = self._lookup_next_layer(Link.__proto__, proto) + + return self._parse_next_layer(protocol, file_, length, alias=proto, packet=packet, + layer=self._exlayer, protocol=self._exproto, + __context__=self._exctx) diff --git a/pcapkit/protocols/link/linux_sll1.py b/pcapkit/protocols/link/linux_sll1.py index aa41ab1fb1..f40901e5b0 100644 --- a/pcapkit/protocols/link/linux_sll1.py +++ b/pcapkit/protocols/link/linux_sll1.py @@ -55,8 +55,9 @@ class LinuxSLL1(LinuxSLL[Data_LinuxSLL1, Schema_LinuxSLL1], """Linux "cooked" capture encapsulation v1 (``LINUX_SLL``). ``protocol_type`` is dispatched as an EtherType, unless ``arphrd_type`` is - in :data:`~pcapkit.protocols.link.linux_sll._ARPHRD_NOT_ETHERTYPE`; in that - case, and when no dissector is registered, the payload is + a key of :attr:`LinuxSLL.__proto__ + `, which dispatches + it instead; with no dissector registered either way, the payload is :class:`~pcapkit.protocols.misc.raw.Raw`. """ @@ -145,7 +146,8 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_LinuxSL protocol_type=schema.protocol_type, ) - _proto = _resolve_next_layer_proto(schema.arphrd_type, schema.protocol_type) + _proto = _resolve_next_layer_proto(self.__proto__, schema.arphrd_type, + schema.protocol_type) return self._decode_next_layer(linux_sll, _proto, length - self.length) def make(self, @@ -167,7 +169,8 @@ def make(self, link_addr_length: Length of the address actually in use, by default ``len(link_addr)``; may exceed 8, up to ``0xFFFF``. protocol_type: Protocol (internet layer); an EtherType unless - ``arphrd_type`` is in the exceptions of :class:`LinuxSLL1`. + ``arphrd_type`` is a key of :attr:`LinuxSLL.__proto__ + `. payload: Payload data. **kwargs: Arbitrary keyword arguments. diff --git a/pcapkit/protocols/link/linux_sll2.py b/pcapkit/protocols/link/linux_sll2.py index 1a604e227d..f27df79d81 100644 --- a/pcapkit/protocols/link/linux_sll2.py +++ b/pcapkit/protocols/link/linux_sll2.py @@ -147,7 +147,8 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_LinuxSL link_addr=schema.link_addr, ) - _proto = _resolve_next_layer_proto(schema.arphrd_type, schema.protocol_type) + _proto = _resolve_next_layer_proto(self.__proto__, schema.arphrd_type, + schema.protocol_type) return self._decode_next_layer(linux_sll2, _proto, length - self.length) def make(self, @@ -164,7 +165,8 @@ def make(self, Args: protocol_type: Protocol (internet layer); an EtherType unless - ``arphrd_type`` is in the exceptions of :class:`LinuxSLL2`. + ``arphrd_type`` is a key of :attr:`LinuxSLL.__proto__ + `. reserved: Reserved, must be zero. interface_index: Index, in the capturing host's own numbering, of the interface the packet was seen on. diff --git a/pcapkit/protocols/link/netlink.py b/pcapkit/protocols/link/netlink.py index 3fbc7fef00..ea1af6e7e2 100644 --- a/pcapkit/protocols/link/netlink.py +++ b/pcapkit/protocols/link/netlink.py @@ -11,12 +11,12 @@ capture whose ``arphrd_type`` is ``ARPHRD_NETLINK`` (an ``nlmon`` interface) -- :class:`~pcapkit.protocols.link.linux_sll.LinuxSLL` dispatches here instead of falling back to :class:`~pcapkit.protocols.misc.raw.Raw` for that -one case; see :data:`~pcapkit.protocols.link.linux_sll._ARPHRD_NOT_ETHERTYPE`. +one case; see :attr:`LinuxSLL.__proto__ `. Netlink is the kernel's own IPC mechanism (routing table dumps, interface notifications and the like), not a "next protocol over this link" the way -every other ``arphrd_type`` exception is -- there is no next-layer dispatch -table here at all, every message's payload is kept as -:class:`~pcapkit.protocols.misc.raw.Raw`. +every other ``arphrd_type`` exception is -- its next-layer registry ships +empty, so every message's payload is kept as +:class:`~pcapkit.protocols.misc.raw.Raw` unless its ``nlmsg_type`` is registered. .. table:: @@ -43,7 +43,7 @@ (``NETLINK_ROUTE``, ``NETLINK_GENERIC``, ...) and is a large surface of its own -- routing table entries, link attributes (``rtattr``/``nlattr`` TLV chains), and so on. The payload is kept as - :class:`~pcapkit.protocols.misc.raw.Raw`, unconditionally. + :class:`~pcapkit.protocols.misc.raw.Raw` unless its ``nlmsg_type`` is registered. * **Multiple messages packed into one read.** The kernel's own netlink(7) page documents that a single read from a netlink socket, and so a single captured frame, commonly carries more than one ``nlmsghdr`` message back to @@ -83,14 +83,16 @@ import sys from typing import TYPE_CHECKING +from pcapkit.const.linux.arphrd import ARPHRD as Enum_ARPHRD from pcapkit.corekit.module import ModuleDescriptor from pcapkit.protocols.data.link.netlink import Netlink as Data_Netlink from pcapkit.protocols.link.link import Link from pcapkit.protocols.schema.link.netlink import Netlink as Schema_Netlink -from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall +from pcapkit.utilities.exceptions import ProtocolError +from pcapkit.utilities.warnings import RegistryWarning, warn if TYPE_CHECKING: - from typing import Any, DefaultDict, NoReturn, Optional, Type + from typing import Any, DefaultDict, Optional, Type from typing_extensions import Literal @@ -136,10 +138,10 @@ class Netlink(Link[Data_Netlink, Schema_Netlink], """This class implements Linux Netlink socket framing. Every message's payload is kept as - :class:`~pcapkit.protocols.misc.raw.Raw`, unconditionally -- see the - module docstring's *Scope* section for why. :attr:`self.__proto__ - ` is kept, as an empty - registry, only so :meth:`register` works the same way every other + :class:`~pcapkit.protocols.misc.raw.Raw` unless its ``nlmsg_type`` is + registered -- see the module docstring's *Scope* section for why. + :attr:`self.__proto__ ` + ships empty, so that :meth:`register` works the same way every other :class:`~pcapkit.protocols.link.link.Link` subclass's does. """ @@ -148,10 +150,9 @@ class Netlink(Link[Data_Netlink, Schema_Netlink], # Defaults. ########################################################################## - #: Protocol index mapping for decoding next layer. Deliberately empty -- - #: see the class docstring -- so every key, including the ``nlmsg_type`` - #: this class dispatches with, resolves through the - #: :class:`~collections.defaultdict` fallback to + #: Protocol index mapping for decoding next layer, keyed by ``nlmsg_type``. + #: Deliberately empty -- see the class docstring -- so every unregistered + #: key resolves through the :class:`~collections.defaultdict` fallback to #: :class:`~pcapkit.protocols.misc.raw.Raw`. __proto__ = collections.defaultdict( lambda: ModuleDescriptor('pcapkit.protocols.misc.raw', 'Raw'), @@ -311,6 +312,45 @@ def make(self, payload=payload, ) + @classmethod + def register(cls, code: 'int', protocol: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]') -> 'None': + r"""Register a new protocol class for a message type. + + Notes: + The fully qualified class name of the new protocol class + should be ``{protocol.module}.{protocol.name}``. + + Arguments: + code: ``nlmsg_type`` of the dissected message + protocol: module descriptor or a + :class:`~pcapkit.protocols.protocol.ProtocolBase` subclass + + Important: + This overrides :meth:`Link.register + ` because + :attr:`self.__proto__ ` is keyed by + ``nlmsg_type`` rather than by EtherType. + + Raises: + pcapkit.utilities.exceptions.RegistryError: If ``protocol`` is not a + class, or not a :class:`~pcapkit.protocols.protocol.ProtocolBase` subclass. + + Warns: + pcapkit.utilities.warnings.RegistryWarning: If this message type is + already registered and the incumbent differs from the + replacement; see :meth:`Protocol.register + ` for the guard. + + """ + 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'message type {code} already registered, overwriting ' + f'{incumbent!r} with {entry!r}', RegistryWarning) + cls.__proto__[code] = entry + ########################################################################## # Data models. ########################################################################## @@ -320,17 +360,17 @@ def __length_hint__(self) -> 'Literal[16]': return 16 @classmethod - def __index__(cls) -> 'NoReturn': # pylint: disable=invalid-index-returned + def __index__(cls) -> 'Enum_ARPHRD': # pylint: disable=invalid-index-returned """Numeral registry index of the protocol. - Raises: - UnsupportedCall: This protocol is reached only through - :class:`~pcapkit.protocols.link.linux_sll.LinuxSLL`'s - ``arphrd_type`` dispatch, not through a link-layer header - type of its own. + Returns: + Numeral registry index of the protocol in Linux ARP hardware + types, the ``arphrd_type`` that + :attr:`LinuxSLL.__proto__ ` + dispatches here. """ - raise UnsupportedCall(f'{cls.__name__!r} object cannot be interpreted as an integer') + return Enum_ARPHRD.NETLINK # type: ignore[return-value] ########################################################################## # Utilities. diff --git a/pcapkit/protocols/link/radiotap.py b/pcapkit/protocols/link/radiotap.py index f67349af79..533cb49228 100644 --- a/pcapkit/protocols/link/radiotap.py +++ b/pcapkit/protocols/link/radiotap.py @@ -11,7 +11,7 @@ whose ``arphrd_type`` is ``ARPHRD_IEEE80211_RADIOTAP`` -- :class:`~pcapkit.protocols.link.linux_sll.LinuxSLL` dispatches here instead of falling back to :class:`~pcapkit.protocols.misc.raw.Raw` for that one -case; see :data:`~pcapkit.protocols.link.linux_sll._ARPHRD_NOT_ETHERTYPE`. +case; see :attr:`LinuxSLL.__proto__ `. Radiotap is not part of the 802.11 standard itself: it is a prefix *in front of* the raw 802.11 MAC frame, written by the capturing tool rather than by the radio. Its structure is a fixed 8-octet start, all little-endian: @@ -70,6 +70,7 @@ import collections from typing import TYPE_CHECKING +from pcapkit.const.linux.arphrd import ARPHRD as Enum_ARPHRD from pcapkit.corekit.decorators import beholder from pcapkit.corekit.module import ModuleDescriptor from pcapkit.protocols.data.link.radiotap import Radiotap as Data_Radiotap @@ -77,10 +78,10 @@ from pcapkit.protocols.protocol import ProtocolBase from pcapkit.protocols.schema.link.radiotap import Radiotap as Schema_Radiotap from pcapkit.protocols.schema.schema import Schema -from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall +from pcapkit.utilities.exceptions import ProtocolError if TYPE_CHECKING: - from typing import Any, DefaultDict, NoReturn, Optional, Sequence, Type + from typing import Any, DefaultDict, Optional, Sequence, Type from typing_extensions import Literal @@ -333,17 +334,17 @@ def __length_hint__(self) -> 'Literal[8]': return 8 @classmethod - def __index__(cls) -> 'NoReturn': # pylint: disable=invalid-index-returned + def __index__(cls) -> 'Enum_ARPHRD': # pylint: disable=invalid-index-returned """Numeral registry index of the protocol. - Raises: - UnsupportedCall: This protocol is reached only through - :class:`~pcapkit.protocols.link.linux_sll.LinuxSLL`'s - ``arphrd_type`` dispatch, not through a link-layer header - type of its own. + Returns: + Numeral registry index of the protocol in Linux ARP hardware + types, the ``arphrd_type`` that + :attr:`LinuxSLL.__proto__ ` + dispatches here. """ - raise UnsupportedCall(f'{cls.__name__!r} object cannot be interpreted as an integer') + return Enum_ARPHRD.IEEE80211_RADIOTAP # type: ignore[return-value] ########################################################################## # Utilities. diff --git a/tests/foundation/registry/test_protocols.py b/tests/foundation/registry/test_protocols.py index b4ef103a8f..8cecea13ad 100644 --- a/tests/foundation/registry/test_protocols.py +++ b/tests/foundation/registry/test_protocols.py @@ -456,6 +456,7 @@ def test_top_level_link_internet_and_transport_protocol_wrappers(self) -> None: from pcapkit.const.reg.apptype import TCP as AppType_TCP from pcapkit.const.reg.apptype import UDP as AppType_UDP from pcapkit.const.ipx.packet import Packet as IPXPacket + from pcapkit.const.linux.arphrd import ARPHRD from pcapkit.const.reg.apptype import TransportProtocol from pcapkit.const.reg.ethertype import EtherType from pcapkit.const.reg.linktype import LinkType @@ -475,6 +476,9 @@ def test_top_level_link_internet_and_transport_protocol_wrappers(self) -> None: (registry.register_ethertype, EtherType.Internet_Protocol_version_4, [(registry.Link, 'register')]), (registry.register_loopback, 2, [(registry.Loopback, 'register')]), + (registry.register_arphrd, ARPHRD.FRAD, [(registry.LinuxSLL, 'register')]), + (registry.register_frad, 0xCC, [(registry.FRAD, 'register')]), + (registry.register_netlink, 16, [(registry.Netlink, 'register')]), (registry.register_transtype, TransType.TCP, [(registry.Internet, 'register')]), (registry.register_ipx, IPXPacket.NCP, [(registry.IPX, 'register')]), ] diff --git a/tests/foundation/registry/test_registrar_code_check_unit.py b/tests/foundation/registry/test_registrar_code_check_unit.py index 915fdfea34..98ab2e17f3 100644 --- a/tests/foundation/registry/test_registrar_code_check_unit.py +++ b/tests/foundation/registry/test_registrar_code_check_unit.py @@ -14,7 +14,7 @@ compares every registry with the snapshot. The registrars that take a plain ``int`` (``NUMBER_REGISTRARS``) accept the ``int``, but still reject :obj:`None` and a :class:`str`. Those keyed by no enum at all -(``INT_ONLY_REGISTRARS``) take nothing but an ``int`` (#2511). +(``INT_ONLY_REGISTRARS``) take nothing but an ``int`` (#2510, #2511). Every registrar, those included, also rejects a member of an enum other than the one it is typed on (#2224). The registry enums compare equal to their @@ -90,6 +90,7 @@ ('register_pcap', ('pcapkit.protocols.misc.pcap.frame:Frame',)), ('register_pcapng', ('pcapkit.protocols.misc.pcapng:PCAPNG',)), ('register_ethertype', ('pcapkit.protocols.link.link:Link',)), + ('register_arphrd', ('pcapkit.protocols.link.linux_sll:LinuxSLL',)), ('register_transtype', ('pcapkit.protocols.internet.internet:Internet',)), ) @@ -121,12 +122,13 @@ 'register_pcap': 'pcapkit.const.reg.linktype:LinkType', 'register_pcapng': 'pcapkit.const.reg.linktype:LinkType', 'register_ethertype': 'pcapkit.const.reg.ethertype:EtherType', + 'register_arphrd': 'pcapkit.const.linux.arphrd:ARPHRD', 'register_transtype': 'pcapkit.const.reg.transtype:TransType', } -#: Registrars that also take a plain ``int``: the port, PPID, packet type or -#: address family used, the classes whose ``__proto__`` they write, and any -#: trailing arguments. +#: Registrars that also take a plain ``int``: the port, PPID, packet type, +#: address family, NLPID or message type used, the classes whose +#: ``__proto__`` they write, and any trailing arguments. NUMBER_REGISTRARS = ( ('register_loopback', 2, ('pcapkit.protocols.link.loopback:Loopback',), ()), ('register_ipx', 17, ('pcapkit.protocols.internet.ipx:IPX',), ()), @@ -135,11 +137,13 @@ ('register_sctp', 60, ('pcapkit.protocols.transport.sctp:SCTP',), ()), ('register_apptype', 80, ('pcapkit.protocols.transport.tcp:TCP', 'pcapkit.protocols.transport.udp:UDP'), ('tcp',)), + ('register_frad', 0xCC, ('pcapkit.protocols.link.frad:FRAD',), ()), + ('register_netlink', 16, ('pcapkit.protocols.link.netlink:Netlink',), ()), ) #: The registrars above whose registry no enum keys, so a plain ``int`` is the -#: only code they take (#2511). -INT_ONLY_REGISTRARS = ('register_loopback',) +#: only code they take (#2510, #2511). +INT_ONLY_REGISTRARS = ('register_loopback', 'register_frad', 'register_netlink') def resolve(path: 'str') -> 'type': @@ -270,13 +274,16 @@ def test_number_registrars_reject_a_bool(self) -> None: registrars._check_code(True, registrars.Enum_AppType, TCP, number=True) def test_int_only_registrars_reject_anything_but_an_int(self) -> None: - """No enum keys Loopback's registry, so only a plain ``int`` is a code (#2511). + """No enum keys Loopback's, FRAD's or Netlink's registry, so only a plain + ``int`` is a code (#2510, #2511). ``_check_code`` with ``kind=None`` is the check, so its message keeps the shape the enum-typed ones have. """ + from pcapkit.const.linux.arphrd import ARPHRD from pcapkit.const.reg.transtype import TransType from pcapkit.foundation.registry import protocols as registrars + from pcapkit.protocols.link.frad import FRAD from pcapkit.protocols.link.loopback import Loopback from pcapkit.protocols.misc.raw import Raw from pcapkit.utilities.exceptions import RegistryError @@ -284,13 +291,18 @@ def test_int_only_registrars_reject_anything_but_an_int(self) -> None: for name, _, targets, extra in NUMBER_REGISTRARS: if name in INT_ONLY_REGISTRARS: self._assert_rejected_unchanged(name, tuple(map(resolve, targets)), Raw, *extra, - codes=(None, '2', 2.0, TransType.IPv4)) + codes=(None, '2', 2.0, TransType.IPv4, ARPHRD.FRAD)) for code, message in ((None, 'None'), (True, 'True'), (TransType.IPv4, 'a member of TransType: ')): with self.subTest(code=code): with self.assertRaisesRegex(RegistryError, rf'^Loopback: code must be an int, not {message}'): registrars._check_code(code, None, Loopback) registrars._check_code(2, None, Loopback) + with self.assertRaisesRegex(RegistryError, r'^FRAD: code must be an int, not None$'): + registrars._check_code(None, None, FRAD) + with self.assertRaisesRegex(RegistryError, + r'^FRAD: code must be an int, not a member of ARPHRD: '): + registrars._check_code(ARPHRD.FRAD, None, FRAD) def test_every_rejection_names_the_destination_registry(self) -> None: """The message says *whose* registry rejected the key (#2268). diff --git a/tests/foundation/registry/test_registry_symmetry_unit.py b/tests/foundation/registry/test_registry_symmetry_unit.py index 878d76db2a..8ef322d249 100644 --- a/tests/foundation/registry/test_registry_symmetry_unit.py +++ b/tests/foundation/registry/test_registry_symmetry_unit.py @@ -77,6 +77,8 @@ def resolve(path: 'str') -> 'Any': 'PCAPNG': 'pcapkit.protocols.misc.pcapng:PCAPNG', 'Link': 'pcapkit.protocols.link.link:Link', 'Loopback': 'pcapkit.protocols.link.loopback:Loopback', + 'LinuxSLL': 'pcapkit.protocols.link.linux_sll:LinuxSLL', + 'FRAD': 'pcapkit.protocols.link.frad:FRAD', 'Internet': 'pcapkit.protocols.internet.internet:Internet', 'TCP': 'pcapkit.protocols.transport.tcp:TCP', 'UDP': 'pcapkit.protocols.transport.udp:UDP', @@ -183,6 +185,9 @@ class ModuleRegistrar(NamedTuple): 'pcapkit.const.reg.ethertype:EtherType.Address_Resolution_Protocol', 'Link', 'pcapkit.protocols.link.link:Link'), ModuleRegistrar('register_loopback', 2, 'Loopback', 'pcapkit.protocols.internet.internet:Internet'), + ModuleRegistrar('register_arphrd', 'pcapkit.const.linux.arphrd:ARPHRD.FRAD', 'LinuxSLL', + 'pcapkit.protocols.link.link:Link'), + ModuleRegistrar('register_frad', 0xCC, 'FRAD', 'pcapkit.protocols.internet.internet:Internet'), ModuleRegistrar('register_transtype', 'pcapkit.const.reg.transtype:TransType.TCP', 'Internet', 'pcapkit.protocols.transport.transport:Transport'), ModuleRegistrar('register_tcp', 80, 'TCP', 'pcapkit.protocols.application.application:Application'), @@ -444,6 +449,7 @@ def test_every_public_registrar_is_covered_or_excused(self) -> None: 'register_traceflow_tcp_callback': 'append-only, nothing to undo with', 'register_icmp_type': 'no schema argument; override and restore are in test_icmp_type_registrar_unit', 'register_ipx': 'IPX.__proto__ ships no entry, so an override has nothing to be restored to', + 'register_netlink': 'Netlink.__proto__ ships no entry, so there is none to restore', } self.assertEqual(sorted(set(registry.__all__) - covered - set(excused)), []) self.assertEqual(sorted((covered | set(excused)) - set(registry.__all__)), []) diff --git a/tests/foundation/registry/test_registry_wrapper_coverage_2511_unit.py b/tests/foundation/registry/test_registry_wrapper_coverage_2511_unit.py index f3b927301d..882688f881 100644 --- a/tests/foundation/registry/test_registry_wrapper_coverage_2511_unit.py +++ b/tests/foundation/registry/test_registry_wrapper_coverage_2511_unit.py @@ -46,8 +46,6 @@ EXCUSED = { 'pcapkit.protocols.protocol.ProtocolBase.__proto__': 'the empty default a class without a map of its own inherits; nothing dispatches by it', - 'pcapkit.protocols.link.frad.FRAD.__proto__': 'wrapper and classmethod owned by #2510', - 'pcapkit.protocols.link.netlink.Netlink.__proto__': 'wrapper and classmethod owned by #2510', } #: ``map path -> (the classmethod its owner inherits, why it fits the map's keys)``. diff --git a/tests/protocols/link/test_frad_unit.py b/tests/protocols/link/test_frad_unit.py index e735afe6f0..eba1e7a58f 100644 --- a/tests/protocols/link/test_frad_unit.py +++ b/tests/protocols/link/test_frad_unit.py @@ -5,7 +5,7 @@ ``ARPHRD_FRAD`` payload that :class:`~pcapkit.protocols.link.linux_sll.LinuxSLL` previously kept whole as :class:`~pcapkit.protocols.misc.raw.Raw` (see -:data:`~pcapkit.protocols.link.linux_sll._ARPHRD_NOT_ETHERTYPE`). Covers the +:attr:`LinuxSLL.__proto__ `). Covers the Q.922 address field, the direct-NLPID and SNAP-encapsulated forms per :rfc:`2427`, and byte-exact round trips. The FRAD-dispatch side of ``LinuxSLL1``/``LinuxSLL2`` is covered in their own test modules. @@ -229,12 +229,16 @@ def test_make_rejects_snap_without_snap_fields(self) -> None: FRAD(nlpid=0x80, snap_oui=b'\x00\x00\x00', snap_pid=0x1_0000) # out of range def test_id_and_index(self) -> None: + """``__index__`` is the ``arphrd_type`` LinuxSLL dispatches here (#2510).""" + from pcapkit.const.linux.arphrd import ARPHRD + from pcapkit.protocols.link.linux_sll import LinuxSLL from pcapkit.protocols.link.frad import FRAD - from pcapkit.utilities.exceptions import UnsupportedCall self.assertEqual(FRAD.id(), ('FRAD',)) - with self.assertRaises(UnsupportedCall): - FRAD.__index__() + self.assertIs(FRAD.__index__(), ARPHRD.FRAD) + self.assertEqual(FRAD.__index__(), 770) + self.assertIs(LinuxSLL._lookup_next_layer(LinuxSLL.__proto__, FRAD.__index__()), + FRAD) def test_truncated_header_rebuilds_as_captured(self) -> None: """A capture ending inside the header must not blow up the parse, and diff --git a/tests/protocols/link/test_gre_unit.py b/tests/protocols/link/test_gre_unit.py index 6d64365974..bec3ae6ab2 100644 --- a/tests/protocols/link/test_gre_unit.py +++ b/tests/protocols/link/test_gre_unit.py @@ -5,7 +5,7 @@ ``ARPHRD_IPGRE``/``ARPHRD_IP6GRE`` payload that :class:`~pcapkit.protocols.link.linux_sll.LinuxSLL` previously kept whole as :class:`~pcapkit.protocols.misc.raw.Raw` (see -:data:`~pcapkit.protocols.link.linux_sll._ARPHRD_NOT_ETHERTYPE`). Covers the +:attr:`LinuxSLL.__proto__ `). Covers the base header [:rfc:`2784`], the optional Key and Sequence Number extensions [:rfc:`2890`], the EtherType-keyed dispatch through :attr:`Link.__proto__ `, and @@ -235,12 +235,19 @@ def test_make_rejects_what_does_not_fit(self) -> None: GRE(**kwargs) def test_id_and_index(self) -> None: + """``__index__`` is ``IPGRE``, which LinuxSLL dispatches here alongside + ``IP6GRE`` (#2510). + """ + from pcapkit.const.linux.arphrd import ARPHRD from pcapkit.protocols.link.gre import GRE - from pcapkit.utilities.exceptions import UnsupportedCall + from pcapkit.protocols.link.linux_sll import LinuxSLL self.assertEqual(GRE.id(), ('GRE',)) - with self.assertRaises(UnsupportedCall): - GRE.__index__() + self.assertIs(GRE.__index__(), ARPHRD.IPGRE) + self.assertEqual(GRE.__index__(), 778) + for arphrd in (ARPHRD.IPGRE, ARPHRD.IP6GRE): + with self.subTest(arphrd=arphrd): + self.assertIs(LinuxSLL._lookup_next_layer(LinuxSLL.__proto__, arphrd), GRE) def test_unregistered_protocol_type_is_kept_raw(self) -> None: from pcapkit.protocols.link.gre import GRE diff --git a/tests/protocols/link/test_linux_sll1_unit.py b/tests/protocols/link/test_linux_sll1_unit.py index a563f1444b..e174a6c5d8 100644 --- a/tests/protocols/link/test_linux_sll1_unit.py +++ b/tests/protocols/link/test_linux_sll1_unit.py @@ -2,8 +2,8 @@ """GitHub issue #2438: the Linux "cooked" capture encapsulation v1 (``LINUX_SLL``). Covers the 16-octet header, the ``arphrd_type``-gated next-layer rule (see -:data:`~pcapkit.protocols.link.linux_sll._ARPHRD_NOT_ETHERTYPE`), link-address -validation, and byte-exact round trips. +:attr:`LinuxSLL.__proto__ `), +link-address validation, and byte-exact round trips. Each test imports :mod:`pcapkit` itself, after :func:`~tests._support.reimport_once_per_class`. @@ -75,8 +75,8 @@ def test_ethernet_arphrd_dispatches_by_ethertype(self) -> None: def test_netlink_arphrd_dispatches_to_netlink(self) -> None: """``ARPHRD_NETLINK`` dispatches to :class:`~pcapkit.protocols.link.netlink.Netlink`, not - :class:`~pcapkit.protocols.misc.raw.Raw` (:issue:`2500`), even though - it too is in :data:`~pcapkit.protocols.link.linux_sll._ARPHRD_NOT_ETHERTYPE`. + :class:`~pcapkit.protocols.misc.raw.Raw` (:issue:`2500`), through + :attr:`LinuxSLL.__proto__ `. """ import struct @@ -132,17 +132,16 @@ def test_frad_arphrd_dispatches_to_frad(self) -> None: self._assert_rebuilds(parsed) def test_frad_protocol_type_is_not_an_ethertype(self) -> None: - """``ARPHRD_FRAD`` is in - :data:`~pcapkit.protocols.link.linux_sll._ARPHRD_NOT_ETHERTYPE`, so its - ``protocol_type`` is never looked up as an EtherType, per tcpdump.org's - LINUX_SLL page (:issue:`2460`). + """``ARPHRD_FRAD`` is a key of + :attr:`LinuxSLL.__proto__ `, + so its ``protocol_type`` is never looked up as an EtherType, per + tcpdump.org's LINUX_SLL page (:issue:`2460`). """ from pcapkit.const.linux.arphrd import ARPHRD - from pcapkit.protocols.link.linux_sll import (_ARPHRD_NOT_ETHERTYPE, - _resolve_next_layer_proto) + from pcapkit.protocols.link.linux_sll import LinuxSLL, _resolve_next_layer_proto - self.assertIn(ARPHRD.FRAD, _ARPHRD_NOT_ETHERTYPE) - proto = _resolve_next_layer_proto(ARPHRD.FRAD, 0x0800) + self.assertIn(ARPHRD.FRAD, LinuxSLL.__proto__) + proto = _resolve_next_layer_proto(LinuxSLL.__proto__, ARPHRD.FRAD, 0x0800) self.assertIs(type(proto), int) self.assertEqual(proto, 0x0800) diff --git a/tests/protocols/link/test_linux_sll2_unit.py b/tests/protocols/link/test_linux_sll2_unit.py index 39d2ef47b9..5059c53a98 100644 --- a/tests/protocols/link/test_linux_sll2_unit.py +++ b/tests/protocols/link/test_linux_sll2_unit.py @@ -2,7 +2,7 @@ """GitHub issue #2438: the Linux "cooked" capture encapsulation v2 (``LINUX_SLL2``). Covers the 20-octet header, the same ``arphrd_type``-gated next-layer rule as -v1 (see :data:`~pcapkit.protocols.link.linux_sll._ARPHRD_NOT_ETHERTYPE`), +v1 (see :attr:`LinuxSLL.__proto__ `), link-address validation, and byte-exact round trips. Each test imports :mod:`pcapkit` itself, after @@ -77,8 +77,8 @@ def test_ethernet_arphrd_dispatches_by_ethertype(self) -> None: def test_netlink_arphrd_dispatches_to_netlink(self) -> None: """``ARPHRD_NETLINK`` dispatches to :class:`~pcapkit.protocols.link.netlink.Netlink`, not - :class:`~pcapkit.protocols.misc.raw.Raw` (:issue:`2500`), even though - it too is in :data:`~pcapkit.protocols.link.linux_sll._ARPHRD_NOT_ETHERTYPE`. + :class:`~pcapkit.protocols.misc.raw.Raw` (:issue:`2500`), through + :attr:`LinuxSLL.__proto__ `. """ import struct @@ -133,17 +133,16 @@ def test_frad_arphrd_dispatches_to_frad(self) -> None: self._assert_rebuilds(parsed) def test_frad_protocol_type_is_not_an_ethertype(self) -> None: - """``ARPHRD_FRAD`` is in - :data:`~pcapkit.protocols.link.linux_sll._ARPHRD_NOT_ETHERTYPE`, so its - ``protocol_type`` is never looked up as an EtherType, per tcpdump.org's - LINUX_SLL2 page (:issue:`2460`). + """``ARPHRD_FRAD`` is a key of + :attr:`LinuxSLL.__proto__ `, + so its ``protocol_type`` is never looked up as an EtherType, per + tcpdump.org's LINUX_SLL2 page (:issue:`2460`). """ from pcapkit.const.linux.arphrd import ARPHRD - from pcapkit.protocols.link.linux_sll import (_ARPHRD_NOT_ETHERTYPE, - _resolve_next_layer_proto) + from pcapkit.protocols.link.linux_sll import LinuxSLL, _resolve_next_layer_proto - self.assertIn(ARPHRD.FRAD, _ARPHRD_NOT_ETHERTYPE) - proto = _resolve_next_layer_proto(ARPHRD.FRAD, 0x0800) + self.assertIn(ARPHRD.FRAD, LinuxSLL.__proto__) + proto = _resolve_next_layer_proto(LinuxSLL.__proto__, ARPHRD.FRAD, 0x0800) self.assertIs(type(proto), int) self.assertEqual(proto, 0x0800) diff --git a/tests/protocols/link/test_linux_sll_arphrd_registry_unit.py b/tests/protocols/link/test_linux_sll_arphrd_registry_unit.py new file mode 100644 index 0000000000..7b88aab280 --- /dev/null +++ b/tests/protocols/link/test_linux_sll_arphrd_registry_unit.py @@ -0,0 +1,337 @@ +# -*- coding: utf-8 -*- +"""GitHub issue #2510: LinuxSLL dispatches its non-EtherType ``arphrd_type`` values +through a registry of its own. + +Covers :attr:`LinuxSLL.__proto__ ` +as shipped, dispatch through it and through the EtherType path from both versions, +the ``__index__`` of the classes it reaches, and the registrars of it and of +:class:`~pcapkit.protocols.link.frad.FRAD`'s and +:class:`~pcapkit.protocols.link.netlink.Netlink`'s registries. + +Each test imports :mod:`pcapkit` itself, after +:func:`~tests._support.reimport_once_per_class`. + +""" + +import struct +import unittest +import warnings + +from tests._support import reimport_once_per_class + +#: An IPv4 header (DF clear, protocol UDP) and an 8-octet UDP header. +IPV4 = (bytes.fromhex('4500001c00010000401100000a0000010a000002') + + bytes.fromhex('3039003500080000')) + +#: The payload each ``arphrd_type`` case carries, and the chain it parses to +#: after the LinuxSLL layer. +CASES = { + 'FRAD': (bytes.fromhex('0c21') + b'\x03\xcc' + IPV4, 'FRAD:IPv4:UDP'), + 'IPGRE': (bytes.fromhex('00000800') + IPV4, 'GRE:IPv4:UDP'), + 'IP6GRE': (bytes.fromhex('00000800') + IPV4, 'GRE:IPv4:UDP'), + 'IEEE80211_RADIOTAP': (bytes.fromhex('00' '00' '0800' '00000000') + b'\xaa' * 10, 'Radiotap:Raw'), + 'NETLINK': (struct.pack(' bytes: + """A ``LINUX_SLL`` header carrying ``protocol_type`` on ``arphrd_type``.""" + return struct.pack('>HHH8sH', 0, arphrd_type, 0, b'', protocol_type) + + +def _sll2(protocol_type: int, arphrd_type: int) -> bytes: + """A ``LINUX_SLL2`` header carrying ``protocol_type`` on ``arphrd_type``.""" + return struct.pack('>HHIHBB8s', protocol_type, 0, 0, arphrd_type, 0, 0, b'') + + +class LinuxSLLRegistryTests(unittest.TestCase): + """Pin ``LinuxSLL.__proto__`` and the dispatch it drives.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _versions(self): # type: ignore[no-untyped-def] + from pcapkit.protocols.link.linux_sll1 import LinuxSLL1 + from pcapkit.protocols.link.linux_sll2 import LinuxSLL2 + + return ((LinuxSLL1, _sll1, 'LINUX_SLL1'), (LinuxSLL2, _sll2, 'LINUX_SLL2')) + + def _snapshot(self): # type: ignore[no-untyped-def] + """Every registry the registrars here write, to put back after a test.""" + from pcapkit.foundation.registry import protocols as registrars + from pcapkit.protocols import __proto__ as names + from pcapkit.protocols.link.frad import FRAD + from pcapkit.protocols.link.linux_sll import LinuxSLL + from pcapkit.protocols.link.netlink import Netlink + + tables = (LinuxSLL.__proto__, FRAD.__proto__, Netlink.__proto__, names, + registrars._CODE_LEASES, registrars._NAME_LEASES) + copies = [dict(table) for table in tables] + + def restore() -> None: + for table, copy in zip(tables, copies): + table.clear() + table.update(copy) + + self.addCleanup(restore) + + def test_the_registry_as_shipped(self) -> None: + """Keyed by ``arphrd_type``, owned by the base, shared by both versions.""" + from pcapkit.const.linux.arphrd import ARPHRD + from pcapkit.corekit.module import ModuleDescriptor + from pcapkit.protocols.link.link import Link + from pcapkit.protocols.link.linux_sll import LinuxSLL + + self.assertEqual(dict(LinuxSLL.__proto__), { + ARPHRD.FRAD: ModuleDescriptor('pcapkit.protocols.link.frad', 'FRAD'), + ARPHRD.IPGRE: ModuleDescriptor('pcapkit.protocols.link.gre', 'GRE'), + ARPHRD.IEEE80211_RADIOTAP: ModuleDescriptor('pcapkit.protocols.link.radiotap', 'Radiotap'), + ARPHRD.IP6GRE: ModuleDescriptor('pcapkit.protocols.link.gre', 'GRE'), + ARPHRD.NETLINK: ModuleDescriptor('pcapkit.protocols.link.netlink', 'Netlink'), + }) + self.assertIn('__proto__', vars(LinuxSLL)) + self.assertIsNot(LinuxSLL.__proto__, Link.__proto__) + self.assertEqual(vars(LinuxSLL)['__proto_key__'], (ARPHRD, False)) + for version, _, _ in self._versions(): + with self.subTest(version=version.__name__): + self.assertNotIn('__proto__', vars(version)) + self.assertIs(version.__proto__, LinuxSLL.__proto__) + + def test_each_registered_arphrd_type_dispatches_through_the_registry(self) -> None: + """``protocol_type`` is never read as an EtherType on these, even when it is one.""" + from pcapkit.const.linux.arphrd import ARPHRD + + for version, header, alias in self._versions(): + for name, (payload, chain) in CASES.items(): + with self.subTest(version=version.__name__, arphrd=name): + raw = header(0x0800, getattr(ARPHRD, name)) + payload + parsed = version(raw, len(raw)) + self.assertEqual(parsed.protochain.chain, f'{alias}:{chain}') + self.assertEqual(parsed.data, raw) + self.assertEqual(version.from_data(parsed.info).data, raw) + + def test_every_other_arphrd_type_dispatches_by_ethertype(self) -> None: + """Ethernet and an ``arphrd_type`` ARPHRD names no member for, alike.""" + from pcapkit.const.reg.ethertype import EtherType + from pcapkit.protocols.misc.raw import Raw + + for version, header, alias in self._versions(): + for arphrd in (1, 0xFFFE): + with self.subTest(version=version.__name__, arphrd=arphrd): + raw = header(0x0800, arphrd) + IPV4 + parsed = version(raw, len(raw)) + self.assertEqual(parsed.protochain.chain, f'{alias}:IPv4:UDP') + + raw = header(0x88CC, arphrd) + IPV4 + parsed = version(raw, len(raw)) + self.assertIsInstance(parsed.payload, Raw) + self.assertIs(parsed.payload.info.protocol, + EtherType.IEEE_Std_802_1AB_Link_Layer_Discovery_Protocol) + + def test_an_empty_payload_is_no_payload_on_either_path(self) -> None: + from pcapkit.const.linux.arphrd import ARPHRD + from pcapkit.protocols.misc.null import NoPayload + + for version, header, _ in self._versions(): + for arphrd in (ARPHRD.FRAD, 1): + with self.subTest(version=version.__name__, arphrd=arphrd): + raw = header(0x0800, arphrd) + self.assertIsInstance(version(raw, len(raw)).payload, NoPayload) + + def test_a_stop_at_this_layer_still_parses_a_registered_arphrd_type(self) -> None: + """The order the NOTE in ``LinuxSLL._import_next_layer`` records. + + Stopping at the link layer or at the version keeps an EtherType payload + :class:`~pcapkit.protocols.misc.raw.Raw`, while a registered + ``arphrd_type`` is still dispatched. + """ + from pcapkit.const.linux.arphrd import ARPHRD + + for version, header, alias in self._versions(): + for stop in ({'layer': 'Link'}, {'protocol': version.__name__}): + with self.subTest(version=version.__name__, stop=stop): + raw = header(0x0800, 1) + IPV4 + self.assertEqual(version(raw, len(raw), **stop).protochain.chain, + f'{alias}:Internet_Protocol_version_4') + + raw = header(0x0800, ARPHRD.NETLINK) + CASES['NETLINK'][0] + self.assertEqual(version(raw, len(raw), **stop).protochain.chain, + f'{alias}:Netlink:Raw') + + def test_index_is_the_arphrd_type_that_dispatches_here(self) -> None: + from pcapkit.const.linux.arphrd import ARPHRD + from pcapkit.protocols.link.frad import FRAD + from pcapkit.protocols.link.gre import GRE + from pcapkit.protocols.link.linux_sll import LinuxSLL + from pcapkit.protocols.link.netlink import Netlink + from pcapkit.protocols.link.radiotap import Radiotap + + for klass, member in ((FRAD, ARPHRD.FRAD), (GRE, ARPHRD.IPGRE), + (Radiotap, ARPHRD.IEEE80211_RADIOTAP), (Netlink, ARPHRD.NETLINK)): + with self.subTest(protocol=klass.__name__): + self.assertIs(klass.__index__(), member) + self.assertIs(LinuxSLL._lookup_next_layer(LinuxSLL.__proto__, klass.__index__()), + klass) + self.assertIs(LinuxSLL._lookup_next_layer(LinuxSLL.__proto__, ARPHRD.IP6GRE), GRE) + + ########################################################################## + # The registrars. + ########################################################################## + + def test_register_arphrd_reaches_dispatch(self) -> None: + """A registered ``arphrd_type`` stops its ``protocol_type`` being an EtherType.""" + from pcapkit.const.linux.arphrd import ARPHRD + from pcapkit.foundation.registry import register_arphrd + from pcapkit.protocols.link.linux_sll import LinuxSLL + from pcapkit.protocols.link.netlink import Netlink + + self._snapshot() + loopback = ARPHRD(772) + self.assertNotIn(loopback, LinuxSLL.__proto__) + register_arphrd(loopback, Netlink) + self.assertIs(LinuxSLL.__proto__[loopback], Netlink) + + for version, header, alias in self._versions(): + with self.subTest(version=version.__name__): + raw = header(0x0800, 772) + CASES['NETLINK'][0] + parsed = version(raw, len(raw)) + self.assertEqual(parsed.protochain.chain, f'{alias}:Netlink:Raw') + self.assertEqual(version.from_data(parsed.info).data, raw) + + def test_register_arphrd_by_name_and_over_a_shipped_entry(self) -> None: + """The ``(module, class_)`` form, a warning naming the key, and a silent restore.""" + from pcapkit.const.linux.arphrd import ARPHRD + from pcapkit.corekit.module import ModuleDescriptor + from pcapkit.foundation.registry import register_arphrd + from pcapkit.protocols.link.linux_sll import LinuxSLL + from pcapkit.protocols.misc.raw import Raw + from pcapkit.utilities.warnings import RegistryWarning + + self._snapshot() + shipped = LinuxSLL.__proto__[ARPHRD.FRAD] + with self.assertWarnsRegex(RegistryWarning, r'^ARP hardware type '): + register_arphrd(ARPHRD.FRAD, 'pcapkit.protocols.misc.raw', 'Raw') + self.assertIs(LinuxSLL.__proto__[ARPHRD.FRAD], Raw) + + for version, header, alias in self._versions(): + with self.subTest(version=version.__name__): + raw = header(0x0800, ARPHRD.FRAD) + CASES['FRAD'][0] + self.assertEqual(version(raw, len(raw)).protochain.chain, f'{alias}:Raw') + + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + register_arphrd(ARPHRD.FRAD, ModuleDescriptor('pcapkit.protocols.link.frad', 'FRAD')) + self.assertIs(LinuxSLL.__proto__[ARPHRD.FRAD], shipped) + with warnings.catch_warnings(): + warnings.simplefilter('error') + register_arphrd(ARPHRD.FRAD, ModuleDescriptor('pcapkit.protocols.link.frad', 'FRAD')) + + def test_register_frad_reaches_dispatch(self) -> None: + """A registered NLPID dispatches, from FRAD alone and from a LinuxSLL frame.""" + from pcapkit.const.linux.arphrd import ARPHRD + from pcapkit.foundation.registry import register_frad + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.link.frad import FRAD + from pcapkit.utilities.warnings import RegistryWarning + + self._snapshot() + frad = bytes.fromhex('0c21') + b'\x03\x81' + IPV4 # 0x81, ISO CLNP: not registered + self.assertEqual(FRAD(frad, len(frad)).protochain.chain, 'FRAD:Raw') + + register_frad(0x81, IPv4) + with warnings.catch_warnings(): + warnings.simplefilter('error') + register_frad(0x81, IPv4) # the entry already stored: silent + self.assertIs(FRAD.__proto__[0x81], IPv4) + self.assertEqual(FRAD(frad, len(frad)).protochain.chain, 'FRAD:IPv4:UDP') + for version, header, alias in self._versions(): + with self.subTest(version=version.__name__): + raw = header(0x0800, ARPHRD.FRAD) + frad + self.assertEqual(version(raw, len(raw)).protochain.chain, + f'{alias}:FRAD:IPv4:UDP') + + with self.assertWarnsRegex(RegistryWarning, r'^NLPID 129 already registered'): + register_frad(0x81, 'pcapkit.protocols.internet.ipv6', 'IPv6') + + def test_register_netlink_reaches_dispatch(self) -> None: + """Netlink ships an empty registry; a registered message type dispatches.""" + from pcapkit.const.linux.arphrd import ARPHRD + from pcapkit.foundation.registry import register_netlink + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.link.netlink import Netlink + from pcapkit.utilities.warnings import RegistryWarning + + self._snapshot() + nlmsg = struct.pack(' None: + """Each refuses before writing anything, naming the registry that refused.""" + from pcapkit.const.linux.arphrd import ARPHRD + from pcapkit.const.reg.ethertype import EtherType + from pcapkit.foundation.registry import register_arphrd, register_frad, register_netlink + from pcapkit.protocols.link.frad import FRAD + from pcapkit.protocols.link.linux_sll import LinuxSLL + from pcapkit.protocols.link.netlink import Netlink + from pcapkit.protocols.misc.raw import Raw + from pcapkit.utilities.exceptions import RegistryError + + cases = ( + (register_arphrd, EtherType(770), r'^LinuxSLL: code must be a member of ARPHRD, ' + r'not a member of EtherType: '), + (register_arphrd, 770, r'^LinuxSLL: code must be a member of ARPHRD, not 770$'), + (register_frad, ARPHRD.FRAD, r'^FRAD: code must be an int, not a member of ARPHRD: '), + (register_frad, True, r'^FRAD: code must be an int, not True$'), + (register_netlink, '3', r"^Netlink: code must be an int, not '3'$"), + ) + tables = (LinuxSLL.__proto__, FRAD.__proto__, Netlink.__proto__) + for register, code, message in cases: + with self.subTest(registrar=register.__name__, code=code): + before = [dict(table) for table in tables] + with self.assertRaisesRegex(RegistryError, message): + register(code, Raw) + self.assertEqual([dict(table) for table in tables], before) + + def test_register_protocol_code_reads_the_owner_pairing(self) -> None: + """A version resolves to ``LinuxSLL``'s pairing; FRAD and Netlink declare none.""" + from pcapkit.const.linux.arphrd import ARPHRD + from pcapkit.const.reg.ethertype import EtherType + from pcapkit.foundation.registry import register_protocol_code + from pcapkit.protocols.link.frad import FRAD + from pcapkit.protocols.link.linux_sll import LinuxSLL + from pcapkit.protocols.link.netlink import Netlink + from pcapkit.protocols.misc.raw import Raw + from pcapkit.utilities.exceptions import RegistryError + + self._snapshot() + for version, _, _ in self._versions(): + with self.subTest(version=version.__name__): + with self.assertRaisesRegex(RegistryError, r'^LinuxSLL: code must be a member of ARPHRD'): + register_protocol_code(Raw, {version: EtherType.Internet_Protocol_version_4}) + register_protocol_code(Raw, {version: ARPHRD(772)}) + self.assertIs(LinuxSLL.__proto__[ARPHRD(772)], Raw) + + for owner in (FRAD, Netlink): + with self.subTest(owner=owner.__name__): + self.assertNotIn('__proto_key__', vars(owner)) + register_protocol_code(Raw, {owner: 0x7F}) + self.assertIs(owner.__proto__[0x7F], Raw) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/link/test_netlink_unit.py b/tests/protocols/link/test_netlink_unit.py index 848c2d23d6..340ff7656b 100644 --- a/tests/protocols/link/test_netlink_unit.py +++ b/tests/protocols/link/test_netlink_unit.py @@ -5,7 +5,7 @@ ``ARPHRD_NETLINK`` payload that :class:`~pcapkit.protocols.link.linux_sll.LinuxSLL` previously kept whole as :class:`~pcapkit.protocols.misc.raw.Raw` (see -:data:`~pcapkit.protocols.link.linux_sll._ARPHRD_NOT_ETHERTYPE`). Covers the +:attr:`LinuxSLL.__proto__ `). Covers the 16-octet ``nlmsghdr`` fields, the byte-order heuristic :func:`~pcapkit.protocols.link.netlink.nlmsg_byteorder` resolves (see :rfc:`3549#section-2.3.2`), and byte-exact round trips. The NETLINK-dispatch @@ -197,12 +197,16 @@ def test_make_explicit_nlmsg_len_is_written_as_given(self) -> None: self.assertEqual(made.info.nlmsg_len, 999) def test_id_and_index(self) -> None: + """``__index__`` is the ``arphrd_type`` LinuxSLL dispatches here (#2510).""" + from pcapkit.const.linux.arphrd import ARPHRD + from pcapkit.protocols.link.linux_sll import LinuxSLL from pcapkit.protocols.link.netlink import Netlink - from pcapkit.utilities.exceptions import UnsupportedCall self.assertEqual(Netlink.id(), ('Netlink',)) - with self.assertRaises(UnsupportedCall): - Netlink.__index__() + self.assertIs(Netlink.__index__(), ARPHRD.NETLINK) + self.assertEqual(Netlink.__index__(), 824) + self.assertIs(LinuxSLL._lookup_next_layer(LinuxSLL.__proto__, Netlink.__index__()), + Netlink) def test_truncated_header_rebuilds_as_captured(self) -> None: """A capture ending inside the header must not blow up the parse, and diff --git a/tests/protocols/link/test_radiotap_unit.py b/tests/protocols/link/test_radiotap_unit.py index 27cc60e8d1..dee14daaee 100644 --- a/tests/protocols/link/test_radiotap_unit.py +++ b/tests/protocols/link/test_radiotap_unit.py @@ -5,7 +5,7 @@ ``arphrd_type`` ``ARPHRD_IEEE80211_RADIOTAP`` payload that :class:`~pcapkit.protocols.link.linux_sll.LinuxSLL` previously kept whole as :class:`~pcapkit.protocols.misc.raw.Raw` (see -:data:`~pcapkit.protocols.link.linux_sll._ARPHRD_NOT_ETHERTYPE`). Covers the +:attr:`LinuxSLL.__proto__ `). Covers the fixed 8-octet start, the present-flags chain (plain and extended), the field-data region being skipped rather than decoded, the 802.11 MAC frame being kept as :class:`~pcapkit.protocols.misc.raw.Raw` (per the module @@ -247,12 +247,16 @@ def test_make_rejects_what_does_not_fit(self) -> None: Radiotap(**kwargs) def test_id_and_index(self) -> None: + """``__index__`` is the ``arphrd_type`` LinuxSLL dispatches here (#2510).""" + from pcapkit.const.linux.arphrd import ARPHRD + from pcapkit.protocols.link.linux_sll import LinuxSLL from pcapkit.protocols.link.radiotap import Radiotap - from pcapkit.utilities.exceptions import UnsupportedCall self.assertEqual(Radiotap.id(), ('Radiotap',)) - with self.assertRaises(UnsupportedCall): - Radiotap.__index__() + self.assertIs(Radiotap.__index__(), ARPHRD.IEEE80211_RADIOTAP) + self.assertEqual(Radiotap.__index__(), 803) + self.assertIs(LinuxSLL._lookup_next_layer(LinuxSLL.__proto__, Radiotap.__index__()), + Radiotap) if __name__ == '__main__': diff --git a/tests/protocols/test_protocol_code_registration_unit.py b/tests/protocols/test_protocol_code_registration_unit.py index 2ec9ec51f9..1ddd8e733c 100644 --- a/tests/protocols/test_protocol_code_registration_unit.py +++ b/tests/protocols/test_protocol_code_registration_unit.py @@ -555,6 +555,7 @@ def test_every_destination_kind_row_is_enforced(self) -> None: reads it: an inherited one would not be its own declaration. """ from pcapkit.const.ipx.packet import Packet + from pcapkit.const.linux.arphrd import ARPHRD from pcapkit.const.reg.apptype import AppType from pcapkit.const.reg.ethertype import EtherType from pcapkit.const.reg.linktype import LinkType @@ -564,6 +565,7 @@ def test_every_destination_kind_row_is_enforced(self) -> None: from pcapkit.protocols.internet.internet import Internet from pcapkit.protocols.internet.ipx import IPX from pcapkit.protocols.link.link import Link + from pcapkit.protocols.link.linux_sll import LinuxSLL from pcapkit.protocols.misc.pcap.frame import Frame from pcapkit.protocols.misc.pcapng import PCAPNG from pcapkit.protocols.misc.raw import Raw @@ -575,6 +577,9 @@ def test_every_destination_kind_row_is_enforced(self) -> None: expected = { # 0x0800 is EtherType.Internet_Protocol_version_4, where Link ships IPv4. Link: (EtherType, False, TransType.ICMP, 0x0800), + # 770 is ARPHRD.FRAD, where LinuxSLL ships FRAD; the EtherType is the + # key of Link's registry, which LinuxSLL's own refuses (#2510). + LinuxSLL: (ARPHRD, False, EtherType.Internet_Protocol_version_4, 770), # 6 is TransType.TCP, where Internet ships TCP. Internet: (TransType, False, EtherType.XEROX_NS_IDP, 6), # TransType.UDP is 17, Packet.NCP: IPX's registry of its own exists so