From abbc6ca2e7be3677169af43fb737b3d9f580f6ad Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Fri, 30 Oct 2026 23:23:56 -0400 Subject: [PATCH] fix(protocols): set NoPayload on terminal extension-header layers - HOPOPT, IPv6_Opts, IPv6_Route, IPv6_Frag, HIP, MH and the generic IPv6_Ext fallback (Shim6) return early in extension mode, before _decode_next_layer sets _next. 'x' in such a layer then raised AttributeError. - Hold NoPayload on that path: the upper layer belongs to the enclosing IPv6 header. - Regression tests build the frames in memory, one per class. --- pcapkit/protocols/internet/hip.py | 4 + pcapkit/protocols/internet/hopopt.py | 4 + pcapkit/protocols/internet/ipv6_ext.py | 4 + pcapkit/protocols/internet/ipv6_frag.py | 4 + pcapkit/protocols/internet/ipv6_opts.py | 4 + pcapkit/protocols/internet/ipv6_route.py | 4 + pcapkit/protocols/internet/mh.py | 4 + tests/protocols/internet/test_hip_unit.py | 32 ++++++++ .../protocols/internet/test_ipv6_ext_unit.py | 79 +++++++++++++++++++ tests/protocols/internet/test_mh_unit.py | 31 ++++++++ 10 files changed, 170 insertions(+) diff --git a/pcapkit/protocols/internet/hip.py b/pcapkit/protocols/internet/hip.py index 588361d644..08c76454ad 100644 --- a/pcapkit/protocols/internet/hip.py +++ b/pcapkit/protocols/internet/hip.py @@ -128,6 +128,7 @@ 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.misc.null import NoPayload 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 @@ -616,6 +617,9 @@ def read(self, length: 'Optional[int]' = None, *, extension: bool = False, **kwa ]) if extension: + # NOTE: As an IPv6 extension header the upper layer belongs to the enclosing + # IPv6 header, so hold NoPayload here so that membership does not raise (#2487). + self._next = NoPayload() # pylint: disable=attribute-defined-outside-init return hip # NOTE: ``_decode_next_layer`` builds the protocol chain from diff --git a/pcapkit/protocols/internet/hopopt.py b/pcapkit/protocols/internet/hopopt.py index 71f824ff1d..c72adf2555 100644 --- a/pcapkit/protocols/internet/hopopt.py +++ b/pcapkit/protocols/internet/hopopt.py @@ -65,6 +65,7 @@ TunnelEncapsulationLimitOption as Data_TunnelEncapsulationLimitOption from pcapkit.protocols.data.internet.hopopt import UnassignedOption as Data_UnassignedOption from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext +from pcapkit.protocols.misc.null import NoPayload 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 @@ -344,6 +345,9 @@ def read(self, length: 'Optional[int]' = None, *, extension: 'bool' = False, **k ) if extension: + # NOTE: As an IPv6 extension header the upper layer belongs to the enclosing + # IPv6 header, so hold NoPayload here so that membership does not raise (#2487). + self._next = NoPayload() # pylint: disable=attribute-defined-outside-init return hopopt return self._decode_next_layer(hopopt, schema.next, length - hopopt.length) diff --git a/pcapkit/protocols/internet/ipv6_ext.py b/pcapkit/protocols/internet/ipv6_ext.py index e4e84afeaf..8edf945028 100644 --- a/pcapkit/protocols/internet/ipv6_ext.py +++ b/pcapkit/protocols/internet/ipv6_ext.py @@ -148,6 +148,7 @@ class for that code instead (see the bottom of the module), so dispatch from pcapkit.const.reg.transtype import TransType as Enum_TransType from pcapkit.protocols.data.internet.ipv6_ext import IPv6_Ext as Data_IPv6_Ext from pcapkit.protocols.internet.internet import Internet +from pcapkit.protocols.misc.null import NoPayload from pcapkit.protocols.protocol import _PT, _ST, ProtocolBase from pcapkit.protocols.schema.internet.ipv6_ext import IPv6_Ext as Schema_IPv6_Ext from pcapkit.protocols.schema.schema import Schema @@ -486,6 +487,9 @@ def read(self, length: 'Optional[int]' = None, *, # pylint: disable=arguments-d ) if extension: + # NOTE: As an IPv6 extension header the upper layer belongs to the enclosing + # IPv6 header, so hold NoPayload here so that membership does not raise (#2487). + self._next = NoPayload() # pylint: disable=attribute-defined-outside-init return cast('_PT', generic_ext) return self._decode_next_layer(cast('_PT', generic_ext), next_header, length - ext_len, payload=rest[max(ext_len - 2, 0):]) diff --git a/pcapkit/protocols/internet/ipv6_frag.py b/pcapkit/protocols/internet/ipv6_frag.py index ccda4ec1c9..173f50d198 100644 --- a/pcapkit/protocols/internet/ipv6_frag.py +++ b/pcapkit/protocols/internet/ipv6_frag.py @@ -29,6 +29,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.ipv6_ext import IPv6_Ext +from pcapkit.protocols.misc.null import NoPayload from pcapkit.protocols.schema.internet.ipv6_frag import IPv6_Frag as Schema_IPv6_Frag from pcapkit.utilities.exceptions import UnsupportedCall @@ -158,6 +159,9 @@ def read(self, length: 'Optional[int]' = None, *, extension: 'bool' = False, # ) if extension: + # NOTE: As an IPv6 extension header the upper layer belongs to the enclosing + # IPv6 header, so hold NoPayload here so that membership does not raise (#2487). + self._next = NoPayload() # pylint: disable=attribute-defined-outside-init return ipv6_frag return self._decode_next_layer(ipv6_frag, schema.next, length - self.length) diff --git a/pcapkit/protocols/internet/ipv6_opts.py b/pcapkit/protocols/internet/ipv6_opts.py index f4d747d408..32181ad03e 100644 --- a/pcapkit/protocols/internet/ipv6_opts.py +++ b/pcapkit/protocols/internet/ipv6_opts.py @@ -65,6 +65,7 @@ TunnelEncapsulationLimitOption as Data_TunnelEncapsulationLimitOption from pcapkit.protocols.data.internet.ipv6_opts import UnassignedOption as Data_UnassignedOption from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext +from pcapkit.protocols.misc.null import NoPayload from pcapkit.protocols.schema.internet.ipv6_opts import CALIPSOOption as Schema_CALIPSOOption from pcapkit.protocols.schema.internet.ipv6_opts import \ HomeAddressOption as Schema_HomeAddressOption @@ -336,6 +337,9 @@ def read(self, length: 'Optional[int]' = None, *, extension: 'bool' = False, # ) if extension: + # NOTE: As an IPv6 extension header the upper layer belongs to the enclosing + # IPv6 header, so hold NoPayload here so that membership does not raise (#2487). + self._next = NoPayload() # pylint: disable=attribute-defined-outside-init return ipv6_opts return self._decode_next_layer(ipv6_opts, schema.next, length - ipv6_opts.length) diff --git a/pcapkit/protocols/internet/ipv6_route.py b/pcapkit/protocols/internet/ipv6_route.py index 8a41d150fd..87199b1e59 100644 --- a/pcapkit/protocols/internet/ipv6_route.py +++ b/pcapkit/protocols/internet/ipv6_route.py @@ -36,6 +36,7 @@ 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.ipv6_ext import IPv6_Ext +from pcapkit.protocols.misc.null import NoPayload 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 RoutingType as Schema_RoutingType @@ -234,6 +235,9 @@ def read(self, length: 'Optional[int]' = None, *, extension: 'bool' = False, # ipv6_route = meth(schema.data, header=schema) if extension: + # NOTE: As an IPv6 extension header the upper layer belongs to the enclosing + # IPv6 header, so hold NoPayload here so that membership does not raise (#2487). + self._next = NoPayload() # pylint: disable=attribute-defined-outside-init return ipv6_route return self._decode_next_layer(ipv6_route, schema.next, length - ipv6_route.length) diff --git a/pcapkit/protocols/internet/mh.py b/pcapkit/protocols/internet/mh.py index 6417c790e2..4b4fb5397c 100644 --- a/pcapkit/protocols/internet/mh.py +++ b/pcapkit/protocols/internet/mh.py @@ -285,6 +285,7 @@ UpdateNotificationMessage as Data_UpdateNotificationMessage from pcapkit.protocols.data.internet.mh import VendorSpecificOption as Data_VendorSpecificOption from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext +from pcapkit.protocols.misc.null import NoPayload from pcapkit.protocols.schema.internet.mh import MH as Schema_MH from pcapkit.protocols.schema.internet.mh import \ AccessNetworkIdentifierOption as Schema_AccessNetworkIdentifierOption @@ -1393,6 +1394,9 @@ def read(self, length: 'Optional[int]' = None, *, version: 'Literal[4, 6]' = 4, mh = meth(schema.data, header=schema) if extension: + # NOTE: As an IPv6 extension header the upper layer belongs to the enclosing + # IPv6 header, so hold NoPayload here so that membership does not raise (#2487). + self._next = NoPayload() # pylint: disable=attribute-defined-outside-init return mh return self._decode_next_layer(mh, schema.next, length - mh.length) diff --git a/tests/protocols/internet/test_hip_unit.py b/tests/protocols/internet/test_hip_unit.py index 5826c09963..1924c26e20 100644 --- a/tests/protocols/internet/test_hip_unit.py +++ b/tests/protocols/internet/test_hip_unit.py @@ -3556,5 +3556,37 @@ def test_hip_echo_response_unsigned_maker_builds_the_response_schema(self) -> No self.assertEqual(hints['return'], 'Schema_EchoResponseUnsignedParameter') +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class HIPExtensionMembershipTests(unittest.TestCase): + """``in`` on a terminal HIP layer parsed from an IPv6 chain (#2487). + + The HIP layer is taken from ``IPv6.extension_headers`` in extension mode. + ``in`` on it answers for its own name and does not raise. The frame is + built in memory; no capture is read. + """ + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_terminal_hip_membership_does_not_raise(self) -> None: + import io # pylint: disable=import-outside-toplevel + import struct # pylint: disable=import-outside-toplevel + import warnings # pylint: disable=import-outside-toplevel + + from pcapkit.protocols.internet.hip import HIP + from pcapkit.protocols.internet.ipv6 import IPv6 + + # HIP: next=59, Hdr Len 4 (40 octets), packet type 1, version 1, checksum, controls + hip_body = struct.pack('!BBBBHH', 59, 4, 0x01, 0x10, 0, 0) + bytes(32) + data = struct.pack('!IHBB', 6 << 28, len(hip_body), 139, 64) + bytes(32) + hip_body + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + ipv6 = IPv6(io.BytesIO(data), len(data)) + (hip,) = [ext for ext in ipv6.extension_headers.values() if isinstance(ext, HIP)] + + self.assertIn('HIP', hip) + self.assertNotIn('ESP', hip) + + if __name__ == '__main__': unittest.main() diff --git a/tests/protocols/internet/test_ipv6_ext_unit.py b/tests/protocols/internet/test_ipv6_ext_unit.py index 27cdcb63df..95f73ee524 100644 --- a/tests/protocols/internet/test_ipv6_ext_unit.py +++ b/tests/protocols/internet/test_ipv6_ext_unit.py @@ -946,3 +946,82 @@ def test_next_is_shared_by_every_subclass(self) -> None: self.assertEqual( IPv6_Frag(io.BytesIO(frag_wire), len(frag_wire), version=6, extension=True).next, TransType.UDP) + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class ExtensionLayerMembershipTests(unittest.TestCase): + """``in`` on a terminal extension-header layer parsed from an IPv6 chain (#2487). + + A layer taken from ``IPv6.extension_headers`` in extension mode holds + ``NoPayload``, so ``in`` answers for its own name and does not raise. Frames + are built in memory; no capture is read. + """ + + def setUp(self) -> None: + reimport_once_per_class(self) + + @staticmethod + def _ipv6(first: int, body: bytes) -> bytes: + return struct.pack('!IHBB', 6 << 28, len(body), first, 64) + bytes(32) + body + + def _parse(self, first: int, body: bytes): # type: ignore[no-untyped-def] + from pcapkit.protocols.internet.ipv6 import IPv6 + + data = self._ipv6(first, body) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + return IPv6(io.BytesIO(data), len(data)) + + def _layer(self, first: int, body: bytes, cls): # type: ignore[no-untyped-def] + ipv6 = self._parse(first, body) + return next(ext for ext in ipv6.extension_headers.values() if isinstance(ext, cls)) + + def test_terminal_hopopt_membership_does_not_raise(self) -> None: + from pcapkit.protocols.internet.hopopt import HOPOPT + + # next=59 (No Next Header), Hdr Ext Len 0, PadN(4) + body = struct.pack('!BB', 59, 0) + bytes([1, 4, 0, 0, 0, 0]) + hopopt = self._layer(0, body, HOPOPT) + + self.assertIn('HOPOPT', hopopt) + self.assertNotIn('ESP', hopopt) + + def test_terminal_ipv6_opts_membership_does_not_raise(self) -> None: + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + + body = struct.pack('!BB', 59, 0) + bytes([1, 4, 0, 0, 0, 0]) + opts = self._layer(60, body, IPv6_Opts) + + self.assertIn('IPv6_Opts', opts) + self.assertNotIn('ESP', opts) + + def test_terminal_ipv6_route_membership_does_not_raise(self) -> None: + from pcapkit.protocols.internet.ipv6_route import IPv6_Route + + # next=59, Hdr Ext Len 0, Routing Type 0, Segments Left 0, reserved + body = struct.pack('!BBBBI', 59, 0, 0, 0, 0) + route = self._layer(43, body, IPv6_Route) + + self.assertIn('IPv6_Route', route) + self.assertNotIn('ESP', route) + + def test_terminal_ipv6_frag_membership_does_not_raise(self) -> None: + from pcapkit.protocols.internet.ipv6_frag import IPv6_Frag + + body = struct.pack('!BBHI', 59, 0, 0, 0x1234) + frag = self._layer(44, body, IPv6_Frag) + + self.assertIn('IPv6_Frag', frag) + self.assertNotIn('ESP', frag) + + def test_terminal_shim6_membership_does_not_raise(self) -> None: + from pcapkit.const.ipv6.extension_header import ExtensionHeader + from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext + + # Shim6 (140) has no dedicated parser: the generic fallback handles it. + # next=59 (No Next Header), Hdr Ext Len 0, padding to 8 octets + body = struct.pack('!BB', 59, 0) + bytes(6) + shim6 = self._layer(int(ExtensionHeader.Shim6), body, IPv6_Ext) + + self.assertIn('IPv6_Ext', shim6) + self.assertNotIn('TCP', shim6) diff --git a/tests/protocols/internet/test_mh_unit.py b/tests/protocols/internet/test_mh_unit.py index 47ed4a2d3f..5f99124308 100644 --- a/tests/protocols/internet/test_mh_unit.py +++ b/tests/protocols/internet/test_mh_unit.py @@ -3647,5 +3647,36 @@ def test_mh_cga_parameters_option_parses(self) -> None: self.assertEqual(option.parameters[0].prefix, 0x20010db8) +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class MHExtensionMembershipTests(unittest.TestCase): + """``in`` on a terminal MH layer parsed from an IPv6 chain (#2487). + + The MH layer is taken from ``IPv6.extension_headers`` in extension mode. + ``in`` on it answers for its own name and does not raise. The frame is + built in memory; no capture is read. + """ + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_terminal_mh_membership_does_not_raise(self) -> None: + import struct # pylint: disable=import-outside-toplevel + import warnings # pylint: disable=import-outside-toplevel + + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.internet.mh import MH + + # MH: next=59, Hdr Len 0 (8 octets), MH type 0, reserved, checksum + mh_body = struct.pack('!BBBBH', 59, 0, 0, 0, 0) + bytes(4) + data = struct.pack('!IHBB', 6 << 28, len(mh_body), 135, 64) + bytes(32) + mh_body + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + ipv6 = IPv6(io.BytesIO(data), len(data)) + (mh,) = [ext for ext in ipv6.extension_headers.values() if isinstance(ext, MH)] + + self.assertIn('MH', mh) + self.assertNotIn('ESP', mh) + + if __name__ == '__main__': unittest.main()