From 1ef0195fef922043a5942d0a320ea9864187fb0e Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Sun, 1 Nov 2026 14:09:30 -0500 Subject: [PATCH] test(generators): enumerate FRAD's NLPID map and ICMP's type registries (#2560) - dispatch.py: a `frad` family over `FRAD.__proto__`, probing each NLPID through a LINUX_SLL capture with `arphrd_type` FRAD, and `PINNED_TARGETS` entries for both codes; `test_cases_cover_every_table_named_in_the_issue` counts them. - options.py: `icmpv4-type` and `icmpv6-type` families over `ICMPv4.__type__` and `ICMPv6.__type__`, with non-degenerate echo, quoted-packet, pointer and MTU arguments; neither writes a capture. - test_option_generator_icmp_2560_unit: pins each generated message's data model and body against RFC 792 and RFC 4443, which catches a retarget between two types of the same layout that the round trip cannot see. --- examples/generators/dispatch.py | 38 +++- examples/generators/options.py | 105 +++++++++++ .../protocols/test_dispatch_registry_unit.py | 23 +-- .../test_option_generator_icmp_2560_unit.py | 167 ++++++++++++++++++ 4 files changed, 322 insertions(+), 11 deletions(-) create mode 100644 tests/protocols/test_option_generator_icmp_2560_unit.py diff --git a/examples/generators/dispatch.py b/examples/generators/dispatch.py index 5a752ded91..0dc1afa0ec 100644 --- a/examples/generators/dispatch.py +++ b/examples/generators/dispatch.py @@ -615,6 +615,36 @@ def _loopback_build(family: 'int') -> 'Any': return _extract_first(_pcap_bytes((_loopback(family, payload),), linktype=0), '.pcap') +############################################################################### +# FRAD -- NLPID. +############################################################################### + + +def _frad_registry() -> 'Any': + from pcapkit.protocols.link.frad import FRAD + return FRAD.__proto__ + + +def _frad_payload(nlpid: 'int') -> 'bytes': + """The registered target's own wire bytes for ``nlpid``.""" + if nlpid == 0xCC: + return _ip4(6, _tcp(9999)) + if nlpid == 0x8E: + return _ip6(6, _tcp(9999)) + raise LookupError(f'frad: no payload builder for NLPID {nlpid!r}') # pragma: no cover + + +def _frad_build(nlpid: 'int') -> 'Any': + """A direct-NLPID FRAD frame, reached as LinuxSLL dispatches it in a capture.""" + from pcapkit.const.linux.arphrd import ARPHRD + from pcapkit.protocols.link.frad import FRAD + from pcapkit.protocols.link.linux_sll1 import LinuxSLL1 + + frad = bytes(FRAD(nlpid=int(nlpid), payload=_frad_payload(nlpid))) + record = bytes(LinuxSLL1(arphrd_type=ARPHRD.FRAD, protocol_type=0, payload=frad)) + return _extract_first(_pcap_bytes((record,), linktype=113), '.pcap') + + ############################################################################### # The families, the cases, and the probe. ############################################################################### @@ -642,7 +672,8 @@ def _link_enum() -> 'Any': #: Every family this module exercises, one per ``__proto__`` table named in -#: GitHub issue #496, and the loopback table #1574 added. +#: GitHub issue #496, the loopback table #1574 added, and FRAD's NLPID table +#: (#2560). FAMILIES = ( Family('link', _link_registry, _link_build, _link_enum), Family('internet', _internet_registry, _internet_build), @@ -652,6 +683,7 @@ def _link_enum() -> 'Any': Family('pcap-frame', _pcap_frame_registry, _pcap_frame_build), Family('pcapng-frame', _pcapng_registry, _pcapng_build), Family('loopback', _loopback_registry, _loopback_build), + Family('frad', _frad_registry, _frad_build), ) #: The families keyed by label, for a caller -- :func:`probe` -- that wants @@ -762,6 +794,10 @@ def _link_enum() -> 'Any': 'loopback/24': ('pcapkit.protocols.internet.ipv6', 'IPv6'), 'loopback/28': ('pcapkit.protocols.internet.ipv6', 'IPv6'), 'loopback/30': ('pcapkit.protocols.internet.ipv6', 'IPv6'), + + # -- FRAD.__proto__ (NLPID, keyed in decimal: 0x8E, 0xCC) ------------------ + 'frad/142': ('pcapkit.protocols.internet.ipv6', 'IPv6'), + 'frad/204': ('pcapkit.protocols.internet.ipv4', 'IPv4'), } diff --git a/examples/generators/options.py b/examples/generators/options.py index cb75275856..41a7c53ae5 100644 --- a/examples/generators/options.py +++ b/examples/generators/options.py @@ -1165,6 +1165,103 @@ def _dccp_rebuild(options: 'Any') -> 'Any': return DCCP(options=options, **DCCP_BASE) +############################################################################### +# ICMP and ICMPv6 -- message types +############################################################################### + +#: The invoking packet an ICMP error message quotes: an IPv4 header and the +#: first eight octets of its UDP datagram [:rfc:`792`]. +ICMPV4_QUOTED = (bytes.fromhex('4500001c00010000401100000a0000010a000002') + + bytes.fromhex('3039003500080000')) + +#: The invoking packet an ICMPv6 error message quotes: an IPv6 header and its +#: UDP header [:rfc:`4443#section-3`]. +ICMPV6_QUOTED = (bytes.fromhex('6000000000081140') + bytes(15) + b'\x01' + bytes(15) + b'\x02' + + bytes.fromhex('3039003500080000')) + +#: Echo arguments shared by both versions. The defaults are all zero and empty, +#: which cannot tell a swapped ``id`` and ``seq``, or a dropped ``data``, apart. +ICMP_ECHO = {'id': 0x1234, 'seq': 0x5678, 'data': b'ping'} + + +def _icmpv4_registry() -> 'Any': + from pcapkit.protocols.internet.icmpv4 import ICMPv4 + return ICMPv4.__type__ + + +def _icmpv4_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.icmp.type import Type as Enum_Type + return { + Enum_Type.Echo_Reply: dict(ICMP_ECHO), + Enum_Type.Echo: dict(ICMP_ECHO), + Enum_Type.Destination_Unreachable: {'data': ICMPV4_QUOTED}, + Enum_Type.Time_Exceeded: {'data': ICMPV4_QUOTED}, + # Octet 8 of the quoted header is its TTL. + Enum_Type.Parameter_Problem: {'pointer': 8, 'data': ICMPV4_QUOTED}, + } + + +def _icmpv4_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.protocols.internet.icmpv4 import ICMPv4 + + # ``body`` has to be a mapping for ``make`` to dispatch on ``__type__``; + # a ``bytes`` body is parsed against the schema registry instead. + return ICMPv4(type=code, code=0, body=dict(kwargs)) + + +def _icmpv4_parse(octets: 'bytes') -> 'Any': + from pcapkit.protocols.internet.icmpv4 import ICMPv4 + return ICMPv4(octets, len(octets)) + + +def _icmp_extract(parsed: 'Any') -> 'Any': + return parsed.info + + +def _icmpv4_rebuild(message: 'Any') -> 'Any': + from pcapkit.protocols.internet.icmpv4 import ICMPv4 + return ICMPv4(type=message.type, code=message.code, checksum=message.checksum, + body=message) + + +def _icmpv6_registry() -> 'Any': + from pcapkit.protocols.internet.icmpv6 import ICMPv6 + return ICMPv6.__type__ + + +def _icmpv6_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.icmpv6.type import Type as Enum_Type + return { + Enum_Type.Destination_Unreachable: {'data': ICMPV6_QUOTED}, + # The IPv6 minimum link MTU [RFC 8200 section 5]. + Enum_Type.Packet_Too_Big: {'mtu': 1280, 'data': ICMPV6_QUOTED}, + Enum_Type.Time_Exceeded: {'data': ICMPV6_QUOTED}, + # Octet 6 of the quoted header is its Next Header. + Enum_Type.Parameter_Problem: {'pointer': 6, 'data': ICMPV6_QUOTED}, + Enum_Type.Echo_Request: dict(ICMP_ECHO), + Enum_Type.Echo_Reply: dict(ICMP_ECHO), + } + + +def _icmpv6_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.protocols.internet.icmpv6 import ICMPv6 + + # With no enclosing IPv6 layer to supply the pseudo-header, the checksum is + # left zero, and the rebuild passes the parsed zero back. + return ICMPv6(type=code, code=0, body=dict(kwargs)) + + +def _icmpv6_parse(octets: 'bytes') -> 'Any': + from pcapkit.protocols.internet.icmpv6 import ICMPv6 + return ICMPv6(octets, len(octets)) + + +def _icmpv6_rebuild(message: 'Any') -> 'Any': + from pcapkit.protocols.internet.icmpv6 import ICMPv6 + return ICMPv6(type=message.type, code=message.code, checksum=message.checksum, + body=message) + + ############################################################################### # HTTP/2 -- frames ############################################################################### @@ -1660,6 +1757,14 @@ def _pcapng_secrets_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': Family('hip-parameter', _hip_registry, _hip_overrides, _hip_build, _hip_parse, _hip_extract, _hip_rebuild, capture='options-internet.pcap', envelope='ipv6', proto=139), + # NOTE: The ICMP families write no capture; their wire form is pinned against + # RFC 792 and RFC 4443 by test_option_generator_icmp_2560_unit instead. + Family('icmpv4-type', _icmpv4_registry, _icmpv4_overrides, + _icmpv4_build, _icmpv4_parse, _icmp_extract, _icmpv4_rebuild, + capture=None, envelope=None), + Family('icmpv6-type', _icmpv6_registry, _icmpv6_overrides, + _icmpv6_build, _icmpv6_parse, _icmp_extract, _icmpv6_rebuild, + capture=None, envelope=None), # The PCAP-NG families write no capture -- see the section comment # above ``PCAPNG_TIME`` for why the construction API's defaults do not # make a reproducible one, and what already covers that space instead. diff --git a/tests/protocols/test_dispatch_registry_unit.py b/tests/protocols/test_dispatch_registry_unit.py index de96208684..4d7c5b899b 100644 --- a/tests/protocols/test_dispatch_registry_unit.py +++ b/tests/protocols/test_dispatch_registry_unit.py @@ -4,7 +4,8 @@ GitHub issue #496: the ``__proto__`` dispatch registries decide which :class:`~pcapkit.protocols.protocol.Protocol` subclass parses the next layer. #496 names the tables, joined by #904's -``Internet.__proto__[TransType.Shim6]`` and #1574's loopback table. +``Internet.__proto__[TransType.Shim6]``, #1574's loopback table and #2560's +FRAD table. :file:`test_registry_runtime.py` checks that a table entry *resolves* to the right class object, and :file:`test_dispatch_bindings_unit.py` checks that a hand-picked set of codes @@ -215,7 +216,7 @@ def test_registry_matches_pinned_target(self) -> None: ) def test_cases_cover_every_table_named_in_the_issue(self) -> None: - """The enumeration finds every entry, across the tables of #496 and #1574's loopback table. + """The enumeration finds every entry, across the tables of #496, #1574 and #2560. A guard on the shape of the result rather than on any one case: if the registries grow or shrink without this module noticing, the per-family @@ -229,20 +230,22 @@ def test_cases_cover_every_table_named_in_the_issue(self) -> None: # address families of Loopback.__proto__, #2438's # Frame.__proto__[LinkType.LINUX_SLL]/[LinkType.LINUX_SLL2] in both the # pcap and pcapng frame tables, #2439's IGMP, #2436's - # Internet.__proto__[TransType.ICMP] and [TransType.IPv6_ICMP], and - # #2489's Internet.__proto__[TransType.DCCP]. - self.assertEqual(len(cases), 57, - 'expected exactly 57 entries across the seven __proto__ ' - 'tables named in GitHub issue #496 and the loopback table ' - 'of #1574; a different count means a registry changed ' - 'shape and this module was not updated to match') + # Internet.__proto__[TransType.ICMP] and [TransType.IPv6_ICMP], #2489's + # Internet.__proto__[TransType.DCCP], and #2560's two NLPIDs of + # FRAD.__proto__. + self.assertEqual(len(cases), 59, + 'expected exactly 59 entries across the seven __proto__ ' + 'tables named in GitHub issue #496, the loopback table ' + 'of #1574 and the FRAD table of #2560; a different count ' + 'means a registry changed shape and this module was not ' + 'updated to match') by_family = {} # type: dict[str, int] for case in cases: by_family[case.family] = by_family.get(case.family, 0) + 1 self.assertEqual(by_family, { 'link': 7, 'internet': 21, 'tcp': 5, 'udp': 4, 'sctp': 2, - 'pcap-frame': 7, 'pcapng-frame': 7, 'loopback': 4, + 'pcap-frame': 7, 'pcapng-frame': 7, 'loopback': 4, 'frad': 2, }) def test_dispatch_reaches_target_or_is_a_recorded_degrade(self) -> None: diff --git a/tests/protocols/test_option_generator_icmp_2560_unit.py b/tests/protocols/test_option_generator_icmp_2560_unit.py new file mode 100644 index 0000000000..ccd5530246 --- /dev/null +++ b/tests/protocols/test_option_generator_icmp_2560_unit.py @@ -0,0 +1,167 @@ +# -*- coding: utf-8 -*- +"""The generator's ICMP and ICMPv6 messages carry their RFC layouts. C.f. #2560. + +The ``icmpv4-type`` and ``icmpv6-type`` families of +:file:`examples/generators/options.py` build one message per code of +``ICMPv4.__type__`` and ``ICMPv6.__type__``. A round trip cannot see an entry +retargeted at a method of the same layout: with Time Exceeded mapped to +``destination_unreachable`` the cycle still closes. So :data:`ICMPV4` and +:data:`ICMPV6` spell each type's data model and body out against :rfc:`792` and +:rfc:`4443` rather than reading them from the generator. + +This module is unit tier: it builds its messages in memory from the generator's +own cases and reads no capture, so it runs on a fresh checkout. + +""" + +from __future__ import annotations + +import importlib +import importlib.util +import sys +import types +import unittest +import warnings +from typing import TYPE_CHECKING + +from tests._tiers import ROOT + +if TYPE_CHECKING: + from typing import Any + +#: Test tier (#1958): tests examples/generators/options.py, which make_samples.py runs +#: on every integration leg. +__tier__ = 'LATEST' + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + +#: An Echo body: identifier 0x1234, sequence number 0x5678, then the data. +ECHO = bytes.fromhex('1234' '5678') + b'ping' +#: An IPv4 header (protocol UDP) and its first eight payload octets, a UDP header. +QUOTED4 = (bytes.fromhex('4500001c00010000401100000a0000010a000002') + + bytes.fromhex('3039003500080000')) +#: An IPv6 header (next header UDP) and a UDP header. +QUOTED6 = (bytes.fromhex('6000000000081140') + bytes(15) + b'\x01' + bytes(15) + b'\x02' + + bytes.fromhex('3039003500080000')) + +#: ICMP type -> the data model it parses into, and its body after the 4-octet +#: header (:rfc:`792`). +ICMPV4 = { + 0: ('Echo', ECHO), # Echo Reply + 3: ('DestinationUnreachable', bytes(4) + QUOTED4), # 4 unused octets + 8: ('Echo', ECHO), # Echo + 11: ('TimeExceeded', bytes(4) + QUOTED4), # 4 unused octets + 12: ('ParameterProblem', b'\x08' + bytes(3) + QUOTED4), # Pointer, 3 unused +} + +#: ICMPv6 type -> the data model it parses into, and its body after the 4-octet +#: header (:rfc:`4443#section-3`, :rfc:`4443#section-4`). +ICMPV6 = { + 1: ('DestinationUnreachable', bytes(4) + QUOTED6), # 4 unused octets + 2: ('PacketTooBig', (1280).to_bytes(4, 'big') + QUOTED6), # 32-bit MTU + 3: ('TimeExceeded', bytes(4) + QUOTED6), # 4 unused octets + 4: ('ParameterProblem', (6).to_bytes(4, 'big') + QUOTED6), # 32-bit Pointer + 128: ('Echo', ECHO), # Echo Request + 129: ('Echo', ECHO), # Echo Reply +} + +#: Family label -> (protocol module and class, data module, expected table). +FAMILIES = { + 'icmpv4-type': (('pcapkit.protocols.internet.icmpv4', 'ICMPv4'), + 'pcapkit.protocols.data.internet.icmpv4', ICMPV4), + 'icmpv6-type': (('pcapkit.protocols.internet.icmpv6', 'ICMPv6'), + 'pcapkit.protocols.data.internet.icmpv6', ICMPV6), +} + + +def _load_generator() -> 'types.ModuleType': + """Load :file:`examples/generators/options.py` by path, under a name of its own. + + Returns: + The generator module. + + Raises: + RuntimeError: If the module cannot be found or loaded. + + """ + path = ROOT / 'examples' / 'generators' / 'options.py' + spec = importlib.util.spec_from_file_location('pcapkit_samples_options_icmp', path) + if spec is None or spec.loader is None: # pragma: no cover + raise RuntimeError(f'cannot load the option generator from {path}') + + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def _ones_complement_sum(octets: bytes) -> int: + """The :rfc:`1071` sum of ``octets``, an odd last octet padded with a zero.""" + padded = octets + bytes(len(octets) % 2) + total = sum(int.from_bytes(padded[i:i + 2], 'big') for i in range(0, len(padded), 2)) + while total >> 16: + total = (total & 0xFFFF) + (total >> 16) + return total + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class ICMPGeneratedMessageTests(unittest.TestCase): + """Each ``icmpv4-type`` and ``icmpv6-type`` case, as the generator builds it.""" + + @classmethod + def setUpClass(cls) -> None: + options = _load_generator() + cls.messages = {} # type: dict[str, dict[int, Any]] + for label in FAMILIES: + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + cls.messages[label] = { + int(outcome.case.code): outcome + for outcome in options.outcomes((options.FAMILY_MAP[label],))} + + def test_every_type_is_generated(self) -> None: + """Each family yields exactly the registered types, and each round-trips.""" + for label, (_, _, expected) in FAMILIES.items(): + with self.subTest(family=label): + self.assertEqual(sorted(self.messages[label]), sorted(expected)) + for code, outcome in self.messages[label].items(): + with self.subTest(type=code): + self.assertEqual(outcome.status, 'OK', outcome.detail) + + def test_each_message_is_laid_out_as_its_rfc_says(self) -> None: + """Type, a zero code, the checksum, then the type's own body.""" + for label, (_, _, expected) in FAMILIES.items(): + for code, outcome in self.messages[label].items(): + with self.subTest(family=label, type=code): + octets = outcome.octets + self.assertEqual(octets[:2], bytes([code, 0])) + self.assertEqual(octets[4:].hex(), expected[code][1].hex()) + + def test_each_icmpv4_checksum_is_the_rfc_1071_sum(self) -> None: + """The whole message, checksum included, sums to all ones (:rfc:`792`).""" + for code, outcome in self.messages['icmpv4-type'].items(): + with self.subTest(type=code): + self.assertEqual(_ones_complement_sum(outcome.octets), 0xFFFF) + + def test_each_message_parses_into_its_types_data_model(self) -> None: + """The data model is the one :data:`ICMPV4` and :data:`ICMPV6` name. + + This is what a retarget between two types of the same layout changes: + Time Exceeded parsed by Destination Unreachable's method yields a + ``DestinationUnreachable``, octets unchanged. + + """ + for label, ((module, name), data_module, expected) in FAMILIES.items(): + protocol = getattr(importlib.import_module(module), name) + data = importlib.import_module(data_module) + for code, outcome in self.messages[label].items(): + with self.subTest(family=label, type=code): + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + parsed = protocol(outcome.octets, len(outcome.octets)) + self.assertIs(type(parsed.info), getattr(data, expected[code][0])) + + +if __name__ == '__main__': + unittest.main()