diff --git a/docs/source/pcapkit/protocols/protocol.rst b/docs/source/pcapkit/protocols/protocol.rst index bb8233f447..555394cc18 100644 --- a/docs/source/pcapkit/protocols/protocol.rst +++ b/docs/source/pcapkit/protocols/protocol.rst @@ -61,6 +61,7 @@ utility arguments and methods of specified protocols. .. automethod:: _make_index .. automethod:: _make_payload + .. automethod:: _lookup_registry .. automethod:: _lookup_next_layer .. automethod:: _decode_next_layer .. automethod:: _import_next_layer diff --git a/pcapkit/protocols/application/httpv2.py b/pcapkit/protocols/application/httpv2.py index f00023f407..5b9afdbf38 100644 --- a/pcapkit/protocols/application/httpv2.py +++ b/pcapkit/protocols/application/httpv2.py @@ -225,7 +225,7 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_HTTP': if schema.type in (Enum_Frame.SETTINGS, Enum_Frame.PING) and schema.stream['sid'] != 0: raise ProtocolError(f'HTTP/2: [Type {schema.type}] invalid format') - name = self.__frame__[schema.type] + name = self._lookup_registry(self.__frame__, schema.type) if isinstance(name, str): meth_name = f'_read_http_{name}' meth = cast('FrameParser', @@ -268,7 +268,7 @@ def make(self, # type: ignore[override] if isinstance(frame, bytes): frame_val = frame # type: bytes | Schema_FrameType elif isinstance(frame, (dict, Data_HTTP)): - name = self.__frame__[type_val] + name = self._lookup_registry(self.__frame__, type_val) if isinstance(name, str): meth_name = f'_make_http_{name}' meth = cast('FrameConstructor', diff --git a/pcapkit/protocols/internet/hopopt.py b/pcapkit/protocols/internet/hopopt.py index fbea14130c..cf4c7c2089 100644 --- a/pcapkit/protocols/internet/hopopt.py +++ b/pcapkit/protocols/internet/hopopt.py @@ -467,7 +467,7 @@ def _read_hopopt_options(self, length: 'int') -> 'Option': for schema in self.__header__.options: dscp = schema.type - name = self.__option__[dscp] + name = self._lookup_registry(self.__option__, dscp) if isinstance(name, str): meth_name = f'_read_opt_{name}' @@ -1275,7 +1275,7 @@ def _make_hopopt_options(self, options: 'list[Schema_Option | tuple[Enum_Option, if code in (Enum_Option.Pad1, Enum_Option.PadN): # ignore padding options by default continue - name = self.__option__[code] # type: str | tuple[OptionParser, OptionConstructor] + name = self._lookup_registry(self.__option__, code) if isinstance(name, str): meth_name = f'_make_opt_{name}' meth = cast('OptionConstructor', @@ -1306,7 +1306,7 @@ def _make_hopopt_options(self, options: 'list[Schema_Option | tuple[Enum_Option, if code in (Enum_Option.Pad1, Enum_Option.PadN): continue - name = self.__option__[code] + name = self._lookup_registry(self.__option__, code) if isinstance(name, str): meth_name = f'_make_opt_{name}' meth = cast('OptionConstructor', diff --git a/pcapkit/protocols/internet/ipv6_opts.py b/pcapkit/protocols/internet/ipv6_opts.py index f7f6ec692b..dd73be6bbd 100644 --- a/pcapkit/protocols/internet/ipv6_opts.py +++ b/pcapkit/protocols/internet/ipv6_opts.py @@ -478,7 +478,7 @@ def _read_ipv6_opts(self, length: 'int') -> 'Option': for schema in self.__header__.options: dscp = schema.type - name = self.__option__[dscp] + name = self._lookup_registry(self.__option__, dscp) if isinstance(name, str): meth_name = f'_read_opt_{name}' @@ -1287,7 +1287,7 @@ def _make_ipv6_opts(self, options: 'list[Schema_Option | tuple[Enum_Option, dict if code in (Enum_Option.Pad1, Enum_Option.PadN): # ignore padding options by default continue - name = self.__option__[code] # type: str | tuple[OptionParser, OptionConstructor] + name = self._lookup_registry(self.__option__, code) if isinstance(name, str): meth_name = f'_make_opt_{name}' meth = cast('OptionConstructor', @@ -1318,7 +1318,7 @@ def _make_ipv6_opts(self, options: 'list[Schema_Option | tuple[Enum_Option, dict if code in (Enum_Option.Pad1, Enum_Option.PadN): continue - name = self.__option__[code] + name = self._lookup_registry(self.__option__, code) if isinstance(name, str): meth_name = f'_make_opt_{name}' meth = cast('OptionConstructor', diff --git a/pcapkit/protocols/internet/ipv6_route.py b/pcapkit/protocols/internet/ipv6_route.py index 179196c5fa..71a26b1065 100644 --- a/pcapkit/protocols/internet/ipv6_route.py +++ b/pcapkit/protocols/internet/ipv6_route.py @@ -201,7 +201,7 @@ def read(self, length: 'Optional[int]' = None, *, extension: 'bool' = False, # length = len(self) schema = self.__header__ - name = self.__routing__[schema.type] + name = self._lookup_registry(self.__routing__, schema.type) if isinstance(name, str): name = f'_read_data_type_{name.lower()}' meth = cast('TypeParser', @@ -260,7 +260,7 @@ def make(self, length = math.ceil((len(data) + 4) / 8) data_val = data.ljust(length * 8 - 4, b'\x00') # type: bytes | Schema_RoutingType elif isinstance(data, (dict, Data_IPv6_Route)): - name = self.__routing__[type_val] + name = self._lookup_registry(self.__routing__, type_val) if isinstance(name, str): name = f'_make_data_type_{name.lower()}' meth = cast('TypeConstructor', diff --git a/pcapkit/protocols/internet/mh.py b/pcapkit/protocols/internet/mh.py index af426c80b0..31d6b2c528 100644 --- a/pcapkit/protocols/internet/mh.py +++ b/pcapkit/protocols/internet/mh.py @@ -675,7 +675,7 @@ def read(self, length: 'Optional[int]' = None, *, version: 'Literal[4, 6]' = 4, length = len(self) schema = self.__header__ - name = self.__message__[schema.type] + name = self._lookup_registry(self.__message__, schema.type) if isinstance(name, str): meth_name = f'_read_msg_{name}' meth = cast('PacketParser', @@ -729,7 +729,7 @@ def make(self, if isinstance(data, bytes): data_val = data # type: bytes | Schema_Packet elif isinstance(data, (dict, Data_MH)): - name = self.__message__[type_val] + name = self._lookup_registry(self.__message__, type_val) if isinstance(name, str): meth_name = f'_make_msg_{name}' meth = cast('PacketConstructor', @@ -1533,7 +1533,7 @@ def _read_mh_options(self, options_schema: 'list[Schema_Option]') -> 'Option': for schema in options_schema: type = schema.type - name = self.__option__[type] + name = self._lookup_registry(self.__option__, type) if isinstance(name, str): meth_name = f'_read_opt_{name}' @@ -2337,7 +2337,7 @@ def _read_cga_extensions(self, extensions_schema: 'list[Schema_CGAExtension]') - for schema in extensions_schema: type = schema.type - name = self.__extension__[type] + name = self._lookup_registry(self.__extension__, type) if isinstance(name, str): meth_name = f'_read_ext_{name}' @@ -3111,7 +3111,7 @@ def _make_mh_options(self, options: 'Option | list[Schema_Option | tuple[Enum_Op data = schema else: code, args = cast('tuple[Enum_Option, dict[str, Any]]', schema) - name = self.__option__[code] + name = self._lookup_registry(self.__option__, code) if isinstance(name, str): meth_name = f'_make_opt_{name}' meth = cast('OptionConstructor', @@ -3125,7 +3125,7 @@ def _make_mh_options(self, options: 'Option | list[Schema_Option | tuple[Enum_Op options_list = [] for code, option in options.items(multi=True): - name = self.__option__[code] + name = self._lookup_registry(self.__option__, code) if isinstance(name, str): meth_name = f'_make_opt_{name}' meth = cast('OptionConstructor', @@ -3780,7 +3780,7 @@ def _make_cga_extensions(self, extensions: 'Extension | list[Schema_CGAExtension data_len = len(schema.pack()) else: code, args = cast('tuple[Enum_CGAExtension, dict[str, Any]]', schema) - name = self.__extension__[code] + name = self._lookup_registry(self.__extension__, code) if isinstance(name, str): meth_name = f'_make_ext_{name}' meth = cast('ExtensionConstructor', @@ -3797,7 +3797,7 @@ def _make_cga_extensions(self, extensions: 'Extension | list[Schema_CGAExtension extensions_list = [] for code, extension in extensions.items(multi=True): - name = self.__extension__[code] + name = self._lookup_registry(self.__extension__, code) if isinstance(name, str): meth_name = f'_make_ext_{name}' meth = cast('ExtensionConstructor', diff --git a/pcapkit/protocols/misc/pcapng.py b/pcapkit/protocols/misc/pcapng.py index 74f4814b34..8d316ceaff 100644 --- a/pcapkit/protocols/misc/pcapng.py +++ b/pcapkit/protocols/misc/pcapng.py @@ -950,7 +950,7 @@ def read(self, length: 'Optional[int]' = None, *, _read: 'bool' = True, self._sect += 1 self._ctx = None - name = self.__block__[schema.type] + name = self._lookup_registry(self.__block__, schema.type) if isinstance(name, str): meth_name = f'_read_block_{name}' meth = cast('BlockParser', @@ -1010,7 +1010,7 @@ def make(self, if isinstance(block, bytes): block_val = block # type: bytes | Schema_BlockType elif isinstance(block, (dict, Data_PCAPNG)): - name = self.__block__[type_val] + name = self._lookup_registry(self.__block__, type_val) if isinstance(name, str): meth_name = f'_make_block_{name}' meth = cast('BlockConstructor', @@ -1876,7 +1876,7 @@ def _read_block_dsb(self, schema: 'Schema_DecryptionSecretsBlock', *, Parsed packet data. """ - name = self.__secrets__[schema.secrets_type] + name = self._lookup_registry(self.__secrets__, schema.secrets_type) if isinstance(name, str): meth_name = f'_read_secrets_{name}' meth = cast('SecretsParser', @@ -2040,7 +2040,7 @@ def _read_pcapng_options(self, options_schema: 'list[Schema_Option]') -> 'Option for schema in options_schema: type = schema.type - name = self.__option__[_option_key(type)] + name = self._lookup_registry(self.__option__, _option_key(type)) if isinstance(name, str): meth_name = f'_read_option_{name}' @@ -3126,7 +3126,7 @@ def _read_nrb_records(self, records_schema: 'list[Schema_NameResolutionRecord]') for schema in records_schema: type = schema.type - name = self.__record__[type] + name = self._lookup_registry(self.__record__, type) if isinstance(name, str): meth_name = f'_read_record_{name}' @@ -3736,7 +3736,7 @@ def _make_block_dsb(self, block: 'Optional[Data_DecryptionSecretsBlock]' = None, if isinstance(secrets_data, bytes): secrets_data_val = secrets_data # type: bytes | Schema_DSBSecrets elif isinstance(secrets_data, (dict, Data_DSBSecrets)): - name = self.__secrets__[secrets_type_val] + name = self._lookup_registry(self.__secrets__, secrets_type_val) if isinstance(name, str): meth_name = f'_make_secrets_{name}' meth = cast('SecretsConstructor', @@ -3925,7 +3925,7 @@ def _make_pcapng_options(self, options: 'Option | list[Schema_Option | tuple[Enu has_endofopt = True continue - name = self.__option__[_option_key(code)] + name = self._lookup_registry(self.__option__, _option_key(code)) if isinstance(name, str): meth_name = f'_make_option_{name}' meth = cast('OptionConstructor', @@ -3952,7 +3952,7 @@ def _make_pcapng_options(self, options: 'Option | list[Schema_Option | tuple[Enu has_endofopt = True continue - name = self.__option__[_option_key(code)] + name = self._lookup_registry(self.__option__, _option_key(code)) if isinstance(name, str): meth_name = f'_make_option_{name}' meth = cast('OptionConstructor', @@ -5353,7 +5353,7 @@ def _make_nrb_records(self, records: 'Record | list[Schema_NameResolutionRecord has_record_end = True continue - name = self.__record__[code] + name = self._lookup_registry(self.__record__, code) if isinstance(name, str): meth_name = f'_make_record_{name}' meth = cast('RecordConstructor', @@ -5379,7 +5379,7 @@ def _make_nrb_records(self, records: 'Record | list[Schema_NameResolutionRecord has_record_end = True continue - name = self.__record__[code] + name = self._lookup_registry(self.__record__, code) if isinstance(name, str): meth_name = f'_make_record_{name}' meth = cast('RecordConstructor', diff --git a/pcapkit/protocols/protocol.py b/pcapkit/protocols/protocol.py index 9736435233..fa0cc07f5f 100644 --- a/pcapkit/protocols/protocol.py +++ b/pcapkit/protocols/protocol.py @@ -58,6 +58,7 @@ _PT = TypeVar('_PT', bound='Data') _ST = TypeVar('_ST', bound='Schema') _CTX = TypeVar('_CTX', bound='ProtocolContext') +_VT = TypeVar('_VT') # readable characters' order list readable = [ord(char) for char in filter(lambda char: not char.isspace(), string.printable)] @@ -1237,6 +1238,42 @@ def _make_payload(cls, data: 'Data') -> 'ProtocolBase': return proto.from_data(data[name]) + @staticmethod + def _lookup_registry(registry: 'DefaultDict[Any, _VT]', code: 'Any') -> '_VT': + """Look up a dispatch registry entry without recording a miss. + + Arguments: + registry: dispatch registry to read, i.e. :attr:`self.__proto__ + ` or one of the per-protocol + ``__option__`` / ``__chunk__`` / ``__block__`` family. Passed in + rather than read from the class, so that a caller reaching the + registry through an instance keeps doing so. + code: registry key to look up, i.e. the wire code being dispatched on + + Returns: + The entry registered for ``code``, or the fallback ``registry`` + declares when ``code`` is not registered. + + Important: + Every one of these registries is a :class:`collections.defaultdict` + held on a *class* attribute, shared by every instance of the class in + the process. So ``registry[code]`` inserts each code it misses, and + parsing one packet carrying an unrecognised code is enough to grow + the registry permanently. + + The inserted value is whatever the default factory would have + produced anyway, so the entry buys nothing. It costs a spurious + "already registered" warning from the next genuine ``register`` call + for that code, and it makes "is this code registered?" + unanswerable by inspection, since the answer depends on what has + been parsed. The fallback is therefore read from the default factory + directly rather than through a lookup that records it. + + """ + if code in registry: + return registry[code] + return cast('Callable[[], _VT]', registry.default_factory)() + @staticmethod def _lookup_next_layer(registry: 'DefaultDict[int, ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]]', proto: 'int') -> 'Type[ProtocolBase]': @@ -1255,31 +1292,29 @@ def _lookup_next_layer(registry: 'DefaultDict[int, ModuleDescriptor[ProtocolBase ``proto`` is not registered. Important: - ``registry`` is a :class:`collections.defaultdict`, so indexing it - with an unregistered code would *insert* that code. It is - class-level -- shared by every instance in the process -- so parsing - a single packet with an unregistered code would grow it, and make - :meth:`self.register ` afterwards report that - code as already registered. The fallback is therefore read from the - default factory rather than through a lookup that records it. - - Resolving a :class:`~pcapkit.corekit.module.ModuleDescriptor` is - still written back, since that is memoisation of an import for a - code that *is* registered rather than a new entry. + The lookup itself is :meth:`self._lookup_registry + `, so a miss does not grow the shared + registry. What this adds is the next-layer-specific resolution step: + a registered code may hold a + :class:`~pcapkit.corekit.module.ModuleDescriptor` rather than a + class, and importing it is written back so the import happens once. + + That write-back is deliberately confined to a *hit*. Memoising the + fallback's resolution under ``proto`` would be exactly the insertion + :meth:`self._lookup_registry ` exists + to avoid. """ - if proto in registry: - protocol = registry[proto] - if isinstance(protocol, ModuleDescriptor): - protocol = protocol.klass - registry[proto] = protocol # update mapping upon import - return protocol - - fallback = cast('Callable[[], ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]]', - registry.default_factory)() - if isinstance(fallback, ModuleDescriptor): - return fallback.klass - return fallback + protocol = ProtocolBase._lookup_registry(registry, proto) + if isinstance(protocol, ModuleDescriptor): + klass = protocol.klass + # a descriptor can also come back from the default factory, and that + # one has no key to memoise under -- writing it back would recreate + # the insertion-on-miss this exists to avoid + if proto in registry: + registry[proto] = klass # update mapping upon import + return klass + return protocol def _decode_next_layer(self, dict_: '_PT', proto: 'int', length: 'Optional[int]' = None, *, packet: 'Optional[dict[str, Any]]' = None) -> '_PT': diff --git a/pcapkit/protocols/transport/sctp.py b/pcapkit/protocols/transport/sctp.py index cc0c8b6c15..3090794ccc 100644 --- a/pcapkit/protocols/transport/sctp.py +++ b/pcapkit/protocols/transport/sctp.py @@ -834,7 +834,7 @@ def _read_sctp_chunks(self) -> 'Chunks': for schema in self.__header__.chunks: code = schema.type - name = self.__chunk__[code] + name = self._lookup_registry(self.__chunk__, code) if isinstance(name, str): meth_name = f'_read_chunk_{name}' @@ -894,7 +894,7 @@ def _make_sctp_chunk(self, code: 'Enum_Chunk', chunk: 'Optional[Data_Chunk]' = N Constructed chunk schema. """ - name = self.__chunk__[code] + name = self._lookup_registry(self.__chunk__, code) if isinstance(name, str): meth_name = f'_make_chunk_{name}' meth = cast('ChunkConstructor', @@ -917,7 +917,7 @@ def _read_sctp_parameters(self, schemas: 'list[Schema_Parameter]') -> 'Parameter for schema in schemas: code = schema.type - name = self.__parameter__[code] + name = self._lookup_registry(self.__parameter__, code) if isinstance(name, str): meth_name = f'_read_param_{name}' @@ -966,7 +966,7 @@ def _make_sctp_parameter(self, code: 'Enum_Parameter', parameter: 'Optional[Data Constructed parameter schema. """ - name = self.__parameter__[code] + name = self._lookup_registry(self.__parameter__, code) if isinstance(name, str): meth_name = f'_make_param_{name}' meth = cast('ParameterConstructor', @@ -989,7 +989,7 @@ def _read_sctp_causes(self, schemas: 'list[Schema_ErrorCause]') -> 'Causes': for schema in schemas: code = schema.code - name = self.__cause__[code] + name = self._lookup_registry(self.__cause__, code) if isinstance(name, str): meth_name = f'_read_cause_{name}' @@ -1038,7 +1038,7 @@ def _make_sctp_cause(self, code: 'Enum_CauseCode', cause: 'Optional[Data_ErrorCa Constructed error cause schema. """ - name = self.__cause__[code] + name = self._lookup_registry(self.__cause__, code) if isinstance(name, str): meth_name = f'_make_cause_{name}' meth = cast('CauseConstructor', diff --git a/pcapkit/protocols/transport/tcp.py b/pcapkit/protocols/transport/tcp.py index 5b2cf82584..1f430dfdc1 100644 --- a/pcapkit/protocols/transport/tcp.py +++ b/pcapkit/protocols/transport/tcp.py @@ -666,7 +666,7 @@ def _read_tcp_options(self, size: 'int') -> 'Option': for schema in self.__header__.options: kind = schema.kind - name = self.__option__[kind] + name = self._lookup_registry(self.__option__, kind) if isinstance(name, str): meth_name = f'_read_mode_{name}' @@ -1403,7 +1403,7 @@ def _read_mode_mp(self, schema: 'Schema_MPTCP', *, options: 'Option') -> 'Data_M """ subtype = schema.subtype - name = self.__mp_option__[subtype] + name = self._lookup_registry(self.__mp_option__, subtype) if isinstance(name, str): meth_name = f'_read_mptcp_{name}' @@ -1945,7 +1945,7 @@ def _make_tcp_options(self, options: 'list[Schema_Option | tuple[Enum_Option, di if code in (Enum_Option.No_Operation, Enum_Option.End_of_Option_List): # ignore padding options by default continue - name = self.__option__[code] + name = self._lookup_registry(self.__option__, code) if isinstance(name, str): meth_name = f'_make_mode_{name}' meth = cast('OptionConstructor', @@ -1976,7 +1976,7 @@ def _make_tcp_options(self, options: 'list[Schema_Option | tuple[Enum_Option, di if code in (Enum_Option.No_Operation, Enum_Option.End_of_Option_List): continue - name = self.__option__[code] + name = self._lookup_registry(self.__option__, code) if isinstance(name, str): meth_name = f'_make_mode_{name}' meth = cast('OptionConstructor', @@ -2572,7 +2572,7 @@ def _make_mode_mp(self, code: 'Enum_Option', opt: 'Optional[Data_MPTCP]' = None, reversed=subtype_reversed, pack=False) subtype_val = Enum_MPTCPOption.get(subtype_val) - name = self.__mp_option__[subtype_val] + name = self._lookup_registry(self.__mp_option__, subtype_val) if isinstance(name, str): meth_name = f'_make_mptcp_{name}' meth = cast('MPOptionConstructor', diff --git a/tests/protocols/application/test_http_unit.py b/tests/protocols/application/test_http_unit.py index 8b3f9416e0..05c5ddec6c 100644 --- a/tests/protocols/application/test_http_unit.py +++ b/tests/protocols/application/test_http_unit.py @@ -282,6 +282,57 @@ def test_httpv2_id_length_make_bytes_and_register_frame_warning(self) -> None: finally: HTTPv2.__dict__['__frame__'][Frame.DATA] = original + def test_unregistered_frame_type_does_not_mutate_the_class_registry(self) -> None: + """Parsing must not write to the shared ``HTTPv2.__frame__``. + + #425's defect on HTTP/2 frame dispatch. ``__frame__`` is a + :class:`collections.defaultdict` on a class attribute shared by every + :class:`~pcapkit.protocols.application.httpv2.HTTP` instance in the + process, so ``__frame__[type]`` inserted every frame type it missed -- + and the value it inserted was ``'none'``, which the default factory + returns anyway. + + Frame type ``0xF0`` sits in the range :rfc:`9113` leaves for extensions, + so an unrecognised frame type is expected traffic rather than a + malformed frame -- HTTP/2 requires an endpoint to ignore one. + + """ + import io + + from pcapkit.const.http.frame import Frame + from pcapkit.protocols.application.httpv2 import HTTP as HTTPv2 + + # Length 13, type 0xF0, no flags, stream 1, then four octets of payload. + # + # 13 is the *whole* frame, header included, which is this library's + # convention rather than :rfc:`9113#section-4.1`'s -- that one counts the + # payload alone, so a real frame with four octets of payload declares 4. + # ``make`` writes ``payload + 9`` (httpv2.py:292) and the readers recover + # the payload as ``length - 9`` (httpv2.py:658,668), and ``read`` rejects + # anything under 9 outright, so a wire-accurate 4 raises here. Declaring + # 13 is what reaches the registry lookup; the mismatch with the RFC is a + # separate defect and not this test's to assert. + packet = bytes.fromhex('00000d' 'f0' '00' '00000001' '61626364') + self.assertEqual(int.from_bytes(packet[:3], 'big'), len(packet)) + + registry = HTTPv2.__dict__['__frame__'] + before = set(registry) + self.assertNotIn(Frame(0xF0), before) + + try: + with mock.patch('pcapkit.protocols.application.httpv2.warn'): + proto = HTTPv2(io.BytesIO(packet), len(packet)) + + # The frame is still parsed, by the fallback the registry declares. + self.assertEqual(proto.info.data, b'abcd') + self.assertEqual(set(registry), before) + + with mock.patch('pcapkit.protocols.application.httpv2.warn') as warn: + HTTPv2.register_frame(Frame(0xF0), 'none') + self.assertEqual(warn.call_count, 0) + finally: + registry.pop(Frame(0xF0), None) + def test_httpv2_frame_readers_cover_successful_frames(self) -> None: from pcapkit.const.http.error_code import ErrorCode from pcapkit.const.http.frame import Frame diff --git a/tests/protocols/internet/test_ipv6_extension_unit.py b/tests/protocols/internet/test_ipv6_extension_unit.py index 22870d5703..cc1e3cb23f 100644 --- a/tests/protocols/internet/test_ipv6_extension_unit.py +++ b/tests/protocols/internet/test_ipv6_extension_unit.py @@ -95,6 +95,69 @@ def test_ipv6_route_register_routing_warns_on_overwrite(self) -> None: finally: IPv6_Route.__dict__['__routing__'][Routing.Source_Route] = original + def test_unregistered_extension_codes_do_not_mutate_the_class_registry(self) -> None: + """Parsing must not write to the shared IPv6 extension registries. + + #425's defect on the three IPv6 extension header registries. Each is a + :class:`collections.defaultdict` on a class attribute shared by every + instance of its protocol in the process, so ``registry[code]`` inserted + every unrecognised option type and routing type a capture carried -- and + the value it inserted was ``'none'``, which the default factory returns + anyway. That made + :meth:`~pcapkit.protocols.internet.hopopt.HOPOPT.register_option` and its + siblings warn about overwriting something nobody registered. + + Option type ``0x1E`` is :rfc:`4727`'s experimentation code and routing + type 253 is :rfc:`4727`'s experimental routing type, so neither is + expected to be registered -- and both really do arrive on the wire, which + is what makes the leak reachable in ordinary use. + + """ + from pcapkit.const.ipv6.option import Option + from pcapkit.const.ipv6.routing import Routing + from pcapkit.protocols.internet.hopopt import HOPOPT + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + from pcapkit.protocols.internet.ipv6_route import IPv6_Route + + src = bytes.fromhex('fe80' + '0000' * 6 + '0001') + dst = bytes.fromhex('fe80' + '0000' * 6 + '0002') + + def datagram(next_header: int, extension: bytes) -> bytes: + return (b'\x60\x00\x00\x00' + len(extension).to_bytes(2, 'big') + + bytes([next_header, 64]) + src + dst + extension) + + for protocol, module, register, code, registry, packet in ( + # next header 0, i.e. hop-by-hop options: one 4-octet option of an + # unregistered type, padded to the mandatory 8 octets. + (HOPOPT, 'hopopt', HOPOPT.register_option, Option(0x1E), + HOPOPT.__dict__['__option__'], datagram(0, bytes([17, 0, 0x1E, 4]) + bytes(4))), + # next header 60, i.e. destination options, carrying the same option + (IPv6_Opts, 'ipv6_opts', IPv6_Opts.register_option, Option(0x1E), + IPv6_Opts.__dict__['__option__'], datagram(60, bytes([17, 0, 0x1E, 4]) + bytes(4))), + # next header 43, i.e. a 16-octet routing header of type 253 + (IPv6_Route, 'ipv6_route', IPv6_Route.register_routing, Routing(253), + IPv6_Route.__dict__['__routing__'], datagram(43, bytes([17, 1, 253, 0]) + bytes(12))), + ): + with self.subTest(protocol=protocol.__name__): + before = set(registry) + self.assertNotIn(code, before) + + try: + IPv6(io.BytesIO(packet), len(packet)) + self.assertEqual(set(registry), before) + + # A second datagram must behave identically; a class-level + # leak from the first would show up here rather than above. + IPv6(io.BytesIO(packet), len(packet)) + self.assertEqual(set(registry), before) + + with mock.patch(f'pcapkit.protocols.internet.{module}.warn') as warn: + register(code, 'none') + self.assertEqual(warn.call_count, 0) + finally: + registry.pop(code, None) + def test_ipv6_frag_index_length_and_make_data(self) -> None: from pcapkit.const.reg.transtype import TransType from pcapkit.protocols.internet.ipv6_frag import IPv6_Frag diff --git a/tests/protocols/internet/test_mh_unit.py b/tests/protocols/internet/test_mh_unit.py index 70c75a8b83..989198e7bc 100644 --- a/tests/protocols/internet/test_mh_unit.py +++ b/tests/protocols/internet/test_mh_unit.py @@ -84,6 +84,68 @@ def test_mh_register_extension_warns_on_overwrite(self) -> None: finally: MH.__dict__['__extension__'][CGAExtension.Multi_Prefix] = original + def test_unregistered_mh_codes_do_not_mutate_the_class_registries(self) -> None: + """Parsing must not write to ``__message__``, ``__option__`` or ``__extension__``. + + #425's defect on MH's three registries. Each is a + :class:`collections.defaultdict` on a class attribute shared by every + :class:`~pcapkit.protocols.internet.mh.MH` instance in the process, so + ``registry[code]`` inserted every unrecognised message type, option type + and CGA extension type -- and the value it inserted was the one the + default factory returns anyway, so it bought nothing while making + :meth:`~pcapkit.protocols.internet.mh.MH.register_message` and its + siblings warn about an overwrite that never happened. + + ``__extension__`` is reached through the construction path rather than a + parse: CGA extensions arrive inside a CGA Parameters option, whose schema + sizes its extension area from ``pkt['length']`` -- a key the nested packet + context does not carry -- so + :meth:`~pcapkit.protocols.internet.mh.MH._read_cga_extensions` is + unreachable from bytes today. The sibling read site in + :meth:`~pcapkit.protocols.internet.mh.MH._make_cga_extensions` had the + identical defect. + + """ + from pcapkit.const.mh.cga_extension import CGAExtension + from pcapkit.const.mh.option import Option + from pcapkit.const.mh.packet import Packet + from pcapkit.protocols.internet.mh import MH + + def parse(hexstr: str) -> None: + raw = bytes.fromhex(hexstr) + self.assertEqual(len(raw) % 8, 0, 'mobility header must be 8-octet aligned') + # the header length field counts 8-octet units after the first, so it + # has to describe the octets actually supplied + self.assertEqual((raw[1] + 1) * 8, len(raw)) + MH(io.BytesIO(raw), len(raw), extension=True) + + for label, register, code, registry, exercise in ( + # next, header length 1 (i.e. 16 octets), type 200, reserved, + # checksum, then ten octets of message body nothing can interpret + ('message', MH.register_message, Packet(200), MH.__dict__['__message__'], + lambda: parse('1101' 'c8' '00' '1234' '00000000000000000000')), + # a Binding Refresh Request carrying one 8-octet option of type 0x40 + ('option', MH.register_option, Option(0x40), MH.__dict__['__option__'], + lambda: parse('1101' '00' '00' '1234' '0000' '4006' '000000000000')), + ('extension', MH.register_extension, CGAExtension(0xFF), + MH.__dict__['__extension__'], + lambda: object.__new__(MH)._make_cga_extensions( + [(CGAExtension(0xFF), dict(data=b''))])), + ): + with self.subTest(registry=label): + before = set(registry) + self.assertNotIn(code, before) + + try: + exercise() + self.assertEqual(set(registry), before) + + with mock.patch('pcapkit.protocols.internet.mh.warn') as warn: + register(code, 'none' if label != 'message' else 'unknown') + self.assertEqual(warn.call_count, 0) + finally: + registry.pop(code, None) + def test_mh_read_make_properties_and_extension_accessors(self) -> None: from pcapkit.const.mh.packet import Packet from pcapkit.const.reg.transtype import TransType diff --git a/tests/protocols/misc/test_pcapng_unit.py b/tests/protocols/misc/test_pcapng_unit.py index cdc6ee43fc..67352b8d71 100644 --- a/tests/protocols/misc/test_pcapng_unit.py +++ b/tests/protocols/misc/test_pcapng_unit.py @@ -130,6 +130,87 @@ def test_pcapng_option_registry_preserves_duplicate_numeric_codes(self) -> None: self.assertEqual(option_map[_option_key(OptionType.pack_flags)], 'pack_flags') self.assertEqual(option_map.get(_option_key(OptionType.get(2, namespace='opt')), 'unknown'), 'unknown') + def test_unregistered_pcapng_codes_do_not_mutate_the_class_registries(self) -> None: + """Parsing must not write to any of PCAP-NG's four dispatch registries. + + #425's defect, four times over. ``__block__``, ``__option__``, + ``__record__`` and ``__secrets__`` are each a + :class:`collections.defaultdict` on a class attribute shared by every + :class:`~pcapkit.protocols.misc.pcapng.PCAPNG` instance in the process, so + ``registry[code]`` inserted every unrecognised block type, option code, + record type and secrets type a capture carried -- values the default + factory returns anyway. + + PCAP-NG makes this the easiest family to hit by accident: the format is + explicitly extensible, a reader is required to skip blocks and options it + does not know, and a capture written by any tool with private extensions + carries codes this library does not register. + + """ + import tempfile + + from pcapkit.const.pcapng.block_type import BlockType + from pcapkit.const.pcapng.option_type import OptionType + from pcapkit.const.pcapng.record_type import RecordType + from pcapkit.const.pcapng.secrets_type import SecretsType + from pcapkit.interface import extract + from pcapkit.protocols.misc.pcapng import PCAPNG, _option_key + + def block(type_: int, body: bytes) -> bytes: + length = 12 + len(body) + return (struct.pack(' bytes: + return (struct.pack(' None: + handle, path = tempfile.mkstemp(suffix='.pcapng') + try: + with os.fdopen(handle, 'wb') as file: + file.write(prologue + trailer) + extract(fin=path, store=True, nofile=True).engine.close() + finally: + os.unlink(path) + + for label, register, code, key, registry, trailer in ( + # An IRIG Timestamp Block: a real, specified block type that this + # library does not implement, so the default 'unknown' applies. + ('block', PCAPNG.register_block, BlockType(0x00000007), BlockType(0x00000007), + PCAPNG.__dict__['__block__'], block(0x00000007, bytes(4))), + # A second section header carrying an option code nothing registers. + ('option', PCAPNG.register_option, OptionType.get(42), + _option_key(OptionType.get(42)), PCAPNG.__dict__['__option__'], + block(0x0A0D0D0A, section + option(42, b'x') + option(0, b''))), + # A Name Resolution Block whose one record has an unregistered type. + ('record', PCAPNG.register_record, RecordType(0x0BAD), RecordType(0x0BAD), + PCAPNG.__dict__['__record__'], + block(0x00000004, struct.pack(' None: from pcapkit.const.pcapng.block_type import BlockType from pcapkit.protocols.data.misc.pcapng import SimplePacketBlock diff --git a/tests/protocols/test_protocol_base_unit.py b/tests/protocols/test_protocol_base_unit.py index a00e32aa04..0e819ed9a2 100644 --- a/tests/protocols/test_protocol_base_unit.py +++ b/tests/protocols/test_protocol_base_unit.py @@ -335,6 +335,43 @@ def __index__(cls) -> int: no_stop._exproto = 'raw' self.assertFalse(no_stop._check_term_threshold()) + def test_lookup_registry_reads_the_fallback_without_recording_it(self) -> None: + """The non-recording lookup is generic over what the registry holds. + + ``__proto__`` maps to protocol classes, but the ``__option__`` / + ``__chunk__`` / ``__block__`` family maps to *method names* -- a + :obj:`str`, or a ``(parser, constructor)`` pair. Those have no + :class:`~pcapkit.corekit.module.ModuleDescriptor` to resolve, so they + want the lookup on its own, which is why it is a helper of its own rather + than something buried inside + :meth:`~pcapkit.protocols.protocol.ProtocolBase._lookup_next_layer`. + + """ + DummyProtocol, _, _ = self._make_protocol_class() + + parser, constructor = object(), object() + registry = collections.defaultdict(lambda: 'donone', { + 2: 'mss', + 3: (parser, constructor), + }) + + # A hit is returned as it is, whichever shape it has. + self.assertEqual(DummyProtocol._lookup_registry(registry, 2), 'mss') + self.assertEqual(DummyProtocol._lookup_registry(registry, 3), + (parser, constructor)) + + # A miss reads the declared fallback and leaves no trace. + self.assertEqual(DummyProtocol._lookup_registry(registry, 156), 'donone') + self.assertNotIn(156, registry) + self.assertEqual(set(registry), {2, 3}) + + # A tuple key -- which is what ``PCAPNG.__option__`` is keyed on, since + # its option codes collide across block namespaces -- is no different. + namespaced = collections.defaultdict(lambda: 'unknown', {('if', 2): 'if_name'}) + self.assertEqual(DummyProtocol._lookup_registry(namespaced, ('if', 2)), 'if_name') + self.assertEqual(DummyProtocol._lookup_registry(namespaced, ('if', 42)), 'unknown') + self.assertEqual(set(namespaced), {('if', 2)}) + def test_lookup_next_layer_reads_the_fallback_without_recording_it(self) -> None: """A missed lookup must not turn into a registration. diff --git a/tests/protocols/transport/test_sctp_unit.py b/tests/protocols/transport/test_sctp_unit.py index e63bad1526..4fdf5e1947 100644 --- a/tests/protocols/transport/test_sctp_unit.py +++ b/tests/protocols/transport/test_sctp_unit.py @@ -1099,6 +1099,63 @@ def test_sub_registries_warn_on_overwrite(self) -> None: finally: registry[code] = original + def test_unregistered_sub_registry_codes_do_not_mutate_the_class(self) -> None: + """Parsing must not write to ``__chunk__``, ``__parameter__`` or ``__cause__``. + + #425's defect, on SCTP's three sub-registries. Each is a + :class:`collections.defaultdict` on a class attribute shared by every + :class:`~pcapkit.protocols.transport.sctp.SCTP` instance in the process, + so ``registry[code]`` inserted every unregistered chunk type, parameter + type and error cause code a capture happened to carry -- and the value it + inserted was ``'donone'``, which the default factory returns anyway. The + entries bought nothing and made + :meth:`~pcapkit.protocols.transport.sctp.SCTP.register_chunk` and its + siblings report an overwrite of something nobody registered. + + The chunk type is unassigned by IANA, and both the parameter type and the + error cause code are drawn from the ``0xFFF0``-and-up range :rfc:`9260` + reserves for IETF-Defined extensions, so nothing is expected to register + them. + + """ + from pcapkit.const.sctp.cause_code import CauseCode + from pcapkit.const.sctp.chunk import Chunk + from pcapkit.const.sctp.parameter import Parameter + from pcapkit.protocols.transport import sctp as sctp_module + from pcapkit.protocols.transport.sctp import SCTP + + header = bytes.fromhex('26ab960c1122334400000000') + for register, code, registry, chunk in ( + # an unassigned chunk type, on its own + (SCTP.register_chunk, Chunk(200), SCTP.__dict__['__chunk__'], + 'c8' '00' '0004'), + # INIT, length 24 = 4 header + 16 fixed + one 4-octet parameter + (SCTP.register_parameter, Parameter(0xFFF0), SCTP.__dict__['__parameter__'], + '01' '00' '0018' '11223344' '0001a000' '000a' '000a' '55667788' 'fff00004'), + # ERROR, length 8 = 4 header + one 4-octet error cause + (SCTP.register_cause, CauseCode(0xFFF0), SCTP.__dict__['__cause__'], + '09' '00' '0008' 'fff00004'), + ): + with self.subTest(register=register.__name__): + before = set(registry) + self.assertNotIn(code, before) + + body = bytes.fromhex(chunk) + # the chunk length field has to be the octets actually supplied, + # or the parser is reading past the end of the packet and the + # test is exercising a code path no real capture reaches + self.assertEqual(int.from_bytes(body[2:4], 'big'), len(body)) + + try: + self._packet(header + body) + self.assertEqual(set(registry), before) + + with mock.patch.object(sctp_module, 'warn') as warned: + register(code, 'donone') + self.assertEqual(warned.call_count, 0) + finally: + registry.pop(code, None) + def test_callable_sub_registry_entries_are_used(self) -> None: from pcapkit.const.sctp.chunk import Chunk from pcapkit.protocols.transport.sctp import SCTP diff --git a/tests/protocols/transport/test_tcp_udp_unit.py b/tests/protocols/transport/test_tcp_udp_unit.py index 0e4e010599..4a4f28f251 100644 --- a/tests/protocols/transport/test_tcp_udp_unit.py +++ b/tests/protocols/transport/test_tcp_udp_unit.py @@ -1,6 +1,7 @@ from __future__ import annotations import datetime +import io import ipaddress import importlib.util import types @@ -1174,6 +1175,97 @@ def test_construction_accepts_bare_integer_ports(self) -> None: self.assertEqual(unnamed.info.srcport.port, 53406) self.assertIs(unnamed.info.dstport, member) + def test_unregistered_option_kind_does_not_mutate_the_class_registry(self) -> None: + """Parsing must not write to the shared ``TCP.__option__``. + + The reproduction from #425: one segment carrying option kind 156, which + nothing registers. ``__option__`` is a + :class:`collections.defaultdict` on a class attribute shared by every + :class:`~pcapkit.protocols.transport.tcp.TCP` instance in the process, so + ``__option__[kind]`` inserted the kind it missed -- and the value it + inserted was ``'donone'``, exactly what the default factory returns + anyway. The entry bought nothing and cost a + :func:`~pcapkit.foundation.registry.protocols.register_tcp_option` call + afterwards a warning about an overwrite that never happened. + + The segment is the issue's, with its data offset corrected from 7 words + to 6: the issue supplies 24 octets but declared a 28-octet header. The + leak reproduces either way -- checked, since a malformed offset driving + the parser somewhere it would not otherwise go would have made this test + pass for the wrong reason -- but a test is worth nothing if the packet it + asserts against could not exist on a wire. + + """ + from pcapkit.const.tcp.option import Option + from pcapkit.foundation.registry.protocols import register_tcp_option + from pcapkit.protocols.transport.tcp import TCP + + # ports, seq, ack, data offset 6 words / flags, window, checksum, urgent + # pointer, then one 4-octet option of kind 156. + packet = (bytes.fromhex('005001bb00000000000000006002ffff00000000') + + bytes([156, 2, 0, 0])) + + registry = TCP.__dict__['__option__'] + before = set(registry) + self.assertNotIn(Option(156), before) + + try: + proto = TCP(io.BytesIO(packet), len(packet)) + + # The header the segment declares is the header it supplies, so the + # option area is read from real octets rather than off the end. + self.assertEqual(proto.info.hdr_len, len(packet)) + + # The option is still parsed, by the fallback the registry declares; + # only the registry write is gone. + self.assertIn(Option(156), proto.info.options) + self.assertEqual(set(registry), before) + + # A second segment must behave identically; a class-level leak from + # the first would show up here rather than above. + TCP(io.BytesIO(packet), len(packet)) + self.assertEqual(set(registry), before) + + with mock.patch('pcapkit.protocols.transport.tcp.warn') as warn: + register_tcp_option(Option(156), 'donone') + self.assertEqual(warn.call_count, 0) + finally: + registry.pop(Option(156), None) + + def test_unregistered_mptcp_subtype_does_not_mutate_the_class_registry(self) -> None: + """``TCP.__mp_option__`` leaks the same way, on the construction path. + + The parse path cannot reach it with an unregistered subtype: the schema + layer resolves an unknown subtype to + :class:`~pcapkit.protocols.schema.transport.tcp.MPTCPUnknown`, whose + ``data`` field sizes itself from ``pkt['length']`` -- a key the nested + packet context does not carry -- so it raises before + :meth:`~pcapkit.protocols.transport.tcp.TCP._read_mode_mp` ever looks the + subtype up. Construction reaches the sibling read site in + :meth:`~pcapkit.protocols.transport.tcp.TCP._make_mode_mp`, which had the + identical defect. + + """ + from pcapkit.const.tcp.mp_tcp_option import MPTCPOption + from pcapkit.const.tcp.option import Option + from pcapkit.foundation.registry.protocols import register_tcp_mp_option + from pcapkit.protocols.transport.tcp import TCP + + registry = TCP.__dict__['__mp_option__'] + before = set(registry) + self.assertNotIn(MPTCPOption(0xF), before) + + try: + schema = object.__new__(TCP)._make_mode_mp(Option.Multipath_TCP, subtype=0xF) + self.assertEqual(schema.test['subtype'], MPTCPOption(0xF)) + self.assertEqual(set(registry), before) + + with mock.patch('pcapkit.protocols.transport.tcp.warn') as warn: + register_tcp_mp_option(MPTCPOption(0xF), 'unknown') + self.assertEqual(warn.call_count, 0) + finally: + registry.pop(MPTCPOption(0xF), None) + if __name__ == '__main__': unittest.main()