From 2e9ecd395bb60e4653a40602747872adc846e0c5 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Wed, 4 Nov 2026 12:38:47 -0500 Subject: [PATCH 1/2] fix(internet): rebuild a whole extension header in the extension mode its parse used (#2768) Parsed with extension=True, as IPv6 parses each header of its chain, every extension header but ESP holds no next layer, yet from_data(info) rebuilt it without extension, so its info gained __next_type__, __next_name__ and nopayload. Internet.from_data now rebuilds an IPv6_Ext subclass whose data model holds no next layer with extension, unless the caller passes it. Shim6's protocol, which IPv6_Ext.read takes from the parse-only alias keyword, is left for a decision on how a rebuild may pass that keyword. Tests: test_ipv6_exthdr_extension_rebuild_2768_unit covers every member, standalone and from IPv6's __exthdr__, through info and to_dict(). --- pcapkit/protocols/internet/internet.py | 14 ++ ...ipv6_exthdr_extension_rebuild_2768_unit.py | 121 ++++++++++++++++++ 2 files changed, 135 insertions(+) create mode 100644 tests/protocols/internet/test_ipv6_exthdr_extension_rebuild_2768_unit.py diff --git a/pcapkit/protocols/internet/internet.py b/pcapkit/protocols/internet/internet.py index e189592b6b..253b4e43de 100644 --- a/pcapkit/protocols/internet/internet.py +++ b/pcapkit/protocols/internet/internet.py @@ -240,6 +240,10 @@ def register(cls, code: 'Enum_TransType', protocol: 'ModuleDescriptor[ProtocolBa def from_data(cls, data: '_PT | dict[str, Any]', **kwargs: 'Any') -> 'Self': """Create protocol instance from data. + An IPv6 extension header parsed alone always holds a next layer in its + data model, so one whose data model holds none is rebuilt with + ``extension``, unless ``kwargs`` give it. + Args: data: Protocol data. **kwargs: Construction keywords, as for :meth:`Protocol.from_data @@ -249,7 +253,17 @@ def from_data(cls, data: '_PT | dict[str, Any]', **kwargs: 'Any') -> 'Self': Protocol instance, cut back to what was captured when the header it was parsed from was cut short (:issue:`1458`). + .. versionchanged:: 1.6.0 + + Such a header is rebuilt with ``extension``, rather than with a next + layer its parse did not hold (:issue:`2768`). + """ + # NOTE: Parsed with ``extension``, every header but ESP leaves its next layer + # to the IPv6 header walking it; ESP's data model is the same either way (#2768). + from pcapkit.protocols.internet.ipv6_ext.protocol import IPv6_Ext # isort: skip # pylint: disable=import-outside-toplevel + if issubclass(cls, IPv6_Ext) and data.get('__next_type__') is None: + kwargs.setdefault('extension', True) return replay_short_read(super().from_data(data, **kwargs), data) ########################################################################## diff --git a/tests/protocols/internet/test_ipv6_exthdr_extension_rebuild_2768_unit.py b/tests/protocols/internet/test_ipv6_exthdr_extension_rebuild_2768_unit.py new file mode 100644 index 0000000000..3b9e905a13 --- /dev/null +++ b/tests/protocols/internet/test_ipv6_exthdr_extension_rebuild_2768_unit.py @@ -0,0 +1,121 @@ +# -*- coding: utf-8 -*- +"""A whole IPv6 extension header rebuilds in the extension mode its parse used. + +GitHub issue #2768: parsed with ``extension=True``, as +:class:`~pcapkit.protocols.internet.ipv6.protocol.IPv6` parses each header of its +chain, every extension header but ESP holds no next layer. ``from_data(info)`` +rebuilt it without ``extension``, so its info gained ``__next_type__``, +``__next_name__`` and ``nopayload``, standalone and from the ``__exthdr__`` infos +of IPv6 alike. +:meth:`Internet.from_data ` +rebuilds a header whose data model holds no next layer with ``extension``. + +Every case builds its own octets and reads no capture. Classes are imported +inside each test, after :func:`~tests._support.reimport_once_per_class`. + +""" + +import unittest +from typing import TYPE_CHECKING + +from tests._support import reimport_once_per_class + +if TYPE_CHECKING: + from typing import Any + +#: Test tier (#1958): hand-picked round trips, one header per class. +__tier__ = 'ALL' + +#: IPv6 Next Header code of each member of the family, as IANA's IPv6 Extension +#: Header Types registry assigns it. IPv6 reaches ``IPv6_Ext`` only as Shim6 (140), +#: whose code it hands over as ``alias``, which the rebuild does not replay +#: (#2768), so ``IPv6_Ext`` is rebuilt here only standalone. +CODES = { + 'HOPOPT': 0, + 'IPv6_Route': 43, + 'IPv6_Frag': 44, + 'ESP': 50, + 'AH': 51, + 'IPv6_Opts': 60, + 'MH': 135, + 'HIP': 139, +} + + +class TestExtensionModeRebuild(unittest.TestCase): + """Pin ``from_data(info).info == info`` for a header parsed with ``extension``.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + @staticmethod + def _family() -> 'dict[str, Any]': + """``IPv6_Ext`` and every class beneath it, every member module imported first.""" + from pcapkit.protocols import internet + from pcapkit.protocols.internet.ipv6_ext.protocol import IPv6_Ext + + # NOTE: the package imports each protocol on first use (#2730). + for name in internet.__all__: + getattr(internet, name) + found = {'IPv6_Ext': IPv6_Ext} # type: dict[str, Any] + stack = list(IPv6_Ext.__subclasses__()) + while stack: + klass = stack.pop() + if klass.__name__ not in found: + found[klass.__name__] = klass + stack.extend(klass.__subclasses__()) + return found + + @staticmethod + def _header(klass: 'Any') -> bytes: + """A whole header of ``klass``, followed by No Next Header.""" + extra = {'data': bytes(6)} if klass.__name__ == 'IPv6_Ext' else {} + return klass(next=59, **extra).data + + def _assert_rebuilds(self, klass: 'Any', info: 'Any', data: bytes) -> None: + """``klass`` rebuilds ``info`` and ``data`` from ``info`` and its dict alike.""" + for source, value in (('info', info), ('to_dict', info.to_dict())): + with self.subTest(source=source): + rebuilt = klass.from_data(value) + self.assertEqual(rebuilt.info, info) + self.assertEqual(rebuilt.data, data) + + def test_every_member_has_a_code(self) -> None: + self.assertEqual(set(self._family()) - {'IPv6_Ext'}, set(CODES)) + + def test_standalone_header_rebuilds_its_info(self) -> None: + for name, klass in sorted(self._family().items()): + with self.subTest(header=name): + header = self._header(klass) + self._assert_rebuilds(klass, klass(header, extension=True).info, header) + + def test_chain_header_rebuilds_its_info(self) -> None: + from pcapkit.protocols.internet.ipv6.protocol import IPv6 + + family = self._family() + for name, code in sorted(CODES.items()): + with self.subTest(header=name): + klass = family[name] + header = self._header(klass) + built = IPv6(next=code, src='2001:db8::1', dst='2001:db8::2', payload=header) + ((proto, info),) = IPv6(built.data).info['__exthdr__'] + self.assertIs(proto, klass) + self._assert_rebuilds(klass, info, header) + + def test_header_parsed_alone_keeps_its_next_layer(self) -> None: + for name, klass in sorted(self._family().items()): + with self.subTest(header=name): + header = self._header(klass) + parsed = klass(header) + self.assertIn('__next_type__', parsed.info) + self._assert_rebuilds(klass, parsed.info, header) + + def test_keywords_override_the_replayed_mode(self) -> None: + for name, klass in sorted(self._family().items()): + with self.subTest(header=name): + parsed = klass(self._header(klass), extension=True) + self.assertIn('__next_type__', klass.from_data(parsed.info, extension=False).info) + + +if __name__ == '__main__': + unittest.main() From ea6b7e564a93c6d532baa5e54853c2dc452496d1 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Wed, 4 Nov 2026 13:08:52 -0500 Subject: [PATCH 2/2] fix(internet): keep a rebuilt Shim6 header's protocol through the alias keyword (#2768) IPv6_Ext.read takes Shim6's protocol from the alias keyword, which a from_data rebuild never passed, so the rebuilt protocol was None. IPv6_Ext._make_data now returns alias=data.protocol, and IPv6_Ext lists alias in __keywords__ so the construction check lets it through. Every member accepts and drops alias, as it does while parsing. Tests: the #2768 module adds the Shim6 alias and in-IPv6 cases, asserts no construction-keyword warning on rebuild, and pins a member dropping alias. --- .../protocols/internet/ipv6_ext/protocol.py | 23 +++++-- ...ipv6_exthdr_extension_rebuild_2768_unit.py | 60 ++++++++++++++++--- 2 files changed, 70 insertions(+), 13 deletions(-) diff --git a/pcapkit/protocols/internet/ipv6_ext/protocol.py b/pcapkit/protocols/internet/ipv6_ext/protocol.py index f22b687933..671155d98d 100644 --- a/pcapkit/protocols/internet/ipv6_ext/protocol.py +++ b/pcapkit/protocols/internet/ipv6_ext/protocol.py @@ -241,6 +241,9 @@ class keywords implement it. code that does not fit in one octet raises :exc:`~pcapkit.utilities.exceptions.ProtocolError`, rather than being handed to the schema unchecked (:issue:`2594`). + * :meth:`from_data` keeps :attr:`protocol`, handing :meth:`read` the + ``alias`` its parse took, rather than rebuilding it as :data:`None` + (:issue:`2768`). """ @@ -250,6 +253,12 @@ class keywords implement it. #: does not declare them would accept and then drop them (:issue:`2286`). __own_make_keywords__ = True + #: :meth:`read` takes ``alias`` out of ``**kwargs`` by name, and + #: :meth:`_make_data` hands it back, so a rebuild keeps the :attr:`protocol` + #: its parse resolved (:issue:`2768`). Every subclass accepts it as well, + #: and drops it, as it does while parsing. + __keywords__ = frozenset({'alias'}) + ########################################################################## # Properties. ########################################################################## @@ -555,9 +564,12 @@ def make(self, *, Nothing calls a protocol's ``make`` positionally; ``__init__`` spreads ``**kwargs`` into it. - ``read`` needs none of this: its ``alias``/``error`` only arrive on - the *parse* path, which ``__init__`` exempts from the construction - check. + ``read`` needs none of this for ``error``: it arrives on the + *parse* path alone, which ``__init__`` exempts from the + construction check. ``alias`` arrives on a rebuild too, from + :meth:`_make_data`, so this class lists it in + :attr:`~pcapkit.protocols.protocol.Protocol.__keywords__` + (:issue:`2768`). """ # NOTE: The schema keeps everything after the two fixed octets as one @@ -688,7 +700,9 @@ def _make_data(cls, data: 'Data_IPv6_Ext') -> 'dict[str, Any]': # type: ignore[ Emits the ``Hdr Ext Len`` octet and the body as they were read, so the rebuild is byte-exact whichever per-protocol length rule :meth:`read` applied, and also for an overrun, where ``data.next`` is - :data:`None` and ``data.declared_next`` keeps the octet. + :data:`None` and ``data.declared_next`` keeps the octet. Hands + :meth:`read` the ``protocol`` it resolved as ``alias``, which picks + that rule and the identity again (:issue:`2768`). Args: data: protocol data @@ -702,6 +716,7 @@ def _make_data(cls, data: 'Data_IPv6_Ext') -> 'dict[str, Any]': # type: ignore[ 'len': data.len, 'data': data.data, 'payload': cls._make_payload(data), + 'alias': data.protocol, } diff --git a/tests/protocols/internet/test_ipv6_exthdr_extension_rebuild_2768_unit.py b/tests/protocols/internet/test_ipv6_exthdr_extension_rebuild_2768_unit.py index 3b9e905a13..b69e852c46 100644 --- a/tests/protocols/internet/test_ipv6_exthdr_extension_rebuild_2768_unit.py +++ b/tests/protocols/internet/test_ipv6_exthdr_extension_rebuild_2768_unit.py @@ -10,12 +10,18 @@ :meth:`Internet.from_data ` rebuilds a header whose data model holds no next layer with ``extension``. +Shim6, which :class:`~pcapkit.protocols.internet.ipv6_ext.protocol.IPv6_Ext` parses, +also lost the ``protocol`` its ``read`` takes from the ``alias`` keyword. Its +``_make_data`` hands ``alias`` back, and its ``__keywords__`` lets the keyword +through the construction check, which every member then accepts and drops. + Every case builds its own octets and reads no capture. Classes are imported inside each test, after :func:`~tests._support.reimport_once_per_class`. """ import unittest +import warnings from typing import TYPE_CHECKING from tests._support import reimport_once_per_class @@ -27,9 +33,8 @@ __tier__ = 'ALL' #: IPv6 Next Header code of each member of the family, as IANA's IPv6 Extension -#: Header Types registry assigns it. IPv6 reaches ``IPv6_Ext`` only as Shim6 (140), -#: whose code it hands over as ``alias``, which the rebuild does not replay -#: (#2768), so ``IPv6_Ext`` is rebuilt here only standalone. +#: Header Types registry assigns it; ``IPv6_Ext`` parses Shim6. IPv6 hands each +#: header its code as ``alias``, and the standalone parses here do the same. CODES = { 'HOPOPT': 0, 'IPv6_Route': 43, @@ -39,6 +44,7 @@ 'IPv6_Opts': 60, 'MH': 135, 'HIP': 139, + 'IPv6_Ext': 140, } @@ -73,21 +79,33 @@ def _header(klass: 'Any') -> bytes: return klass(next=59, **extra).data def _assert_rebuilds(self, klass: 'Any', info: 'Any', data: bytes) -> None: - """``klass`` rebuilds ``info`` and ``data`` from ``info`` and its dict alike.""" + """``klass`` rebuilds ``info`` and ``data`` from ``info`` and its dict alike. + + A keyword ``_make_data`` returns that no signature declares is warned of + and fails the case too. + + """ + from pcapkit.utilities.warnings import UnknownFieldWarning + for source, value in (('info', info), ('to_dict', info.to_dict())): with self.subTest(source=source): - rebuilt = klass.from_data(value) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + rebuilt = klass.from_data(value) + self.assertEqual([str(item.message) for item in caught + if issubclass(item.category, UnknownFieldWarning)], []) self.assertEqual(rebuilt.info, info) self.assertEqual(rebuilt.data, data) def test_every_member_has_a_code(self) -> None: - self.assertEqual(set(self._family()) - {'IPv6_Ext'}, set(CODES)) + self.assertEqual(set(self._family()), set(CODES)) def test_standalone_header_rebuilds_its_info(self) -> None: for name, klass in sorted(self._family().items()): with self.subTest(header=name): header = self._header(klass) - self._assert_rebuilds(klass, klass(header, extension=True).info, header) + parsed = klass(header, extension=True, alias=CODES[name]) + self._assert_rebuilds(klass, parsed.info, header) def test_chain_header_rebuilds_its_info(self) -> None: from pcapkit.protocols.internet.ipv6.protocol import IPv6 @@ -106,16 +124,40 @@ def test_header_parsed_alone_keeps_its_next_layer(self) -> None: for name, klass in sorted(self._family().items()): with self.subTest(header=name): header = self._header(klass) - parsed = klass(header) + parsed = klass(header, alias=CODES[name]) self.assertIn('__next_type__', parsed.info) self._assert_rebuilds(klass, parsed.info, header) def test_keywords_override_the_replayed_mode(self) -> None: for name, klass in sorted(self._family().items()): with self.subTest(header=name): - parsed = klass(self._header(klass), extension=True) + parsed = klass(self._header(klass), extension=True, alias=CODES[name]) self.assertIn('__next_type__', klass.from_data(parsed.info, extension=False).info) + def test_shim6_keeps_its_protocol(self) -> None: + from pcapkit.const.ipv6.extension_header import ExtensionHeader + from pcapkit.protocols.internet.ipv6_ext.protocol import IPv6_Ext + + header = self._header(IPv6_Ext) + parsed = IPv6_Ext(header, extension=True, alias=CODES['IPv6_Ext']) # type: Any + self.assertIs(parsed.info.protocol, ExtensionHeader.Shim6) + for source, value in (('info', parsed.info), ('to_dict', parsed.info.to_dict())): + with self.subTest(source=source): + self.assertIs(IPv6_Ext.from_data(value).info.protocol, ExtensionHeader.Shim6) + + def test_a_member_accepts_and_drops_alias(self) -> None: + from pcapkit.utilities.warnings import UnknownFieldWarning + + for name, klass in sorted(self._family().items()): + if name == 'IPv6_Ext': + continue + with self.subTest(header=name): + with warnings.catch_warnings(): + warnings.simplefilter('error', UnknownFieldWarning) + built = klass(next=59, alias=CODES['IPv6_Ext']) + self.assertEqual(built.data, self._header(klass)) + self.assertEqual(built.info, klass(next=59).info) + if __name__ == '__main__': unittest.main()