From de7177fdec2c2192a190c617a62671d07e30fc96 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Mon, 28 Sep 2026 23:09:42 -0400 Subject: [PATCH] fix(ipv6): parse unrecognised extension headers generically (#891) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Added IPv6_GenericExt, a generic RFC 6564 §4 parser for IPv6 extension headers: it reads the guaranteed next-header/Hdr-Ext-Len octets and derives next/length/protocol from the actual parse, applying AH's and IPv6-Frag's own length rules where they diverge. Gated to version==6, since it is registered into the Internet-wide dispatch table for Shim6 and would otherwise also activate for an IPv4 payload carrying protocol byte 140. - IPv6._decode_next_layer now stops its walk structurally -- on any layer whose parsed info carries no `next` field -- instead of reading `info.next` unconditionally and crashing the whole packet when it hits ESP, BIT-EMU, 253, 254, or any other IANA code with no dedicated parser (the actual #891 defect). IPv6._import_next_layer substitutes IPv6_GenericExt for a recognised header whose own parser raises. - IPv6_GenericExt.alias is hyphenated ('IPv6-GenericExt'), matching its siblings, so IPv6's `alias.lstrip('IPv6-')` keys the packet dict as 'genericext' rather than '_genericext'. - An overrun warns in the house SchemaWarning convention, then stops the walk rather than clipping: the length also decides where the next header starts, so a clipped skip distance would fabricate a layer. - Exported IPv6_GenericExt from the pcapkit, pcapkit.protocols, pcapkit.protocols.data and pcapkit.protocols.schema aggregators. - Re-pointed the #889 test pinning the pre-fix whole-packet collapse, and added coverage for both dispatch paths, the AH/Frag rules, the overrun stop, the version gate, and an unimplemented terminal code (BIT-EMU). Build/test: tests/protocols/internet/ (260 passed, 868 subtests) and tests/project/{test_public_api,test_isort_clean,test_module_isolation}.py via the worktree's own pcapkit; coverage on the touched files rose from 81% to 94% with the new tests added. `make isort` clean; the two remaining mypy findings and the pylint findings pre-exist on main. --- .../pcapkit/protocols/internet/index.rst | 1 + .../protocols/internet/ipv6_generic_ext.rst | 72 +++ examples/generators/dispatch.py | 13 + pcapkit/__init__.py | 2 +- pcapkit/protocols/__init__.py | 2 +- pcapkit/protocols/data/__init__.py | 3 + pcapkit/protocols/data/internet/__init__.py | 6 + .../data/internet/ipv6_generic_ext.py | 45 ++ pcapkit/protocols/internet/__init__.py | 3 +- pcapkit/protocols/internet/ipv6.py | 148 ++++- .../protocols/internet/ipv6_generic_ext.py | 509 ++++++++++++++++++ pcapkit/protocols/schema/__init__.py | 1 + pcapkit/protocols/schema/internet/__init__.py | 6 + .../schema/internet/ipv6_generic_ext.py | 42 ++ .../internet/test_ipv6_generic_ext_unit.py | 368 +++++++++++++ tests/protocols/internet/test_mh_unit.py | 56 +- .../protocols/test_dispatch_registry_unit.py | 15 +- 17 files changed, 1258 insertions(+), 34 deletions(-) create mode 100644 docs/source/pcapkit/protocols/internet/ipv6_generic_ext.rst create mode 100644 pcapkit/protocols/data/internet/ipv6_generic_ext.py create mode 100644 pcapkit/protocols/internet/ipv6_generic_ext.py create mode 100644 pcapkit/protocols/schema/internet/ipv6_generic_ext.py create mode 100644 tests/protocols/internet/test_ipv6_generic_ext_unit.py diff --git a/docs/source/pcapkit/protocols/internet/index.rst b/docs/source/pcapkit/protocols/internet/index.rst index 3314c15674..2bd20e2805 100644 --- a/docs/source/pcapkit/protocols/internet/index.rst +++ b/docs/source/pcapkit/protocols/internet/index.rst @@ -16,6 +16,7 @@ internet layer, with detailed implementation and methods. ipv4 ipv6 ipv6_frag + ipv6_generic_ext ipv6_opts ipv6_route hopopt diff --git a/docs/source/pcapkit/protocols/internet/ipv6_generic_ext.rst b/docs/source/pcapkit/protocols/internet/ipv6_generic_ext.rst new file mode 100644 index 0000000000..508a38e5a8 --- /dev/null +++ b/docs/source/pcapkit/protocols/internet/ipv6_generic_ext.rst @@ -0,0 +1,72 @@ +IPv6_GenericExt - Generic IPv6 Extension Header +================================================ + +.. module:: pcapkit.protocols.internet.ipv6_generic_ext + +:mod:`pcapkit.protocols.internet.ipv6_generic_ext` contains +:class:`~pcapkit.protocols.internet.ipv6_generic_ext.IPv6_GenericExt` +only, which implements a **generic** extractor for IPv6 extension +headers [*]_, standing in for one whenever the header's own dedicated +parser is unavailable or has failed. :rfc:`6564#section-4` guarantees, +with an RFC 2119 **MUST**, that any IPv6 extension header defined +since April 2012 carries the same first two octets: + +======= ========= ===================== ===================================== +Octets Bits Name Description +======= ========= ===================== ===================================== + 0 0 ``next`` Next Header + 1 8 ``len`` Hdr Ext Len (8-octet units, + excluding the first 8 octets) + 2 16 ``payload`` Header-specific content +======= ========= ===================== ===================================== + +so those two octets are parseable without knowing anything else about +the header. See the module docstring below for the closed exception +table (``IPv6-Frag`` and ``AH`` each use their own length rule; ``ESP`` +has a dedicated parser whose own info reports no next header rather than +lacking one, and ``BIT-EMU``, ``253`` and ``254`` have no dedicated +parser at all -- none of the four ever reaches this class), the two ways +this class is dispatched to, and why an overrun stops the walk instead of +clipping it. + +.. autoclass:: pcapkit.protocols.internet.ipv6_generic_ext.IPv6_GenericExt + :no-members: + :show-inheritance: + + .. autoproperty:: name + .. autoproperty:: alias + .. autoproperty:: length + .. autoproperty:: protocol + .. autoproperty:: next + .. autoproperty:: payload + .. autoproperty:: protochain + + .. automethod:: read + .. automethod:: make + + .. automethod:: _make_data + + .. automethod:: __post_init__ + .. automethod:: __index__ + +Header Schemas +-------------- + +.. module:: pcapkit.protocols.schema.internet.ipv6_generic_ext + +.. autoclass:: pcapkit.protocols.schema.internet.ipv6_generic_ext.IPv6_GenericExt + :members: + :show-inheritance: + +Data Models +----------- + +.. module:: pcapkit.protocols.data.internet.ipv6_generic_ext + +.. autoclass:: pcapkit.protocols.data.internet.ipv6_generic_ext.IPv6_GenericExt + :members: + :show-inheritance: + +.. rubric:: Footnotes + +.. [*] :rfc:`6564` diff --git a/examples/generators/dispatch.py b/examples/generators/dispatch.py index 9044e22be3..1049873e8e 100644 --- a/examples/generators/dispatch.py +++ b/examples/generators/dispatch.py @@ -392,6 +392,13 @@ def _internet_payload(code: 'int') -> 'bytes': if code == TransType.OSPFIGP: from pcapkit.protocols.link.ospf import OSPF return bytes(OSPF()) + if code == TransType.Shim6: + # #904: no dedicated dissector exists for Shim6 -- IPv6_GenericExt + # parses only the two octets RFC 6564 §4 guarantees (next header, + # Hdr Ext Len), so this is hand-built rather than constructed + # through a class: next=TCP(6), Hdr Ext Len=0 -> an 8-octet header, + # six of them padding. + return bytes([int(TransType.TCP), 0]) + b'\x00' * 6 + _tcp(9999) raise LookupError(f'internet: no payload builder for {code!r}') # pragma: no cover @@ -621,6 +628,12 @@ def _internet_enum() -> 'Any': 'internet/HIP': ('pcapkit.protocols.internet.hip', 'HIP'), 'internet/SCTP': ('pcapkit.protocols.transport.sctp', 'SCTP'), 'internet/OSPFIGP': ('pcapkit.protocols.link.ospf', 'OSPF'), + # #904: Shim6 (140) previously had no entry at all, and the default + # factory made it resolve to Raw. It is now registered directly at + # IPv6_GenericExt, which parses the RFC 6564 §4 generic layout it has + # never had a dedicated dissector for -- a deliberate addition, not a + # regression. + 'internet/Shim6': ('pcapkit.protocols.internet.ipv6_generic_ext', 'IPv6_GenericExt'), # -- TCP.__proto__ (port) -------------------------------------------------- 'tcp/20': ('pcapkit.protocols.application.ftp', 'FTP_DATA'), diff --git a/pcapkit/__init__.py b/pcapkit/__init__.py index 54ed700ad0..a635ad8e60 100644 --- a/pcapkit/__init__.py +++ b/pcapkit/__init__.py @@ -116,7 +116,7 @@ 'L2TPv2', 'OSPF', 'RARP', 'S_Tag', 'VLAN', 'AH', 'ESP', 'IP', 'IPsec', 'IPv4', 'IPv6', 'IPX', # Internet Layer - 'HIP', 'HOPOPT', 'IPv6_Frag', 'IPv6_Opts', 'IPv6_Route', 'MH', + 'HIP', 'HOPOPT', 'IPv6_Frag', 'IPv6_GenericExt', 'IPv6_Opts', 'IPv6_Route', 'MH', # IPv6 Extension Header 'TCP', 'UDP', 'SCTP', # Transport Layer diff --git a/pcapkit/protocols/__init__.py b/pcapkit/protocols/__init__.py index c2a0f815e0..1f35252a95 100644 --- a/pcapkit/protocols/__init__.py +++ b/pcapkit/protocols/__init__.py @@ -57,7 +57,7 @@ 'AH', 'ESP', 'IP', 'IPsec', 'IPv4', 'IPv6', 'IPX', # IPv6 Extension Header - 'HIP', 'HOPOPT', 'IPv6_Frag', 'IPv6_Opts', + 'HIP', 'HOPOPT', 'IPv6_Frag', 'IPv6_GenericExt', 'IPv6_Opts', 'IPv6_Route', 'MH', # Transport Layer diff --git a/pcapkit/protocols/data/__init__.py b/pcapkit/protocols/data/__init__.py index 30a4540f75..7f4a488943 100644 --- a/pcapkit/protocols/data/__init__.py +++ b/pcapkit/protocols/data/__init__.py @@ -117,6 +117,9 @@ # IPv6 Fragment Header 'IPv6_Frag', + # Generic IPv6 Extension Header + 'IPv6_GenericExt', + # IPv6 Destination Options Header 'IPv6_Opts', 'IPv6_Opts_RPLFlags', 'IPv6_Opts_MPLFlags', 'IPv6_Opts_DFFFlags', diff --git a/pcapkit/protocols/data/internet/__init__.py b/pcapkit/protocols/data/internet/__init__.py index 44866423a0..266e1f4108 100644 --- a/pcapkit/protocols/data/internet/__init__.py +++ b/pcapkit/protocols/data/internet/__init__.py @@ -131,6 +131,9 @@ # IPv6 Fragment Header from pcapkit.protocols.data.internet.ipv6_frag import IPv6_Frag +# Generic IPv6 Extension Header +from pcapkit.protocols.data.internet.ipv6_generic_ext import IPv6_GenericExt + # IPv6 Destination Options from pcapkit.protocols.data.internet.ipv6_opts import CALIPSOOption as IPv6_Opts_CALIPSOOption from pcapkit.protocols.data.internet.ipv6_opts import DFFFlags as IPv6_Opts_DFFFlags @@ -266,6 +269,9 @@ # IPv6 Fragment Header 'IPv6_Frag', + # Generic IPv6 Extension Header + 'IPv6_GenericExt', + # IPv6 Destination Options Header 'IPv6_Opts', 'IPv6_Opts_RPLFlags', 'IPv6_Opts_MPLFlags', 'IPv6_Opts_DFFFlags', diff --git a/pcapkit/protocols/data/internet/ipv6_generic_ext.py b/pcapkit/protocols/data/internet/ipv6_generic_ext.py new file mode 100644 index 0000000000..d031c85289 --- /dev/null +++ b/pcapkit/protocols/data/internet/ipv6_generic_ext.py @@ -0,0 +1,45 @@ +# -*- coding: utf-8 -*- +"""data model for generically-parsed IPv6 extension headers""" + +from typing import TYPE_CHECKING + +from pcapkit.corekit.infoclass import info_final +from pcapkit.protocols.data.protocol import Protocol + +if TYPE_CHECKING: + from typing import Optional + + from pcapkit.const.ipv6.extension_header import ExtensionHeader + from pcapkit.const.reg.transtype import TransType + +__all__ = ['IPv6_GenericExt'] + + +@info_final +class IPv6_GenericExt(Protocol): + """Data model for a generically-parsed IPv6 extension header. + + See :class:`pcapkit.protocols.internet.ipv6_generic_ext.IPv6_GenericExt` + for how each field below is derived. + + """ + + #: The extension header this instance stands in for -- the numeric code + #: the caller dispatched on, resolved to its + #: :class:`~pcapkit.const.ipv6.extension_header.ExtensionHeader` member. + #: :data:`None` when ``alias`` named no such member (:meth:`read + #: ` + #: sets it so on a lookup miss, and the class property at + #: :attr:`~pcapkit.protocols.internet.ipv6_generic_ext.IPv6_GenericExt.protocol` + #: is typed to match). + protocol: 'Optional[ExtensionHeader]' + #: Next header, parsed off the wire. :data:`None` when the declared + #: length would have overrun what remained and the walk stopped instead + #: of trusting it. + next: 'Optional[TransType]' + #: Length of this extension header, in octets, actually consumed. + length: 'int' + #: Original parsing error, if this instance was reached as a + #: :func:`~pcapkit.utilities.decorators.beholder` fallback rather than by + #: direct dispatch. + error: 'Optional[Exception]' diff --git a/pcapkit/protocols/internet/__init__.py b/pcapkit/protocols/internet/__init__.py index e00b2deb73..57a38c0cee 100644 --- a/pcapkit/protocols/internet/__init__.py +++ b/pcapkit/protocols/internet/__init__.py @@ -25,6 +25,7 @@ from pcapkit.protocols.internet.hip import HIP from pcapkit.protocols.internet.hopopt import HOPOPT from pcapkit.protocols.internet.ipv6_frag import IPv6_Frag +from pcapkit.protocols.internet.ipv6_generic_ext import IPv6_GenericExt from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts from pcapkit.protocols.internet.ipv6_route import IPv6_Route from pcapkit.protocols.internet.mh import MH @@ -39,6 +40,6 @@ __all__ = [ 'ETHERTYPE', # Protocol Numbers 'AH', 'ESP', 'IP', 'IPsec', 'IPv4', 'IPv6', 'IPX', # Internet Layer - 'HIP', 'HOPOPT', 'IPv6_Frag', + 'HIP', 'HOPOPT', 'IPv6_Frag', 'IPv6_GenericExt', 'IPv6_Opts', 'IPv6_Route', 'MH', # IPv6 Extension Header ] diff --git a/pcapkit/protocols/internet/ipv6.py b/pcapkit/protocols/internet/ipv6.py index 100d90992b..74cd582625 100644 --- a/pcapkit/protocols/internet/ipv6.py +++ b/pcapkit/protocols/internet/ipv6.py @@ -56,6 +56,67 @@ class IPv6(IP[Data_IPv6, Schema_IPv6], schema=Schema_IPv6, data=Data_IPv6): """This class implements Internet Protocol version 6.""" + ########################################################################## + # Defaults. + ########################################################################## + + #: Extension header codes that have a *dedicated* parser class in this + #: package whose own layout follows :rfc:`6564#section-4`'s generic + #: ``next`` + ``Hdr Ext Len`` format (see the module docstring of + #: :mod:`pcapkit.protocols.internet.ipv6_generic_ext` for the exception + #: table in full). When that dedicated parser raises, + #: :meth:`_import_next_layer` substitutes + #: :class:`~pcapkit.protocols.internet.ipv6_generic_ext.IPv6_GenericExt` + #: instead of letting the generic + #: :func:`~pcapkit.utilities.decorators.beholder` fall back to plain + #: :class:`~pcapkit.protocols.misc.raw.Raw`, which has no ``next`` field + #: and used to crash the whole packet at :meth:`_decode_next_layer`'s + #: ``proto = info.next`` (GitHub issue #891). + #: + #: :attr:`~pcapkit.const.ipv6.extension_header.ExtensionHeader.Shim6` + #: is deliberately absent, even though its wire format also conforms: + #: this package has never had a dedicated parser for it to begin with + #: (``pcapkit/protocols/internet/NotImplemented/shim6.py`` is a 0-byte + #: placeholder, excluded from the wheel by ``MANIFEST.in``), so there is + #: no "own parser" here for it to raise from -- ``Shim6`` reaches + #: :class:`IPv6_GenericExt` by *direct* registration instead (see the + #: bottom of :mod:`pcapkit.protocols.internet.ipv6_generic_ext`), which + #: already produces exactly this class without needing this set to name + #: it. ``ESP``, ``BIT-EMU``, ``253`` and ``254`` are absent, but not for + #: the same reason as each other, and not because a generic fallback + #: would help them: + #: + #: * ``ESP`` *does* have a dedicated, registered parser + #: (:class:`~pcapkit.protocols.internet.esp.ESP`) -- it is excluded + #: because :rfc:`4303` places the real Next Header byte inside the + #: encrypted trailer, so its own info always *carries* a ``next`` + #: attribute (unlike ``BIT-EMU``/``253``/``254`` below), just one that is + #: :data:`None` whenever the payload could not be decrypted -- which, + #: with no key material available to a generic parse, is always. The + #: :meth:`_decode_next_layer` walk below still ends there, one iteration + #: later, because :data:`None` fails + #: :class:`~pcapkit.const.ipv6.extension_header.ExtensionHeader`'s + #: constructor at the top of the loop -- the *existing* end-of-chain + #: path, unrelated to the structural check this set exists for. + #: * ``BIT-EMU``, ``253`` and ``254`` have no dedicated parser at all, so + #: they resolve to plain :class:`~pcapkit.protocols.misc.raw.Raw`, whose + #: info has no ``next`` *attribute* -- this is what the structural check + #: catches. + #: + #: :meth:`_decode_next_layer`'s walk stops cleanly at whichever of these + #: (or any other IANA code this package has not implemented) it meets, + #: and keeps this layer's own header intact instead of losing the whole + #: packet as it used to. + __generic_ext_codes__ = frozenset({ + Enum_ExtensionHeader.HOPOPT, + Enum_ExtensionHeader.IPv6_Route, + Enum_ExtensionHeader.IPv6_Opts, + Enum_ExtensionHeader.Mobility_Header, + Enum_ExtensionHeader.HIP, + Enum_ExtensionHeader.IPv6_Frag, + Enum_ExtensionHeader.AH, + }) + ########################################################################## # Properties. ########################################################################## @@ -335,7 +396,6 @@ def _decode_next_layer(self, ipv6: 'Data_IPv6', proto: 'Optional[int]' = None, # record protocol name # self._protos = ProtoChain(name, chain, alias) _protos.append(next_) - proto = info.next # update header & payload length hdr_len += next_.length # type: ignore[assignment] @@ -350,6 +410,45 @@ def _decode_next_layer(self, ipv6: 'Data_IPv6', proto: 'Optional[int]' = None, # is what gets handed to ``super()._decode_next_layer`` below payload = payload[next_.length:] + # GitHub issue #891: a layer with no ``next`` field cannot + # safely continue the walk. This is a *structural* check -- + # does the parsed info even carry a ``next`` attribute? -- not + # a fixed set of codes, and deliberately so: HOPOPT, IPv6-Route, + # IPv6-Opts, MH, HIP, IPv6-Frag and AH all have dedicated + # parsers whose data carries ``next``, and Shim6 and any + # recognised header whose own parser raised are both handled by + # :class:`~pcapkit.protocols.internet.ipv6_generic_ext.IPv6_GenericExt`, + # which also carries ``next`` (possibly :data:`None`, on an + # overrun -- see its module docstring). Every IANA extension + # header code this package has not implemented a dedicated + # parser for -- today that is ``BIT-EMU``, ``253`` and ``254``, + # and tomorrow it is whatever IANA assigns next -- has no + # generic fallback either (see :attr:`__generic_ext_codes__`'s + # docstring for why), so :meth:`_import_next_layer` returns a + # plain :class:`~pcapkit.protocols.misc.raw.Raw`, whose info + # carries no ``next`` at all. Reading ``info.next`` on that + # unconditionally is what used to raise ``AttributeError`` + # here and let a further-out :func:`~pcapkit.utilities.decorators.beholder` + # catch it and degrade the *whole* packet -- the actual #891 + # defect, for every code nobody has implemented. Stopping here + # instead keeps this layer's own fields (still recorded above, + # in ``self._exthdr`` and in the packet dict) and reports no + # further next header, exactly like the overrun case. + # + # This has to run -- and, on a hit, has to set ``proto`` -- + # *before* the fragment-header special case below: IPv6-Frag + # always carries a real ``next`` (the ``hasattr`` check above + # never actually fires for it), and that ``next`` is the real + # transport layer's code, which the fragment branch's own + # ``break`` must leave in ``proto`` for the final + # ``super()._decode_next_layer`` call below the loop to dispatch + # to correctly. + if not hasattr(info, 'next'): + proto = None + break + + proto = info.next + # keep original data after fragment header if ex_proto == Enum_ExtensionHeader.IPv6_Frag: ipv6.__update__({ @@ -387,6 +486,35 @@ def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *, # Returns: Instance of next layer. + Notes: + If the dedicated parser for a code in :attr:`__generic_ext_codes__` + raises, this substitutes + :class:`~pcapkit.protocols.internet.ipv6_generic_ext.IPv6_GenericExt` + for it rather than letting the exception reach the + :func:`~pcapkit.utilities.decorators.beholder` decorating this + method, which would otherwise substitute plain + :class:`~pcapkit.protocols.misc.raw.Raw` -- and ``Raw`` has no + ``next`` field, which is what used to crash the whole packet at + :meth:`_decode_next_layer`'s ``proto = info.next`` (GitHub issue + #891). Every other exception -- including one raised by + ``IPv6_GenericExt`` itself, or by ``ESP``'s own dedicated + parser -- still reaches ``beholder`` unchanged, so *this + method's own* behaviour for anything outside that closed set is + exactly what it was before this method learned the + substitution: a plain ``Raw`` for that one layer. What changed + for ``BIT-EMU``, ``253`` and ``254`` -- which have no dedicated + parser at all, so they were *already* reaching plain ``Raw`` + with no exception involved -- is one level up: + :meth:`_decode_next_layer` now stops its walk structurally on + any layer whose info carries no ``next`` attribute, ``Raw`` + included, instead of reading ``info.next`` unconditionally and + crashing the whole packet. ``ESP`` is unaffected either way: its + info always carries a ``next`` (:data:`None`, since :rfc:`4303` + encrypts the real value), so neither this substitution nor that + structural check ever engages for it, and the walk ends after it + exactly as it always has -- see :attr:`__generic_ext_codes__`'s + docstring for the full distinction. + """ if TYPE_CHECKING: protocol: 'Type[ProtocolBase]' @@ -407,7 +535,19 @@ def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *, # else: protocol = self._lookup_next_layer(self.__proto__, proto) - next_ = protocol(file_, length, version=version, extension=extension, # type: ignore[abstract] - alias=proto, packet=packet, layer=self._exlayer, protocol=self._exproto, - __context__=self._exctx) + try: + next_ = protocol(file_, length, version=version, extension=extension, # type: ignore[abstract] + alias=proto, packet=packet, layer=self._exlayer, protocol=self._exproto, + __context__=self._exctx) + except Exception as exc: + from pcapkit.protocols.internet.ipv6_generic_ext import \ + IPv6_GenericExt # isort: skip # pylint: disable=import-outside-toplevel + + if not (extension and protocol is not IPv6_GenericExt + and proto in self.__generic_ext_codes__): + raise + + next_ = IPv6_GenericExt(file_, length, version=version, extension=extension, + alias=proto, error=exc, packet=packet, layer=self._exlayer, + protocol=self._exproto, __context__=self._exctx) return next_ diff --git a/pcapkit/protocols/internet/ipv6_generic_ext.py b/pcapkit/protocols/internet/ipv6_generic_ext.py new file mode 100644 index 0000000000..3c8650d29d --- /dev/null +++ b/pcapkit/protocols/internet/ipv6_generic_ext.py @@ -0,0 +1,509 @@ +# -*- coding: utf-8 -*- +"""IPv6_GenericExt - Generic IPv6 Extension Header +====================================================== + +.. module:: pcapkit.protocols.internet.ipv6_generic_ext + +:mod:`pcapkit.protocols.internet.ipv6_generic_ext` contains +:class:`~pcapkit.protocols.internet.ipv6_generic_ext.IPv6_GenericExt` +only, which implements a **generic** extractor for IPv6 extension +headers, standing in for one whenever the header's own dedicated +parser is unavailable or has failed. + +Why this is safe in general +---------------------------- + +:rfc:`6564#section-4` is Standards Track and says, with an RFC 2119 +**MUST**, that any IPv6 extension header defined from April 2012 +onward carries the same first two octets: + +======= ========= ===================== ===================================== +Octets Bits Name Description +======= ========= ===================== ===================================== + 0 0 ``next`` Next Header + 1 8 ``len`` Hdr Ext Len (8-octet units, + excluding the first 8 octets) + 2 16 ``payload`` Header-specific content +======= ========= ===================== ===================================== + +so the first two octets of a *conforming* header are parseable without +knowing anything else about it. :rfc:`6564#section-5` is explicit that +this is **not retroactive** -- *"[i]t applies only to newly defined +extension headers"* -- which is why the pre-existing headers need the +closed exception table below rather than being assumed to conform. +(:rfc:`8200#section-4.8` restates the same layout, but without a MUST, +and mislabels the generic length field as *"Length of the Destination +Options header"*; cite :rfc:`6564#section-4`, not that section.) + +The exception table, verified against IANA's ``protocol-numbers-1.csv`` +*IPv6 Extension Header* column and each cited RFC: + +======================================================= =========================================== +Header(s) Length rule +======================================================= =========================================== +``HOPOPT``, ``IPv6-Route``, ``IPv6-Opts``, ``MH``, ``(octet[1] + 1) * 8`` -- the :rfc:`6564` +``HIP``, ``Shim6`` generic rule +``IPv6-Frag`` constant ``8``; octet[1] is Reserved, not a + length (:rfc:`8200#section-4.5`) +``AH`` ``(octet[1] + 2) * 4`` -- :rfc:`4302#section-2.2` + counts in 4-octet units with a bias of 2, + not :rfc:`6564`'s 8-octet units and bias of 1 +``ESP`` terminal -- has a dedicated, registered parser + (:class:`~pcapkit.protocols.internet.esp.ESP`), + but its Next Header byte is inside the + encrypted trailer (:rfc:`4303`), so its own + info's ``next`` is :data:`None` rather than a + value to continue on +``BIT-EMU``, ``253``, ``254`` terminal -- no dedicated parser exists, so no + next header field is ever read at all, either +======================================================= =========================================== + +This table classifies all twelve IANA-registered codes by *wire format* +alone. ``Shim6`` conforms to it (:rfc:`5533`), but this package has never +had a dedicated parser class for it to begin with -- see "Two entry paths" +below for how it reaches this class regardless, by direct dispatch rather +than by a parser of its own failing. + +:rfc:`8200#section-4.5` says outright that Encapsulating Security +Payload *"is not considered an extension header"*; :rfc:`4303` puts its +Next Header inside the encrypted trailer, with no length field anywhere +in the cleartext part; and 253/254 are reserved for private +experimentation (:rfc:`3692`) with no wire format at all. None of the +four is reachable through this class, by construction -- see +:meth:`pcapkit.protocols.internet.ipv6.IPv6._import_next_layer`. Each of +them still enters :meth:`IPv6._decode_next_layer +`'s walk (it is a +real :class:`~pcapkit.const.ipv6.extension_header.ExtensionHeader` member), +but the four part ways there: ``ESP`` resolves to its own dedicated parser, +whose info carries a ``next`` that is simply :data:`None` -- so the walk +ends the ordinary way, ``ExtensionHeader(None)`` failing at the top of the +next iteration, exactly as it did before this class existed. ``BIT-EMU``, +``253`` and ``254`` have no dedicated parser and resolve to plain +:class:`~pcapkit.protocols.misc.raw.Raw`, whose info has no ``next`` +*attribute* at all; for these three (and any future IANA code nobody has +implemented yet) the walk stops on a *structural* check -- does the parsed +layer carry a ``next`` at all? -- rather than on a list of codes. + +Two entry paths +---------------- + +This class is reached two different ways: + +1. **An unrecognised protocol number.** :class:`~pcapkit.const.ipv6.extension_header.ExtensionHeader.Shim6` + is a real IANA extension header (:rfc:`5533`) that this package has never + had a dedicated parser for, so + :meth:`pcapkit.protocols.internet.internet.Internet._lookup_next_layer` + used to default it to plain :class:`~pcapkit.protocols.misc.raw.Raw` -- + which has no ``next`` field, so the walk in + :meth:`IPv6._decode_next_layer ` + crashed on it (GitHub issue #891). This module registers itself for that + code instead (see the bottom of the module), so dispatch reaches a + working generic parser directly. That registration is global -- shared + by every :class:`~pcapkit.protocols.internet.internet.Internet` + subclass -- so :meth:`__post_init__` gates on ``version == 6`` to keep + it from also activating for an IPv4 payload that happens to carry + protocol number 140; see its docstring for what that would otherwise do. +2. **A recognised header whose own parser raises.** This is the actual #891 + defect: ``HOPOPT``, ``IPv6-Route``, ``IPv6-Opts``, ``MH``, ``HIP``, + ``IPv6-Frag`` and ``AH`` all have dedicated classes, and when one of + *those* raises, :func:`~pcapkit.utilities.decorators.beholder` + (:meth:`Protocol._import_next_layer `) + would ordinarily catch it and substitute plain + :class:`~pcapkit.protocols.misc.raw.Raw` -- which loses the ``next`` + field the same way, and crashes the walk exactly as Shim6 did. + :meth:`IPv6._import_next_layer ` + catches that failure itself, one layer in from ``beholder``, and + substitutes this class instead: the bad header costs only itself, not + the rest of the chain. + +The overrun guard +------------------- + +The house convention for a declared length that does not fit what remains +is warn-and-clip: emit a :class:`~pcapkit.utilities.warnings.SchemaWarning` +naming what was declared against what is left, then read only what is +left -- + +* :func:`pcapkit.protocols.schema.misc.pcapng.bounded_option` + (``pcapkit/protocols/schema/misc/pcapng.py:373``) +* :func:`pcapkit.protocols.schema.misc.pcapng.bounded_area` + (``pcapkit/protocols/schema/misc/pcapng.py:443``) +* :meth:`pcapkit.corekit.fields.field.FieldBase.pack ` + (``pcapkit/corekit/fields/field.py:507``) +* :meth:`pcapkit.protocols.misc.pcapng.PCAPNG.read_frame` + (``pcapkit/protocols/misc/pcapng.py:1119``) + +This class follows that convention's *warning* and deliberately diverges on +the *action*. Clipping is right when the declared length only governs how +much of the current object to read -- there is always a well-defined "what +is left" to fall back to. Here, the declared length also decides *where the +next header starts*; a clipped skip distance points at whatever bytes +happen to be at the end of the buffer, which are not a header. Continuing +the walk from there would fabricate a layer and record a false entry in +:class:`~pcapkit.corekit.protochain.ProtoChain`, which reads as a parsed +fact rather than as the guess it would be. So :meth:`read` below warns in +the same wording as the sites above, then **stops the walk**: this instance +absorbs every remaining octet and reports :data:`None` for ``next``, which +is what the caller's loop already reads as "no more extension headers" and +ends on honestly, at the bad header, instead of inventing what follows it. + +""" +from typing import TYPE_CHECKING, overload + +from pcapkit.const.ipv6.extension_header import ExtensionHeader as Enum_ExtensionHeader +from pcapkit.const.reg.transtype import TransType as Enum_TransType +from pcapkit.protocols.data.internet.ipv6_generic_ext import IPv6_GenericExt as Data_IPv6_GenericExt +from pcapkit.protocols.internet.internet import Internet +from pcapkit.protocols.schema.internet.ipv6_generic_ext import \ + IPv6_GenericExt as Schema_IPv6_GenericExt +from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall, stacklevel +from pcapkit.utilities.warnings import SchemaWarning, warn + +if TYPE_CHECKING: + from typing import IO, Any, NoReturn, Optional + + from typing_extensions import Literal + + from pcapkit.corekit.protochain import ProtoChain + from pcapkit.protocols.protocol import ProtocolBase + +__all__ = ['IPv6_GenericExt'] + + +class IPv6_GenericExt(Internet[Data_IPv6_GenericExt, Schema_IPv6_GenericExt], + schema=Schema_IPv6_GenericExt, data=Data_IPv6_GenericExt): + """This class implements a generic IPv6 extension header parser. + + See the module docstring for the RFC citations backing the length + rules below, the two ways this class gets dispatched to, and the + reasoning for stopping rather than clipping on an overrun. + + """ + + ########################################################################## + # Properties. + ########################################################################## + + @property + def name(self) -> 'Literal["Generic IPv6 Extension Header"]': + """Name of current protocol.""" + return 'Generic IPv6 Extension Header' + + @property + def alias(self) -> 'Literal["IPv6-GenericExt"]': + """Acronym of corresponding protocol. + + Hyphenated, like :attr:`IPv6_Frag.alias + ` and + :attr:`IPv6_Opts.alias `, + and for the same reason: :meth:`IPv6._decode_next_layer + ` builds the + packet-dict key by ``self.alias.lstrip('IPv6-').lower()`` -- + :meth:`str.lstrip` strips a character *set*, not a prefix, so the + default (class-name) alias ``'IPv6_GenericExt'`` would strip to + ``'_GenericExt'`` and key the dict as ``_genericext``. The hyphen + makes every leading character (``I``, ``P``, ``v``, ``6``, ``-``) a + member of the set being stripped, same as the siblings, giving + ``genericext``. + + """ + return 'IPv6-GenericExt' + + @property + def length(self) -> 'int': + """Header length of current protocol.""" + return self._info.length + + @property + def protocol(self) -> 'Optional[Enum_ExtensionHeader]': + """The extension header this instance stands in for. + + This is **not** the base :attr:`Protocol.protocol + ` meaning ("name + of next layer protocol"); it is deliberately repointed, on the + owner's ruling for GitHub issue #891, at this instance's *own* + identity -- which header's format it parsed, e.g. + :attr:`~pcapkit.const.ipv6.extension_header.ExtensionHeader.HOPOPT` + or :attr:`~pcapkit.const.ipv6.extension_header.ExtensionHeader.Shim6`. + That identity is resolved per instance from the numeric code handed + to the constructor (``alias``), which + :meth:`pcapkit.protocols.internet.ipv6.IPv6._decode_next_layer` + already holds before it dispatches (``ipv6.py:327``). See + :meth:`__index__` for why the *class-level* identity cannot be + made to work the same way. + + """ + return self._info.protocol + + @property + def next(self) -> 'Optional[Enum_TransType]': + """Next header, as parsed off the wire. + + :data:`None` when the declared length would have overrun what + remained of the chain and the walk stopped instead of trusting it + -- see the module docstring's "overrun guard" section. + + """ + return self._info.next + + @property + def payload(self) -> 'ProtocolBase | NoReturn': + """Payload of current instance. + + Raises: + UnsupportedCall: if the protocol is used as an IPv6 extension header + + """ + if self._extf: + raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute 'payload'") + return self._next + + @property + def protochain(self) -> 'ProtoChain | NoReturn': + """Protocol chain of current instance. + + Raises: + UnsupportedCall: if the protocol is used as an IPv6 extension header + + """ + if self._extf: + raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute 'protochain'") + return super().protochain + + ########################################################################## + # Methods. + ########################################################################## + + def read(self, length: 'Optional[int]' = None, *, error: 'Optional[Exception]' = None, + alias: 'Optional[int]' = None, version: 'Literal[4, 6]' = 6, # pylint: disable=arguments-differ,unused-argument + extension: 'bool' = False, **kwargs: 'Any') -> 'Data_IPv6_GenericExt': # pylint: disable=unused-argument + """Read a generically-parsed IPv6 extension header. + + Args: + length: Length of packet data. + error: Parsing error, if reached as a + :func:`~pcapkit.utilities.decorators.beholder` fallback. + alias: Numeric extension header code this instance stands in + for, e.g. ``0`` for ``HOPOPT`` or ``140`` for ``Shim6``. + version: IP protocol version. + extension: If the protocol is used as an IPv6 extension header. + **kwargs: Arbitrary keyword arguments. + + Returns: + Parsed packet data. + + """ + if length is None: + length = len(self) + schema = self.__header__ + + ext_code = None # type: Optional[Enum_ExtensionHeader] + if alias is not None: + try: + ext_code = Enum_ExtensionHeader(alias) + except ValueError: + ext_code = None + + if ext_code == Enum_ExtensionHeader.IPv6_Frag: + # RFC 8200 §4.5: octet 1 is Reserved, not a length -- the + # fragment header is always exactly 8 octets, and RFC 6564 §5 + # says explicitly that it predates and does not follow the + # generic format. + nominal = 8 + elif ext_code == Enum_ExtensionHeader.AH: + # RFC 4302 §2.2: Payload Len is in 4-octet units, excluding the + # first 8 octets -- a different unit and a different bias from + # RFC 6564's Hdr Ext Len. + nominal = (schema.len + 2) * 4 + else: + # RFC 6564 §4 (MUST): Hdr Ext Len is in 8-octet units, + # excluding the first 8 octets. Also the fallback when ``alias`` + # named no known extension header at all, which is the best + # generic guess available. + nominal = (schema.len + 1) * 8 + + if nominal > length: + # See the module docstring's "overrun guard" section for why + # this warns like the house convention but stops rather than + # clips. + warn(f'IPv6: extension header declares a length of {nominal} octet(s) with ' + f'{length} octet(s) left in the chain; stopping the walk instead of ' + f'skipping past it', SchemaWarning, stacklevel=stacklevel()) + ext_len = length + next_header = None # type: Optional[Enum_TransType] + else: + ext_len = nominal + next_header = schema.next + + generic_ext = Data_IPv6_GenericExt( + protocol=ext_code, + next=next_header, + length=ext_len, + error=error, + ) + + if extension: + return generic_ext + return self._decode_next_layer(generic_ext, next_header, length - ext_len) + + def make(self, + next: 'Enum_TransType | int' = Enum_TransType.UDP, # pylint: disable=redefined-builtin + len: 'int' = 0, # pylint: disable=redefined-builtin + payload: 'bytes | ProtocolBase | Any' = b'', + **kwargs: 'Any') -> 'Schema_IPv6_GenericExt': + """Make (construct) packet data. + + Args: + next: Next header type. + len: Raw ``Hdr Ext Len`` octet to emit -- the caller's + responsibility to size correctly, since this class does not + know, at construction time, which per-protocol rule (see + the module docstring) the octet is meant to satisfy. + payload: Payload of current instance. + **kwargs: Arbitrary keyword arguments. + + Returns: + Constructed packet data. + + """ + return Schema_IPv6_GenericExt( + next=next, + len=len, + payload=payload, + ) + + ########################################################################## + # Data models. + ########################################################################## + + @overload + def __post_init__(self, file: 'IO[bytes] | bytes', length: 'Optional[int]' = ..., *, # pylint: disable=arguments-differ + version: 'Literal[4, 6]' = ..., extension: 'bool' = ..., + **kwargs: 'Any') -> 'None': ... + @overload + def __post_init__(self, **kwargs: 'Any') -> 'None': ... # pylint: disable=arguments-differ + + def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None, length: 'Optional[int]' = None, *, # pylint: disable=arguments-differ + version: 'Literal[4, 6]' = 6, extension: 'bool' = False, + **kwargs: 'Any') -> 'None': + """Post initialisation hook. + + Args: + file: Source packet stream. + length: Length of packet data. + version: IP protocol version. + extension: If the protocol is used as an IPv6 extension header. + **kwargs: Arbitrary keyword arguments. + + Raises: + ProtocolError: If ``version`` is not ``6``. + + See Also: + For construction argument, please refer to :meth:`make`. + + Note: + This class is registered into :attr:`Internet.__proto__ + ` -- + shared by *every* :class:`~pcapkit.protocols.internet.internet.Internet` + subclass, :class:`~pcapkit.protocols.internet.ipv4.IPv4` included -- + so that :attr:`~pcapkit.const.ipv6.extension_header.ExtensionHeader.Shim6` + resolves to a working parser instead of defaulting to + :class:`~pcapkit.protocols.misc.raw.Raw`. Without this guard, an + IPv4 packet whose protocol byte happens to be 140 would reach + this class too and walk an IPv6-style extension-header chain out + of an IPv4 payload -- verified: it would read + ``IPv4:IPv6-GenericExt:...`` instead of the ``IPv4:Shim6`` a + plain, non-continuing ``Raw`` gives today. Rejecting here sends + construction back through :func:`~pcapkit.utilities.decorators.beholder` + at the *caller's* layer, which substitutes that same ``Raw`` -- + i.e. this restores exactly the pre-existing, version-agnostic + behaviour rather than inventing a new one, and needs no + IPv4-specific code of its own. + + """ + if version != 6: + raise ProtocolError( + f'{self.__class__.__name__}: only valid for IPv6, got version={version}') + + #: bool: If the protocol is used as an IPv6 extension header. + self._extf = extension + + # call super __post_init__ + super().__post_init__(file, length, version=version, extension=extension, **kwargs) # type: ignore[arg-type] + + def __length_hint__(self) -> 'Literal[2]': + """Return an estimated length for the object.""" + return 2 + + @classmethod + def __index__(cls) -> 'NoReturn': + """Numeral registry index of the protocol. + + Raises: + UnsupportedCall: This protocol has no *class-level* registry + entry. Unlike :meth:`Raw.__index__ + `, which raises + because :class:`~pcapkit.protocols.misc.raw.Raw` has no + identity to report at all, this class *does* have one -- + see :attr:`protocol` -- but it is resolved per instance, + not per class: one :class:`IPv6_GenericExt` stands in for + :attr:`~pcapkit.const.ipv6.extension_header.ExtensionHeader.HOPOPT`, + :attr:`~pcapkit.const.ipv6.extension_header.ExtensionHeader.Shim6` + and any other RFC 6564-conforming code alike, while + :meth:`__index__` is a ``@classmethod`` with nowhere to put + a value that differs per instance. + + """ + raise UnsupportedCall(f'{cls.__name__!r} object cannot be interpreted as an integer') + + ########################################################################## + # Utilities. + ########################################################################## + + @classmethod + def _make_data(cls, data: 'Data_IPv6_GenericExt') -> 'dict[str, Any]': # type: ignore[override] + """Create key-value pairs from ``data`` for protocol construction. + + Inverts whichever per-protocol length rule :meth:`read` applied, + using ``data.protocol`` -- the extension header this instance stood + in for -- to pick the same rule back. Round-tripping a ``data.next`` + of :data:`None` (the overrun case) is not supported: there is no + octet value that both satisfies the rule and still fits, which is + exactly why :meth:`read` stopped rather than clipped. + + Args: + data: protocol data + + Returns: + Key-value pairs for protocol construction. + + """ + if data.protocol == Enum_ExtensionHeader.IPv6_Frag: + len_octet = 0 # constant length; the octet itself is Reserved + elif data.protocol == Enum_ExtensionHeader.AH: + len_octet = data.length // 4 - 2 + else: + len_octet = data.length // 8 - 1 + + return { + 'next': data.next, + 'len': len_octet, + 'payload': cls._make_payload(data), + } + + +# NOTE: Registered by direct assignment into ``Internet.__proto__`` -- the +# same mechanism ``pcapkit.protocols.internet.internet`` uses to pre-populate +# every other entry -- rather than via the ``code=`` keyword of +# ``ProtocolBase.__init_subclass__``. The two are equivalent in effect, but +# ``code=`` resolves through ``register_protocol_code``, which imports +# ``pcapkit.foundation.registry.protocols`` *at class-definition time*, i.e. +# while ``pcapkit.protocols.internet`` (which imports this module) is still +# being built. Nothing else in this package's ``code=`` usage does that from +# inside ``pcapkit.protocols.internet`` itself, and doing so here completed a +# cycle back into a not-yet-finished ``pcapkit.protocols.internet`` through +# ``pcapkit.foundation.extraction`` -- measured as ``ImportError: cannot +# import name 'Extractor' from partially initialized module +# 'pcapkit.foundation.extraction'`` on a bare ``import pcapkit``. Plain +# dict assignment carries no import of its own, so it cannot re-trigger that. +Internet.__proto__[Enum_TransType.Shim6] = IPv6_GenericExt diff --git a/pcapkit/protocols/schema/__init__.py b/pcapkit/protocols/schema/__init__.py index 0cfe662cf7..c35e1d4ecd 100644 --- a/pcapkit/protocols/schema/__init__.py +++ b/pcapkit/protocols/schema/__init__.py @@ -68,6 +68,7 @@ 'IPv4_TROption', 'IPv4_RTRALTOption', 'IPv4_QSOption', 'IPv4_QuickStartRequestOption', 'IPv4_QuickStartReportOption', 'IPv6_Frag', + 'IPv6_GenericExt', 'IPv6_Opts', 'IPv6_Opts_UnassignedOption', 'IPv6_Opts_PadOption', 'IPv6_Opts_TunnelEncapsulationLimitOption', 'IPv6_Opts_RouterAlertOption', 'IPv6_Opts_CALIPSOOption', 'IPv6_Opts_SMFIdentificationBasedDPDOption', diff --git a/pcapkit/protocols/schema/internet/__init__.py b/pcapkit/protocols/schema/internet/__init__.py index 61e7aa14b7..7d6096c240 100644 --- a/pcapkit/protocols/schema/internet/__init__.py +++ b/pcapkit/protocols/schema/internet/__init__.py @@ -131,6 +131,9 @@ # IPv6 Fragment Header from pcapkit.protocols.schema.internet.ipv6_frag import IPv6_Frag +# Generic IPv6 Extension Header +from pcapkit.protocols.schema.internet.ipv6_generic_ext import IPv6_GenericExt + # IPv6 Destination Options from pcapkit.protocols.schema.internet.ipv6_opts import CALIPSOOption as IPv6_Opts_CALIPSOOption from pcapkit.protocols.schema.internet.ipv6_opts import \ @@ -263,6 +266,9 @@ # IPv6 Fragment Header 'IPv6_Frag', + # Generic IPv6 Extension Header + 'IPv6_GenericExt', + # IPv6 Destination Options 'IPv6_Opts', 'IPv6_Opts_UnassignedOption', 'IPv6_Opts_PadOption', 'IPv6_Opts_TunnelEncapsulationLimitOption', diff --git a/pcapkit/protocols/schema/internet/ipv6_generic_ext.py b/pcapkit/protocols/schema/internet/ipv6_generic_ext.py new file mode 100644 index 0000000000..5f5f6fb126 --- /dev/null +++ b/pcapkit/protocols/schema/internet/ipv6_generic_ext.py @@ -0,0 +1,42 @@ +# -*- coding: utf-8 -*- +# mypy: disable-error-code=assignment +"""header schema for generically-parsed IPv6 extension headers""" + +from typing import TYPE_CHECKING + +from pcapkit.const.reg.transtype import TransType as Enum_TransType +from pcapkit.corekit.fields.misc import PayloadField +from pcapkit.corekit.fields.numbers import EnumField, UInt8Field +from pcapkit.protocols.schema.schema import Schema, schema_final + +__all__ = ['IPv6_GenericExt'] + +if TYPE_CHECKING: + from pcapkit.protocols.protocol import ProtocolBase + + +@schema_final +class IPv6_GenericExt(Schema): + """Header schema for a generically-parsed IPv6 extension header. + + Only the two octets :rfc:`6564#section-4` guarantees are parsed at this + layer -- ``next`` and the raw ``Hdr Ext Len`` octet. Combining them into + an actual skip distance is per protocol (a constant for ``IPv6-Frag``, + 4-octet units for ``AH``, 8-octet units for the rest), so that part is + done in :meth:`pcapkit.protocols.internet.ipv6_generic_ext.IPv6_GenericExt.read`, + which knows which protocol this instance stands in for; this schema does + not. + + """ + + #: Next header. + next: 'Enum_TransType' = EnumField(length=1, namespace=Enum_TransType) + #: Raw ``Hdr Ext Len`` octet -- see the class docstring for why its + #: interpretation is not fixed here. + len: 'int' = UInt8Field() + #: Everything after the two fixed octets; opaque at this layer. + payload: 'bytes' = PayloadField() + + if TYPE_CHECKING: + def __init__(self, next: 'Enum_TransType | int', len: 'int', + payload: 'bytes | ProtocolBase | Schema') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements diff --git a/tests/protocols/internet/test_ipv6_generic_ext_unit.py b/tests/protocols/internet/test_ipv6_generic_ext_unit.py new file mode 100644 index 0000000000..8cf7685a9c --- /dev/null +++ b/tests/protocols/internet/test_ipv6_generic_ext_unit.py @@ -0,0 +1,368 @@ +from __future__ import annotations + +import importlib.util +import io +import struct +import unittest +import warnings + +from tests._support import purge_modules + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + + +def _udp_bytes(payload: bytes = b'') -> bytes: + """A minimal, structurally valid UDP datagram (checksum not enforced on read).""" + return struct.pack('>HHHH', 12345, 53, 8 + len(payload), 0) + payload + + +def _ipv6_bytes(next_code: int, ext_and_payload: bytes) -> bytes: + """A minimal IPv6 header (version 6, ``::1`` -> ``::1``) wrapping ``ext_and_payload``.""" + header = struct.pack('>IHBB', 6 << 28, len(ext_and_payload), next_code, 64) + header += (b'\x00' * 15 + b'\x01') * 2 # src = dst = ::1 + return header + ext_and_payload + + +def _ipv4_bytes(proto_byte: int, payload: bytes = b'') -> bytes: + """A minimal, valid IPv4 header (``127.0.0.1`` -> ``127.0.0.1``, no options).""" + header = struct.pack('>BBHHHBBH4s4s', 0x45, 0, 20 + len(payload), 0, 0, 64, proto_byte, 0, + b'\x7f\x00\x00\x01', b'\x7f\x00\x00\x01') + return header + payload + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class IPv6GenericExtUnitTests(unittest.TestCase): + """Tests for GitHub issue #891: a generic RFC 6564 parser for IPv6 + extension headers, so a failed or unimplemented one costs only itself + rather than the whole packet. + """ + + def setUp(self) -> None: + purge_modules(['pcapkit']) + + # -- direct construction: the per-protocol length rules ----------------- + + def test_generic_rule_applies_to_hopopt_route_opts_mh_hip(self) -> None: + """RFC 6564 §4: ``(octet[1] + 1) * 8``, for every conformer except + ``IPv6-Frag`` and ``AH``, which have their own rule (tested below). + """ + from pcapkit.const.ipv6.extension_header import ExtensionHeader + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.ipv6_generic_ext import IPv6_GenericExt + + for code in (ExtensionHeader.HOPOPT, ExtensionHeader.IPv6_Route, + ExtensionHeader.IPv6_Opts, ExtensionHeader.Mobility_Header, + ExtensionHeader.HIP): + with self.subTest(code=code): + raw = bytes([int(TransType.UDP), 1]) + b'\x00' * 14 # (1+1)*8 == 16 + inst = IPv6_GenericExt(io.BytesIO(raw), len(raw), extension=True, + alias=int(code)) + self.assertEqual(inst.next, TransType.UDP) + self.assertEqual(inst.length, 16) + self.assertEqual(inst.protocol, code) + + def test_ah_rule_is_four_octet_units_with_bias_two(self) -> None: + """RFC 4302 §2.2: ``(octet[1] + 2) * 4``, not RFC 6564's 8-octet units.""" + from pcapkit.const.ipv6.extension_header import ExtensionHeader + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.ipv6_generic_ext import IPv6_GenericExt + + raw = bytes([int(TransType.TCP), 2]) + b'\x00' * 14 # (2+2)*4 == 16 + inst = IPv6_GenericExt(io.BytesIO(raw), len(raw), extension=True, + alias=int(ExtensionHeader.AH)) + self.assertEqual(inst.next, TransType.TCP) + self.assertEqual(inst.length, 16) + self.assertEqual(inst.protocol, ExtensionHeader.AH) + + def test_frag_rule_is_a_constant_eight_octets(self) -> None: + """RFC 8200 §4.5: octet[1] is Reserved for IPv6-Frag, not a length -- + the header is always exactly 8 octets regardless of its value. + """ + from pcapkit.const.ipv6.extension_header import ExtensionHeader + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.ipv6_generic_ext import IPv6_GenericExt + + raw = bytes([int(TransType.UDP), 200]) + b'\x00' * 14 # 200 must be ignored + inst = IPv6_GenericExt(io.BytesIO(raw), len(raw), extension=True, + alias=int(ExtensionHeader.IPv6_Frag)) + self.assertEqual(inst.next, TransType.UDP) + self.assertEqual(inst.length, 8) + self.assertEqual(inst.protocol, ExtensionHeader.IPv6_Frag) + + def test_overrun_warns_and_stops_instead_of_clipping(self) -> None: + """The owner's ruling: warn in the house convention's wording, then + stop the walk (absorb what remains, report no next header) rather + than clip-and-continue -- a clipped skip distance would point at + trailing buffer bytes, not at a header. + """ + from pcapkit.const.ipv6.extension_header import ExtensionHeader + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.ipv6_generic_ext import IPv6_GenericExt + from pcapkit.utilities.warnings import SchemaWarning + + # (250 + 1) * 8 == 2008, but only 8 octets are actually available. + raw = bytes([int(TransType.UDP), 250]) + b'\x00' * 6 + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + inst = IPv6_GenericExt(io.BytesIO(raw), len(raw), extension=True, + alias=int(ExtensionHeader.HOPOPT)) + + self.assertIsNone(inst.next) + self.assertEqual(inst.length, len(raw)) # absorbed everything, nothing skipped past + schema_warnings = [w for w in caught if issubclass(w.category, SchemaWarning)] + self.assertEqual(len(schema_warnings), 1) + message = str(schema_warnings[0].message) + self.assertIn('declares a length of 2008 octet(s)', message) + self.assertIn('8 octet(s) left in the chain', message) + self.assertIn('stopping the walk', message) + + # -- __index__ and the guarded extension-mode accessors ------------------ + + def test_index_raises_because_no_class_level_identity_exists(self) -> None: + from pcapkit.protocols.internet.ipv6_generic_ext import IPv6_GenericExt + from pcapkit.utilities.exceptions import UnsupportedCall + + with self.assertRaises(UnsupportedCall): + IPv6_GenericExt.__index__() + + def test_extension_mode_blocks_payload_and_protochain_but_not_protocol(self) -> None: + from pcapkit.const.ipv6.extension_header import ExtensionHeader + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.ipv6_generic_ext import IPv6_GenericExt + from pcapkit.utilities.exceptions import UnsupportedCall + + raw = bytes([int(TransType.UDP), 1]) + b'\x00' * 14 + inst = IPv6_GenericExt(io.BytesIO(raw), len(raw), extension=True, + alias=int(ExtensionHeader.HOPOPT)) + + with self.assertRaises(UnsupportedCall): + _ = inst.payload + with self.assertRaises(UnsupportedCall): + _ = inst.protochain + + # unlike the base ``Protocol.protocol`` meaning, this is the + # instance's own identity and stays readable regardless of ``_extf``. + self.assertEqual(inst.protocol, ExtensionHeader.HOPOPT) + self.assertEqual(inst.next, TransType.UDP) + self.assertEqual(inst.length, 16) + + # -- round trip: make()/_make_data() invert the same rule ---------------- + + def test_make_and_make_data_round_trip_the_generic_and_ah_rules(self) -> None: + from pcapkit.const.ipv6.extension_header import ExtensionHeader + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.ipv6_generic_ext import IPv6_GenericExt + + for code, len_octet, total in ( + (ExtensionHeader.HOPOPT, 1, 16), + (ExtensionHeader.AH, 2, 16), + (ExtensionHeader.IPv6_Frag, 0, 8), + ): + with self.subTest(code=code): + raw = bytes([int(TransType.UDP), len_octet]) + b'\x00' * (total - 2) + inst = IPv6_GenericExt(io.BytesIO(raw), len(raw), extension=True, + alias=int(code)) + values = IPv6_GenericExt._make_data(inst.info) + self.assertEqual(values['next'], TransType.UDP) + self.assertEqual(values['len'], len_octet) + + # -- entry path 2: a recognised header's own parser raises --------------- + + def test_mh_own_parser_failure_falls_back_and_chain_reaches_real_udp(self) -> None: + """The actual #891 defect, with real bytes after the bad header this + time: proves the walk does not merely avoid crashing but genuinely + resumes at the real next layer. + """ + from pcapkit.const.mh.packet import Packet + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.internet.ipv6_generic_ext import IPv6_GenericExt + from pcapkit.protocols.internet.mh import MH, FastBindingAcknowledgmentStatus + from pcapkit.protocols.transport.udp import UDP + + mh_raw = bytearray(bytes(MH( + next=TransType.UDP, chksum=b'\x12\x34', type=Packet.Fast_Binding_Acknowledgment, + data={'status': FastBindingAcknowledgmentStatus.Insufficient_resources, + 'key_mngt': True, 'seq': 0x1234, 'lifetime': 40, 'options': []}))) + mh_raw[6] = 50 # unassigned FastBindingAcknowledgmentStatus byte -- MH.read raises + + udp_payload = _udp_bytes(b'hello') + raw = _ipv6_bytes(int(TransType.Mobility_Header), bytes(mh_raw) + udp_payload) + ipv6 = IPv6(io.BytesIO(raw), len(raw)) + + exthdrs = list(ipv6.extension_headers.items(multi=True)) + self.assertEqual(len(exthdrs), 1) + genext = exthdrs[0][1] + self.assertIsInstance(genext, IPv6_GenericExt) + self.assertEqual(genext.next, TransType.UDP) + self.assertEqual(genext.length, len(mh_raw)) + self.assertIsInstance(genext.info.error, Exception) + + self.assertIsInstance(ipv6.payload, UDP) + self.assertEqual(bytes(ipv6.payload.payload), b'hello') + self.assertEqual(str(ipv6.protochain), 'IPv6:IPv6-GenericExt:UDP:Raw') + + def test_overrun_inside_a_real_chain_stops_the_walk_honestly(self) -> None: + """Same failing MH message, but its own ``Header Len`` octet is also + corrupted to a value the generic fallback cannot trust either -- + the walk must warn and stop, not invent a layer from trailing bytes. + """ + from pcapkit.const.mh.packet import Packet + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.internet.ipv6_generic_ext import IPv6_GenericExt + from pcapkit.protocols.internet.mh import MH, FastBindingAcknowledgmentStatus + from pcapkit.utilities.warnings import SchemaWarning + + mh_raw = bytearray(bytes(MH( + next=TransType.UDP, chksum=b'\x12\x34', type=Packet.Fast_Binding_Acknowledgment, + data={'status': FastBindingAcknowledgmentStatus.Insufficient_resources, + 'key_mngt': True, 'seq': 0x1234, 'lifetime': 40, 'options': []}))) + mh_raw[6] = 50 # MH.read raises + mh_raw[1] = 250 # and the generic fallback's own length would overrun + + trailing = _udp_bytes(b'should-not-be-reached') + raw = _ipv6_bytes(int(TransType.Mobility_Header), bytes(mh_raw) + trailing) + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + ipv6 = IPv6(io.BytesIO(raw), len(raw)) + + self.assertTrue(any(issubclass(w.category, SchemaWarning) for w in caught)) + + exthdrs = list(ipv6.extension_headers.items(multi=True)) + self.assertEqual(len(exthdrs), 1) + genext = exthdrs[0][1] + self.assertIsInstance(genext, IPv6_GenericExt) + self.assertIsNone(genext.next) + # absorbed everything, including the bytes that would have been a + # perfectly good UDP datagram -- the point is that it is not trusted + self.assertEqual(genext.length, len(mh_raw) + len(trailing)) + self.assertEqual(str(ipv6.protochain), 'IPv6:IPv6-GenericExt') + + # -- entry path 1: an unrecognised protocol number (Shim6) --------------- + + def test_shim6_has_no_dedicated_parser_and_dispatches_directly(self) -> None: + """``Shim6`` (140) is a real IANA extension header this package has + never implemented, so it used to default to plain ``Raw`` via + ``Internet.__proto__``'s defaultdict -- registered here directly + instead, so no exception is even involved. + """ + from pcapkit.const.ipv6.extension_header import ExtensionHeader + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.internet import Internet + from pcapkit.protocols.internet.ipv6_generic_ext import IPv6_GenericExt + + self.assertIs(Internet.__proto__[TransType.Shim6], IPv6_GenericExt) + + def test_shim6_packet_parses_generically_and_chain_reaches_real_udp(self) -> None: + 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_generic_ext import IPv6_GenericExt + from pcapkit.protocols.transport.udp import UDP + + shim6_ext = bytes([int(TransType.UDP), 1]) + b'\x00' * 14 # 16 octets, generic rule + udp_payload = _udp_bytes(b'shim6-ok') + raw = _ipv6_bytes(int(ExtensionHeader.Shim6), shim6_ext + udp_payload) + ipv6 = IPv6(io.BytesIO(raw), len(raw)) + + exthdrs = list(ipv6.extension_headers.items(multi=True)) + self.assertEqual(len(exthdrs), 1) + code, genext = exthdrs[0] + self.assertEqual(code, ExtensionHeader.Shim6) + self.assertIsInstance(genext, IPv6_GenericExt) + self.assertIsNone(genext.info.error) # direct dispatch, not a beholder fallback + self.assertIsInstance(ipv6.payload, UDP) + self.assertEqual(bytes(ipv6.payload.payload), b'shim6-ok') + self.assertEqual(str(ipv6.protochain), 'IPv6:IPv6-GenericExt:UDP:Raw') + + # -- alias, so the packet dict and the chain segment are not '_genericext' -- + + def test_alias_is_hyphenated_like_its_siblings(self) -> None: + from pcapkit.const.ipv6.extension_header import ExtensionHeader + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.ipv6_generic_ext import IPv6_GenericExt + + raw = bytes([int(TransType.UDP), 1]) + b'\x00' * 14 + inst = IPv6_GenericExt(io.BytesIO(raw), len(raw), extension=True, + alias=int(ExtensionHeader.HOPOPT)) + self.assertEqual(inst.alias, 'IPv6-GenericExt') + # this is exactly the computation ``IPv6._decode_next_layer`` performs + # to key its packet dict -- ``str.lstrip`` strips a character *set*, + # so the hyphen matters, not just the dash-free text either side of it. + self.assertEqual(inst.alias.lstrip('IPv6-').lower(), 'genericext') + + # -- the version gate: this class only ever activates for IPv6 ---------- + + def test_version_gate_rejects_non_ipv6_construction(self) -> None: + from pcapkit.const.ipv6.extension_header import ExtensionHeader + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.ipv6_generic_ext import IPv6_GenericExt + from pcapkit.utilities.exceptions import ProtocolError + + raw = bytes([int(TransType.UDP), 1]) + b'\x00' * 14 + with self.assertRaises(ProtocolError): + IPv6_GenericExt(io.BytesIO(raw), len(raw), version=4, extension=True, + alias=int(ExtensionHeader.HOPOPT)) + + def test_shim6_registration_does_not_leak_into_ipv4_parsing(self) -> None: + """GitHub issue #891 cross-review: registering ``IPv6_GenericExt`` + into the shared ``Internet.__proto__`` for Shim6 must not make an + IPv4 packet whose protocol byte happens to be 140 walk an IPv6-style + extension-header chain. The version gate on ``__post_init__`` sends + construction back through the caller's own ``@beholder``, which + restores exactly the pre-existing ``Raw`` fallback -- verified + against a real :class:`~pcapkit.protocols.internet.ipv4.IPv4` packet, + not just the class in isolation. + """ + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.misc.raw import Raw + + raw = _ipv4_bytes(int(TransType.Shim6), b'\x11\x01' + b'\x00' * 14) + ipv4 = IPv4(io.BytesIO(raw), len(raw)) + + self.assertIsInstance(ipv4.payload, Raw) + # the enum member's own name, not a chain walked out of an IPv4 + # payload -- this is the exact pre-#891-registration rendering. + self.assertEqual(str(ipv4.protochain), 'IPv4:Shim6') + + # -- item 1 (cross-review): an unimplemented terminal code must not ------ + # -- collapse the whole packet ------------------------------------------- + + def test_unimplemented_terminal_code_stops_the_walk_not_the_packet(self) -> None: + """``BIT-EMU`` (147) has no dedicated parser and is not one of the + RFC 6564 conformers, so it resolves to plain :class:`Raw`, whose + info carries no ``next`` -- the exact #891 signature if the walk + read ``info.next`` on it unconditionally. The structural check in + :meth:`IPv6._decode_next_layer + ` must stop + there instead, keeping this packet's own header (source, destination, + hop limit) intact rather than losing it to a further-out + ``@beholder``. ``253``/``254`` are the same code path (also + unregistered, also resolve to ``Raw``); this file covers one to keep + the test proportionate, per the review's own framing. + """ + 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.misc.raw import Raw + + raw = _ipv6_bytes(int(TransType.BIT_EMU), b'\x11\x01' + b'\x00' * 14) + ipv6 = IPv6(io.BytesIO(raw), len(raw)) + + # the header this class exists to protect -- lost entirely under the + # pre-#891 defect, since the whole IPv6 layer became a bare ``Raw``. + self.assertEqual(str(ipv6.src), '::1') + self.assertEqual(str(ipv6.dst), '::1') + + exthdrs = list(ipv6.extension_headers.items(multi=True)) + self.assertEqual(len(exthdrs), 1) + code, terminal = exthdrs[0] + self.assertEqual(code, ExtensionHeader.BIT_EMU) + self.assertIsInstance(terminal, Raw) + + self.assertIsInstance(ipv6.payload, Raw) + self.assertEqual(str(ipv6.protochain), 'IPv6:Raw:Raw') diff --git a/tests/protocols/internet/test_mh_unit.py b/tests/protocols/internet/test_mh_unit.py index 2365e98438..17fb0648d0 100644 --- a/tests/protocols/internet/test_mh_unit.py +++ b/tests/protocols/internet/test_mh_unit.py @@ -1698,31 +1698,35 @@ def wrap_in_ipv4(mh_payload: bytes) -> IPv4: self.assertIsInstance(ip.payload, Raw) self.assertEqual(bytes(ip.payload), bytes(lra_raw)) - with self.subTest('over IPv6, the same raise costs the whole packet (pre-existing defect)'): + with self.subTest('over IPv6, the raise now costs only the MH slot (GitHub issue #891)'): # Mobility Header is *also* one of IPv6's own chained extension # headers (``Enum_ExtensionHeader.Mobility_Header``, value 135 -- # ``pcapkit/const/ipv6/extension_header.py:42``), so IPv6 never # reaches the fallback through the base ``_import_next_layer`` at - # all. It walks its own ``@beholder``-decorated override in - # ``_decode_next_layer``, gets a ``Raw`` fallback back for the MH - # slot exactly as IPv4 does, and then does - # ``proto = info.next`` (``pcapkit/protocols/internet/ipv6.py:338``) - # on it regardless. ``Raw``'s info carries no ``.next``, so *that* - # line raises ``AttributeError`` inside ``IPv6.read`` -- and it is - # this second, unrelated exception that a further-out - # ``@beholder`` (Ethernet's, here) actually catches, degrading - # the whole IPv6 packet -- header fields and the MH message - # both -- to ``Raw``. + # all -- it walks its own ``@beholder``-decorated override + # (``pcapkit/protocols/internet/ipv6.py:_import_next_layer``). # - # This walk defect predates this fix and already fires on a - # truncated or otherwise malformed extension header on - # unmodified ``main``; what this fix adds is a route to it from - # a *well-formed* packet that merely names a byte its RFC leaves - # unassigned, which is a far likelier way to arrive here than a - # malformed header. This subtest pins that as it stands today, - # deliberately, so that whoever fixes ipv6.py:338 sees this - # assertion fail and knows to revisit it -- it is not a claim - # that losing the whole packet is the desired behaviour. + # Before GitHub issue #891, that override let any exception from + # MH's own parser reach ``@beholder`` unchanged, which substituted + # plain ``Raw`` for the MH slot; ``Raw``'s info carries no + # ``.next``, so ``proto = info.next`` + # (``pcapkit/protocols/internet/ipv6.py:_decode_next_layer``) + # raised ``AttributeError`` on it regardless -- and it was *that*, + # unrelated, exception that a further-out ``@beholder`` + # (Ethernet's) actually caught, degrading the whole IPv6 packet, + # header fields and MH message alike, to ``Raw``. + # + # #891 gives ``IPv6._import_next_layer`` its own catch, one layer + # in from ``@beholder``: for an extension header whose wire + # format RFC 6564 guarantees (Mobility Header among them), the MH + # slot becomes + # :class:`~pcapkit.protocols.internet.ipv6_generic_ext.IPv6_GenericExt` + # instead of ``Raw``. It parses the two guaranteed octets + # generically -- next header ``UDP`` and a length that exactly + # matches this ``fback_raw`` payload -- so ``IPv6.read`` returns + # normally: IPv6's own header fields (source, destination, hop + # limit) survive, and the chain now ends honestly at the bad + # header's replacement rather than swallowing the packet. def wrap_in_ethernet_ipv6(mh_payload: bytes) -> Ethernet: ipv6_header = struct.pack( '>IHBB', 6 << 28, len(mh_payload), int(TransType.Mobility_Header), 64, @@ -1731,9 +1735,17 @@ def wrap_in_ethernet_ipv6(mh_payload: bytes) -> Ethernet: raw = eth_header + ipv6_header + mh_payload return Ethernet(io.BytesIO(raw), len(raw)) + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.internet.ipv6_generic_ext import IPv6_GenericExt + eth = wrap_in_ethernet_ipv6(bytes(fback_raw)) - self.assertIsInstance(eth.payload, Raw) - self.assertEqual(str(eth.protochain), 'Ethernet:Internet_Protocol_version_6') + self.assertIsInstance(eth.payload, IPv6) + self.assertEqual(str(eth.protochain), 'Ethernet:IPv6:IPv6-GenericExt') + + genext = next(iter(eth.payload.extension_headers.items(multi=True)))[1] + self.assertIsInstance(genext, IPv6_GenericExt) + self.assertEqual(genext.next, TransType.UDP) + self.assertEqual(genext.length, len(fback_raw)) def test_mh_message_flags_pack_each_bit_independently(self) -> None: """Regression test: a cleared flag must not be emitted as a set bit. diff --git a/tests/protocols/test_dispatch_registry_unit.py b/tests/protocols/test_dispatch_registry_unit.py index b4653d7672..1bb114bbfc 100644 --- a/tests/protocols/test_dispatch_registry_unit.py +++ b/tests/protocols/test_dispatch_registry_unit.py @@ -2,7 +2,8 @@ """Every ``__proto__`` dispatch-registry entry, enumerated rather than hand-picked. GitHub issue #496: :file:`pcapkit/foundation/registry/protocols.py` documents -seven registries -- 38 entries in total -- that decide which +seven registries -- 39 entries in total as of #904, which added +``Internet.__proto__[TransType.Shim6]`` -- that decide which :class:`~pcapkit.protocols.protocol.Protocol` subclass parses the next layer. Nothing walked all of them. :file:`test_registry_runtime.py` checks that a table entry *resolves* to the right class object, and @@ -211,7 +212,7 @@ def test_registry_currently_matches_pinned_target(self) -> None: ) def test_cases_cover_every_table_named_in_the_issue(self) -> None: - """The enumeration finds all 38 entries the issue counted, across all seven tables. + """The enumeration finds all 39 entries the issue counted, across all seven tables. 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 @@ -219,8 +220,12 @@ def test_cases_cover_every_table_named_in_the_issue(self) -> None: """ cases = self.dispatch.cases() - self.assertEqual(len(cases), 38, - 'expected exactly 38 entries across the seven __proto__ ' + # #904: 38 -> 39, and the internet count 16 -> 17 below with it -- one + # new entry, Internet.__proto__[TransType.Shim6], registered at + # IPv6_GenericExt where it previously had none at all (defaulted to + # Raw). Diffed against origin/main: no other table changed shape. + self.assertEqual(len(cases), 39, + 'expected exactly 39 entries across the seven __proto__ ' 'tables named in GitHub issue #496; a different count ' 'means a registry changed shape and this module was not ' 'updated to match') @@ -229,7 +234,7 @@ def test_cases_cover_every_table_named_in_the_issue(self) -> None: for case in cases: by_family[case.family] = by_family.get(case.family, 0) + 1 self.assertEqual(by_family, { - 'link': 7, 'internet': 16, 'tcp': 4, 'udp': 3, 'sctp': 2, + 'link': 7, 'internet': 17, 'tcp': 4, 'udp': 3, 'sctp': 2, 'pcap-frame': 3, 'pcapng-frame': 3, })