diff --git a/pcapkit/protocols/internet/ipv6.py b/pcapkit/protocols/internet/ipv6.py index 475906526..f7c926d28 100644 --- a/pcapkit/protocols/internet/ipv6.py +++ b/pcapkit/protocols/internet/ipv6.py @@ -392,6 +392,21 @@ def _make_data(cls, data: 'Data_IPv6') -> 'dict[str, Any]': # type: ignore[over 'trailer': data.get('trailer', b''), } + @staticmethod + def _exthdr_key(alias: 'str') -> 'str': + """Key an extension header's info is written under in :class:`Data_IPv6`. + + Args: + alias: :attr:`~pcapkit.protocols.protocol.ProtocolBase.alias` of + the extension header's parser. + + Returns: + The key; :meth:`str.lstrip` strips a character *set*, which every + parser's alias is chosen to suit. + + """ + return alias.lstrip('IPv6-').lower() + @classmethod def _lookup_exthdr(cls, data: 'Data_IPv6') -> 'tuple[tuple[str, ...], tuple[tuple[Type[ProtocolBase], Any], ...]]': # pylint: disable=line-too-long """Recover the extension header chain of ``data`` without ``__exthdr__``. @@ -412,6 +427,9 @@ def _lookup_exthdr(cls, data: 'Data_IPv6') -> 'tuple[tuple[str, ...], tuple[tupl predecessor, and ``data.items(multi=True)`` yields each of them in turn. """ + from pcapkit.protocols.internet.ipv6_ext import \ + IPv6_Ext # isort: skip # pylint: disable=import-outside-toplevel + names = [] # type: list[str] chain = [] # type: list[tuple[Type[ProtocolBase], Any]] proto = data.get('next') @@ -429,8 +447,19 @@ def _lookup_exthdr(cls, data: 'Data_IPv6') -> 'tuple[tuple[str, ...], tuple[tupl # carrying ``next`` ends the chain, and so does a fragment header. if 'next' not in info: break + + # NOTE: A header whose dedicated parser raised was parsed by + # :class:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext` instead, + # c.f. :meth:`_import_next_layer`, and written under that class's + # key rather than the dedicated one's. Rebuilding it with the + # dedicated class would look for fields only that class parses + # (:issue:`1473`), so the key decides the parser. + if key == cls._exthdr_key(IPv6_Ext.alias.fget(None)): # type: ignore[attr-defined] + parser = IPv6_Ext # type: Type[ProtocolBase] + else: + parser = cls._lookup_next_layer(cls.__proto__, int(ex_proto)) names.append(key) - chain.append((cls._lookup_next_layer(cls.__proto__, int(ex_proto)), info)) + chain.append((parser, info)) proto = info['next'] if ex_proto == Enum_ExtensionHeader.IPv6_Frag: @@ -528,7 +557,7 @@ def _decode_next_layer(self, ipv6: 'Data_IPv6', proto: 'Optional[int]' = None, # keeps its predecessor's info rather than overwriting it, and # :meth:`Info.to_dict ` # writes every one of them, in wire order (:issue:`1453`). - name = next_.alias.lstrip('IPv6-').lower() + name = self._exthdr_key(next_.alias) ipv6.__update__(OrderedMultiDict([(name, info)])) # record protocol name diff --git a/pcapkit/utilities/decorators.py b/pcapkit/utilities/decorators.py index 878868df6..07974caa5 100644 --- a/pcapkit/utilities/decorators.py +++ b/pcapkit/utilities/decorators.py @@ -149,7 +149,16 @@ def behold(*args: 'P.args', **kwargs: 'P.kwargs') -> 'R_beholder': # turning a next-layer parse failure that should have degraded to # Raw into a crash. Reachable once something is registered on an # SCTP payload protocol identifier, which NGAP is. - file_ = self._get_payload() + # + # A ``payload`` passed to the decorated method is the octets that + # method was asked to parse, and wins: IPv6 hands its upper layer + # the payload *past* the extension headers, so falling back to the + # whole IPv6 payload would hold those headers twice, once in the + # extension header chain and once more in this ``Raw`` + # (:issue:`1471`). + file_ = cast('Optional[bytes]', kwargs.get('payload')) + if file_ is None: + file_ = self._get_payload() # NOTE: ``alias=proto`` matches what ``_import_next_layer`` passes, so # a payload that failed to parse still reports the code it arrived diff --git a/tests/protocols/internet/test_ipv6_exthdr_cut_rebuild_unit.py b/tests/protocols/internet/test_ipv6_exthdr_cut_rebuild_unit.py new file mode 100644 index 000000000..7b268426c --- /dev/null +++ b/tests/protocols/internet/test_ipv6_exthdr_cut_rebuild_unit.py @@ -0,0 +1,118 @@ +# -*- coding: utf-8 -*- +"""A cut IPv6 extension header chain rebuilds byte-exact, from info and dict. + +GitHub issue #1471: an upper layer cut short after an extension header fell +back to :class:`~pcapkit.protocols.misc.raw.Raw` over the *whole* IPv6 payload, +since :func:`~pcapkit.utilities.decorators.beholder` ignored the ``payload`` it +was handed, so the rebuild wrote the extension headers twice -- Destination +Options, TCP cut at 57 octets rebuilt 65. + +GitHub issue #1473: a header whose dedicated parser raised is parsed by +:class:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext` and written under its +``ext`` key, but the ``to_dict`` rebuild chose the dedicated class from the +``next`` code, which then missed fields only it parses -- ``AttributeError: +'HOPOPT' object has no attribute 'options'``. + +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 struct +import unittest +import warnings + +from tests._support import reimport_once_per_class + +HOP, TCP, DST, AH, NONXT, HIP = 0, 6, 60, 51, 59, 139 + +#: A TCP header and one octet of data. +SEGMENT = struct.pack('!HHIIBBHHH', 1, 2, 0, 0, 0x50, 2, 0, 0, 0) + b'x' + + +def opts(nxt: int) -> bytes: + """An 8-octet Hop-by-Hop or Destination Options header: one PadN.""" + return bytes([nxt, 0, 1, 4, 0, 0, 0, 0]) + + +def auth(nxt: int) -> bytes: + """A 24-octet Authentication Header with a 12-octet ICV.""" + return bytes([nxt, 4, 0, 0]) + bytes(20) + + +def hip(nxt: int) -> bytes: + """A 40-octet HIPv2 header with no parameters.""" + return bytes([nxt, 4, 1, 0x21, 0, 0, 0, 0]) + bytes(32) + + +#: First next-header code and IPv6 payload of each chain. +CHAINS = { + 'dst-tcp': (DST, opts(TCP) + SEGMENT), + 'hop-tcp': (HOP, opts(TCP) + SEGMENT), + 'hop-dst-tcp': (HOP, opts(DST) + opts(TCP) + SEGMENT), + 'dst-ah-tcp': (DST, opts(AH) + auth(TCP) + SEGMENT), + 'hop-nonxt': (HOP, bytes([NONXT, 1, 1, 12]) + bytes(12)), + 'hip': (HIP, hip(NONXT)), + 'dst-hip': (DST, opts(HIP) + hip(NONXT)), +} + + +def packet(name: str) -> bytes: + """The whole IPv6 packet of ``CHAINS[name]``.""" + code, body = CHAINS[name] + return struct.pack('!IHBB', 6 << 28, len(body), code, 64) + bytes(32) + body + + +class TestIPv6ExtensionHeaderCutRebuild(unittest.TestCase): + """Pin the rebuild of an extension header chain cut at every octet.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + quiet = warnings.catch_warnings() + quiet.__enter__() # pylint: disable=unnecessary-dunder-call + self.addCleanup(quiet.__exit__, None, None, None) + warnings.simplefilter('ignore') + + def test_every_cut_rebuilds_from_info_and_dict(self) -> None: + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.link.ethernet import Ethernet + + for name in CHAINS: + full = packet(name) + for cut in range(40, len(full) + 1): + for cls, data in ((IPv6, full[:cut]), (Ethernet, bytes(12) + b'\x86\xdd' + full[:cut])): + proto = cls(data, len(data)) + with self.subTest(chain=name, cut=cut, layer=cls.__name__, form='parse'): + self.assertEqual(proto.data, data) + with self.subTest(chain=name, cut=cut, layer=cls.__name__, form='info'): + self.assertEqual(cls.from_data(proto.info).data, data) + with self.subTest(chain=name, cut=cut, layer=cls.__name__, form='dict'): + self.assertEqual(cls.from_data(proto.info.to_dict()).data, data) + + def test_raw_upper_layer_starts_past_the_extension_headers(self) -> None: + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.misc.raw import Raw + + data = packet('dst-tcp')[:57] + ip6 = IPv6(data, len(data)) + self.assertIsInstance(ip6.payload, Raw) + self.assertEqual(ip6.payload.data, data[48:]) + self.assertEqual(len(IPv6.from_data(ip6.info).data), 57) + + def test_dict_rebuild_keeps_the_generic_parser(self) -> None: + from pcapkit.const.ipv6.extension_header import ExtensionHeader + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext + + for cut in (48, 50, 52, 55): + data = packet('hop-nonxt')[:cut] + with self.subTest(cut=cut): + ip6 = IPv6(data, len(data)) + self.assertIs(type(ip6._exthdr[ExtensionHeader.HOPOPT]), IPv6_Ext) + _, chain = IPv6._lookup_exthdr(ip6.info.to_dict()) + self.assertEqual([parser for parser, _ in chain], [IPv6_Ext]) + self.assertEqual(IPv6.from_data(ip6.info.to_dict()).data, data) + + +if __name__ == '__main__': + unittest.main()