diff --git a/pcapkit/foundation/registry/protocols.py b/pcapkit/foundation/registry/protocols.py index b4ec54b768..780e68dfc6 100644 --- a/pcapkit/foundation/registry/protocols.py +++ b/pcapkit/foundation/registry/protocols.py @@ -536,6 +536,23 @@ def register_protocol_code(protocol: 'Type[ProtocolBase]', code: 'Any') -> 'None destination.__name__, key) +def _check_code(code: 'Any') -> 'None': + """Reject a registration key that is not an enum member. + + The registrars typed on an enum key call this before touching any + registry, so a bad ``code`` leaves every registry as it was (:issue:`2186`). + + Args: + code: Registration key handed to the registrar. + + Raises: + RegistryError: If ``code`` is not an enum member. + + """ + if not isinstance(code, (enum.Enum, aenum.Enum)): + raise RegistryError(f'code must be an enum member, not {code!r}') + + ############################################################################### # Top-Level Registries ############################################################################### @@ -571,7 +588,13 @@ def register_linktype(code: 'LinkType', module: 'str | ModuleDescriptor[Protocol * :func:`pcapkit.foundation.registry.protocols.register_pcap` * :func:`pcapkit.foundation.registry.protocols.register_pcapng` + Raises: + RegistryError: If ``code`` is not an enum member, or ``module`` + is not a :class:`~pcapkit.protocols.protocol.ProtocolBase` + subclass. + """ + _check_code(code) if isinstance(module, str): module = cast('ModuleDescriptor[ProtocolBase]', ModuleDescriptor(module, class_)) @@ -609,7 +632,13 @@ def register_pcap(code: 'LinkType', module: 'str | ModuleDescriptor[ProtocolBase :class:`~pcapkit.protocols.protocol.ProtocolBase` subclass class\_: class name + Raises: + RegistryError: If ``code`` is not an enum member, or ``module`` + is not a :class:`~pcapkit.protocols.protocol.ProtocolBase` + subclass. + """ + _check_code(code) if isinstance(module, str): module = cast('ModuleDescriptor[ProtocolBase]', ModuleDescriptor(module, class_)) @@ -646,7 +675,13 @@ def register_pcapng(code: 'LinkType', module: 'str | ModuleDescriptor[ProtocolBa :class:`~pcapkit.protocols.protocol.ProtocolBase` subclass class\_: class name + Raises: + RegistryError: If ``code`` is not an enum member, or ``module`` + is not a :class:`~pcapkit.protocols.protocol.ProtocolBase` + subclass. + """ + _check_code(code) if isinstance(module, str): module = cast('ModuleDescriptor[ProtocolBase]', ModuleDescriptor(module, class_)) @@ -688,7 +723,13 @@ def register_ethertype(code: 'EtherType', module: 'str | ModuleDescriptor[Protoc :class:`~pcapkit.protocols.protocol.ProtocolBase` subclass class\_: class name + Raises: + RegistryError: If ``code`` is not an enum member, or ``module`` + is not a :class:`~pcapkit.protocols.protocol.ProtocolBase` + subclass. + """ + _check_code(code) if isinstance(module, str): module = cast('ModuleDescriptor[ProtocolBase]', ModuleDescriptor(module, class_)) @@ -730,7 +771,13 @@ def register_transtype(code: 'TransType', module: 'str | ModuleDescriptor[Protoc :class:`~pcapkit.protocols.protocol.ProtocolBase` subclass class\_: class name + Raises: + RegistryError: If ``code`` is not an enum member, or ``module`` + is not a :class:`~pcapkit.protocols.protocol.ProtocolBase` + subclass. + """ + _check_code(code) if isinstance(module, str): module = cast('ModuleDescriptor[ProtocolBase]', ModuleDescriptor(module, class_)) @@ -762,10 +809,11 @@ def register_ipv4_option(code: 'IPv4_OptionNumber', meth: 'str | tuple[IPv4_Opti It should be a subclass of :class:`pcapkit.protocols.schema.internet.ipv4.Option`. Raises: - RegistryError: If ``meth`` is a name with no matching - ``_read_opt_*`` method. + RegistryError: If ``code`` is not an enum member, or ``meth`` + is a name with no matching ``_read_opt_*`` method. """ + _check_code(code) if isinstance(meth, str) and not hasattr(IPv4, f'_read_opt_{meth}'): raise RegistryError('method must be a valid IPv4 option parser function') @@ -794,10 +842,11 @@ def register_hip_parameter(code: 'HIP_Parameter', meth: 'str | tuple[HIP_Paramet It should be a subclass of :class:`pcapkit.protocols.schema.internet.hip.Parameter`. Raises: - RegistryError: If ``meth`` is a name with no matching - ``_read_param_*`` method. + RegistryError: If ``code`` is not an enum member, or ``meth`` + is a name with no matching ``_read_param_*`` method. """ + _check_code(code) if isinstance(meth, str) and not hasattr(HIP, f'_read_param_{meth}'): raise RegistryError('method must be a valid HIP parameter parser function') @@ -826,10 +875,11 @@ def register_hopopt_option(code: 'IPv6_Option', meth: 'str | tuple[HOPOPT_Option It should be a subclass of :class:`pcapkit.protocols.schema.internet.hopopt.Option`. Raises: - RegistryError: If ``meth`` is a name with no matching - ``_read_opt_*`` method. + RegistryError: If ``code`` is not an enum member, or ``meth`` + is a name with no matching ``_read_opt_*`` method. """ + _check_code(code) if isinstance(meth, str) and not hasattr(HOPOPT, f'_read_opt_{meth}'): raise RegistryError('method must be a valid HOPOPT option parser function') @@ -858,10 +908,11 @@ def register_ipv6_opts_option(code: 'IPv6_Option', meth: 'str | tuple[IPv6_Opts_ It should be a subclass of :class:`pcapkit.protocols.schema.internet.ipv6_opts.Option`. Raises: - RegistryError: If ``meth`` is a name with no matching - ``_read_opt_*`` method. + RegistryError: If ``code`` is not an enum member, or ``meth`` + is a name with no matching ``_read_opt_*`` method. """ + _check_code(code) 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') @@ -890,10 +941,11 @@ def register_ipv6_route_routing(code: 'IPv6_Routing', meth: 'str | tuple[IPv6_Ro It should be a subclass of :class:`pcapkit.protocols.schema.internet.ipv6_route.RoutingType`. Raises: - RegistryError: If ``meth`` is a name with no matching - ``_read_data_type_*`` method. + RegistryError: If ``code`` is not an enum member, or ``meth`` + is a name with no matching ``_read_data_type_*`` method. """ + _check_code(code) 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') @@ -922,10 +974,11 @@ def register_mh_message(code: 'MH_Packet', meth: 'str | tuple[MH_PacketParser, M It should be a subclass of :class:`pcapkit.protocols.schema.internet.mh.Packet`. Raises: - RegistryError: If ``meth`` is a name with no matching - ``_read_msg_*`` method. + RegistryError: If ``code`` is not an enum member, or ``meth`` + is a name with no matching ``_read_msg_*`` method. """ + _check_code(code) if isinstance(meth, str) and not hasattr(MH, f'_read_msg_{meth}'): raise RegistryError('method must be a valid MH message type parser function') @@ -954,10 +1007,11 @@ def register_mh_option(code: 'MH_Option', meth: 'str | tuple[MH_OptionParser, MH It should be a subclass of :class:`pcapkit.protocols.schema.internet.mh.Option`. Raises: - RegistryError: If ``meth`` is a name with no matching - ``_read_opt_*`` method. + RegistryError: If ``code`` is not an enum member, or ``meth`` + is a name with no matching ``_read_opt_*`` method. """ + _check_code(code) if isinstance(meth, str) and not hasattr(MH, f'_read_opt_{meth}'): raise RegistryError('method must be a valid MH option parser function') @@ -986,8 +1040,8 @@ def register_mh_extension(code: 'MH_CGAExtension', meth: 'str | tuple[MH_Extensi It should be a subclass of :class:`pcapkit.protocols.schema.internet.mh.CGAExtension`. Raises: - RegistryError: If ``meth`` is a name with no matching - ``_read_ext_*`` method. + RegistryError: If ``code`` is not an enum member, or ``meth`` + is a name with no matching ``_read_ext_*`` method. .. versionchanged:: 1.5.0 @@ -995,6 +1049,7 @@ def register_mh_extension(code: 'MH_CGAExtension', meth: 'str | tuple[MH_Extensi that take one (:issue:`516`). """ + _check_code(code) if isinstance(meth, str) and not hasattr(MH, f'_read_ext_{meth}'): raise RegistryError('method must be a valid MH CGA extension function') @@ -1258,10 +1313,11 @@ def register_tcp_option(code: 'TCP_Option', meth: 'str | tuple[TCP_OptionParser, It should be a subclass of :class:`pcapkit.protocols.schema.transport.tcp.Option`. Raises: - RegistryError: If ``meth`` is a name with no matching - ``_read_mode_*`` method. + RegistryError: If ``code`` is not an enum member, or ``meth`` + is a name with no matching ``_read_mode_*`` method. """ + _check_code(code) if isinstance(meth, str) and not hasattr(TCP, f'_read_mode_{meth}'): raise RegistryError('method must be a TCP option parser function') @@ -1290,10 +1346,11 @@ def register_tcp_mp_option(code: 'TCP_MPTCPOption', meth: 'str | tuple[TCP_MPOpt It should be a subclass of :class:`pcapkit.protocols.schema.transport.tcp.MPTCP`. Raises: - RegistryError: If ``meth`` is a name with no matching - ``_read_mptcp_*`` method. + RegistryError: If ``code`` is not an enum member, or ``meth`` + is a name with no matching ``_read_mptcp_*`` method. """ + _check_code(code) if isinstance(meth, str) and not hasattr(TCP, f'_read_mptcp_{meth}'): raise RegistryError('method must be a MPTCP option parser function') @@ -1410,10 +1467,11 @@ def register_http_frame(code: 'HTTP_Frame', meth: 'str | tuple[HTTP_FrameParser, It should be a subclass of :class:`pcapkit.protocols.schema.application.httpv2.FrameType`. Raises: - RegistryError: If ``meth`` is a name with no matching - ``_read_http_*`` method. + RegistryError: If ``code`` is not an enum member, or ``meth`` + is a name with no matching ``_read_http_*`` method. """ + _check_code(code) if isinstance(meth, str) and not hasattr(HTTPv2, f'_read_http_{meth}'): raise RegistryError('method must be a frame parser function') @@ -1447,10 +1505,11 @@ def register_pcapng_block(code: 'PCAPNG_BlockType', meth: 'str | tuple[PCAPNG_Bl It should be a subclass of :class:`pcapkit.protocols.schema.misc.pcapng.BlockType`. Raises: - RegistryError: If ``meth`` is a name with no matching - ``_read_block_*`` method. + RegistryError: If ``code`` is not an enum member, or ``meth`` + is a name with no matching ``_read_block_*`` method. """ + _check_code(code) if isinstance(meth, str) and not hasattr(PCAPNG, f'_read_block_{meth}'): raise RegistryError('method must be a block parser function') @@ -1479,10 +1538,11 @@ def register_pcapng_option(code: 'PCAPNG_OptionType', meth: 'str | tuple[PCAPNG_ It should be a subclass of :class:`pcapkit.protocols.schema.misc.pcapng.Option`. Raises: - RegistryError: If ``meth`` is a name with no matching - ``_read_option_*`` method. + RegistryError: If ``code`` is not an enum member, or ``meth`` + is a name with no matching ``_read_option_*`` method. """ + _check_code(code) if isinstance(meth, str) and not hasattr(PCAPNG, f'_read_option_{meth}'): raise RegistryError('method must be a option parser function') @@ -1511,10 +1571,11 @@ def register_pcapng_record(code: 'PCAPNG_RecordType', meth: 'str | tuple[PCAPNG_ It should be a subclass of :class:`pcapkit.protocols.schema.misc.pcapng.NameResolutionRecord`. Raises: - RegistryError: If ``meth`` is a name with no matching - ``_read_record_*`` method. + RegistryError: If ``code`` is not an enum member, or ``meth`` + is a name with no matching ``_read_record_*`` method. """ + _check_code(code) if isinstance(meth, str) and not hasattr(PCAPNG, f'_read_record_{meth}'): raise RegistryError('method must be a name resolution record parser function') @@ -1543,10 +1604,11 @@ def register_pcapng_secrets(code: 'PCAPNG_SecretsType', meth: 'str | tuple[PCAPN It should be a subclass of :class:`pcapkit.protocols.schema.misc.pcapng.DSBSecrets`. Raises: - RegistryError: If ``meth`` is a name with no matching - ``_read_secrets_*`` method. + RegistryError: If ``code`` is not an enum member, or ``meth`` + is a name with no matching ``_read_secrets_*`` method. """ + _check_code(code) if isinstance(meth, str) and not hasattr(PCAPNG, f'_read_secrets_{meth}'): raise RegistryError('method must be a decryption secrets parser function') diff --git a/tests/foundation/registry/test_registrar_code_check_unit.py b/tests/foundation/registry/test_registrar_code_check_unit.py new file mode 100644 index 0000000000..71f6f871a7 --- /dev/null +++ b/tests/foundation/registry/test_registrar_code_check_unit.py @@ -0,0 +1,146 @@ +# -*- coding: utf-8 -*- +"""A registrar handed a ``code`` that is not an enum member changes nothing. C.f. #2186. + +The registrars keyed on an enum code log ``code.name`` once the registration is +done, so an ``int`` or :obj:`None` failed there with :exc:`AttributeError` after +the protocol's table had been written. Each registrar rejects such a ``code`` +with :exc:`~pcapkit.utilities.exceptions.RegistryError` before it touches +anything. + +Every case snapshots each registry the registrars here write -- the dispatch and +method tables, the schema tables, :data:`pcapkit.protocols.__proto__` and the +name leases -- calls the registrar with an otherwise valid argument list, and +compares every registry with the snapshot. The registrars that take a plain +``int`` port or PPID (``register_tcp``, ``register_udp``, ``register_sctp`` and +``register_apptype``) are out of scope: an ``int`` is a valid code there. + +""" + +from __future__ import annotations + +import importlib +import importlib.util +import unittest +import warnings + +from tests._support import reimport_once_per_class + +#: Test tier (#1958): a pure-Python guard sweep over the registrars. +__tier__ = 'LATEST' + +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None + for name in ('tbtrim', 'aenum', 'chardet', 'dictdumper')) + +#: Registrars keyed on an enum code and given a parser method, with the protocol +#: and schema classes holding the tables they write. +METHOD_REGISTRARS = ( + ('register_ipv4_option', 'pcapkit.protocols.internet.ipv4:IPv4', + 'pcapkit.protocols.schema.internet.ipv4:Option'), + ('register_hip_parameter', 'pcapkit.protocols.internet.hip:HIP', + 'pcapkit.protocols.schema.internet.hip:Parameter'), + ('register_hopopt_option', 'pcapkit.protocols.internet.hopopt:HOPOPT', + 'pcapkit.protocols.schema.internet.hopopt:Option'), + ('register_ipv6_opts_option', 'pcapkit.protocols.internet.ipv6_opts:IPv6_Opts', + 'pcapkit.protocols.schema.internet.ipv6_opts:Option'), + ('register_ipv6_route_routing', 'pcapkit.protocols.internet.ipv6_route:IPv6_Route', + 'pcapkit.protocols.schema.internet.ipv6_route:RoutingType'), + ('register_mh_message', 'pcapkit.protocols.internet.mh:MH', + 'pcapkit.protocols.schema.internet.mh:Packet'), + ('register_mh_option', 'pcapkit.protocols.internet.mh:MH', + 'pcapkit.protocols.schema.internet.mh:Option'), + ('register_mh_extension', 'pcapkit.protocols.internet.mh:MH', + 'pcapkit.protocols.schema.internet.mh:CGAExtension'), + ('register_tcp_option', 'pcapkit.protocols.transport.tcp:TCP', + 'pcapkit.protocols.schema.transport.tcp:Option'), + ('register_tcp_mp_option', 'pcapkit.protocols.transport.tcp:TCP', + 'pcapkit.protocols.schema.transport.tcp:MPTCP'), + ('register_http_frame', 'pcapkit.protocols.application.httpv2:HTTP', + 'pcapkit.protocols.schema.application.httpv2:FrameType'), + ('register_pcapng_block', 'pcapkit.protocols.misc.pcapng:PCAPNG', + 'pcapkit.protocols.schema.misc.pcapng:BlockType'), + ('register_pcapng_option', 'pcapkit.protocols.misc.pcapng:PCAPNG', + 'pcapkit.protocols.schema.misc.pcapng:Option'), + ('register_pcapng_record', 'pcapkit.protocols.misc.pcapng:PCAPNG', + 'pcapkit.protocols.schema.misc.pcapng:NameResolutionRecord'), + ('register_pcapng_secrets', 'pcapkit.protocols.misc.pcapng:PCAPNG', + 'pcapkit.protocols.schema.misc.pcapng:DSBSecrets'), +) + +#: Registrars keyed on an enum code and given a protocol class, with the classes +#: whose ``__proto__`` they write. +MODULE_REGISTRARS = ( + ('register_linktype', ('pcapkit.protocols.misc.pcap.frame:Frame', + 'pcapkit.protocols.misc.pcapng:PCAPNG')), + ('register_pcap', ('pcapkit.protocols.misc.pcap.frame:Frame',)), + ('register_pcapng', ('pcapkit.protocols.misc.pcapng:PCAPNG',)), + ('register_ethertype', ('pcapkit.protocols.link.link:Link',)), + ('register_transtype', ('pcapkit.protocols.internet.internet:Internet',)), +) + +#: Codes that are not enum members: the issue's ``int``, :obj:`None`, and a +#: :class:`str` (the PCAP-NG option codes are a string enum). +BAD_CODES = (250, None, '250') + + +def resolve(path: 'str') -> 'type': + module, _, name = path.partition(':') + return getattr(importlib.import_module(module), name) + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class RegistrarCodeCheckTests(unittest.TestCase): + def setUp(self) -> None: + reimport_once_per_class(self) + + def _registries(self, classes: 'tuple[type, ...]') -> 'dict[tuple[str, str], dict]': + """Every dict-valued class attribute of ``classes``, plus the module-wide registries.""" + from pcapkit.foundation.registry import protocols as registrars + from pcapkit.protocols import __proto__ as protocol_registry + + found = {('pcapkit.protocols', '__proto__'): protocol_registry, + ('registrars', '_CODE_LEASES'): registrars._CODE_LEASES} + for cls in classes: + for attr, value in vars(cls).items(): + if isinstance(value, dict): + found[(cls.__qualname__, attr)] = value + return found + + def _assert_rejected_unchanged(self, name: 'str', classes: 'tuple[type, ...]', *args: 'object', + **kwargs: 'object') -> None: + from pcapkit.foundation.registry import protocols as registrars + from pcapkit.utilities.exceptions import RegistryError + + registries = self._registries(classes) + for code in BAD_CODES: + with self.subTest(registrar=name, code=code): + before = {key: dict(table) for key, table in registries.items()} + try: + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + with self.assertRaises(RegistryError): + getattr(registrars, name)(code, *args, **kwargs) + after = {key: dict(table) for key, table in registries.items()} + for key, table in before.items(): + self.assertEqual(after[key], table, f'{name}({code!r}) changed {key}') + finally: + for key, table in before.items(): + registries[key].clear() + registries[key].update(table) + + def test_method_registrars_reject_a_non_enum_code_before_registering(self) -> None: + def parser(*args: 'object', **kwargs: 'object') -> None: ... + + for name, proto, schema in METHOD_REGISTRARS: + proto_cls, schema_cls = resolve(proto), resolve(schema) + self._assert_rejected_unchanged(name, (proto_cls, schema_cls), + (parser, parser), schema=schema_cls) + + def test_module_registrars_reject_a_non_enum_code_before_registering(self) -> None: + from pcapkit.protocols.misc.raw import Raw + + for name, targets in MODULE_REGISTRARS: + self._assert_rejected_unchanged(name, tuple(map(resolve, targets)), Raw) + + +if __name__ == '__main__': + unittest.main()