diff --git a/docs/source/pcapkit/protocols/internet/index.rst b/docs/source/pcapkit/protocols/internet/index.rst index 2bd20e2805..9c40ef532d 100644 --- a/docs/source/pcapkit/protocols/internet/index.rst +++ b/docs/source/pcapkit/protocols/internet/index.rst @@ -15,8 +15,8 @@ internet layer, with detailed implementation and methods. ip ipv4 ipv6 + ipv6_ext 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_ext.rst similarity index 64% rename from docs/source/pcapkit/protocols/internet/ipv6_generic_ext.rst rename to docs/source/pcapkit/protocols/internet/ipv6_ext.rst index 508a38e5a8..85feba9861 100644 --- a/docs/source/pcapkit/protocols/internet/ipv6_generic_ext.rst +++ b/docs/source/pcapkit/protocols/internet/ipv6_ext.rst @@ -1,11 +1,18 @@ -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 +IPv6_Ext - IPv6 Extension Header +================================ + +.. module:: pcapkit.protocols.internet.ipv6_ext + +:mod:`pcapkit.protocols.internet.ipv6_ext` contains +:class:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext` +only, which serves two roles at once (GitHub issue #917): it is the +shared **base class** of all eight IPv6 extension headers this package +implements -- supplying them the ``extension``-mode contract, i.e. the +guards that make :attr:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext.payload`, +:attr:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext.protocol` and +:attr:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext.protochain` +unavailable on a header parsed as part of an IPv6 chain -- and it +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 @@ -29,7 +36,7 @@ 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 +.. autoclass:: pcapkit.protocols.internet.ipv6_ext.IPv6_Ext :no-members: :show-inheritance: @@ -52,18 +59,18 @@ clipping it. Header Schemas -------------- -.. module:: pcapkit.protocols.schema.internet.ipv6_generic_ext +.. module:: pcapkit.protocols.schema.internet.ipv6_ext -.. autoclass:: pcapkit.protocols.schema.internet.ipv6_generic_ext.IPv6_GenericExt +.. autoclass:: pcapkit.protocols.schema.internet.ipv6_ext.IPv6_Ext :members: :show-inheritance: Data Models ----------- -.. module:: pcapkit.protocols.data.internet.ipv6_generic_ext +.. module:: pcapkit.protocols.data.internet.ipv6_ext -.. autoclass:: pcapkit.protocols.data.internet.ipv6_generic_ext.IPv6_GenericExt +.. autoclass:: pcapkit.protocols.data.internet.ipv6_ext.IPv6_Ext :members: :show-inheritance: diff --git a/examples/generators/dispatch.py b/examples/generators/dispatch.py index 1049873e8e..37d6fc0192 100644 --- a/examples/generators/dispatch.py +++ b/examples/generators/dispatch.py @@ -393,7 +393,7 @@ def _internet_payload(code: 'int') -> 'bytes': from pcapkit.protocols.link.ospf import OSPF return bytes(OSPF()) if code == TransType.Shim6: - # #904: no dedicated dissector exists for Shim6 -- IPv6_GenericExt + # #904: no dedicated dissector exists for Shim6 -- IPv6_Ext # 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, @@ -630,10 +630,10 @@ def _internet_enum() -> 'Any': '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 + # IPv6_Ext, 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'), + 'internet/Shim6': ('pcapkit.protocols.internet.ipv6_ext', 'IPv6_Ext'), # -- TCP.__proto__ (port) -------------------------------------------------- 'tcp/20': ('pcapkit.protocols.application.ftp', 'FTP_DATA'), diff --git a/pcapkit/__init__.py b/pcapkit/__init__.py index a635ad8e60..0a604df051 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_GenericExt', 'IPv6_Opts', 'IPv6_Route', 'MH', + 'HIP', 'HOPOPT', 'IPv6_Frag', 'IPv6_Ext', '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 1f35252a95..0edb972133 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_GenericExt', 'IPv6_Opts', + 'HIP', 'HOPOPT', 'IPv6_Frag', 'IPv6_Ext', 'IPv6_Opts', 'IPv6_Route', 'MH', # Transport Layer diff --git a/pcapkit/protocols/data/__init__.py b/pcapkit/protocols/data/__init__.py index 7f4a488943..0c46fab999 100644 --- a/pcapkit/protocols/data/__init__.py +++ b/pcapkit/protocols/data/__init__.py @@ -117,8 +117,8 @@ # IPv6 Fragment Header 'IPv6_Frag', - # Generic IPv6 Extension Header - 'IPv6_GenericExt', + # IPv6 Extension Header (base + generic fallback) + 'IPv6_Ext', # IPv6 Destination Options Header 'IPv6_Opts', diff --git a/pcapkit/protocols/data/internet/__init__.py b/pcapkit/protocols/data/internet/__init__.py index 266e1f4108..5db9559ff9 100644 --- a/pcapkit/protocols/data/internet/__init__.py +++ b/pcapkit/protocols/data/internet/__init__.py @@ -131,8 +131,8 @@ # 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 Extension Header (base + generic fallback) +from pcapkit.protocols.data.internet.ipv6_ext import IPv6_Ext # IPv6 Destination Options from pcapkit.protocols.data.internet.ipv6_opts import CALIPSOOption as IPv6_Opts_CALIPSOOption @@ -269,8 +269,8 @@ # IPv6 Fragment Header 'IPv6_Frag', - # Generic IPv6 Extension Header - 'IPv6_GenericExt', + # IPv6 Extension Header (base + generic fallback) + 'IPv6_Ext', # IPv6 Destination Options Header 'IPv6_Opts', diff --git a/pcapkit/protocols/data/internet/ipv6_generic_ext.py b/pcapkit/protocols/data/internet/ipv6_ext.py similarity index 82% rename from pcapkit/protocols/data/internet/ipv6_generic_ext.py rename to pcapkit/protocols/data/internet/ipv6_ext.py index d031c85289..07cd2ec341 100644 --- a/pcapkit/protocols/data/internet/ipv6_generic_ext.py +++ b/pcapkit/protocols/data/internet/ipv6_ext.py @@ -12,14 +12,14 @@ from pcapkit.const.ipv6.extension_header import ExtensionHeader from pcapkit.const.reg.transtype import TransType -__all__ = ['IPv6_GenericExt'] +__all__ = ['IPv6_Ext'] @info_final -class IPv6_GenericExt(Protocol): +class IPv6_Ext(Protocol): """Data model for a generically-parsed IPv6 extension header. - See :class:`pcapkit.protocols.internet.ipv6_generic_ext.IPv6_GenericExt` + See :class:`pcapkit.protocols.internet.ipv6_ext.IPv6_Ext` for how each field below is derived. """ @@ -28,9 +28,9 @@ class IPv6_GenericExt(Protocol): #: 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` + #: :attr:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext.protocol` #: is typed to match). protocol: 'Optional[ExtensionHeader]' #: Next header, parsed off the wire. :data:`None` when the declared diff --git a/pcapkit/protocols/internet/__init__.py b/pcapkit/protocols/internet/__init__.py index 57a38c0cee..f7e1ed14f0 100644 --- a/pcapkit/protocols/internet/__init__.py +++ b/pcapkit/protocols/internet/__init__.py @@ -25,7 +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_ext import IPv6_Ext 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 @@ -40,6 +40,6 @@ __all__ = [ 'ETHERTYPE', # Protocol Numbers 'AH', 'ESP', 'IP', 'IPsec', 'IPv4', 'IPv6', 'IPX', # Internet Layer - 'HIP', 'HOPOPT', 'IPv6_Frag', 'IPv6_GenericExt', + 'HIP', 'HOPOPT', 'IPv6_Frag', 'IPv6_Ext', 'IPv6_Opts', 'IPv6_Route', 'MH', # IPv6 Extension Header ] diff --git a/pcapkit/protocols/internet/ah.py b/pcapkit/protocols/internet/ah.py index 8dcad7e9d5..5546c22c5c 100644 --- a/pcapkit/protocols/internet/ah.py +++ b/pcapkit/protocols/internet/ah.py @@ -29,6 +29,7 @@ from pcapkit.const.reg.transtype import TransType as Enum_TransType from pcapkit.protocols.data.internet.ah import AH as Data_AH from pcapkit.protocols.internet.ipsec import IPsec +from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext from pcapkit.protocols.schema.internet.ah import AH as Schema_AH from pcapkit.utilities.exceptions import UnsupportedCall @@ -46,9 +47,20 @@ __all__ = ['AH'] -class AH(IPsec[Data_AH, Schema_AH], +class AH(IPsec[Data_AH, Schema_AH], IPv6_Ext[Data_AH, Schema_AH], schema=Schema_AH, data=Data_AH): - """This class implements Authentication Header.""" + """This class implements Authentication Header. + + Double-inherited (GitHub issue #917): ``AH`` is both a member of the + IPsec family and an IPv6 extension header -- IANA's + ``protocol-numbers-1.csv`` marks it ``Y`` in the *IPv6 Extension Header* + column (:rfc:`4302`), and this package's own + :class:`~pcapkit.const.ipv6.extension_header.ExtensionHeader` registry + agrees. :class:`~pcapkit.protocols.internet.ipsec.IPsec` is first in the + bases so that its :meth:`~pcapkit.protocols.internet.ipsec.IPsec.id` + keeps precedence. + + """ ########################################################################## # Properties. @@ -59,6 +71,25 @@ def name(self) -> 'Literal["Authentication Header"]': """Name of corresponding protocol.""" return 'Authentication Header' + @property + def alias(self) -> 'Literal["AH"]': + """Acronym of corresponding protocol. + + Spelled out rather than left to + :attr:`ProtocolBase.alias `'s + class-name default, because + :class:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext` now sits + between this class and that default in the MRO and carries a concrete + ``'IPv6-Ext'`` of its own (GitHub issue #917). Inheriting it would + rename this header in every + :class:`~pcapkit.corekit.protochain.ProtoChain` string and in + :meth:`IPv6._decode_next_layer + `'s packet + dict key. The value is exactly what the default produced before. + + """ + return 'AH' + @property def length(self) -> 'int': """Header length of current protocol.""" diff --git a/pcapkit/protocols/internet/esp.py b/pcapkit/protocols/internet/esp.py index d997dd8697..9572633d18 100644 --- a/pcapkit/protocols/internet/esp.py +++ b/pcapkit/protocols/internet/esp.py @@ -176,9 +176,10 @@ from pcapkit.corekit.infoclass import Info, info_final from pcapkit.protocols.data.internet.esp import ESP as Data_ESP from pcapkit.protocols.internet.ipsec import IPsec +from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext from pcapkit.protocols.schema.internet.esp import ESP as Schema_ESP from pcapkit.protocols.schema.schema import Schema -from pcapkit.utilities.exceptions import ProtocolError, ProtocolUnbound +from pcapkit.utilities.exceptions import ProtocolError, ProtocolUnbound, UnsupportedCall from pcapkit.utilities.warnings import ProtocolWarning, warn __all__ = ['ESP', 'ESPStatus', 'Cipher', 'Integrity', 'CipherSuite', 'IntegritySuite', @@ -187,7 +188,7 @@ if TYPE_CHECKING: from enum import IntEnum as StdlibEnum from ipaddress import IPv4Address, IPv6Address - from typing import IO, Any, Optional, Type + from typing import IO, Any, NoReturn, Optional, Type from aenum import IntEnum as AenumEnum from typing_extensions import Literal @@ -962,9 +963,26 @@ def __repr__(self) -> 'str': ############################################################################## -class ESP(IPsec[Data_ESP, Schema_ESP], +class ESP(IPsec[Data_ESP, Schema_ESP], IPv6_Ext[Data_ESP, Schema_ESP], schema=Schema_ESP, data=Data_ESP): - """This class implements Encapsulating Security Payload.""" + """This class implements Encapsulating Security Payload. + + Double-inherited (GitHub issue #917), mirroring + :class:`~pcapkit.protocols.internet.ah.AH`: IANA's + ``protocol-numbers-1.csv`` marks ``ESP`` ``Y`` in the *IPv6 Extension + Header* column (:rfc:`4303`), and this package's own + :class:`~pcapkit.const.ipv6.extension_header.ExtensionHeader` registry + agrees (``ESP = 50``), so it must honour the same extension-mode contract + as its siblings. :attr:`payload` and :attr:`protochain` come from + :class:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext`; :attr:`protocol` + is spelled out below for the reason given there. + + Note: + :rfc:`8200#section-4.5` says outright that ESP "is not considered an + extension header". The library follows IANA's registry rather than + that sentence, on the owner's ruling for GitHub issue #895. + + """ ########################################################################## # Properties. @@ -975,6 +993,36 @@ def name(self) -> 'Literal["Encapsulating Security Payload"]': """Name of corresponding protocol.""" return 'Encapsulating Security Payload' + @property + def alias(self) -> 'Literal["ESP"]': + """Acronym of corresponding protocol. + + Spelled out rather than left to + :attr:`ProtocolBase.alias `'s + class-name default, because + :class:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext` now sits + between this class and that default in the MRO and carries a concrete + ``'IPv6-Ext'`` of its own. Inheriting it would rename this header in + every :class:`~pcapkit.corekit.protochain.ProtoChain` string and in + :meth:`IPv6._decode_next_layer + `'s packet + dict key. The value is exactly what the default produced before. + + """ + return 'ESP' + + @property + def protocol(self) -> 'Optional[str] | NoReturn': + """Name of next layer protocol (if any). + + 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 'protocol'") + return super().protocol + @property def length(self) -> 'int': """Length of the ESP header, payload, trailer and ICV. diff --git a/pcapkit/protocols/internet/hip.py b/pcapkit/protocols/internet/hip.py index e929057b1b..bb1033c4e2 100644 --- a/pcapkit/protocols/internet/hip.py +++ b/pcapkit/protocols/internet/hip.py @@ -120,6 +120,7 @@ from pcapkit.protocols.data.internet.hip import UnassignedParameter as Data_UnassignedParameter from pcapkit.protocols.data.internet.hip import ViaRVSParameter as Data_ViaRVSParameter from pcapkit.protocols.internet.internet import Internet +from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext from pcapkit.protocols.schema.internet.hip import HIP as Schema_HIP from pcapkit.protocols.schema.internet.hip import AckDataParameter as Schema_AckDataParameter from pcapkit.protocols.schema.internet.hip import ACKParameter as Schema_ACKParameter @@ -254,10 +255,38 @@ class Locator(TypedDict): spi: 'NotRequired[int]' -class HIP(Internet[Data_HIP, Schema_HIP], +class HIP(IPv6_Ext[Data_HIP, Schema_HIP], Internet[Data_HIP, Schema_HIP], schema=Schema_HIP, data=Data_HIP): """This class implements Host Identity Protocol. + Double-inherited, per the maintainer's convention on GitHub pull request + #924: a header that is *only* usable as an extension header inherits + :class:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext` alone, while one + that is also usable as a standalone protocol names + :class:`~pcapkit.protocols.internet.internet.Internet` as well. HIP is + both, on two independent grounds: + + * :rfc:`7401#section-5.1` states that "the HIP header is logically an + IPv6 extension header", and IANA lists protocol 139 in its *IPv6 + Extension Header Types* registry -- 11 entries, HIP among them. + * :rfc:`7401#appendix-C.2`, "IPv4 HIP Packet (I1 Packet)", works a + checksum for an **IPv4** header carrying ``Next Header: 139`` with + ``Payload Protocol: 59``. HIP therefore travels directly as an IPv4 + payload, exactly as :class:`~pcapkit.protocols.internet.ah.AH` and + :class:`~pcapkit.protocols.internet.esp.ESP` do. + + The second ground is what separates HIP from + :class:`~pcapkit.protocols.internet.mh.MH` and Shim6, which are protocols + in their own right but cannot appear under IPv4: + :rfc:`6275#section-6.1.1` defines the Mobility Header checksum over a + pseudo-header of IPv6 header fields with no IPv4 variant, and Mobile IPv4 + carries its equivalent messages over UDP port 434 (:rfc:`5944`) rather + than as protocol 135. + + ``Internet`` is already reached transitively through ``IPv6_Ext``; naming + it is what records the classification, so a future reader can tell a + deliberate standalone protocol from a header that merely inherits one. + This class currently supports parsing of the following HIP parameters, which are registered in the :attr:`self.__parameter__ ` attribute: diff --git a/pcapkit/protocols/internet/hopopt.py b/pcapkit/protocols/internet/hopopt.py index 8aa2debcb2..507fb0a79c 100644 --- a/pcapkit/protocols/internet/hopopt.py +++ b/pcapkit/protocols/internet/hopopt.py @@ -64,7 +64,7 @@ from pcapkit.protocols.data.internet.hopopt import \ TunnelEncapsulationLimitOption as Data_TunnelEncapsulationLimitOption from pcapkit.protocols.data.internet.hopopt import UnassignedOption as Data_UnassignedOption -from pcapkit.protocols.internet.internet import Internet +from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext from pcapkit.protocols.schema.internet.hopopt import HOPOPT as Schema_HOPOPT from pcapkit.protocols.schema.internet.hopopt import CALIPSOOption as Schema_CALIPSOOption from pcapkit.protocols.schema.internet.hopopt import HomeAddressOption as Schema_HomeAddressOption @@ -120,7 +120,7 @@ __all__ = ['HOPOPT'] -class HOPOPT(Internet[Data_HOPOPT, Schema_HOPOPT], +class HOPOPT(IPv6_Ext[Data_HOPOPT, Schema_HOPOPT], schema=Schema_HOPOPT, data=Data_HOPOPT): """This class implements IPv6 Hop-by-Hop Options. @@ -222,6 +222,25 @@ def name(self) -> 'Literal["IPv6 Hop-by-Hop Options"]': """Name of current protocol.""" return 'IPv6 Hop-by-Hop Options' + @property + def alias(self) -> 'Literal["HOPOPT"]': + """Acronym of corresponding protocol. + + Spelled out rather than left to + :attr:`ProtocolBase.alias `'s + class-name default, because + :class:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext` now sits + between this class and that default in the MRO and carries a concrete + ``'IPv6-Ext'`` of its own (GitHub issue #917). Inheriting it would + rename this header in every + :class:`~pcapkit.corekit.protochain.ProtoChain` string and in + :meth:`IPv6._decode_next_layer + `'s packet + dict key. The value is exactly what the default produced before. + + """ + return 'HOPOPT' + @property def length(self) -> 'int': """Header length of current protocol.""" diff --git a/pcapkit/protocols/internet/ipv6.py b/pcapkit/protocols/internet/ipv6.py index 74cd582625..a0bbed7a38 100644 --- a/pcapkit/protocols/internet/ipv6.py +++ b/pcapkit/protocols/internet/ipv6.py @@ -63,10 +63,10 @@ class IPv6(IP[Data_IPv6, Schema_IPv6], #: 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 + #: :mod:`pcapkit.protocols.internet.ipv6_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` + #: :class:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext` #: 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 @@ -79,8 +79,8 @@ class IPv6(IP[Data_IPv6, Schema_IPv6], #: (``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 + #: :class:`IPv6_Ext` by *direct* registration instead (see the + #: bottom of :mod:`pcapkit.protocols.internet.ipv6_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 @@ -417,7 +417,7 @@ def _decode_next_layer(self, ipv6: 'Data_IPv6', proto: 'Optional[int]' = None, # 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`, + # :class:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext`, # 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 @@ -489,7 +489,7 @@ def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *, # 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` + :class:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext` for it rather than letting the exception reach the :func:`~pcapkit.utilities.decorators.beholder` decorating this method, which would otherwise substitute plain @@ -497,7 +497,7 @@ def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *, # ``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 + ``IPv6_Ext`` 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 @@ -540,14 +540,14 @@ def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *, # 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 + from pcapkit.protocols.internet.ipv6_ext import \ + IPv6_Ext # isort: skip # pylint: disable=import-outside-toplevel - if not (extension and protocol is not IPv6_GenericExt + if not (extension and protocol is not IPv6_Ext 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) + next_ = IPv6_Ext(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_ext.py similarity index 58% rename from pcapkit/protocols/internet/ipv6_generic_ext.py rename to pcapkit/protocols/internet/ipv6_ext.py index 3c8650d29d..82446ffd75 100644 --- a/pcapkit/protocols/internet/ipv6_generic_ext.py +++ b/pcapkit/protocols/internet/ipv6_ext.py @@ -1,14 +1,17 @@ # -*- coding: utf-8 -*- -"""IPv6_GenericExt - Generic IPv6 Extension Header -====================================================== +"""IPv6_Ext - IPv6 Extension Header +==================================== -.. module:: pcapkit.protocols.internet.ipv6_generic_ext +.. module:: pcapkit.protocols.internet.ipv6_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. +:mod:`pcapkit.protocols.internet.ipv6_ext` contains +:class:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext` +only, which serves two roles at once (GitHub issue #917): it is the +shared **base class** of every IPv6 extension header in this package, +and it implements a **generic** extractor for IPv6 extension headers, +standing in for one whenever the header's own dedicated parser is +unavailable or has failed. See the class docstring for the division +between the two. Why this is safe in general ---------------------------- @@ -58,14 +61,25 @@ 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 +This table classifies by *wire format* alone, over the twelve codes +:class:`~pcapkit.const.ipv6.extension_header.ExtensionHeader` enumerates. +Note that IANA's own *IPv6 Extension Header Types* registry has **eleven**: +the twelfth, ``BIT_EMU`` (147), comes from this package generating that +enumeration out of the *Protocol Numbers* registry's extension-header +column instead, where 147 is flagged ``Y`` while the extension-header +registry omits it. That discrepancy is GitHub issue #925 and is not this +class's to resolve; the classification below holds either way, since 147 +is not reachable through here. + +``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 +:rfc:`8200#section-4.5` sets Encapsulating Security Payload aside -- +*"For this purpose,"* it writes, ESP *"is not considered an extension +header"*, and the sentence after it lists ESP among *"examples of +upper-layer headers"*; :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 @@ -148,36 +162,93 @@ ends on honestly, at the bad header, instead of inventing what follows it. """ -from typing import TYPE_CHECKING, overload +from typing import TYPE_CHECKING, Generic, cast, 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.data.internet.ipv6_ext import IPv6_Ext as Data_IPv6_Ext from pcapkit.protocols.internet.internet import Internet -from pcapkit.protocols.schema.internet.ipv6_generic_ext import \ - IPv6_GenericExt as Schema_IPv6_GenericExt +from pcapkit.protocols.protocol import _PT, _ST +from pcapkit.protocols.schema.internet.ipv6_ext import IPv6_Ext as Schema_IPv6_Ext 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 import IO, Any, NoReturn, Optional, Protocol from typing_extensions import Literal from pcapkit.corekit.protochain import ProtoChain from pcapkit.protocols.protocol import ProtocolBase -__all__ = ['IPv6_GenericExt'] + class _NextHeaderData(Protocol): + """The one field every IPv6 extension header's data model carries. + :rfc:`8200#section-4.1` puts a Next Header octet first in every + extension header, and all eight implemented ones record it under this + name -- which is what lets :attr:`IPv6_Ext.next` be a *shared* member + rather than a fallback-role-only one. Narrower than + :class:`~pcapkit.protocols.data.internet.ipv6_ext.IPv6_Ext` on + purpose: ``length`` is deliberately absent, because + :class:`~pcapkit.protocols.data.internet.ipv6_frag.IPv6_Frag` has no + such field (its length is the constant 8, and its own + :attr:`~pcapkit.protocols.internet.ipv6_frag.IPv6_Frag.length` + property supplies it). + + """ + + next: 'Optional[Enum_TransType]' + +__all__ = ['IPv6_Ext'] -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. + +class IPv6_Ext(Internet[_PT, _ST], Generic[_PT, _ST], + schema=Schema_IPv6_Ext, data=Data_IPv6_Ext): + """This class implements a generic IPv6 extension header parser, and is + the shared base of every IPv6 extension header in this package. 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. + The two roles + -------------- + + This one class plays both, on the owner's ruling for GitHub issue #917: + + 1. **The concrete fallback parser** for an :rfc:`6564`-conforming header + this package has no dedicated class for, or whose dedicated class + raised -- which is what :meth:`read`, :meth:`make`, :attr:`name`, + :attr:`alias` and the ``schema=``/``data=`` above implement. + 2. **The base class** of the eight implemented extension headers + (:class:`~pcapkit.protocols.internet.hopopt.HOPOPT`, + :class:`~pcapkit.protocols.internet.ipv6_route.IPv6_Route`, + :class:`~pcapkit.protocols.internet.ipv6_frag.IPv6_Frag`, + :class:`~pcapkit.protocols.internet.ipv6_opts.IPv6_Opts`, + :class:`~pcapkit.protocols.internet.hip.HIP`, + :class:`~pcapkit.protocols.internet.mh.MH`, + :class:`~pcapkit.protocols.internet.ah.AH` and + :class:`~pcapkit.protocols.internet.esp.ESP`), which is what the + ``_extf`` guards on :attr:`payload`, :attr:`protocol` and + :attr:`protochain` are for. + + It is generic in its data and schema types -- exactly like + :class:`~pcapkit.protocols.internet.ipsec.IPsec`, the other base in this + package -- so that a subclass keeps its *own* ``_PT``/``_ST`` instead of + inheriting this class's. ``AH`` and ``ESP`` therefore double-inherit two + identically-parameterised generic bases, ``IPsec[…]`` and ``IPv6_Ext[…]``. + + Warning: + A subclass **must** define :attr:`name`, :attr:`alias`, + :attr:`protocol`, :attr:`length` and :meth:`__index__` itself. All five + carry this class's *fallback-role* answers, which are wrong for a + header that has 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. Nothing in the language enforces the + override, so ``tests/protocols/internet/test_ipv6_ext_unit.py`` + enforces it instead, over every subclass discovered at runtime. + """ ########################################################################## @@ -185,14 +256,25 @@ class IPv6_GenericExt(Internet[Data_IPv6_GenericExt, Schema_IPv6_GenericExt], ########################################################################## @property - def name(self) -> 'Literal["Generic IPv6 Extension Header"]': - """Name of current protocol.""" - return 'Generic IPv6 Extension Header' + def name(self) -> 'str': + """Name of current protocol. + + Annotated ``str`` rather than as the ``Literal`` every *leaf* protocol + in this package uses, because this one is also a base: a ``Literal`` + here makes each subclass's own ``Literal`` an incompatible override + (measured: eight ``[override]`` errors from mypy, one per subclass). + The value returned is still the single fallback-role constant. + + """ + return 'IPv6 Extension Header' @property - def alias(self) -> 'Literal["IPv6-GenericExt"]': + def alias(self) -> 'str': """Acronym of corresponding protocol. + Annotated ``str`` rather than ``Literal``, for the reason given on + :attr:`name`. + Hyphenated, like :attr:`IPv6_Frag.alias ` and :attr:`IPv6_Opts.alias `, @@ -200,51 +282,91 @@ def alias(self) -> 'Literal["IPv6-GenericExt"]': ` 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``. + default (class-name) alias ``'IPv6_Ext'`` would strip to ``'_Ext'`` + and key the dict as ``_ext``. The hyphen makes every leading + character (``I``, ``P``, ``v``, ``6``, ``-``) a member of the set + being stripped, same as the siblings, giving ``ext``. """ - return 'IPv6-GenericExt' + return 'IPv6-Ext' @property def length(self) -> 'int': - """Header length of current protocol.""" - return self._info.length + """Header length of current protocol. + + Fallback-role member: it reads this module's *own* data model, so a + subclass whose data model records its length elsewhere -- or not at + all, as :class:`~pcapkit.protocols.data.internet.ipv6_frag.IPv6_Frag` + does not -- must override it. All eight implemented headers do, and + ``tests/protocols/internet/test_ipv6_ext_unit.py`` holds them to it. + + """ + return cast('Data_IPv6_Ext', self._info).length @property - def protocol(self) -> 'Optional[Enum_ExtensionHeader]': + def protocol(self) -> 'Optional[Enum_ExtensionHeader] | Optional[str]': """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. + In the *fallback* role -- i.e. when this instance parsed this + module's own schema -- 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 + already holds before it dispatches (``ipv6.py:388``). See :meth:`__index__` for why the *class-level* identity cannot be - made to work the same way. + made to work the same way. It is deliberately *not* ``_extf``-guarded + in this role: that same call passes ``extension=True`` for every + extension header in a chain, so guarding it would make the identity + unreadable in precisely the case it exists for. + + In the *base* role it falls through to ``super()``, restoring the + ordinary :class:`~pcapkit.protocols.protocol.ProtocolBase` meaning. + That fall-through is load-bearing rather than tidy: all eight + subclasses implement their own ``_extf``-guarded ``protocol`` as + ``return super().protocol``, and this class sits between them and + :class:`~pcapkit.protocols.protocol.ProtocolBase` in the MRO -- so + without the discriminator below, every one of those eight would end + up reading ``self._info.protocol`` off a data model that has no such + field. Measured: ``AttributeError`` on all eight. + + The discriminator is the *class-level* + :attr:`~pcapkit.protocols.protocol.ProtocolBase.__data__` rather than + an :func:`isinstance` test on ``self._info``, because a ``make``-only + instance has no ``_info`` at all -- reading it here to decide which + role we are in turns + :attr:`ProtocolBase.protocol `, + which only ever needed ``self._protos``, into an ``AttributeError`` + (measured: ``tests/protocols/internet/test_ipv6_extension_unit.py`` + constructs exactly that instance). """ - return self._info.protocol + if self.__data__ is Data_IPv6_Ext: + return cast('Data_IPv6_Ext', self._info).protocol + return super().protocol @property def next(self) -> 'Optional[Enum_TransType]': """Next header, as parsed off the wire. + Shared by every subclass rather than fallback-role-only -- see + :class:`_NextHeaderData` for why that is sound. Note this is an + *addition* for the eight implemented headers: none of them declared a + ``next`` property of its own before GitHub issue #917, so reading one + raised :exc:`AttributeError`, and nothing could have depended on a + value it never returned. + :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 + return cast('_NextHeaderData', self._info).next @property def payload(self) -> 'ProtocolBase | NoReturn': @@ -256,7 +378,7 @@ def payload(self) -> 'ProtocolBase | NoReturn': """ if self._extf: raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute 'payload'") - return self._next + return super().payload @property def protochain(self) -> 'ProtoChain | NoReturn': @@ -274,28 +396,48 @@ def protochain(self) -> 'ProtoChain | NoReturn': # 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 + def read(self, length: 'Optional[int]' = None, *, # pylint: disable=arguments-differ + extension: 'bool' = False, **kwargs: 'Any') -> '_PT': """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. + **kwargs: Arbitrary keyword arguments, two of which are this + class's own and are supplied by :meth:`IPv6._import_next_layer + ` + (``ipv6.py:540,550``) rather than typed by a caller: + + * ``alias`` -- the numeric extension header code this instance + stands in for, e.g. ``0`` for ``HOPOPT`` or ``140`` for + ``Shim6``. + * ``error`` -- the parsing error, when this instance was + reached as a :func:`~pcapkit.utilities.decorators.beholder` + fallback rather than by direct dispatch. Returns: Parsed packet data. + Note: + Those two are read out of ``**kwargs`` rather than declared as + parameters, and ``version`` is not declared either, because this + class is *also* the base of eight subclasses whose own ``read`` + accepts none of the three. Declaring them here asserts of the whole + family an interface only the fallback has, and both linters say so: + mypy reports eight ``Signature of "read" incompatible with + supertype`` ``[override]`` errors, pylint ``arguments-differ``. + ``version`` costs nothing to drop in any case -- this method never + read it (it carried ``# pylint: disable=unused-argument`` for + exactly that reason); the version gate lives in + :meth:`__post_init__`, which still takes it explicitly. + """ + alias = kwargs.get('alias') # type: Optional[int] + error = kwargs.get('error') # type: Optional[Exception] + if length is None: length = len(self) - schema = self.__header__ + schema = cast('Schema_IPv6_Ext', self.__header__) ext_code = None # type: Optional[Enum_ExtensionHeader] if alias is not None: @@ -335,7 +477,7 @@ def read(self, length: 'Optional[int]' = None, *, error: 'Optional[Exception]' = ext_len = nominal next_header = schema.next - generic_ext = Data_IPv6_GenericExt( + generic_ext = Data_IPv6_Ext( protocol=ext_code, next=next_header, length=ext_len, @@ -343,14 +485,14 @@ def read(self, length: 'Optional[int]' = None, *, error: 'Optional[Exception]' = ) if extension: - return generic_ext - return self._decode_next_layer(generic_ext, next_header, length - ext_len) + return cast('_PT', generic_ext) + return self._decode_next_layer(cast('_PT', generic_ext), next_header, length - ext_len) - def make(self, + 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': + **kwargs: 'Any') -> '_ST': """Make (construct) packet data. Args: @@ -365,12 +507,43 @@ def make(self, Returns: Constructed packet data. + Note: + **Keyword-only**, and that matters in two directions at once. + + The three stay *declared*, unlike :meth:`read`'s own keywords, + because :func:`~pcapkit.protocols.protocol._check_construction_keywords` + builds its allowlist from :func:`inspect.signature` of ``make``. + Hiding them in ``**kwargs`` was tried and **breaks construction**: + measured as ``UnsupportedCall: IPv6_Ext: unexpected keyword(s): + 'len' (did you mean 'length'?), 'next', 'payload'`` on a plain + ``IPv6_Ext(next=..., len=..., payload=...)``. The ``__keywords__`` + escape hatch would cover that, but it is unioned down the MRO, so + it would widen the allowlist -- and weaken that misspelling check -- + for all eight subclasses too. + + They are keyword-*only* because this class is also a base, and each + subclass's ``make`` puts different names in the same positions. As + positional parameters they drew 18 pylint ``arguments-renamed`` + warnings (``ESP.make``: ``next`` -> ``spi``, ``len`` -> ``seq``, + ``payload`` -> ``next``; ``IPv6_Route.make``: ``next`` -> ``dst``, + ``len`` -> ``next``, ``payload`` -> ``next_default``; and two each + for the other six) and eight mypy ``[override]`` errors. With no + positional parameters there is no position to disagree about: + both counts drop to zero, and the allowlist is unaffected because + ``_declared_keywords`` collects ``KEYWORD_ONLY`` parameters too. + Nothing calls a protocol's ``make`` positionally -- ``__init__`` + spreads ``**kwargs`` into it (``protocol.py:607``). + + ``read`` needs none of this: its ``alias``/``error`` only ever + arrive on the *parse* path, which ``__init__`` exempts from the + construction check outright. + """ - return Schema_IPv6_GenericExt( + return cast('_ST', Schema_IPv6_Ext( next=next, len=len, payload=payload, - ) + )) ########################################################################## # Data models. @@ -396,13 +569,25 @@ def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None, length: 'Opt **kwargs: Arbitrary keyword arguments. Raises: - ProtocolError: If ``version`` is not ``6``. + ProtocolError: If this is the *fallback* parser (see the class + docstring's "two roles") and ``version`` is not ``6``. See Also: For construction argument, please refer to :meth:`make`. Note: - This class is registered into :attr:`Internet.__proto__ + The version gate below is scoped to the fallback role, by the same + ``__data__`` discriminator :attr:`protocol` uses. Applying it to + subclasses would break the two that are *not* IPv6-only: + :class:`~pcapkit.protocols.internet.ah.AH` and + :class:`~pcapkit.protocols.internet.esp.ESP` both default to + ``version=4`` and are perfectly valid under IPv4, so an + unconditional gate here rejects them outright (measured: + ``ProtocolError: ESP: only valid for IPv6, got version=4`` from + ``tests/protocols/internet/test_esp_unit.py``). + + The gate itself is unchanged in effect for the fallback, and is + there because 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 -- @@ -412,7 +597,7 @@ def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None, length: 'Opt 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 + ``IPv4:IPv6-Ext:...`` 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`` -- @@ -421,7 +606,7 @@ def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None, length: 'Opt IPv4-specific code of its own. """ - if version != 6: + if self.__data__ is Data_IPv6_Ext and version != 6: raise ProtocolError( f'{self.__class__.__name__}: only valid for IPv6, got version={version}') @@ -431,8 +616,14 @@ def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None, length: 'Opt # 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.""" + def __length_hint__(self) -> 'int': + """Return an estimated length for the object. + + Two octets -- all :rfc:`6564#section-4` guarantees. Annotated ``int`` + rather than ``Literal[2]``, for the reason given on :attr:`name`: every + subclass has a longer fixed header and its own ``Literal``. + + """ return 2 @classmethod @@ -446,7 +637,7 @@ def __index__(cls) -> 'NoReturn': 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 + not per class: one :class:`IPv6_Ext` 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 @@ -461,7 +652,7 @@ def __index__(cls) -> 'NoReturn': ########################################################################## @classmethod - def _make_data(cls, data: 'Data_IPv6_GenericExt') -> 'dict[str, Any]': # type: ignore[override] + def _make_data(cls, data: 'Data_IPv6_Ext') -> 'dict[str, Any]': # type: ignore[override] """Create key-value pairs from ``data`` for protocol construction. Inverts whichever per-protocol length rule :meth:`read` applied, @@ -506,4 +697,4 @@ def _make_data(cls, data: 'Data_IPv6_GenericExt') -> 'dict[str, Any]': # type: # 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 +Internet.__proto__[Enum_TransType.Shim6] = IPv6_Ext diff --git a/pcapkit/protocols/internet/ipv6_frag.py b/pcapkit/protocols/internet/ipv6_frag.py index 27c798cbd5..1af01bd09f 100644 --- a/pcapkit/protocols/internet/ipv6_frag.py +++ b/pcapkit/protocols/internet/ipv6_frag.py @@ -28,7 +28,7 @@ from pcapkit.const.reg.transtype import TransType as Enum_TransType from pcapkit.protocols.data.internet.ipv6_frag import IPv6_Frag as Data_IPv6_Frag -from pcapkit.protocols.internet.internet import Internet +from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext from pcapkit.protocols.schema.internet.ipv6_frag import IPv6_Frag as Schema_IPv6_Frag from pcapkit.utilities.exceptions import UnsupportedCall @@ -46,7 +46,7 @@ __all__ = ['IPv6_Frag'] -class IPv6_Frag(Internet[Data_IPv6_Frag, Schema_IPv6_Frag], +class IPv6_Frag(IPv6_Ext[Data_IPv6_Frag, Schema_IPv6_Frag], schema=Schema_IPv6_Frag, data=Data_IPv6_Frag): """This class implements Fragment Header for IPv6.""" diff --git a/pcapkit/protocols/internet/ipv6_opts.py b/pcapkit/protocols/internet/ipv6_opts.py index dbf8bba64d..3eb42e6ff9 100644 --- a/pcapkit/protocols/internet/ipv6_opts.py +++ b/pcapkit/protocols/internet/ipv6_opts.py @@ -64,7 +64,7 @@ from pcapkit.protocols.data.internet.ipv6_opts import \ TunnelEncapsulationLimitOption as Data_TunnelEncapsulationLimitOption from pcapkit.protocols.data.internet.ipv6_opts import UnassignedOption as Data_UnassignedOption -from pcapkit.protocols.internet.internet import Internet +from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext from pcapkit.protocols.schema.internet.ipv6_opts import CALIPSOOption as Schema_CALIPSOOption from pcapkit.protocols.schema.internet.ipv6_opts import \ HomeAddressOption as Schema_HomeAddressOption @@ -124,7 +124,7 @@ __all__ = ['IPv6_Opts'] -class IPv6_Opts(Internet[Data_IPv6_Opts, Schema_IPv6_Opts], +class IPv6_Opts(IPv6_Ext[Data_IPv6_Opts, Schema_IPv6_Opts], schema=Schema_IPv6_Opts, data=Data_IPv6_Opts): """This class implements Destination Options for IPv6. diff --git a/pcapkit/protocols/internet/ipv6_route.py b/pcapkit/protocols/internet/ipv6_route.py index 7a5be1d8a0..924165a253 100644 --- a/pcapkit/protocols/internet/ipv6_route.py +++ b/pcapkit/protocols/internet/ipv6_route.py @@ -35,7 +35,7 @@ from pcapkit.protocols.data.internet.ipv6_route import SourceRoute as Data_SourceRoute from pcapkit.protocols.data.internet.ipv6_route import Type2 as Data_Type2 from pcapkit.protocols.data.internet.ipv6_route import UnknownType as Data_UnknownType -from pcapkit.protocols.internet.internet import Internet +from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext from pcapkit.protocols.schema.internet.ipv6_route import RPL as Schema_RPL from pcapkit.protocols.schema.internet.ipv6_route import IPv6_Route as Schema_IPv6_Route from pcapkit.protocols.schema.internet.ipv6_route import SourceRoute as Schema_SourceRoute @@ -67,7 +67,7 @@ __all__ = ['IPv6_Route'] -class IPv6_Route(Internet[Data_IPv6_Route, Schema_IPv6_Route], +class IPv6_Route(IPv6_Ext[Data_IPv6_Route, Schema_IPv6_Route], schema=Schema_IPv6_Route, data=Data_IPv6_Route): """This class implements Routing Header for IPv6. diff --git a/pcapkit/protocols/internet/mh.py b/pcapkit/protocols/internet/mh.py index edfec040ab..e69552635c 100644 --- a/pcapkit/protocols/internet/mh.py +++ b/pcapkit/protocols/internet/mh.py @@ -275,7 +275,7 @@ from pcapkit.protocols.data.internet.mh import \ UpdateNotificationMessage as Data_UpdateNotificationMessage from pcapkit.protocols.data.internet.mh import VendorSpecificOption as Data_VendorSpecificOption -from pcapkit.protocols.internet.internet import Internet +from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext from pcapkit.protocols.schema.internet.mh import MH as Schema_MH from pcapkit.protocols.schema.internet.mh import \ AccessNetworkIdentifierOption as Schema_AccessNetworkIdentifierOption @@ -890,7 +890,7 @@ def _missing_(cls, value: 'int') -> 'NoReturn': raise EnumValueError('%r is not a valid %s' % (value, cls.__name__)) -class MH(Internet[Data_MH, Schema_MH], +class MH(IPv6_Ext[Data_MH, Schema_MH], schema=Schema_MH, data=Data_MH): """This class implements Mobility Header. @@ -1371,6 +1371,25 @@ def name(self) -> 'Literal["Mobility Header"]': """Name of current protocol.""" return 'Mobility Header' + @property + def alias(self) -> 'Literal["MH"]': + """Acronym of corresponding protocol. + + Spelled out rather than left to + :attr:`ProtocolBase.alias `'s + class-name default, because + :class:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext` now sits + between this class and that default in the MRO and carries a concrete + ``'IPv6-Ext'`` of its own (GitHub issue #917). Inheriting it would + rename this header in every + :class:`~pcapkit.corekit.protochain.ProtoChain` string and in + :meth:`IPv6._decode_next_layer + `'s packet + dict key. The value is exactly what the default produced before. + + """ + return 'MH' + @property def length(self) -> 'int': """Header length of current protocol.""" diff --git a/pcapkit/protocols/schema/__init__.py b/pcapkit/protocols/schema/__init__.py index c35e1d4ecd..360de9e1cf 100644 --- a/pcapkit/protocols/schema/__init__.py +++ b/pcapkit/protocols/schema/__init__.py @@ -68,7 +68,7 @@ 'IPv4_TROption', 'IPv4_RTRALTOption', 'IPv4_QSOption', 'IPv4_QuickStartRequestOption', 'IPv4_QuickStartReportOption', 'IPv6_Frag', - 'IPv6_GenericExt', + 'IPv6_Ext', '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 7d6096c240..3eef94ddef 100644 --- a/pcapkit/protocols/schema/internet/__init__.py +++ b/pcapkit/protocols/schema/internet/__init__.py @@ -131,8 +131,8 @@ # 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 Extension Header (base + generic fallback) +from pcapkit.protocols.schema.internet.ipv6_ext import IPv6_Ext # IPv6 Destination Options from pcapkit.protocols.schema.internet.ipv6_opts import CALIPSOOption as IPv6_Opts_CALIPSOOption @@ -266,8 +266,8 @@ # IPv6 Fragment Header 'IPv6_Frag', - # Generic IPv6 Extension Header - 'IPv6_GenericExt', + # IPv6 Extension Header (base + generic fallback) + 'IPv6_Ext', # IPv6 Destination Options 'IPv6_Opts', diff --git a/pcapkit/protocols/schema/internet/ipv6_generic_ext.py b/pcapkit/protocols/schema/internet/ipv6_ext.py similarity index 91% rename from pcapkit/protocols/schema/internet/ipv6_generic_ext.py rename to pcapkit/protocols/schema/internet/ipv6_ext.py index 5f5f6fb126..406972cb35 100644 --- a/pcapkit/protocols/schema/internet/ipv6_generic_ext.py +++ b/pcapkit/protocols/schema/internet/ipv6_ext.py @@ -9,21 +9,21 @@ from pcapkit.corekit.fields.numbers import EnumField, UInt8Field from pcapkit.protocols.schema.schema import Schema, schema_final -__all__ = ['IPv6_GenericExt'] +__all__ = ['IPv6_Ext'] if TYPE_CHECKING: from pcapkit.protocols.protocol import ProtocolBase @schema_final -class IPv6_GenericExt(Schema): +class IPv6_Ext(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`, + done in :meth:`pcapkit.protocols.internet.ipv6_ext.IPv6_Ext.read`, which knows which protocol this instance stands in for; this schema does not. diff --git a/tests/protocols/internet/test_ipv6_ext_unit.py b/tests/protocols/internet/test_ipv6_ext_unit.py new file mode 100644 index 0000000000..260a51eedb --- /dev/null +++ b/tests/protocols/internet/test_ipv6_ext_unit.py @@ -0,0 +1,781 @@ +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 IPv6ExtUnitTests(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_ext import IPv6_Ext + + 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_Ext(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_ext import IPv6_Ext + + raw = bytes([int(TransType.TCP), 2]) + b'\x00' * 14 # (2+2)*4 == 16 + inst = IPv6_Ext(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_ext import IPv6_Ext + + raw = bytes([int(TransType.UDP), 200]) + b'\x00' * 14 # 200 must be ignored + inst = IPv6_Ext(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_ext import IPv6_Ext + 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_Ext(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_ext import IPv6_Ext + from pcapkit.utilities.exceptions import UnsupportedCall + + with self.assertRaises(UnsupportedCall): + IPv6_Ext.__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_ext import IPv6_Ext + from pcapkit.utilities.exceptions import UnsupportedCall + + raw = bytes([int(TransType.UDP), 1]) + b'\x00' * 14 + inst = IPv6_Ext(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_ext import IPv6_Ext + + 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_Ext(io.BytesIO(raw), len(raw), extension=True, + alias=int(code)) + values = IPv6_Ext._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_ext import IPv6_Ext + 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_Ext) + 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-Ext: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_ext import IPv6_Ext + 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_Ext) + 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-Ext') + + # -- 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_ext import IPv6_Ext + + self.assertIs(Internet.__proto__[TransType.Shim6], IPv6_Ext) + + 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_ext import IPv6_Ext + 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_Ext) + 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-Ext:UDP:Raw') + + # -- alias, so the packet dict and the chain segment are not '_ext' ------- + + 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_ext import IPv6_Ext + + raw = bytes([int(TransType.UDP), 1]) + b'\x00' * 14 + inst = IPv6_Ext(io.BytesIO(raw), len(raw), extension=True, + alias=int(ExtensionHeader.HOPOPT)) + self.assertEqual(inst.alias, 'IPv6-Ext') + # 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(), 'ext') + + # -- 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_ext import IPv6_Ext + from pcapkit.utilities.exceptions import ProtocolError + + raw = bytes([int(TransType.UDP), 1]) + b'\x00' * 14 + with self.assertRaises(ProtocolError): + IPv6_Ext(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_Ext`` + 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') + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class IPv6ExtSharedBaseContractTests(unittest.TestCase): + """Tests for GitHub issue #917: ``IPv6_Ext`` serves two roles at once -- + the shared base of every IPv6 extension header, *and* the concrete + fallback parser for one this package does not implement. + + That merge removes a safety net. Before it, each of the eight implemented + headers inherited :class:`~pcapkit.protocols.internet.internet.Internet` + directly, whose ``name``/``alias``/``protocol``/``length`` are generic or + abstract and whose ``__index__`` is :func:`abc.abstractmethod` -- so + forgetting one was loud. After it, the same omission silently inherits this + module's own *fallback-role* answers, which are wrong for a header that has + an identity: it reports itself as ``IPv6 Extension Header`` / ``IPv6-Ext``, + reads its length off a data model that is not its own, and ``__index__`` + raises instead of returning its IANA number. + + The eight are correct today only because all eight happen to shadow every + one of those members. Nothing in the language enforces that, so this class + does -- over every subclass discovered at runtime rather than a hard-coded + list, so a ninth added later is held to the same contract without anyone + remembering to come back here. + """ + + #: The members a subclass must define *itself*. Each carries a + #: fallback-role answer on the base that is wrong for a real header, and + #: each was measured to break concretely if inherited: ``alias`` renames the + #: 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__') + + #: The family as it stands, so a member silently *leaving* the base is + #: caught as well as one joining without the overrides. Named rather than + #: counted: a count alone cannot tell a swap from a no-op. + KNOWN_MEMBERS = frozenset({ + 'HOPOPT', 'IPv6_Route', 'IPv6_Frag', 'IPv6_Opts', 'HIP', 'MH', 'AH', 'ESP', + }) + + def setUp(self) -> None: + purge_modules(['pcapkit']) + + @staticmethod + def _subclasses(base: type) -> 'list[type]': + """Every subclass of ``base``, transitively. + + :meth:`type.__subclasses__` is one level deep, so an indirect subclass + -- which nothing forbids -- would otherwise escape this whole class. + """ + found = {} # type: dict[str, type] + stack = list(base.__subclasses__()) + while stack: + klass = stack.pop() + if klass.__qualname__ in found: + continue + found[klass.__qualname__] = klass + stack.extend(klass.__subclasses__()) + return [found[key] for key in sorted(found)] + + def _family(self) -> 'list[type]': + """The discovered family, with every member module imported first.""" + import pcapkit.protocols.internet # noqa: F401 # populates __subclasses__ + from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext + + return self._subclasses(IPv6_Ext) + + @staticmethod + def _constant(klass: type, member: str) -> 'str | None': + """What ``klass``'s ``member`` property answers with, given no instance + state. + + Resolved through the MRO with :func:`getattr` rather than read out of + ``klass.__dict__``, and called rather than having its ``Literal`` + annotation parsed. Both choices matter: the annotation is the + *documented* value where this is the value the library actually + answers with, and resolving through the MRO is what makes an + *inherited* fallback-role answer visible -- looking only at + ``__dict__`` would return :data:`None` for a class that failed to + override, which is indistinguishable from a pass. + + :data:`None` for a property that genuinely needs an instance -- + ``HIP.alias`` is ``f'HIPv{self._info.version}'``, derived from the + parsed header, so it has no constant to check. + """ + try: + return getattr(klass, member).fget(None) + except Exception: # pylint: disable=broad-exception-caught + return None + + def test_the_family_is_exactly_the_eight_known_members(self) -> None: + """Importing :mod:`pcapkit.protocols.internet` must turn up all eight, + and only those eight. This is what keeps the per-subclass tests below + from passing vacuously on a walk that discovered nothing. + """ + self.assertEqual({klass.__qualname__ for klass in self._family()}, + set(self.KNOWN_MEMBERS)) + + #: Which family members are *also* standalone protocols, and so name a + #: second base explicitly rather than only reaching one transitively. + #: + #: The maintainer's convention, from GitHub pull request #924: a header + #: usable *only* as an extension header inherits ``IPv6_Ext`` alone; one + #: usable as a standalone protocol names ``Internet`` (or ``IPsec``) as + #: well. The three here each travel as an IPv4 payload on a primary + #: source -- ``AH`` :rfc:`4302#section-3.1.1`, ``ESP`` + #: :rfc:`4303#section-3.1.1`, ``HIP`` :rfc:`7401#appendix-C.2`, whose + #: worked example is an IPv4 header carrying ``Next Header: 139``. + #: + #: ``MH`` is deliberately absent. It is a protocol in its own right, but + #: :rfc:`6275#section-6.1.1` defines its checksum over a pseudo-header of + #: IPv6 header fields with no IPv4 variant, and Mobile IPv4 carries the + #: equivalent messages over UDP port 434 (:rfc:`5944`) rather than as + #: protocol 135 -- so it cannot appear under IPv4 and stays extension-only. + STANDALONE_MEMBERS = frozenset({'AH', 'ESP', 'HIP'}) + + def test_standalone_members_name_a_second_base_and_the_rest_do_not(self) -> None: + """Pins the classification, which prose alone cannot keep honest. + + Asserted against ``__bases__`` rather than ``__mro__``, because every + member reaches :class:`~pcapkit.protocols.internet.internet.Internet` + transitively through ``IPv6_Ext`` -- so an ``__mro__`` check passes for + all eight and tests nothing. What the convention encodes is the + *declaration*: naming the base is how a deliberate standalone protocol + is told apart from a header that merely inherits one. + + """ + from pcapkit.protocols.internet.internet import Internet + from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext + + for klass in self._family(): + with self.subTest(cls=klass.__qualname__): + named = {base for base in klass.__bases__ + if base is not IPv6_Ext and issubclass(base, Internet)} + if klass.__qualname__ in self.STANDALONE_MEMBERS: + self.assertTrue( + named, + f'{klass.__qualname__} is classified as also usable as a standalone ' + f'protocol, so it must name Internet (or a subclass such as IPsec) ' + f'as an explicit base; it names only ' + f'{tuple(b.__name__ for b in klass.__bases__)}', + ) + else: + self.assertFalse( + named, + f'{klass.__qualname__} is classified as usable only as an extension ' + f'header, so IPv6_Ext should be its sole Internet-derived base; it ' + f'also names {tuple(b.__name__ for b in named)}', + ) + + def test_every_standalone_member_is_in_the_family(self) -> None: + """Guards the pin above from going vacuous if a name is misspelled.""" + self.assertLessEqual(self.STANDALONE_MEMBERS, frozenset(self.KNOWN_MEMBERS)) + + def test_every_subclass_defines_the_required_members_itself(self) -> None: + """Checked against ``__dict__``, not :func:`getattr`. + + :func:`getattr` cannot tell an override from an inherited fallback-role + answer -- which is the entire failure mode -- so it would report all + five present on a subclass that defines none of them. + """ + for klass in self._family(): + for member in self.REQUIRED_OVERRIDES: + with self.subTest(klass=klass.__qualname__, member=member): + # ``assertTrue`` rather than ``assertIn``: the latter dumps + # the whole class ``mappingproxy`` into the failure, which + # buries the message that says what to do about it. + self.assertTrue( + member in klass.__dict__, + f'{klass.__qualname__} inherits {member!r} from IPv6_Ext, whose ' + f'value is the generic fallback parser\'s and is wrong for a ' + f'header with an identity of its own; define it on ' + f'{klass.__qualname__} itself', + ) + + def test_every_subclass_index_returns_a_real_transtype(self) -> None: + """The base's :meth:`__index__` raises ``UnsupportedCall`` -- one + instance stands in for many codes, and a :func:`classmethod` has + nowhere to put a per-instance value. A subclass has exactly one code, + so it must return it. + """ + from pcapkit.const.reg.transtype import TransType as Enum_TransType + + for klass in self._family(): + with self.subTest(klass=klass.__qualname__): + index = klass.__index__() # must not raise + self.assertIsInstance(index, Enum_TransType) + self.assertIs(index, Enum_TransType(int(index))) + + def test_every_subclass_alias_survives_the_packet_dict_key_computation(self) -> None: + """:meth:`IPv6._decode_next_layer + ` keys its + packet dict by ``alias.lstrip('IPv6-').lower()`` (``ipv6.py:391``). + :meth:`str.lstrip` strips a character *set*, so the hyphen in + ``IPv6-Route``/``IPv6-Frag``/``IPv6-Opts`` is what lets ``IPv6`` fall + away cleanly, where the underscore of a class-name default like + ``IPv6_Route`` leaves ``_route``. + + Asserted as "no underscore, and the key is usable" rather than as + "contains a hyphen", because four of the eight legitimately carry none: + ``HOPOPT``, ``MH``, ``AH`` and ``ESP`` have no ``IPv6`` prefix for the + strip to bite on, so demanding a hyphen of them would be demanding a + rename rather than testing an invariant. + """ + for klass in self._family(): + alias = self._constant(klass, 'alias') + if alias is None: # HIP -- see ``_constant`` + continue + with self.subTest(klass=klass.__qualname__, alias=alias): + self.assertNotIn('_', alias) + key = alias.lstrip('IPv6-').lower() + self.assertTrue(key) + self.assertFalse(key.startswith(('_', '-'))) + if alias.startswith('IPv6'): + self.assertTrue(alias.startswith('IPv6-')) + + def test_no_subclass_answers_with_the_fallback_identity(self) -> None: + """The concrete failure the contract exists to prevent, asserted + directly rather than only through the ``__dict__`` check: no real + header may end up reporting the fallback parser's own identity. + """ + from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext + + base_name = self._constant(IPv6_Ext, 'name') + base_alias = self._constant(IPv6_Ext, 'alias') + self.assertEqual(base_name, 'IPv6 Extension Header') + self.assertEqual(base_alias, 'IPv6-Ext') + + for klass in self._family(): + with self.subTest(klass=klass.__qualname__): + self.assertNotEqual(self._constant(klass, 'name'), base_name) + self.assertNotEqual(self._constant(klass, 'alias'), base_alias) + + def test_the_base_itself_still_has_no_class_level_identity(self) -> None: + """The other half of the contract: the fallback genuinely has none, so + it must keep raising rather than be handed a placeholder index to + satisfy the rule above. + """ + from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext + from pcapkit.utilities.exceptions import UnsupportedCall + + with self.assertRaises(UnsupportedCall): + IPv6_Ext.__index__() + + def test_ah_and_esp_double_inherit_and_linearise(self) -> None: + """Both are IPsec members *and* IPv6 extension headers -- IANA marks + each ``Y`` in the *IPv6 Extension Header* column -- so each carries two + identically-parameterised generic bases. The MRO is pinned because it + is what makes the ``super()`` delegation in their ``protocol`` guards + land on :class:`~pcapkit.protocols.protocol.ProtocolBase` rather than + on the base's fallback-role ``protocol``. + """ + from pcapkit.protocols.internet.ah import AH + from pcapkit.protocols.internet.esp import ESP + from pcapkit.protocols.internet.internet import Internet + from pcapkit.protocols.internet.ipsec import IPsec + from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext + + for klass in (AH, ESP): + with self.subTest(klass=klass.__qualname__): + self.assertEqual( + [base.__qualname__ for base in klass.__mro__[:5]], + [klass.__qualname__, 'IPsec', 'IPv6_Ext', 'Internet', 'ProtocolBase']) + self.assertTrue(issubclass(klass, IPsec)) + self.assertTrue(issubclass(klass, IPv6_Ext)) + self.assertTrue(issubclass(klass, Internet)) + + def test_esp_extension_mode_is_no_longer_dead(self) -> None: + """GitHub issue #895: ``ESP`` accepted ``extension=``, stored it in + ``_extf`` and never read it. Subclassing the base makes ``_extf`` + load-bearing for + :attr:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext.payload` and + :attr:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext.protochain`, which + ``ESP`` inherits, and ``ESP.protocol`` supplies the third guard by hand. + """ + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.esp import ESP + from pcapkit.utilities.exceptions import UnsupportedCall + + wire = struct.pack('>II', 0x66, 1) + b'\x00' * 16 + inst = ESP(io.BytesIO(wire), len(wire), version=6, extension=True) + for attr in ('payload', 'protocol', 'protochain'): + with self.subTest(attr=attr, extension=True): + with self.assertRaises(UnsupportedCall): + getattr(inst, attr) + + # ... and with ``extension=False`` the same three stay open, so the + # guards read ``_extf`` rather than refusing unconditionally. + plain = ESP(io.BytesIO(wire), len(wire), version=6) + self.assertEqual(str(plain.protochain).split(':')[0], 'ESP') + self.assertEqual(plain.alias, 'ESP') + self.assertIs(ESP.__index__(), TransType.ESP) + + def test_base_role_protocol_delegates_past_the_fallback_answer(self) -> None: + """The trap this change had to avoid, pinned directly. + + Every subclass implements ``protocol`` as ``return super().protocol``, + and ``IPv6_Ext`` now sits between it and + :class:`~pcapkit.protocols.protocol.ProtocolBase` in the MRO. Without + the ``__data__`` discriminator in + :attr:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext.protocol`, that + ``super()`` lands on the *fallback-role* body, which reads + ``self._info.protocol`` -- a field none of the eight data models has. + Measured before the discriminator was added: ``AttributeError`` on all + eight. Both parenting shapes are exercised, since ``AH``/``ESP`` reach + the base through ``IPsec``. + """ + from pcapkit.const.ipv6.extension_header import ExtensionHeader + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.ah import AH + from pcapkit.protocols.internet.esp import ESP + from pcapkit.protocols.internet.hopopt import HOPOPT + from pcapkit.protocols.internet.mh import MH + + udp = _udp_bytes(b'hi') + cases = [ + (HOPOPT, bytes([int(TransType.UDP), 0]) + b'\x01\x04\x00\x00\x00\x00' + udp), + (MH, bytes([int(TransType.UDP), 0, 0, 0]) + b'\x00' * 4 + udp), + (AH, bytes([int(TransType.UDP), 2, 0, 0]) + b'\x00' * 12 + udp), + (ESP, struct.pack('>II', 0x66, 1) + b'\x00' * 16), + ] + for klass, wire in cases: + with self.subTest(klass=klass.__qualname__): + inst = klass(io.BytesIO(wire), len(wire), version=6) + protocol = inst.protocol + # the ``ProtocolBase`` meaning -- a chain entry name, i.e. a + # ``str`` -- never the fallback's ``ExtensionHeader`` identity. + self.assertNotIsInstance(protocol, ExtensionHeader) + self.assertIsInstance(protocol, str) + + def test_standalone_fallback_decodes_its_payload_and_reports_its_chain(self) -> None: + """With ``extension=False`` the fallback is an ordinary protocol: the + two guards open up and :meth:`read` continues into the next layer + instead of returning early. + """ + from pcapkit.const.ipv6.extension_header import ExtensionHeader + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext + from pcapkit.protocols.transport.udp import UDP + + raw = bytes([int(TransType.UDP), 0]) + b'\x00' * 6 + _udp_bytes(b'hi') + inst = IPv6_Ext(io.BytesIO(raw), len(raw), alias=int(ExtensionHeader.HOPOPT)) + + self.assertIsInstance(inst.payload, UDP) + self.assertEqual(str(inst.protochain), 'IPv6-Ext:UDP:Raw') + self.assertEqual(inst.protocol, ExtensionHeader.HOPOPT) + self.assertEqual(inst.length, 8) + + def test_make_emits_the_two_guaranteed_octets(self) -> None: + """:meth:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext.make` reads + ``next``/``len``/``payload`` out of ``**kwargs`` rather than declaring + them (see its docstring), so exercise the construction path that + actually supplies them. + """ + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext + + built = IPv6_Ext(next=TransType.TCP, len=0, payload=b'\x00' * 6, version=6) + self.assertEqual(bytes(built)[:2], bytes([int(TransType.TCP), 0])) + self.assertEqual(len(bytes(built)), 8) + + # the defaults, when the caller supplies nothing at all + default = IPv6_Ext(version=6) + self.assertEqual(bytes(default)[:2], bytes([int(TransType.UDP), 0])) + + def test_an_absent_or_unregistered_alias_uses_the_generic_rule(self) -> None: + """``alias`` names the code this instance stands in for. When it is + missing, or is a number IANA has not registered as an extension header, + :rfc:`6564#section-4`'s ``(octet[1] + 1) * 8`` is the best available + guess and ``protocol`` reports :data:`None` rather than inventing a + member. + """ + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext + + raw = bytes([int(TransType.UDP), 1]) + b'\x00' * 14 # (1+1)*8 == 16 + for label, kwargs in (('absent', {}), ('unregistered', {'alias': 200})): + with self.subTest(alias=label): + inst = IPv6_Ext(io.BytesIO(raw), len(raw), extension=True, **kwargs) + self.assertIsNone(inst.protocol) + self.assertEqual(inst.length, 16) + self.assertEqual(inst.next, TransType.UDP) + # ``read`` with no explicit length falls back to ``len(self)`` + self.assertEqual(inst.read().length, 16) + + def test_length_hint_is_the_two_octets_rfc_6564_guarantees(self) -> None: + """Two, not a real header length: those two octets are all + :rfc:`6564#section-4` promises of a header this class has never seen. + """ + from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext + + self.assertEqual(object.__new__(IPv6_Ext).__length_hint__(), 2) + + def test_next_is_now_shared_by_every_subclass(self) -> None: + """None of the eight declared a ``next`` property before #917, so + reading one raised :exc:`AttributeError`. They inherit the base's now, + which is sound because :rfc:`8200#section-4.1` puts a Next Header octet + first in every extension header and all eight record it under that name. + """ + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.hopopt import HOPOPT + from pcapkit.protocols.internet.ipv6_frag import IPv6_Frag + + hopopt_wire = bytes([int(TransType.UDP), 0]) + b'\x01\x04\x00\x00\x00\x00' + self.assertEqual( + HOPOPT(io.BytesIO(hopopt_wire), len(hopopt_wire), version=6, + extension=True).next, TransType.UDP) + + frag_wire = bytes([int(TransType.UDP), 0]) + struct.pack('>HI', 0, 0) + self.assertEqual( + IPv6_Frag(io.BytesIO(frag_wire), len(frag_wire), version=6, + extension=True).next, TransType.UDP) diff --git a/tests/protocols/internet/test_ipv6_generic_ext_unit.py b/tests/protocols/internet/test_ipv6_generic_ext_unit.py deleted file mode 100644 index 8cf7685a9c..0000000000 --- a/tests/protocols/internet/test_ipv6_generic_ext_unit.py +++ /dev/null @@ -1,368 +0,0 @@ -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 17fb0648d0..d63d948c07 100644 --- a/tests/protocols/internet/test_mh_unit.py +++ b/tests/protocols/internet/test_mh_unit.py @@ -1720,7 +1720,7 @@ def wrap_in_ipv4(mh_payload: bytes) -> IPv4: # 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` + # :class:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext` # 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 @@ -1736,14 +1736,14 @@ def wrap_in_ethernet_ipv6(mh_payload: bytes) -> Ethernet: return Ethernet(io.BytesIO(raw), len(raw)) from pcapkit.protocols.internet.ipv6 import IPv6 - from pcapkit.protocols.internet.ipv6_generic_ext import IPv6_GenericExt + from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext eth = wrap_in_ethernet_ipv6(bytes(fback_raw)) self.assertIsInstance(eth.payload, IPv6) - self.assertEqual(str(eth.protochain), 'Ethernet:IPv6:IPv6-GenericExt') + self.assertEqual(str(eth.protochain), 'Ethernet:IPv6:IPv6-Ext') genext = next(iter(eth.payload.extension_headers.items(multi=True)))[1] - self.assertIsInstance(genext, IPv6_GenericExt) + self.assertIsInstance(genext, IPv6_Ext) self.assertEqual(genext.next, TransType.UDP) self.assertEqual(genext.length, len(fback_raw)) diff --git a/tests/protocols/test_dispatch_registry_unit.py b/tests/protocols/test_dispatch_registry_unit.py index 1bb114bbfc..3c836a7276 100644 --- a/tests/protocols/test_dispatch_registry_unit.py +++ b/tests/protocols/test_dispatch_registry_unit.py @@ -222,7 +222,7 @@ def test_cases_cover_every_table_named_in_the_issue(self) -> None: cases = self.dispatch.cases() # #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 + # IPv6_Ext 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__ '