diff --git a/docs/source/ext.rst b/docs/source/ext.rst index 46b73f34a5..f74b7357f1 100644 --- a/docs/source/ext.rst +++ b/docs/source/ext.rst @@ -244,13 +244,9 @@ available extensible items and the helper registry functions: +-------------------+------------------------------------------------------------+-----------------------------------------------------------------------------------------------+----------------------------------------------------------------------------+ | Protocol Type | Protocol Class | Extensable Registry | Registry Function | +===================+============================================================+===============================================================================================+============================================================================+ -| | | :func:`IPv4._read_opt_${name} ` | | -+ + :class:`~pcapkit.protocols.internet.ipv4.IPv4` +-----------------------------------------------------------------------------------------------+ :func:`~pcapkit.foundation.registry.protocols.register_ipv4_option` + -| | | :func:`IPv4._make_opt_${name} ` | | +| | :class:`~pcapkit.protocols.internet.ipv4.IPv4` | :attr:`IPv4.__option__ ` | :func:`~pcapkit.foundation.registry.protocols.register_ipv4_option` | + +------------------------------------------------------------+-----------------------------------------------------------------------------------------------+----------------------------------------------------------------------------+ -| | | :func:`HIP._read_param_${name} ` | | -+ + :class:`~pcapkit.protocols.internet.hip.HIP` +-----------------------------------------------------------------------------------------------+ :func:`~pcapkit.foundation.registry.protocols.register_hip_parameter` + -| | | :func:`HIP._make_param_${name} ` | | +| | :class:`~pcapkit.protocols.internet.hip.HIP` | :attr:`HIP.__parameter__ ` | :func:`~pcapkit.foundation.registry.protocols.register_hip_parameter` | + +------------------------------------------------------------+-----------------------------------------------------------------------------------------------+----------------------------------------------------------------------------+ | | :class:`~pcapkit.protocols.internet.hopopt.HOPOPT` | :attr:`HOPOPT.__option__ ` | :func:`~pcapkit.foundation.registry.protocols.register_hopopt_option` | + Internet Layer +------------------------------------------------------------+-----------------------------------------------------------------------------------------------+----------------------------------------------------------------------------+ diff --git a/docs/source/pcapkit/protocols/internet/hip.rst b/docs/source/pcapkit/protocols/internet/hip.rst index a0eac3560e..7b81d9d512 100644 --- a/docs/source/pcapkit/protocols/internet/hip.rst +++ b/docs/source/pcapkit/protocols/internet/hip.rst @@ -146,6 +146,9 @@ Octets Bits Name Description .. automethod:: _make_param_via_rvs .. automethod:: _make_param_relay_hmac + .. autoattribute:: __parameter__ + :no-value: + .. automethod:: __post_init__ .. automethod:: __index__ diff --git a/docs/source/pcapkit/protocols/internet/ipv4.rst b/docs/source/pcapkit/protocols/internet/ipv4.rst index fbac300ab3..4dddf9e613 100644 --- a/docs/source/pcapkit/protocols/internet/ipv4.rst +++ b/docs/source/pcapkit/protocols/internet/ipv4.rst @@ -82,6 +82,9 @@ Octets Bits Name Description .. automethod:: _make_opt_rtralt .. automethod:: _make_opt_qs + .. autoattribute:: __option__ + :no-value: + .. automethod:: __index__ Header Schemas diff --git a/pcapkit/foundation/registry/protocols.py b/pcapkit/foundation/registry/protocols.py index a0da7f3789..f009c79509 100644 --- a/pcapkit/foundation/registry/protocols.py +++ b/pcapkit/foundation/registry/protocols.py @@ -358,13 +358,13 @@ def register_transtype(code: 'TransType', module: 'str | ModuleDescriptor[Protoc register_protocol(module) -# NOTE: pcapkit.protocols.internet.internet.IPv4 +# NOTE: pcapkit.protocols.internet.ipv4.IPv4.__option__ def register_ipv4_option(code: 'IPv4_OptionNumber', meth: 'str | tuple[IPv4_OptionParser, IPv4_OptionConstructor]', *, schema: 'Optional[Type[Schema_IPv4_Option]]' = None) -> 'None': """Register an option parser. The function will register the given option parser to the - :data:`pcapkit.protocols.internet.internet.IPv4` internal registry. + :data:`pcapkit.protocols.internet.ipv4.IPv4.__option__` registry. Args: code: :class:`IPv4 ` option code as @@ -383,13 +383,13 @@ def register_ipv4_option(code: 'IPv4_OptionNumber', meth: 'str | tuple[IPv4_Opti logger.debug('registered IPv4 option parser: %s', code.name) -# NOTE: pcapkit.protocols.internet.hip.HIP +# NOTE: pcapkit.protocols.internet.hip.HIP.__parameter__ def register_hip_parameter(code: 'HIP_Parameter', meth: 'str | tuple[HIP_ParameterParser, HIP_ParameterConstructor]', *, schema: 'Optional[Type[Schema_HIP_Parameter]]' = None) -> 'None': """Register a parameter parser. The function will register the given parameter parser to the - :data:`pcapkit.protocols.internet.hip.HIP` internal registry. + :data:`pcapkit.protocols.internet.hip.HIP.__parameter__` registry. Args: code: :class:`~pcapkit.protocols.internet.hip.HIP` parameter code as diff --git a/pcapkit/protocols/internet/hip.py b/pcapkit/protocols/internet/hip.py index 4c16368f5b..803be3fa2d 100644 --- a/pcapkit/protocols/internet/hip.py +++ b/pcapkit/protocols/internet/hip.py @@ -30,6 +30,7 @@ .. [*] https://en.wikipedia.org/wiki/Host_Identity_Protocol """ +import collections import datetime import ipaddress import math @@ -204,7 +205,7 @@ from datetime import timedelta from enum import IntEnum as StdlibEnum from ipaddress import IPv6Address - from typing import IO, Any, Callable, NoReturn, Optional, Type + from typing import IO, Any, Callable, DefaultDict, NoReturn, Optional, Type from aenum import IntEnum as AenumEnum from mypy_extensions import DefaultArg, KwArg, NamedArg @@ -256,8 +257,8 @@ class HIP(Internet[Data_HIP, Schema_HIP], """This class implements Host Identity Protocol. This class currently supports parsing of the following HIP parameters, - which are directly mapped to the :class:`pcapkit.const.hip.parameter.Parameter` - enumeration: + which are registered in the :attr:`self.__parameter__ ` + attribute: .. list-table:: :header-rows: 1 @@ -363,6 +364,71 @@ class HIP(Internet[Data_HIP, Schema_HIP], """ + ########################################################################## + # Defaults. + ########################################################################## + + #: DefaultDict[Enum_Parameter, str | tuple[ParameterParser, ParameterConstructor]]: + #: Parameter code to method mapping, c.f. :meth:`_read_hip_param` and/or + #: :meth:`_make_hip_param`. Method names are expected to be referred to the + #: class by ``_read_param_${name}`` and/or ``_make_param_${name}``, and if + #: such name not found, the value should then be a method that can parse the + #: parameter by itself. + __parameter__ = collections.defaultdict( + lambda: 'unassigned', + { + Enum_Parameter.ESP_INFO: 'esp_info', # [RFC 7402] 65 + Enum_Parameter.R1_Counter: 'r1_counter', # [RFC 5201] 128, v1 only + Enum_Parameter.R1_COUNTER: 'r1_counter', # [RFC 7401] 129 + Enum_Parameter.LOCATOR_SET: 'locator_set', # [RFC 8046] 193 + Enum_Parameter.PUZZLE: 'puzzle', # [RFC 7401] 257 + Enum_Parameter.SOLUTION: 'solution', # [RFC 7401] 321 + Enum_Parameter.SEQ: 'seq', # [RFC 7401] 385 + Enum_Parameter.ACK: 'ack', # [RFC 7401] 449 + Enum_Parameter.DH_GROUP_LIST: 'dh_group_list', # [RFC 7401] 511 + Enum_Parameter.DIFFIE_HELLMAN: 'diffie_hellman', # [RFC 7401] 513 + Enum_Parameter.HIP_TRANSFORM: 'hip_transform', # [RFC 5201] 577, v1 only + Enum_Parameter.HIP_CIPHER: 'hip_cipher', # [RFC 7401] 579 + Enum_Parameter.NAT_TRAVERSAL_MODE: 'nat_traversal_mode', # [RFC 5770] 608 + Enum_Parameter.TRANSACTION_PACING: 'transaction_pacing', # [RFC 5770] 610 + Enum_Parameter.ENCRYPTED: 'encrypted', # [RFC 7401] 641 + Enum_Parameter.HOST_ID: 'host_id', # [RFC 7401] 705 + Enum_Parameter.HIT_SUITE_LIST: 'hit_suite_list', # [RFC 7401] 715 + Enum_Parameter.CERT: 'cert', # [RFC 7401][RFC 8002] 768 + Enum_Parameter.NOTIFICATION: 'notification', # [RFC 7401] 832 + Enum_Parameter.ECHO_REQUEST_SIGNED: 'echo_request_signed', # [RFC 7401] 897 + Enum_Parameter.REG_INFO: 'reg_info', # [RFC 8003] 930 + Enum_Parameter.REG_REQUEST: 'reg_request', # [RFC 8003] 932 + Enum_Parameter.REG_RESPONSE: 'reg_response', # [RFC 8003] 934 + Enum_Parameter.REG_FAILED: 'reg_failed', # [RFC 8003] 936 + Enum_Parameter.REG_FROM: 'reg_from', # [RFC 5770] 950 + Enum_Parameter.ECHO_RESPONSE_SIGNED: 'echo_response_signed', # [RFC 7401] 961 + Enum_Parameter.TRANSPORT_FORMAT_LIST: 'transport_format_list', # [RFC 7401] 2049 + Enum_Parameter.ESP_TRANSFORM: 'esp_transform', # [RFC 7402] 4095 + Enum_Parameter.SEQ_DATA: 'seq_data', # [RFC 6078] 4481 + Enum_Parameter.ACK_DATA: 'ack_data', # [RFC 6078] 4545 + Enum_Parameter.PAYLOAD_MIC: 'payload_mic', # [RFC 6078] 4577 + Enum_Parameter.TRANSACTION_ID: 'transaction_id', # [RFC 6078] 4580 + Enum_Parameter.OVERLAY_ID: 'overlay_id', # [RFC 6079] 4592 + Enum_Parameter.ROUTE_DST: 'route_dst', # [RFC 6028] 4601 + Enum_Parameter.HIP_TRANSPORT_MODE: 'hip_transport_mode', # [RFC 6261] 7680 + Enum_Parameter.HIP_MAC: 'hip_mac', # [RFC 7401] 61505 + Enum_Parameter.HIP_MAC_2: 'hip_mac_2', # [RFC 7401] 61569 + Enum_Parameter.HIP_SIGNATURE_2: 'hip_signature_2', # [RFC 7401] 61633 + Enum_Parameter.HIP_SIGNATURE: 'hip_signature', # [RFC 7401] 61697 + Enum_Parameter.ECHO_REQUEST_UNSIGNED: 'echo_request_unsigned', # [RFC 7401] 63661 + Enum_Parameter.ECHO_RESPONSE_UNSIGNED: 'echo_response_unsigned', # [RFC 7401] 63425 + Enum_Parameter.RELAY_FROM: 'relay_from', # [RFC 5770] 63998 + Enum_Parameter.RELAY_TO: 'relay_to', # [RFC 5770] 64002 + Enum_Parameter.OVERLAY_TTL: 'overlay_ttl', # [RFC 6079] 64011 + Enum_Parameter.ROUTE_VIA: 'route_via', # [RFC 6028] 64017 + Enum_Parameter.FROM: 'from', # [RFC 8004] 65498 + Enum_Parameter.RVS_HMAC: 'rvs_hmac', # [RFC 8004] 65500 + Enum_Parameter.VIA_RVS: 'via_rvs', # [RFC 8004] 65502 + Enum_Parameter.RELAY_HMAC: 'relay_hmac', # [RFC 5770] 65520 + }, + ) # type: DefaultDict[Enum_Parameter | int, str | tuple[ParameterParser, ParameterConstructor]] + ########################################################################## # Properties. ########################################################################## @@ -575,20 +641,13 @@ def register_parameter(cls, code: 'Enum_Parameter', meth: 'str | tuple[Parameter """Register a parameter parser. Args: - code: IPv4 option code. - meth: Method name or callable to parse and/or construct the option. + code: HIP parameter code. + meth: Method name or callable to parse and/or construct the parameter. """ - name = code.name.lower() - if hasattr(cls, f'_read_param_{name}'): + if code in cls.__parameter__: warn(f'parameter {code} already registered, overwriting', RegistryWarning) - - if isinstance(meth, str): - meth = (getattr(cls, f'_read_param_{meth}', cls._read_param_unassigned), # type: ignore[arg-type] - getattr(cls, f'_make_param_{meth}', cls._make_param_unassigned)) # type: ignore[arg-type] - - setattr(cls, f'_read_param_{name}', meth[0]) - setattr(cls, f'_make_param_{name}', meth[1]) + cls.__parameter__[code] = meth ########################################################################## # Data models. @@ -683,10 +742,14 @@ def _read_hip_param(self, length: 'int', *, version: 'int') -> 'Parameter': # p for schema in self.__header__.param: dscp = schema.type + name = self._lookup_registry(self.__parameter__, dscp) - meth_name = f'_read_param_{dscp.name.lower()}' - meth = cast('ParameterParser', - getattr(self, meth_name, self._read_param_unassigned)) + if isinstance(name, str): + meth_name = f'_read_param_{name}' + meth = cast('ParameterParser', + getattr(self, meth_name, self._read_param_unassigned)) + else: + meth = name[0] data = meth(schema, version=version, options=options) # record parameter data @@ -2824,9 +2887,13 @@ def _make_hip_param(self, parameters: 'list[Schema_Parameter | tuple[Enum_Parame total_length += len(schema_packed) else: code, args = cast('tuple[Enum_Parameter, dict[str, Any]]', schema) - meth_name = f'_make_param_{code.name.lower()}' - meth = cast('ParameterConstructor', - getattr(self, meth_name, self._make_param_unassigned)) + name = self._lookup_registry(self.__parameter__, code) + if isinstance(name, str): + meth_name = f'_make_param_{name}' + meth = cast('ParameterConstructor', + getattr(self, meth_name, self._make_param_unassigned)) + else: + meth = name[1] data = meth(code, version=version, **args) # type: Schema_Parameter data_packed = data.pack() @@ -2837,9 +2904,13 @@ def _make_hip_param(self, parameters: 'list[Schema_Parameter | tuple[Enum_Parame parameters_list = [] for code, param in parameters.items(multi=True): - meth_name = f'_make_param_{code.name.lower()}' - meth = cast('ParameterConstructor', - getattr(self, meth_name, self._make_param_unassigned)) + name = self._lookup_registry(self.__parameter__, code) + if isinstance(name, str): + meth_name = f'_make_param_{name}' + meth = cast('ParameterConstructor', + getattr(self, meth_name, self._make_param_unassigned)) + else: + meth = name[1] data = meth(code, param, version=version) data_packed = data.pack() diff --git a/pcapkit/protocols/internet/ipv4.py b/pcapkit/protocols/internet/ipv4.py index ea2cae7e99..fc45c9ef76 100644 --- a/pcapkit/protocols/internet/ipv4.py +++ b/pcapkit/protocols/internet/ipv4.py @@ -34,6 +34,7 @@ .. [*] https://en.wikipedia.org/wiki/IPv4 """ +import collections import datetime import ipaddress import math @@ -105,7 +106,7 @@ from datetime import timedelta from enum import IntEnum as StdlibEnum from ipaddress import IPv4Address - from typing import Any, Callable, Optional, Type + from typing import Any, Callable, DefaultDict, Optional, Type from aenum import IntEnum as AenumEnum from mypy_extensions import DefaultArg, KwArg, NamedArg @@ -128,8 +129,8 @@ class IPv4(IP[Data_IPv4, Schema_IPv4], """This class implements Internet Protocol version 4. This class currently supports parsing of the following IPv4 options, - which are directly mapped to the :class:`pcapkit.const.ipv4.option_number.OptionNumber` - enumeration: + which are registered in the :attr:`self.__option__ ` + attribute: .. list-table:: :header-rows: 1 @@ -182,6 +183,36 @@ class IPv4(IP[Data_IPv4, Schema_IPv4], """ + ########################################################################## + # Defaults. + ########################################################################## + + #: DefaultDict[Enum_OptionNumber, str | tuple[OptionParser, OptionConstructor]]: + #: Option code to method mapping, c.f. :meth:`_read_ipv4_options` and/or + #: :meth:`_make_ipv4_options`. Method names are expected to be referred + #: to the class by ``_read_opt_${name}`` and/or ``_make_opt_${name}``, and + #: if such name not found, the value should then be a method that can parse + #: the option by itself. + __option__ = collections.defaultdict( + lambda: 'unassigned', + { + Enum_OptionNumber.EOOL: 'eool', # [RFC 791] 0 + Enum_OptionNumber.NOP: 'nop', # [RFC 791] 1 + Enum_OptionNumber.SEC: 'sec', # [RFC 1108] 130 + Enum_OptionNumber.LSR: 'lsr', # [RFC 791] 131 + Enum_OptionNumber.TS: 'ts', # [RFC 791] 68 + Enum_OptionNumber.E_SEC: 'e_sec', # [RFC 1108] 133 + Enum_OptionNumber.RR: 'rr', # [RFC 791] 7 + Enum_OptionNumber.SID: 'sid', # [RFC 791][RFC 6814] 136 + Enum_OptionNumber.SSR: 'ssr', # [RFC 791] 137 + Enum_OptionNumber.MTUP: 'mtup', # [RFC 1063][RFC 1191] 11 + Enum_OptionNumber.MTUR: 'mtur', # [RFC 1063][RFC 1191] 12 + Enum_OptionNumber.TR: 'tr', # [RFC 1393][RFC 6814] 82 + Enum_OptionNumber.RTRALT: 'rtralt', # [RFC 2113] 148 + Enum_OptionNumber.QS: 'qs', # [RFC 4782] 25 + }, + ) # type: DefaultDict[Enum_OptionNumber | int, str | tuple[OptionParser, OptionConstructor]] + ########################################################################## # Properties. ########################################################################## @@ -447,16 +478,9 @@ def register_option(cls, code: 'Enum_OptionNumber', meth: 'str | tuple[OptionPar meth: Method name or callable to parse and/or construct the option. """ - name = code.name.lower() - if hasattr(cls, f'_read_opt_{name}'): + if code in cls.__option__: warn(f'option {code} already registered, overwriting', RegistryWarning) - - if isinstance(meth, str): - meth = (getattr(cls, f'_read_opt_{meth}', cls._read_opt_unassigned), # type: ignore[arg-type] - getattr(cls, f'_make_opt_{meth}', cls._make_opt_unassigned)) # type: ignore[arg-type] - - setattr(cls, f'_read_opt_{name}', meth[0]) - setattr(cls, f'_make_opt_{name}', meth[1]) + cls.__option__[code] = meth ########################################################################## # Data models. @@ -563,11 +587,14 @@ def _read_ipv4_options(self, length: 'int') -> 'Option': for schema in self.__header__.options: kind = schema.type - name = kind.name.lower() + name = self._lookup_registry(self.__option__, kind) - meth_name = f'_read_opt_{name}' - meth = cast('OptionParser', - getattr(self, meth_name, self._read_opt_unassigned)) + if isinstance(name, str): + meth_name = f'_read_opt_{name}' + meth = cast('OptionParser', + getattr(self, meth_name, self._read_opt_unassigned)) + else: + meth = name[0] data = meth(schema, options=options) # record option data @@ -1204,9 +1231,13 @@ def _make_ipv4_options(self, options: 'list[Schema_Option | tuple[Enum_OptionNum if code in (Enum_OptionNumber.NOP, Enum_OptionNumber.EOOL): # ignore padding options by default continue - name = f'_make_opt_{code.name.lower()}' - meth = cast('OptionConstructor', - getattr(self, name, self._make_opt_unassigned)) + name = self._lookup_registry(self.__option__, code) + if isinstance(name, str): + meth_name = f'_make_opt_{name}' + meth = cast('OptionConstructor', + getattr(self, meth_name, self._make_opt_unassigned)) + else: + meth = name[1] data = meth(code, **args) data_len = len(data.pack()) @@ -1231,9 +1262,13 @@ def _make_ipv4_options(self, options: 'list[Schema_Option | tuple[Enum_OptionNum if code in (Enum_OptionNumber.NOP, Enum_OptionNumber.EOOL): continue - name = f'_make_opt_{code.name.lower()}' - meth = cast('OptionConstructor', - getattr(self, name, self._make_opt_unassigned)) + name = self._lookup_registry(self.__option__, code) + if isinstance(name, str): + meth_name = f'_make_opt_{name}' + meth = cast('OptionConstructor', + getattr(self, meth_name, self._make_opt_unassigned)) + else: + meth = name[1] data = meth(code, option) data_len = len(data.pack()) diff --git a/tests/protocols/internet/test_hip_unit.py b/tests/protocols/internet/test_hip_unit.py index 2db211c245..33b6eb7947 100644 --- a/tests/protocols/internet/test_hip_unit.py +++ b/tests/protocols/internet/test_hip_unit.py @@ -230,53 +230,200 @@ def test_hip_read_make_post_init_and_parameter_list_edges(self) -> None: def test_hip_register_parameter_warns_on_overwrite(self) -> None: from pcapkit.const.hip.parameter import Parameter - from pcapkit.protocols.data.internet import hip as hip_data from pcapkit.protocols.internet.hip import HIP - from pcapkit.protocols.schema.internet import hip as hip_schema - read_name = f'_read_param_{Parameter.ESP_INFO.name.lower()}' - make_name = f'_make_param_{Parameter.ESP_INFO.name.lower()}' - original_read = getattr(HIP, read_name) - original_make = getattr(HIP, make_name) + registry = HIP.__dict__['__parameter__'] + + original = registry[Parameter.ESP_INFO] try: with mock.patch('pcapkit.protocols.internet.hip.warn') as warn: HIP.register_parameter(Parameter.ESP_INFO, 'esp_info') warn.assert_called_once() - self.assertIs(getattr(HIP, read_name), original_read) - self.assertIs(getattr(HIP, make_name), original_make) + self.assertEqual(registry[Parameter.ESP_INFO], 'esp_info') + finally: + registry[Parameter.ESP_INFO] = original + + # An unregistered code carries no entry, so registering one is not an + # overwrite. The setattr form could not tell the two apart: it keyed the + # warning on ``hasattr(cls, f'_read_param_{name}')``, which is true of + # every shipped handler as well as of anything a user had installed. + custom = Parameter.Unassigned_65501 + self.assertNotIn(custom, registry) + try: + with mock.patch('pcapkit.protocols.internet.hip.warn') as warn: + HIP.register_parameter(custom, 'unassigned') + warn.assert_not_called() + self.assertEqual(registry[custom], 'unassigned') + + with mock.patch('pcapkit.protocols.internet.hip.warn') as warn: + HIP.register_parameter(custom, 'unassigned') + warn.assert_called_once() finally: - setattr(HIP, read_name, original_read) - setattr(HIP, make_name, original_make) + registry.pop(custom, None) + + def test_hip_register_parameter_dispatches_a_registered_callable_pair(self) -> None: + """A ``(parser, constructor)`` pair must reach both dispatch directions. + + The pair is called with the signatures :data:`ParameterParser` and + :data:`ParameterConstructor` declare -- ``(schema, *, version, options)`` + and ``(code, param=None, *, version, **kwargs)`` -- i.e. as plain + callables rather than as methods with an implicit ``self``. That is the + calling convention every other dispatch family uses, and the reason the + registry form replaced ``setattr``: installing the callable on the class + made it a descriptor, so dispatch passed ``self`` as the first positional + argument and a handler written to the declared signature could not be + called at all. + + """ + from pcapkit.const.hip.parameter import Parameter + from pcapkit.corekit.multidict import OrderedMultiDict + from pcapkit.protocols.data.internet import hip as hip_data + from pcapkit.protocols.internet.hip import HIP + from pcapkit.protocols.schema.internet import hip as hip_schema + registry = HIP.__dict__['__parameter__'] custom = Parameter.Unassigned_65501 - custom_read_name = f'_read_param_{custom.name.lower()}' - custom_make_name = f'_make_param_{custom.name.lower()}' - original_custom_read = getattr(HIP, custom_read_name, None) - original_custom_make = getattr(HIP, custom_make_name, None) + seen = [] # type: list[str] - def read_param(packet, *, version, options): - return hip_data.UnassignedParameter(type=packet.type, critical=False, - length=4, contents=packet.value) + def read_param(schema, *, version, options): + seen.append(f'read/v{version}') + return hip_data.UnassignedParameter(type=schema.type, critical=False, + length=4 + schema.len, + contents=schema.value) def make_param(code, param=None, *, version, contents=b'', **kwargs): + seen.append(f'make/v{version}') + if param is not None: + contents = param.contents return hip_schema.UnassignedParameter(type=code, len=len(contents), value=contents) + self.assertNotIn(custom, registry) try: - with mock.patch('pcapkit.protocols.internet.hip.warn') as warn: - HIP.register_parameter(custom, (read_param, make_param)) - warn.assert_not_called() - self.assertIs(getattr(HIP, custom_read_name), read_param) - self.assertIs(getattr(HIP, custom_make_name), make_param) + HIP.register_parameter(custom, (read_param, make_param)) + self.assertEqual(registry[custom], (read_param, make_param)) + + schema = hip_schema.UnassignedParameter(type=custom, len=4, + value=b'abcd') + proto = object.__new__(HIP) + proto.__header__ = types.SimpleNamespace(param=[schema]) + parsed = proto._read_hip_param(len(schema), version=2) + self.assertEqual(seen, ['read/v2']) + self.assertEqual(parsed[custom].contents, b'abcd') + + # the list-of-tuples branch of the constructor + seen.clear() + made_list, list_len = proto._make_hip_param( + [(custom, {'contents': b'wxyz'})], version=2) + self.assertEqual(seen, ['make/v2']) + self.assertEqual(made_list[0].value, b'wxyz') + # 4 octets of type and length, 4 of value, 4 of padding to a multiple + # of 8 + self.assertEqual(list_len, 12) + + # ... and the OrderedMultiDict branch, which the two halves of an + # issue like this are equally easy to fix one of and forget the other + seen.clear() + made_dict, dict_len = proto._make_hip_param( + OrderedMultiDict([(custom, parsed[custom])]), version=2) + self.assertEqual(seen, ['make/v2']) + self.assertEqual(made_dict[0].value, b'abcd') + self.assertEqual(dict_len, 12) finally: - if original_custom_read is None: - delattr(HIP, custom_read_name) - else: - setattr(HIP, custom_read_name, original_custom_read) - if original_custom_make is None: - delattr(HIP, custom_make_name) - else: - setattr(HIP, custom_make_name, original_custom_make) + registry.pop(custom, None) + + def test_hip_parameter_registry_covers_every_shipped_handler(self) -> None: + """Every ``_read_param_*`` / ``_make_param_*`` pair must be reachable. + + The setattr form derived the handler name from ``code.name.lower()``, so + a handler was reachable by construction and "what is registered?" had no + direct answer. Under the registry the mapping is explicit data, which + means a handler added without its registry entry becomes dead code -- + this pins the two sides together. + + """ + from pcapkit.const.hip.parameter import Parameter + from pcapkit.protocols.internet.hip import HIP + + registry = HIP.__dict__['__parameter__'] + fallback = registry.default_factory() + self.assertEqual(fallback, 'unassigned') + + registered = set(registry.values()) | {fallback} + self.assertEqual( + {name[len('_read_param_'):] for name in vars(HIP) + if name.startswith('_read_param_')}, + registered, + ) + self.assertEqual( + {name[len('_make_param_'):] for name in vars(HIP) + if name.startswith('_make_param_')}, + registered, + ) + + # ``R1_Counter`` (128, HIPv1) and ``R1_COUNTER`` (129, HIPv2) are distinct + # codes whose names differ only in case, so both have to be keyed + # explicitly -- ``code.name.lower()`` collapsed them for free. + self.assertEqual(registry[Parameter.R1_Counter], 'r1_counter') + self.assertEqual(registry[Parameter.R1_COUNTER], 'r1_counter') + + # every key is a real parameter code, and every value names real methods + for code, name in registry.items(): + with self.subTest(code=code): + self.assertIsInstance(code, Parameter) + self.assertTrue(hasattr(HIP, f'_read_param_{name}')) + self.assertTrue(hasattr(HIP, f'_make_param_{name}')) + + def test_hip_unregistered_parameter_code_does_not_mutate_the_registry(self) -> None: + """Parsing must not write to the shared HIP parameter registry. + + :attr:`HIP.__parameter__ ` + is a :class:`collections.defaultdict` on a class attribute shared by + every instance in the process, so ``registry[code]`` would insert each + code it missed -- the leak #428 swept out of the sixteen registries that + already existed. This one is new, so the guard has to be pinned here too: + the reads go through + :meth:`~pcapkit.protocols.protocol.ProtocolBase._lookup_registry`. + + Code 65499 is unassigned in IANA's registry and 4650 is + ``RELAYED_ADDRESS``, which is assigned but has no parser, so both take + the fallback. + + """ + from pcapkit.const.hip.parameter import Parameter + from pcapkit.corekit.multidict import OrderedMultiDict + from pcapkit.protocols.internet.hip import HIP + from pcapkit.protocols.schema.internet import hip as hip_schema + + registry = HIP.__dict__['__parameter__'] + proto = object.__new__(HIP) + + for code in (Parameter.Unassigned_65499, Parameter.RELAYED_ADDRESS): + with self.subTest(code=code): + before = set(registry) + self.assertNotIn(code, before) + + schema = hip_schema.UnassignedParameter(type=code, len=4, + value=b'abcd') + proto.__header__ = types.SimpleNamespace(param=[schema]) + proto._read_hip_param(len(schema), version=2) + self.assertEqual(set(registry), before) + + # both constructor branches, since either can leak on its own + proto._make_hip_param([(code, {'contents': b'abcd'})], version=2) + self.assertEqual(set(registry), before) + proto._make_hip_param( + OrderedMultiDict([(code, types.SimpleNamespace(contents=b'abcd'))]), + version=2) + self.assertEqual(set(registry), before) + + # a leak would make the next genuine registration warn + with mock.patch('pcapkit.protocols.internet.hip.warn') as warn: + try: + HIP.register_parameter(code, 'unassigned') + warn.assert_not_called() + finally: + registry.pop(code, None) def test_hip_parameter_readers_cover_simple_models_and_guards(self) -> None: from pcapkit.const.hip.certificate import Certificate diff --git a/tests/protocols/internet/test_ipv4_unit.py b/tests/protocols/internet/test_ipv4_unit.py index 4e1ac9713d..808a079b2a 100644 --- a/tests/protocols/internet/test_ipv4_unit.py +++ b/tests/protocols/internet/test_ipv4_unit.py @@ -162,58 +162,204 @@ def test_ipv4_properties_read_and_make_cover_packet_paths(self) -> None: def test_ipv4_register_option_warns_on_overwrite(self) -> None: from pcapkit.const.ipv4.option_number import OptionNumber - from pcapkit.protocols.data.internet import ipv4 as ipv4_data from pcapkit.protocols.internet.ipv4 import IPv4 - from pcapkit.protocols.schema.internet import ipv4 as ipv4_schema - read_name = f'_read_opt_{OptionNumber.EOOL.name.lower()}' - make_name = f'_make_opt_{OptionNumber.EOOL.name.lower()}' - original_read = getattr(IPv4, read_name) - original_make = getattr(IPv4, make_name) + registry = IPv4.__dict__['__option__'] + + original = registry[OptionNumber.EOOL] try: with mock.patch('pcapkit.protocols.internet.ipv4.warn') as warn: IPv4.register_option(OptionNumber.EOOL, 'eool') warn.assert_called_once() - self.assertIs(getattr(IPv4, read_name), original_read) - self.assertIs(getattr(IPv4, make_name), original_make) + self.assertEqual(registry[OptionNumber.EOOL], 'eool') finally: - setattr(IPv4, read_name, original_read) - setattr(IPv4, make_name, original_make) + registry[OptionNumber.EOOL] = original + # An unregistered code carries no entry, so registering one is not an + # overwrite. The setattr form could not tell the two apart: it keyed the + # warning on ``hasattr(cls, f'_read_opt_{name}')``, which is true of every + # shipped handler as well as of anything a user had already installed. custom = OptionNumber.get(31) - custom_read_name = f'_read_opt_{custom.name.lower()}' - custom_make_name = f'_make_opt_{custom.name.lower()}' - original_custom_read = getattr(IPv4, custom_read_name, None) - original_custom_make = getattr(IPv4, custom_make_name, None) + self.assertNotIn(custom, registry) + try: + with mock.patch('pcapkit.protocols.internet.ipv4.warn') as warn: + IPv4.register_option(custom, 'unassigned') + warn.assert_not_called() + self.assertEqual(registry[custom], 'unassigned') + + with mock.patch('pcapkit.protocols.internet.ipv4.warn') as warn: + IPv4.register_option(custom, 'unassigned') + warn.assert_called_once() + finally: + registry.pop(custom, None) + + def test_ipv4_register_option_dispatches_a_registered_callable_pair(self) -> None: + """A ``(parser, constructor)`` pair must reach both dispatch directions. + + The pair is called with the signatures :data:`OptionParser` and + :data:`OptionConstructor` declare -- ``(schema, *, options)`` and + ``(code, option=None, **kwargs)`` -- i.e. as plain callables rather than + as methods with an implicit ``self``. That is the calling convention + every other dispatch family uses, and the reason the registry form + replaced ``setattr``: installing the callable on the class made it a + descriptor, so dispatch passed ``self`` as the first positional argument + and a handler written to the declared signature could not be called at + all. + + """ + from pcapkit.const.ipv4.option_number import OptionNumber + from pcapkit.corekit.multidict import OrderedMultiDict + from pcapkit.protocols.data.internet import ipv4 as ipv4_data + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.schema.internet import ipv4 as ipv4_schema + + registry = IPv4.__dict__['__option__'] + custom = OptionNumber.get(31) + proto = object.__new__(IPv4) + seen = [] # type: list[str] def read_option(schema, *, options): + seen.append('read') return ipv4_data.UnassignedOption( code=schema.type, - type=IPv4._read_ipv4_opt_type(schema.type), + type=proto._read_ipv4_opt_type(schema.type), length=schema.length, data=schema.data, ) def make_option(code, option=None, *, data=b'', **kwargs): + seen.append('make') if option is not None: data = option.data - return ipv4_schema.UnassignedOption(type=code, length=len(data), data=data) + return ipv4_schema.UnassignedOption(type=code, length=len(data) + 2, + data=data) + self.assertNotIn(custom, registry) try: - with mock.patch('pcapkit.protocols.internet.ipv4.warn') as warn: - IPv4.register_option(custom, (read_option, make_option)) - warn.assert_not_called() - self.assertIs(getattr(IPv4, custom_read_name), read_option) - self.assertIs(getattr(IPv4, custom_make_name), make_option) + IPv4.register_option(custom, (read_option, make_option)) + self.assertEqual(registry[custom], (read_option, make_option)) + + proto.__header__ = types.SimpleNamespace( + options=[ipv4_schema.UnassignedOption(type=custom, length=4, + data=b'xx')], + ) + parsed = proto._read_ipv4_options(4) + self.assertEqual(seen, ['read']) + self.assertEqual(parsed[custom].data, b'xx') + + # the list-of-tuples branch of the constructor + seen.clear() + made_list, list_len = proto._make_ipv4_options([(custom, {'data': b'yy'})]) + self.assertEqual(seen, ['make']) + self.assertEqual(made_list[0].data, b'yy') + self.assertEqual(list_len, 4) + + # ... and the OrderedMultiDict branch, which the two halves of an + # issue like this are equally easy to fix one of and forget the other + seen.clear() + made_dict, dict_len = proto._make_ipv4_options( + OrderedMultiDict([(custom, parsed[custom])])) + self.assertEqual(seen, ['make']) + self.assertEqual(made_dict[0].data, b'xx') + self.assertEqual(dict_len, 4) finally: - if original_custom_read is None: - delattr(IPv4, custom_read_name) - else: - setattr(IPv4, custom_read_name, original_custom_read) - if original_custom_make is None: - delattr(IPv4, custom_make_name) - else: - setattr(IPv4, custom_make_name, original_custom_make) + registry.pop(custom, None) + + def test_ipv4_option_registry_covers_every_shipped_handler(self) -> None: + """Every ``_read_opt_*`` / ``_make_opt_*`` pair must be reachable. + + The setattr form derived the handler name from ``code.name.lower()``, so + a handler was reachable by construction and "what is registered?" had no + direct answer. Under the registry the mapping is explicit data, which + means a handler added without its registry entry becomes dead code -- + this pins the two sides together. + + """ + from pcapkit.const.ipv4.option_number import OptionNumber + from pcapkit.protocols.internet.ipv4 import IPv4 + + registry = IPv4.__dict__['__option__'] + fallback = registry.default_factory() + self.assertEqual(fallback, 'unassigned') + + registered = set(registry.values()) | {fallback} + shipped = {name[len('_read_opt_'):] for name in vars(IPv4) + if name.startswith('_read_opt_')} + self.assertEqual(shipped, registered) + self.assertEqual( + {name[len('_make_opt_'):] for name in vars(IPv4) + if name.startswith('_make_opt_')}, + registered, + ) + + # every key is a real option number, and every value names real methods + for code, name in registry.items(): + with self.subTest(code=code): + self.assertIsInstance(code, OptionNumber) + self.assertTrue(hasattr(IPv4, f'_read_opt_{name}')) + self.assertTrue(hasattr(IPv4, f'_make_opt_{name}')) + + def test_ipv4_unregistered_option_code_does_not_mutate_the_registry(self) -> None: + """Parsing must not write to the shared IPv4 option registry. + + :attr:`IPv4.__option__ ` + is a :class:`collections.defaultdict` on a class attribute shared by + every instance in the process, so ``registry[code]`` would insert each + code it missed -- the leak #428 swept out of the sixteen registries that + already existed. This one is new, so the guard has to be pinned here too: + the reads go through + :meth:`~pcapkit.protocols.protocol.ProtocolBase._lookup_registry`. + + Option number 31 is unassigned in IANA's registry and 134 is ``CIPSO``, + which is assigned but has no parser, so both take the fallback. + + """ + from pcapkit.const.ipv4.option_number import OptionNumber + from pcapkit.corekit.multidict import OrderedMultiDict + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.schema.internet import ipv4 as ipv4_schema + + registry = IPv4.__dict__['__option__'] + proto = object.__new__(IPv4) + + for code in (OptionNumber.get(31), OptionNumber.CIPSO): + with self.subTest(code=code): + before = set(registry) + self.assertNotIn(code, before) + + proto.__header__ = types.SimpleNamespace( + options=[ipv4_schema.UnassignedOption(type=code, length=4, + data=b'xx')], + ) + proto._read_ipv4_options(4) + self.assertEqual(set(registry), before) + + # both constructor branches, since either can leak on its own + proto._make_ipv4_options([(code, {'data': b'xx'})]) + self.assertEqual(set(registry), before) + + # The OrderedMultiDict branch cannot get as far as building the + # option: :meth:`IPv4._make_opt_unassigned + # ` + # declares ``data`` keyword-only with no default, where every + # sibling fallback constructor defaults it to ``b''``, so it + # raises before reading the payload out of ``option.data``. That + # predates the registry migration and is left alone here; what + # this asserts is that the *lookup* which runs first still does + # not insert. + with self.assertRaises(TypeError): + proto._make_ipv4_options(OrderedMultiDict([ + (code, types.SimpleNamespace(data=b'xx')), + ])) + self.assertEqual(set(registry), before) + + # a leak would make the next genuine registration warn + with mock.patch('pcapkit.protocols.internet.ipv4.warn') as warn: + try: + IPv4.register_option(code, 'unassigned') + warn.assert_not_called() + finally: + registry.pop(code, None) def test_ipv4_option_constructors_cover_common_and_error_branches(self) -> None: from pcapkit.const.ipv4.option_number import OptionNumber