diff --git a/docs/source/pcapkit/protocols/internet/hip.rst b/docs/source/pcapkit/protocols/internet/hip.rst index 1ef6482aff..ecc9777e89 100644 --- a/docs/source/pcapkit/protocols/internet/hip.rst +++ b/docs/source/pcapkit/protocols/internet/hip.rst @@ -37,6 +37,8 @@ Octets Bits Name Description .. autoproperty:: protocol .. autoproperty:: protochain + .. automethod:: id + .. automethod:: register_parameter .. automethod:: read diff --git a/docs/source/pcapkit/protocols/internet/hopopt.rst b/docs/source/pcapkit/protocols/internet/hopopt.rst index d519ec44a1..1747e3821c 100644 --- a/docs/source/pcapkit/protocols/internet/hopopt.rst +++ b/docs/source/pcapkit/protocols/internet/hopopt.rst @@ -28,6 +28,8 @@ Octets Bits Name Description .. autoproperty:: protocol .. autoproperty:: protochain + .. automethod:: id + .. automethod:: register_option .. automethod:: read diff --git a/docs/source/pcapkit/protocols/internet/ipv6_ext.rst b/docs/source/pcapkit/protocols/internet/ipv6_ext.rst index 05ba9ae3d6..ecc0deca48 100644 --- a/docs/source/pcapkit/protocols/internet/ipv6_ext.rst +++ b/docs/source/pcapkit/protocols/internet/ipv6_ext.rst @@ -45,6 +45,8 @@ instead of clipping it. .. autoproperty:: payload .. autoproperty:: protochain + .. automethod:: id + .. automethod:: read .. automethod:: make diff --git a/docs/source/pcapkit/protocols/internet/ipv6_frag.rst b/docs/source/pcapkit/protocols/internet/ipv6_frag.rst index a34c8a38f6..432178c96e 100644 --- a/docs/source/pcapkit/protocols/internet/ipv6_frag.rst +++ b/docs/source/pcapkit/protocols/internet/ipv6_frag.rst @@ -31,6 +31,8 @@ Octets Bits Name Description .. autoproperty:: protocol .. autoproperty:: protochain + .. automethod:: id + .. automethod:: read .. automethod:: make diff --git a/docs/source/pcapkit/protocols/internet/ipv6_opts.rst b/docs/source/pcapkit/protocols/internet/ipv6_opts.rst index 6df99df980..5c8ad36168 100644 --- a/docs/source/pcapkit/protocols/internet/ipv6_opts.rst +++ b/docs/source/pcapkit/protocols/internet/ipv6_opts.rst @@ -28,6 +28,8 @@ Octets Bits Name Description .. autoproperty:: protocol .. autoproperty:: protochain + .. automethod:: id + .. automethod:: register_option .. automethod:: read diff --git a/docs/source/pcapkit/protocols/internet/ipv6_route.rst b/docs/source/pcapkit/protocols/internet/ipv6_route.rst index f1e24199b9..7c356d991e 100644 --- a/docs/source/pcapkit/protocols/internet/ipv6_route.rst +++ b/docs/source/pcapkit/protocols/internet/ipv6_route.rst @@ -29,6 +29,8 @@ Octets Bits Name Description .. autoproperty:: protocol .. autoproperty:: protochain + .. automethod:: id + .. automethod:: register_routing .. automethod:: read diff --git a/docs/source/pcapkit/protocols/internet/mh.rst b/docs/source/pcapkit/protocols/internet/mh.rst index c5169f92cb..76a9768ba3 100644 --- a/docs/source/pcapkit/protocols/internet/mh.rst +++ b/docs/source/pcapkit/protocols/internet/mh.rst @@ -36,6 +36,8 @@ Octets Bits Name Description .. autoproperty:: protocol .. autoproperty:: protochain + .. automethod:: id + .. automethod:: register_message .. automethod:: register_option .. automethod:: register_extension diff --git a/pcapkit/protocols/internet/hip.py b/pcapkit/protocols/internet/hip.py index 163c328aaf..588361d644 100644 --- a/pcapkit/protocols/internet/hip.py +++ b/pcapkit/protocols/internet/hip.py @@ -747,6 +747,16 @@ def register_parameter(cls, code: 'Enum_Parameter', meth: 'str | tuple[Parameter warn(f'parameter {code} already registered, overwriting', RegistryWarning) cls.__parameter__[code] = meth + @classmethod + def id(cls) -> 'tuple[Literal["HIP"]]': # type: ignore[override] + """Index ID of the protocol. + + Returns: + Index ID of the protocol. + + """ + return ('HIP',) + ########################################################################## # Data models. ########################################################################## diff --git a/pcapkit/protocols/internet/hopopt.py b/pcapkit/protocols/internet/hopopt.py index 78da396c2a..71f824ff1d 100644 --- a/pcapkit/protocols/internet/hopopt.py +++ b/pcapkit/protocols/internet/hopopt.py @@ -417,6 +417,16 @@ def register_option(cls, code: 'Enum_Option', meth: 'str | tuple[OptionParser, O warn(f'option {code} already registered, overwriting', RegistryWarning) cls.__option__[code] = meth + @classmethod + def id(cls) -> 'tuple[Literal["HOPOPT"]]': # type: ignore[override] + """Index ID of the protocol. + + Returns: + Index ID of the protocol. + + """ + return ('HOPOPT',) + ########################################################################## # Data models. ########################################################################## diff --git a/pcapkit/protocols/internet/ipv6_ext.py b/pcapkit/protocols/internet/ipv6_ext.py index 7bed48c0f2..e4e84afeaf 100644 --- a/pcapkit/protocols/internet/ipv6_ext.py +++ b/pcapkit/protocols/internet/ipv6_ext.py @@ -223,14 +223,15 @@ class keywords implement it. Warning: A subclass **must** define :attr:`name`, :attr:`alias`, - :attr:`protocol`, :attr:`length` and :meth:`__index__` itself. Each - carries this class's *fallback-role* answer, which is wrong for a - header with an identity of its own: it would report itself as - ``IPv6 Extension Header`` / ``IPv6-Ext``, read its length off a data - model that is not its own, and :meth:`__index__` would raise rather - than return its IANA number. The language cannot enforce the override, - so ``tests/protocols/internet/test_ipv6_ext_unit.py`` does, over every - subclass discovered at runtime. + :attr:`protocol`, :attr:`length`, :meth:`__index__` and :meth:`id` + itself. Each carries this class's *fallback-role* answer, which is + wrong for a header with an identity of its own: it would report itself + as ``IPv6 Extension Header`` / ``IPv6-Ext``, read its length off a data + model that is not its own, :meth:`__index__` would raise rather than + return its IANA number, and :meth:`id` would match every member's name + rather than narrow to its own class name. The language cannot enforce + the override, so ``tests/protocols/internet/test_ipv6_ext_unit.py`` + does, over every subclass discovered at runtime. .. versionadded:: 1.5.0 @@ -379,6 +380,22 @@ def protochain(self) -> 'ProtoChain | NoReturn': # Methods. ########################################################################## + @classmethod + def id(cls) -> 'tuple[Literal["HOPOPT"], Literal["IPv6_Route"], Literal["IPv6_Frag"], Literal["ESP"], Literal["AH"], Literal["IPv6_Opts"], Literal["MH"], Literal["HIP"]]': # pylint: disable=line-too-long + """Index ID of the protocol. + + Returns: + Index ID of every direct concrete subclass, each of which narrows + its own :meth:`id` to itself. A header parsed by this class as the + generic fallback therefore matches every one of these names. + + Note: + ``== 'IPv6_Ext'`` is :data:`False`, as ``== 'IP'`` is on + :class:`~pcapkit.protocols.internet.ipv4.IPv4`; ``== IPv6_Ext`` is :data:`True`. + + """ + return ('HOPOPT', 'IPv6_Route', 'IPv6_Frag', 'ESP', 'AH', 'IPv6_Opts', 'MH', 'HIP') + def read(self, length: 'Optional[int]' = None, *, # pylint: disable=arguments-differ extension: 'bool' = False, **kwargs: 'Any') -> '_PT': """Read a generically-parsed IPv6 extension header. diff --git a/pcapkit/protocols/internet/ipv6_frag.py b/pcapkit/protocols/internet/ipv6_frag.py index 0f323784fa..ccda4ec1c9 100644 --- a/pcapkit/protocols/internet/ipv6_frag.py +++ b/pcapkit/protocols/internet/ipv6_frag.py @@ -217,6 +217,16 @@ def make(self, payload=payload, ) + @classmethod + def id(cls) -> 'tuple[Literal["IPv6_Frag"]]': # type: ignore[override] + """Index ID of the protocol. + + Returns: + Index ID of the protocol. + + """ + return ('IPv6_Frag',) + ########################################################################## # Data models. ########################################################################## diff --git a/pcapkit/protocols/internet/ipv6_opts.py b/pcapkit/protocols/internet/ipv6_opts.py index dca1af49a1..f4d747d408 100644 --- a/pcapkit/protocols/internet/ipv6_opts.py +++ b/pcapkit/protocols/internet/ipv6_opts.py @@ -409,6 +409,16 @@ def register_option(cls, code: 'Enum_Option', meth: 'str | tuple[OptionParser, O warn(f'option {code} already registered, overwriting', RegistryWarning) cls.__option__[code] = meth + @classmethod + def id(cls) -> 'tuple[Literal["IPv6_Opts"]]': # type: ignore[override] + """Index ID of the protocol. + + Returns: + Index ID of the protocol. + + """ + return ('IPv6_Opts',) + ########################################################################## # Data models. ########################################################################## diff --git a/pcapkit/protocols/internet/ipv6_route.py b/pcapkit/protocols/internet/ipv6_route.py index cf25b985d1..8a41d150fd 100644 --- a/pcapkit/protocols/internet/ipv6_route.py +++ b/pcapkit/protocols/internet/ipv6_route.py @@ -410,6 +410,16 @@ def register_routing(cls, code: 'Enum_Routing', meth: 'str | tuple[TypeParser, T warn(f'routing {code} already registered, overwriting', RegistryWarning) cls.__routing__[code] = meth + @classmethod + def id(cls) -> 'tuple[Literal["IPv6_Route"]]': # type: ignore[override] + """Index ID of the protocol. + + Returns: + Index ID of the protocol. + + """ + return ('IPv6_Route',) + ########################################################################## # Data models. ########################################################################## diff --git a/pcapkit/protocols/internet/mh.py b/pcapkit/protocols/internet/mh.py index f28046f4f4..6417c790e2 100644 --- a/pcapkit/protocols/internet/mh.py +++ b/pcapkit/protocols/internet/mh.py @@ -1584,6 +1584,16 @@ def register_extension(cls, code: 'Enum_CGAExtension', meth: 'str | tuple[Extens warn(f'extension {code} already registered, overwriting', RegistryWarning) cls.__extension__[code] = meth + @classmethod + def id(cls) -> 'tuple[Literal["MH"]]': # type: ignore[override] + """Index ID of the protocol. + + Returns: + Index ID of the protocol. + + """ + return ('MH',) + ########################################################################## # Data models. ########################################################################## diff --git a/tests/protocols/internet/test_ipv6_ext_unit.py b/tests/protocols/internet/test_ipv6_ext_unit.py index dfc46f1ba1..27cdcb63df 100644 --- a/tests/protocols/internet/test_ipv6_ext_unit.py +++ b/tests/protocols/internet/test_ipv6_ext_unit.py @@ -299,6 +299,110 @@ def test_alias_is_hyphenated_like_its_siblings(self) -> None: # so the hyphen matters, not just the dash-free text either side of it. self.assertEqual(inst.alias.lstrip('IPv6-').lower(), 'ext') + # -- id(): the family composes its members, each member narrows ------- + + def test_id_composes_every_direct_concrete_subclass(self) -> None: + """Derived from the live hierarchy, so a new member cannot drift.""" + import inspect + + import pcapkit.protocols # noqa: F401 # pylint: disable=unused-import + from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext + + members = [klass for klass in IPv6_Ext.__subclasses__() if not inspect.isabstract(klass)] + self.assertCountEqual(IPv6_Ext.id(), [klass.__name__ for klass in members]) + for klass in members: + with self.subTest(cls=klass.__qualname__): + self.assertEqual(klass.id(), (klass.__name__,)) + self.assertIn(klass.__name__.upper(), IPv6_Ext.expand_comp('IPv6_Ext')) + + @staticmethod + def _member_packets() -> 'dict[str, bytes]': + """One real IPv6 packet per family member, keyed by its class name.""" + from pcapkit.const.ipv6.extension_header import ExtensionHeader as EH + from pcapkit.const.reg.transtype import TransType + + udp = _udp_bytes(b'ok') + nxt = int(TransType.UDP) + hit = bytes(15) + headers = { + 'HOPOPT': (EH.HOPOPT, bytes([nxt, 0]) + bytes(6) + udp), + 'IPv6_Opts': (EH.IPv6_Opts, bytes([nxt, 0]) + bytes(6) + udp), + 'IPv6_Frag': (EH.IPv6_Frag, bytes([nxt]) + bytes(7) + udp), + 'IPv6_Route': (EH.IPv6_Route, bytes([nxt, 1, 253, 0]) + bytes(12) + udp), + 'MH': (EH.Mobility_Header, bytes([nxt, 0, 0, 0]) + bytes(4) + udp), + 'AH': (EH.AH, bytes([nxt, 2, 0, 0]) + bytes(12) + udp), + 'ESP': (EH.ESP, struct.pack('>II', 0x66, 1) + bytes(16)), + 'HIP': (EH.HIP, bytes([nxt, 4, 0x01, 0x21]) + b'\x12\x34\x00\x00' + + hit + b'\x01' + hit + b'\x02' + udp), + } + return {name: _ipv6_bytes(int(code), body) for name, (code, body) in headers.items()} + + def test_selector_matches_by_member_and_by_family(self) -> None: + """A member name matches only that member; the family name matches all.""" + from pcapkit.protocols.internet.ipv6 import IPv6 + + packets = self._member_packets() + for name, raw in packets.items(): + ipv6 = IPv6(io.BytesIO(raw), len(raw)) + with self.subTest(header=name): + (_, ext), = ipv6.extension_headers.items(multi=True) + self.assertEqual(type(ext).__name__, name) + self.assertIn(name, ipv6.protochain) + self.assertIn('IPv6_Ext', ipv6.protochain) + for other in packets: + if other != name: + self.assertNotIn(other, ipv6.protochain) + if name in ('AH', 'ESP'): + self.assertIn('IPsec', ipv6.protochain) + else: + self.assertNotIn('IPsec', ipv6.protochain) + + def test_eq_on_the_family_name_is_false_like_ip_and_ipsec(self) -> None: + """``==`` tests a layer's own :meth:`id`, which omits a family name.""" + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext + + ah = bytes([int(TransType.UDP), 2, 0, 0]) + bytes(12) + _udp_bytes(b'ok') + raw = _ipv4_bytes(int(TransType.AH), ah) + ipv4 = IPv4(io.BytesIO(raw), len(raw)) + self.assertTrue(ipv4 == 'IPv4') + self.assertFalse(ipv4 == 'IP') + self.assertTrue(ipv4.payload == 'AH') + self.assertFalse(ipv4.payload == 'IPsec') + + for name, raw in self._member_packets().items(): + (_, ext), = IPv6(io.BytesIO(raw), len(raw)).extension_headers.items(multi=True) + with self.subTest(header=name): + self.assertTrue(ext == name) + self.assertFalse(ext == 'IPv6_Ext') + + generic = bytes([int(TransType.UDP), 0]) + bytes(6) + fallback = IPv6_Ext(io.BytesIO(generic), len(generic), extension=True, alias=0) + self.assertFalse(fallback == 'IPv6_Ext') + self.assertTrue(fallback == IPv6_Ext) + + def test_fallback_for_a_malformed_member_matches_every_member_name(self) -> None: + """A HOPOPT that its own parser rejects is parsed by ``IPv6_Ext``.""" + from pcapkit.const.ipv6.extension_header import ExtensionHeader + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext + + # Router Alert declaring 10 octets of data in a 6-octet option area + hopopt = bytes([int(TransType.UDP), 0, 0x05, 10]) + bytes(4) + raw = _ipv6_bytes(int(ExtensionHeader.HOPOPT), hopopt + _udp_bytes(b'ok')) + ipv6 = IPv6(io.BytesIO(raw), len(raw)) + + (_, ext), = ipv6.extension_headers.items(multi=True) + self.assertIs(type(ext), IPv6_Ext) + self.assertIsInstance(ext.info.error, Exception) + self.assertEqual(str(ipv6.protochain), 'IPv6:IPv6-Ext:UDP:Raw') + for name in (*IPv6_Ext.id(), 'IPv6_Ext', 'IPsec'): + with self.subTest(selector=name): + self.assertIn(name, ipv6.protochain) + # -- the version gate: this class only ever activates for IPv6 ---------- def test_version_gate_rejects_non_ipv6_construction(self) -> None: @@ -438,8 +542,9 @@ class IPv6ExtSharedBaseContractTests(unittest.TestCase): #: header in every :class:`~pcapkit.corekit.protochain.ProtoChain` string #: and in the packet dict key, ``protocol`` reads a field the subclass's #: data model does not have, ``length`` likewise (``IPv6_Frag``'s data model - #: has no ``length`` field at all), and ``__index__`` raises. - REQUIRED_OVERRIDES = ('name', 'alias', 'protocol', 'length', '__index__') + #: has no ``length`` field at all), ``__index__`` raises, and ``id`` matches + #: every member's name. + REQUIRED_OVERRIDES = ('name', 'alias', 'protocol', 'length', '__index__', 'id') #: The expected family, so a member silently *leaving* the base is #: caught as well as one joining without the overrides. Named rather than