From 254034005005a9b630601291777b8a5390fe2d68 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Sat, 10 Oct 2026 12:40:55 -0400 Subject: [PATCH] fix(protocols): bound the IPv6 extension-header chain at 32 headers (#1604) - Inside IPv4, each extension header dissected the next recursively, so a long chain hit the recursion limit and beholder left the rest raw, silently, at a depth that varied by interpreter; IPv6's own walk had no bound at all. EXTENSION_HEADER_LIMIT (32) now caps both: past it the rest stays Raw, with one ProtocolWarning. - pypcapfile's _past_extension_headers and _upper_layer stop at the same depth. - dpkt, scapy and tshark walk the chain in their own parsers; where they disagree is recorded as Gap rows in a new agreement module (#1609, #1606). --- .../pcapkit/protocols/internet/internet.rst | 17 ++ pcapkit/protocols/internet/internet.py | 127 ++++++++- pcapkit/protocols/internet/ipv6.py | 36 ++- pcapkit/toolkit/pypcapfile.py | 19 +- ...est_ext_chain_limit_agreement_1604_unit.py | 245 ++++++++++++++++++ .../test_ipv6_ext_chain_limit_1604_unit.py | 216 +++++++++++++++ ...st_pypcapfile_ext_chain_limit_1604_unit.py | 149 +++++++++++ 7 files changed, 779 insertions(+), 30 deletions(-) create mode 100644 tests/foundation/engines/test_ext_chain_limit_agreement_1604_unit.py create mode 100644 tests/protocols/internet/test_ipv6_ext_chain_limit_1604_unit.py create mode 100644 tests/toolkit/test_pypcapfile_ext_chain_limit_1604_unit.py diff --git a/docs/source/pcapkit/protocols/internet/internet.rst b/docs/source/pcapkit/protocols/internet/internet.rst index 54fd11b296..507dfc5808 100644 --- a/docs/source/pcapkit/protocols/internet/internet.rst +++ b/docs/source/pcapkit/protocols/internet/internet.rst @@ -18,8 +18,25 @@ which is a base class for internet layer protocols, eg. :class:`~pcapkit.protoco .. automethod:: _decode_next_layer .. automethod:: _import_next_layer + .. automethod:: _next_layer_class + .. automethod:: _next_exthdr_depth .. autoattribute:: __layer__ :no-value: .. autoattribute:: __proto__ :no-value: + .. autoattribute:: _exthdr_depth + +Extension Header Chains +----------------------- + +A chain of IPv6 extension headers is dissected to at most +:data:`~pcapkit.protocols.internet.internet.EXTENSION_HEADER_LIMIT` headers in +a row, whether it follows IPv6 or another layer such as IPv4. What follows is +kept as :class:`~pcapkit.protocols.misc.raw.Raw`, with one +:exc:`~pcapkit.utilities.warnings.ProtocolWarning` (:issue:`1604`). The +``pypcapfile`` toolkit stops at the same depth. The ``dpkt``, ``scapy`` and +``pyshark`` engines leave the walk to their own parsers, which do not stop +there, so they may still find the upper layer past it (:issue:`1609`). + +.. autodata:: pcapkit.protocols.internet.internet.EXTENSION_HEADER_LIMIT diff --git a/pcapkit/protocols/internet/internet.py b/pcapkit/protocols/internet/internet.py index 7cb51dbe0e..c604632024 100644 --- a/pcapkit/protocols/internet/internet.py +++ b/pcapkit/protocols/internet/internet.py @@ -15,13 +15,15 @@ import io from typing import TYPE_CHECKING, Generic, cast +from pcapkit.const.ipv6.extension_header import ExtensionHeader as Enum_ExtensionHeader from pcapkit.const.reg.transtype import TransType as Enum_TransType from pcapkit.corekit.module import ModuleDescriptor from pcapkit.corekit.protochain import ProtoChain from pcapkit.protocols.protocol import _PT, _ST, ProtocolBase from pcapkit.protocols.schema.schema import keep_short_read, replay_short_read from pcapkit.utilities.decorators import beholder -from pcapkit.utilities.warnings import RegistryWarning, warn +from pcapkit.utilities.exceptions import stacklevel +from pcapkit.utilities.warnings import ProtocolWarning, RegistryWarning, warn if TYPE_CHECKING: from typing import IO, Any, Optional, Type @@ -30,6 +32,24 @@ __all__ = ['Internet'] +#: Most IPv6 extension headers dissected in one chain, i.e. one after another +#: with no other layer between them (:issue:`1604`). The walk stops there, and +#: what follows the last of them is kept as +#: :class:`~pcapkit.protocols.misc.raw.Raw`, with one +#: :exc:`~pcapkit.utilities.warnings.ProtocolWarning`. That holds whether the +#: chain follows IPv6, which walks it in a loop +#: (:meth:`IPv6._decode_next_layer +#: `), or another +#: layer such as IPv4, where each header dissects the next one itself +#: (:meth:`Internet._import_next_layer`). The toolkit adapters that walk a +#: chain themselves stop at the same depth. +#: +#: :rfc:`8200#section-4.1` allows each extension header at most once, and the +#: Destination Options header at most twice, so no compliant chain gets close. +#: Without the bound, a chain dissected header by header stops wherever +#: Python's recursion limit falls, which depends on the interpreter. +EXTENSION_HEADER_LIMIT = 32 + class Internet(ProtocolBase[_PT, _ST], Generic[_PT, _ST]): # pylint: disable=abstract-method """Abstract base class for internet layer protocol family. @@ -118,6 +138,12 @@ class Internet(ProtocolBase[_PT, _ST], Generic[_PT, _ST]): # pylint: disable=ab }, ) + #: int: How many IPv6 extension headers in a row the next layer follows: + #: this layer's place in such a chain, or how many headers + #: :class:`~pcapkit.protocols.internet.ipv6.IPv6` walked, else ``0``, c.f. + #: :data:`EXTENSION_HEADER_LIMIT` and :meth:`_next_exthdr_depth`. + _exthdr_depth = 0 + ########################################################################## # Properties. ########################################################################## @@ -197,6 +223,10 @@ def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None, file: Source packet stream. length: Length of packet data. extension: If the protocol is used as an IPv6 extension header. + _exthdr_depth (int): Where this layer is in a chain of IPv6 + extension headers, counting from ``1``, or ``0`` if it is not + in one. Handed over by :meth:`_import_next_layer` while + parsing, c.f. :data:`EXTENSION_HEADER_LIMIT`. **kwargs: Arbitrary keyword arguments. Notes: @@ -211,6 +241,9 @@ def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None, (:issue:`1458`). """ + # NOTE: Set before parsing, since parsing is what dissects the next layer. + self._exthdr_depth = cast('int', kwargs.pop('_exthdr_depth', 0)) + super().__post_init__(file, length, extension=extension, **kwargs) # type: ignore[arg-type] keep_short_read(self) @@ -237,6 +270,77 @@ def _read_protos(self, size: 'int') -> 'Enum_TransType': _prot = Enum_TransType.get(_byte) return _prot + def _next_exthdr_depth(self, proto: 'Optional[int]', extension: 'bool' = False) -> 'int': + """Place the next layer in its chain of IPv6 extension headers. + + Arguments: + proto: next layer protocol index + extension: if the next layer is parsed as an extension header by + :class:`~pcapkit.protocols.internet.ipv6.IPv6`'s walk, which + counts the chain itself, c.f. :meth:`IPv6._decode_next_layer + ` + + Returns: + One more than :attr:`_exthdr_depth` if ``proto`` names an IPv6 + extension header, or ``0`` if it names none, so that another layer + between two headers starts a new chain. Past + :data:`EXTENSION_HEADER_LIMIT`, the next layer is left raw. + + """ + if extension: + return 0 + try: + Enum_ExtensionHeader(proto) + except ValueError: + return 0 + return self._exthdr_depth + 1 + + def _next_layer_class(self, proto: 'Optional[int]', file_: 'bytes', length: 'int', + extension: 'bool' = False) -> 'tuple[Type[ProtocolBase], dict[str, Any]]': + r"""Pick the class the next layer is parsed with. + + Arguments: + proto: next layer protocol index + file\_: payload of the current layer + length: valid (*non-padding*) length + extension: if is extension header, c.f. :meth:`_next_exthdr_depth` + + Returns: + :class:`~pcapkit.protocols.misc.null.NoPayload` if there is + nothing to parse; :class:`~pcapkit.protocols.misc.raw.Raw` if the + parse stops at this layer, or the next layer is an IPv6 extension + header past :data:`EXTENSION_HEADER_LIMIT` in a row; otherwise the + class registered for ``proto``. With it, the keywords only that next + layer is constructed with: the warning given, as ``error``, or the + next layer's place in its chain, as ``_exthdr_depth``. + + Warns: + ProtocolWarning: If the next layer is past + :data:`EXTENSION_HEADER_LIMIT` (:issue:`1604`). + + """ + if min(length, len(file_)) == 0: + from pcapkit.protocols.misc.null import NoPayload # isort: skip # pylint: disable=import-outside-toplevel + return NoPayload, {} + if self._sigterm: + from pcapkit.protocols.misc.raw import Raw # isort: skip # pylint: disable=import-outside-toplevel + return Raw, {} + + depth = self._next_exthdr_depth(proto, extension) + if depth > EXTENSION_HEADER_LIMIT: + from pcapkit.protocols.misc.raw import Raw # isort: skip # pylint: disable=import-outside-toplevel + + message = (f'{self.alias}: a chain of IPv6 extension headers reached the limit of ' + f'{EXTENSION_HEADER_LIMIT} headers; the rest of the packet, from Next ' + f'Header {int(cast("int", proto))}, is left undissected') + warn(message, ProtocolWarning, stacklevel=stacklevel()) + return Raw, {'error': message} + + protocol = self._lookup_next_layer(self.__proto__, cast('int', proto)) + if depth and issubclass(protocol, Internet): + return protocol, {'_exthdr_depth': depth} + return protocol, {} + def _decode_next_layer(self, dict_: '_PT', proto: 'Optional[int]' = None, # pylint: disable=arguments-differ length: 'Optional[int]' = None, *, packet: 'Optional[dict[str, Any]]' = None, version: 'Literal[4, 6]' = 4, ipv6_exthdr: 'Optional[ProtoChain]' = None, @@ -304,10 +408,15 @@ def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *, # Returns: Instance of next layer. - """ - if TYPE_CHECKING: - protocol: 'Type[ProtocolBase]' + Notes: + An IPv6 extension header past :data:`EXTENSION_HEADER_LIMIT` in a + row is not dissected: it and the rest of the packet are kept as + :class:`~pcapkit.protocols.misc.raw.Raw`, with one + :exc:`~pcapkit.utilities.warnings.ProtocolWarning` (:issue:`1604`). + Each header dissects the next one from here, so the chain's depth + is handed to the next layer, c.f. :meth:`_next_layer_class`. + """ if payload is None: file_ = self.__header__.get_payload() else: @@ -315,13 +424,7 @@ def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *, # if length is None: length = len(file_) - if min(length, len(file_)) == 0: - from pcapkit.protocols.misc.null import NoPayload as protocol # isort: skip # pylint: disable=import-outside-toplevel - elif self._sigterm: - from pcapkit.protocols.misc.raw import Raw as protocol # isort: skip # pylint: disable=import-outside-toplevel - else: - protocol = self._lookup_next_layer(self.__proto__, proto) - + protocol, extra = self._next_layer_class(proto, file_, length, extension) return self._parse_next_layer(protocol, file_, length, version=version, extension=extension, alias=proto, packet=packet, layer=self._exlayer, - protocol=self._exproto, __context__=self._exctx) + protocol=self._exproto, __context__=self._exctx, **extra) diff --git a/pcapkit/protocols/internet/ipv6.py b/pcapkit/protocols/internet/ipv6.py index 884136c8a8..16cb16bd05 100644 --- a/pcapkit/protocols/internet/ipv6.py +++ b/pcapkit/protocols/internet/ipv6.py @@ -35,6 +35,7 @@ from pcapkit.corekit.multidict import MultiDict, OrderedMultiDict from pcapkit.corekit.protochain import ProtoChain from pcapkit.protocols.data.internet.ipv6 import IPv6 as Data_IPv6 +from pcapkit.protocols.internet.internet import EXTENSION_HEADER_LIMIT from pcapkit.protocols.internet.ip import IP from pcapkit.protocols.schema.internet.ipv6 import IPv6 as Schema_IPv6 from pcapkit.protocols.schema.internet.ipv6 import jumbo_payload_length @@ -530,7 +531,7 @@ def _decode_next_layer(self, ipv6: 'Data_IPv6', proto: 'Optional[int]' = None, hdr_len = self.length # header length raw_len = ipv6.payload if length is None else length # payload length - _protos = [] # ProtoChain buffer + _protos = [] # type: list[ProtocolBase] # ProtoChain buffer _exthdr = [] # (parser class, info) per extension header split = None # type: Optional[tuple[int, int]] # at first Fragment header opaque = False # walk ended at a non-zero Fragment Offset @@ -543,6 +544,15 @@ def _decode_next_layer(self, ipv6: 'Data_IPv6', proto: 'Optional[int]' = None, except ValueError: break + # NOTE: At most :data:`EXTENSION_HEADER_LIMIT + # ` + # headers are walked, as at most that many are dissected after any + # other layer. This header and the rest are then the upper layer, + # which :meth:`_import_next_layer` keeps raw, with one warning + # (:issue:`1604`). + if len(_protos) >= EXTENSION_HEADER_LIMIT: + break + # # directly break when No Next Header occurs # if proto.name == 'IPv6-NoNxt': # proto = None @@ -646,6 +656,7 @@ def _decode_next_layer(self, ipv6: 'Data_IPv6', proto: 'Optional[int]' = None, # of a slice of UDP payload (:issue:`1545`). Reassembly takes # ``fragment`` above, which this leaves as it was. ipv6_exthdr = ProtoChain.from_list(_protos) # type: ignore[arg-type] + self._exthdr_depth = len(_protos) # the headers the upper layer follows return super()._decode_next_layer(ipv6, None if opaque else proto, raw_len, packet=packet, ipv6_exthdr=ipv6_exthdr, payload=payload) @@ -689,10 +700,15 @@ def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *, # ends after it -- see :attr:`__generic_ext_codes__`'s docstring for the full distinction. - """ - if TYPE_CHECKING: - protocol: 'Type[ProtocolBase]' + The upper layer past a walk cut at :data:`EXTENSION_HEADER_LIMIT + ` is the + next extension header, and is kept raw with one + :exc:`~pcapkit.utilities.warnings.ProtocolWarning`, as in + :meth:`Internet._import_next_layer + ` + (:issue:`1604`). + """ if payload is None: file_ = self.__header__.get_payload() else: @@ -700,19 +716,11 @@ def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *, # if length is None: length = len(file_) - if min(length, len(file_)) == 0: - from pcapkit.protocols.misc.null import \ - NoPayload as protocol # isort: skip # pylint: disable=import-outside-toplevel - elif self._sigterm: - from pcapkit.protocols.misc.raw import \ - Raw as protocol # isort: skip # pylint: disable=import-outside-toplevel - else: - protocol = self._lookup_next_layer(self.__proto__, proto) - + protocol, extra = self._next_layer_class(proto, file_, length, extension) try: next_ = self._parse_next_layer(protocol, file_, length, version=version, extension=extension, alias=proto, packet=packet, layer=self._exlayer, - protocol=self._exproto, __context__=self._exctx) + protocol=self._exproto, __context__=self._exctx, **extra) except Exception as exc: from pcapkit.protocols.internet.ipv6_ext import \ IPv6_Ext # isort: skip # pylint: disable=import-outside-toplevel diff --git a/pcapkit/toolkit/pypcapfile.py b/pcapkit/toolkit/pypcapfile.py index c73034f60d..882fb84bab 100644 --- a/pcapkit/toolkit/pypcapfile.py +++ b/pcapkit/toolkit/pypcapfile.py @@ -73,6 +73,7 @@ from pcapkit.foundation.reassembly.data.ip import Packet as IP_Packet from pcapkit.foundation.reassembly.data.tcp import Packet as TCP_Packet from pcapkit.foundation.traceflow.data.tcp import Packet as TF_TCP_Packet +from pcapkit.protocols.internet.internet import EXTENSION_HEADER_LIMIT from pcapkit.protocols.internet.ipv4 import IPv4 as Protocol_IPv4 from pcapkit.protocols.schema.internet.ipv6 import jumbo_payload_length from pcapkit.protocols.transport.tcp import TCP as Protocol_TCP @@ -659,7 +660,10 @@ def _past_extension_headers(protocol: 'int', data: 'bytes') -> 'Optional[tuple[i default engine leaves one of them undissected. An AH is ``(Payload Length + 2) * 4`` octets long (:rfc:`4302#section-2.2`), and no shorter than :data:`AH_HEADER_LEN`; any other is read by - :func:`_default_extension_header` (:issue:`1597`). + :func:`_default_extension_header` (:issue:`1597`). Past + :data:`~pcapkit.protocols.internet.internet.EXTENSION_HEADER_LIMIT` + of them in a row, the default engine leaves the rest undissected too + (:issue:`1604`). Note: Inside IPv4, the default engine decodes what an IPv6 Fragment header @@ -668,7 +672,11 @@ def _past_extension_headers(protocol: 'int', data: 'bytes') -> 'Optional[tuple[i they are there. """ + depth = 0 while protocol in IPV4_EXTENSION_HEADERS: + if depth == EXTENSION_HEADER_LIMIT: + return None + depth += 1 if protocol == Enum_TransType.AH: length = (data[1] + 2) * 4 if len(data) >= AH_HEADER_LEN else 0 if length < AH_HEADER_LEN or len(data) < length: @@ -704,7 +712,9 @@ def _upper_layer(version: 'int', data: 'bytes') -> 'Optional[tuple[int, bytes]]' Fragment header as if that header were not there. A later fragment of either has no upper-layer header, since from its offset on it carries a slice of the payload (:rfc:`791#section-3.2`, :rfc:`8200#section-4.5`). - Nor does a truncated header or header chain. + Nor does a truncated header or header chain, or a chain of more than + :data:`~pcapkit.protocols.internet.internet.EXTENSION_HEADER_LIMIT` + extension headers (:issue:`1604`). Note: Each of these rules is the default engine's own, as are the ones @@ -737,10 +747,11 @@ def _upper_layer(version: 'int', data: 'bytes') -> 'Optional[tuple[int, bytes]]' if not extent: extent = jumbo_payload_length(data[6], data[IPV6_HEADER_LEN:]) or 0 data = data[:IPV6_HEADER_LEN + extent] - nxt, offset = data[6], IPV6_HEADER_LEN + nxt, offset, depth = data[6], IPV6_HEADER_LEN, 0 while nxt in IPV6_EXTENSION_HEADERS: - if len(data) < offset + 8: + if depth == EXTENSION_HEADER_LIMIT or len(data) < offset + 8: return None + depth += 1 if nxt == Enum_ExtensionHeader.IPv6_Frag: # fixed length, second octet reserved if struct.unpack_from('!H', data, offset + 2)[0] >> 3: # a later fragment return None diff --git a/tests/foundation/engines/test_ext_chain_limit_agreement_1604_unit.py b/tests/foundation/engines/test_ext_chain_limit_agreement_1604_unit.py new file mode 100644 index 0000000000..613ba92488 --- /dev/null +++ b/tests/foundation/engines/test_ext_chain_limit_agreement_1604_unit.py @@ -0,0 +1,245 @@ +# -*- coding: utf-8 -*- +"""Every engine stops a chain of IPv6 extension headers where ``default`` does. C.f. #1604. + +The ``default`` engine dissects at most +:data:`~pcapkit.protocols.internet.internet.EXTENSION_HEADER_LIMIT` extension +headers in a row, and leaves the rest raw, so the TCP behind a longer chain is +not found. Each engine here reads one capture per chain length -- 30, 32, 33 and +1,000 Destination Options headers, then a TCP segment -- carried in IPv4 and in +IPv6, with TCP reassembly and flow tracing on, and its ``tcp`` and ``trace`` +inputs are compared with ``default``'s, as in +:mod:`tests.foundation.engines.test_engine_agreement_runtime`. + +``pypcapfile`` walks the chain itself and stops at the same depth. It leaves TCP +over IPv6 out, so it is compared over IPv4 only, and ``declined`` checks that +it warns of leaving out TCP over IPv6 exactly where ``default`` finds some. It +imports on Python 3.11 only, and is skipped elsewhere. + +The other parsers walk the chain themselves, and cannot be told where to stop, +so where they find the TCP past the limit is recorded in :attr:`KNOWN_FAILURES` +under #1609 rather than hidden. Measured on dpkt 1.9.8, scapy 2.7.0 and tshark 4.6.9. + +Every capture is built in a temporary directory, so no fixture is read. + +""" + +from __future__ import annotations + +import importlib.util +import ipaddress +import os +import tempfile +import unittest +import warnings +from typing import TYPE_CHECKING +from unittest import mock + +from tests._support import close_extractor, reimport_once_per_class +from tests.foundation import _roundtrip as harness +from tests.foundation._roundtrip import Gap, Outcome + +if TYPE_CHECKING: + from typing import Any, Optional + +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None + for name in ('tbtrim', 'aenum', 'chardet', 'dictdumper')) + +#: Chain lengths: under the limit, at it, one past it, and far past it. +CHAINS = (30, 32, 33, 1000) +#: What carries the chain. +CARRIERS = ('ipv4', 'ipv6') +#: Every engine compared with ``default``, as :attr:`Extractor.__engine__ +#: ` names them. ``pypcap`` +#: and ``pcap_ct`` decline both reassembly and flow tracing, so are left out. +ENGINES = ('dpkt', 'scapy', 'pyshark', 'pypcapfile') +#: Engines whose flow-tracing input carries no TCP octets, c.f. +#: :data:`tests.foundation.engines.test_engine_agreement_runtime.NO_TCP_OCTETS`. +NO_TCP_OCTETS = ('pyshark',) + + +def capture(carrier: 'str', count: 'int') -> 'bytes': + """A PCAP savefile of one frame: ``count`` Destination Options headers, then TCP.""" + headers = b''.join(bytes([60 if index < count - 1 else 6, 0, 1, 4, 0, 0, 0, 0]) + for index in range(count)) + payload = headers + harness.tcp(b'payload', seq=1) + if carrier == 'ipv4': + frame = harness.ethernet(harness.ipv4(payload, proto=60), 0x0800) + else: + frame = harness.ethernet(harness.ipv6(payload, nxt=60), 0x86DD) + return harness.pcap([(1, 0, frame)]) + + +def _fields(packet: 'Any', skip: 'tuple[str, ...]' = ()) -> 'dict[str, Any]': + return {key: bytes(value) if isinstance(value, bytearray) else value + for key, value in packet.items() if key not in skip} + + +class _Run: + """One engine's reading of one capture.""" + + def __init__(self, engine: 'str', path: 'str') -> None: + from pcapkit import extract + from pcapkit.foundation.reassembly.reassembly import ReassemblyBase + from pcapkit.foundation.traceflow.traceflow import TraceFlowBase + from pcapkit.utilities.exceptions import UnsupportedCall + from pcapkit.utilities.warnings import AttributeWarning, EngineWarning + + self.error = self.fallback = None # type: Optional[str] + #: What the ``pypcapfile`` toolkit warned it leaves out, if anything. + self.declined = None # type: Optional[str] + #: Whether the engine declines TCP reassembly, or flow tracing. + self.no_tcp = self.no_trace = False + self.inputs = {'tcp': {}, 'trace': {}} # type: dict[str, dict[int, list[dict[str, Any]]]] + inputs = self.inputs + reassemble, trace = ReassemblyBase.__call__, TraceFlowBase.__call__ + + def spy_reassembly(this: 'Any', packet: 'Any') -> 'None': + inputs['tcp'].setdefault(packet.num, []).append(_fields(packet)) + return reassemble(this, packet) + + def spy_trace(this: 'Any', packet: 'Any') -> 'None': + inputs['trace'].setdefault(packet.index, []).append(_fields(packet, ('frame',))) + return trace(this, packet) + + with warnings.catch_warnings(), tempfile.TemporaryDirectory() as tracedir, \ + mock.patch.object(ReassemblyBase, '__call__', spy_reassembly), \ + mock.patch.object(TraceFlowBase, '__call__', spy_trace), \ + mock.patch('pcapkit.foundation.extraction.warn') as warn, \ + mock.patch('pcapkit.toolkit.pypcapfile.warn') as toolkit_warn: + warnings.simplefilter('ignore') + try: + extractor = extract(fin=path, nofile=True, engine=engine, reassembly=True, + tcp=True, trace=True, trace_fout=tracedir, trace_format='json') + except Exception as exc: # pylint: disable=broad-except + self.error = f'{type(exc).__name__}: {exc}' + return + close_extractor(extractor) + fallen = [str(call.args[0]) for call in warn.call_args_list + if len(call.args) > 1 and call.args[1] is EngineWarning] + if fallen or extractor._exnam != engine: # pylint: disable=protected-access + self.fallback = '; '.join(fallen) or f'ran as {extractor._exnam}' # pylint: disable=protected-access + return + self.declined = '; '.join(str(call.args[0]) for call in toolkit_warn.call_args_list + if len(call.args) > 1 and call.args[1] is AttributeWarning + and 'is left out of TCP reassembly' in str(call.args[0])) or None + for name in ('reassembly', 'trace'): + try: + found = getattr(getattr(extractor, name), 'tcp', None) + except UnsupportedCall: + found = None + setattr(self, 'no_tcp' if name == 'reassembly' else 'no_trace', found is None) + + +_RUNS = {} # type: dict[tuple[str, str], _Run] + + +def _over_ipv4(record: 'dict[str, Any]') -> 'bool': + address = record['bufid'][0] if 'bufid' in record else record['src'] + return isinstance(address, ipaddress.IPv4Address) + + +def _difference(got: 'dict[int, Any]', want: 'dict[int, Any]') -> 'Optional[str]': + for number in sorted(set(got) | set(want)): + if number not in got: + return f'frame {number}: missing, default has {len(want[number])} input(s)' + if number not in want: + return f'frame {number}: extra, default has none' + if got[number] != want[number]: + return f'frame {number}: {got[number]!r:.120} != {want[number]!r:.120}' + return None + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class TestExtensionHeaderChainAgreement(harness.RoundTripBase): + """Every engine that runs stops the chain where ``default`` does, or is a recorded gap.""" + + STATUSES = ('ERROR', 'INPUTS', 'DECLINED') + KNOWN_FAILURES = ( + Gap(1609, 'dpkt walks the whole extension header chain in a loop (dpkt.ip6.IP6.unpack), ' + 'so finds the TCP past EXTENSION_HEADER_LIMIT that the default engine leaves raw', + 'INPUTS', 'extra, default has none', ('ipv6-33/dpkt/*', 'ipv6-1000/dpkt/*')), + Gap(1609, 'scapy dissects each extension header behind the last, with no limit of its own, ' + 'so finds the TCP past EXTENSION_HEADER_LIMIT; it gives up near 245 layers at the ' + 'default recursion limit, and so agrees at 1,000', + 'INPUTS', 'extra, default has none', ('ipv6-33/scapy/*',)), + Gap(1609, 'tshark dissects up to about 496 extension headers, so finds the TCP past ' + 'EXTENSION_HEADER_LIMIT, carried in IPv4 or IPv6; at 1,000 it gives up and agrees', + 'INPUTS', 'extra, default has none', ('ipv4-33/pyshark/trace', 'ipv6-33/pyshark/trace')), + Gap(1606, 'dpkt and scapy dissect no IPv6 extension header carried in IPv4 ' + '(dpkt keeps the payload of IPv4 protocol 60 as bytes, scapy as Raw), so find no ' + 'TCP behind one, where the default engine reads past it at any length up to ' + 'EXTENSION_HEADER_LIMIT', + 'INPUTS', 'missing, default has', ('ipv4-30/dpkt/*', 'ipv4-32/dpkt/*', + 'ipv4-30/scapy/*', 'ipv4-32/scapy/*')), + ) + + @classmethod + def setUpClass(cls) -> None: + super().setUpClass() + cls.tmp = tempfile.TemporaryDirectory() # pylint: disable=consider-using-with + for carrier in CARRIERS: + for count in CHAINS: + with open(cls.path(carrier, count), 'wb') as file: + file.write(capture(carrier, count)) + + @classmethod + def tearDownClass(cls) -> None: + _RUNS.clear() + cls.tmp.cleanup() + super().tearDownClass() + + @classmethod + def path(cls, carrier: 'str', count: 'int') -> 'str': + return os.path.join(cls.tmp.name, f'{carrier}-{count}.pcap') + + def setUp(self) -> None: + super().setUp() + reimport_once_per_class(self) + + def labels(self) -> 'list[str]': + return [f'{carrier}-{count}/{engine}/{aspect}' + for carrier in CARRIERS for count in CHAINS for engine in ENGINES + for aspect in ('tcp', 'trace', 'declined') + if aspect != 'declined' or engine == 'pypcapfile'] + + def run_of(self, engine: 'str', name: 'str') -> '_Run': + if (engine, name) not in _RUNS: + carrier, count = name.split('-') + _RUNS[engine, name] = _Run(engine, self.path(carrier, int(count))) + return _RUNS[engine, name] + + def outcome(self, label: 'str') -> 'Outcome': + name, engine, aspect = label.split('/') + run = self.run_of(engine, name) + if run.fallback is not None: + return Outcome('SKIP', f'{engine} did not run: {run.fallback}') + if run.error is not None: + return Outcome('ERROR', run.error) + base = self.run_of('default', name) + if base.error is not None or base.fallback is not None: + return Outcome('ERROR', f'default: {base.error or base.fallback}') + + over_ipv6 = any(not _over_ipv4(records[0]) for records in base.inputs['tcp'].values()) + if aspect == 'declined': + if (run.declined is not None) != over_ipv6: + return Outcome('DECLINED', f'warned {run.declined!r}, but default finds TCP over IPv6: ' + f'{over_ipv6}') + return Outcome('OK') + if (aspect == 'tcp' and run.no_tcp) or (aspect == 'trace' and run.no_trace): + return Outcome('SKIP', f'{engine} declines {aspect}') + + mine, theirs = run.inputs[aspect], base.inputs[aspect] + if run.declined is not None: + theirs = {number: records for number, records in theirs.items() if _over_ipv4(records[0])} + if aspect == 'trace' and engine in NO_TCP_OCTETS: + def strip(inputs: 'dict[int, Any]') -> 'dict[int, Any]': + return {number: [{key: value for key, value in record.items() + if key not in ('header', 'payload')} for record in records] + for number, records in inputs.items()} + mine, theirs = strip(mine), strip(theirs) + diff = _difference(mine, theirs) + return Outcome('INPUTS', diff) if diff else Outcome('OK') + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/internet/test_ipv6_ext_chain_limit_1604_unit.py b/tests/protocols/internet/test_ipv6_ext_chain_limit_1604_unit.py new file mode 100644 index 0000000000..c9c6f08fd2 --- /dev/null +++ b/tests/protocols/internet/test_ipv6_ext_chain_limit_1604_unit.py @@ -0,0 +1,216 @@ +# -*- coding: utf-8 -*- +"""A chain of IPv6 extension headers is dissected to 32 headers, and no further. + +GitHub issue #1604: inside IPv4, each IPv6 extension header dissects the next +one itself, so a long chain ran into Python's recursion limit, and +:func:`~pcapkit.utilities.decorators.beholder` caught the ``RecursionError`` +and left the rest raw, silently. Where that happened depended on the +interpreter and the stack -- after 79 headers on Python 3.11, and between 85 +and 90 on 3.14, from a script at the default limit -- and IPv6, which walks its +chain in a loop, went on to any depth. + +Now both stop at :data:`~pcapkit.protocols.internet.internet.EXTENSION_HEADER_LIMIT` +headers in a row: what follows is kept as +:class:`~pcapkit.protocols.misc.raw.Raw`, with one +:exc:`~pcapkit.utilities.warnings.ProtocolWarning`, however long the chain and +whatever the recursion limit. + +Every case builds its own octets in memory and reads no capture. Classes are +imported inside each test, after :func:`~tests._support.reimport_once_per_class`. + +""" + +import sys +import unittest +import warnings + +from tests._support import reimport_once_per_class +from tests.foundation import _roundtrip as wire + +#: The TCP segment every chain ends in. +SEGMENT = wire.tcp(b'payload', seq=1) + +#: Ethernet II EtherTypes of the two carriers. +ETHERTYPE = {'ipv4': 0x0800, 'ipv6': 0x86DD} + + +def options(nxt: int) -> bytes: + """An 8-octet Destination Options header: Hdr Ext Len 0 and one PadN.""" + return bytes([nxt, 0, 1, 4, 0, 0, 0, 0]) + + +def ah(nxt: int) -> bytes: + """A 12-octet Authentication Header, its fixed part only (Payload Length 1).""" + return bytes([nxt, 1, 0, 0]) + bytes.fromhex('0000100000000001') + + +def chain(count: int, *, mixed: bool = False, last: int = 6) -> bytes: + """``count`` extension headers, then :data:`SEGMENT`. + + Destination Options headers only, or alternately Destination Options and AH + when ``mixed``, the first and the last always Destination Options. Each + names the next, and the last names ``last``. + + """ + kinds = [ah if mixed and index % 2 and index < count - 1 else options for index in range(count)] + codes = [51 if kind is ah else 60 for kind in kinds] + return b''.join(kind(nxt) for kind, nxt in zip(kinds, codes[1:] + [last])) + SEGMENT + + +def ip(carrier: str, payload: bytes) -> bytes: + """An IPv4 or IPv6 packet whose first header after its own is Destination Options.""" + if carrier == 'ipv4': + return wire.ipv4(payload, proto=60) + return wire.ipv6(payload, nxt=60) + + +def frame(carrier: str, count: int, *, mixed: bool = False) -> bytes: + """An Ethernet frame of IPv4 or IPv6 whose payload is :func:`chain`.""" + return wire.ethernet(ip(carrier, chain(count, mixed=mixed)), ETHERTYPE[carrier]) + + +def raw_of(parsed: object) -> object: + """The first :class:`~pcapkit.protocols.misc.raw.Raw` layer of ``parsed``, if any.""" + from pcapkit.protocols.misc.null import NoPayload + from pcapkit.protocols.misc.raw import Raw + + layer = parsed + while not isinstance(layer, (Raw, NoPayload)): + layer = layer.payload # type: ignore[attr-defined] + return layer if isinstance(layer, Raw) else None + + +class _Depth: + """Lower the recursion limit to ``margin`` frames above the caller's.""" + + def __init__(self, margin: int) -> None: + self.margin = margin + + def __enter__(self) -> '_Depth': + depth, frame_ = 0, sys._getframe() # pylint: disable=protected-access + while frame_ is not None: + depth, frame_ = depth + 1, frame_.f_back + self.saved = sys.getrecursionlimit() + sys.setrecursionlimit(depth + self.margin) + return self + + def __exit__(self, *exc: object) -> None: + sys.setrecursionlimit(self.saved) + + +class ExtensionHeaderChainLimitTests(unittest.TestCase): + """C.f. #1604.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def dissect(self, data: bytes) -> 'tuple[object, list[str]]': + """The frame ``data`` parses to, and the messages of the ProtocolWarnings it gave.""" + from pcapkit.protocols.link.ethernet import Ethernet + from pcapkit.utilities.warnings import ProtocolWarning + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + parsed = Ethernet(data) + return parsed, [str(item.message) for item in caught if issubclass(item.category, ProtocolWarning)] + + def summary(self, data: bytes) -> 'tuple[object, ...]': + """What a cut chain comes to: its layers, the raw remainder and the warnings.""" + parsed, messages = self.dissect(data) + layer = raw_of(parsed) + if layer is None: + return (str(parsed.protochain), None, None, tuple(messages)) + return (str(parsed.protochain), layer.info.protocol, len(layer.data), tuple(messages)) + + def test_the_limit_is_32(self) -> None: + from pcapkit.protocols.internet.internet import EXTENSION_HEADER_LIMIT + + self.assertEqual(EXTENSION_HEADER_LIMIT, 32) + + def test_a_chain_of_32_is_dissected_to_the_tcp(self) -> None: + for carrier in ETHERTYPE: + for mixed in (False, True): + with self.subTest(carrier=carrier, mixed=mixed): + parsed, messages = self.dissect(frame(carrier, 32, mixed=mixed)) + self.assertIn('TCP', parsed) + names = str(parsed.protochain).split(':') + self.assertEqual(names.count('IPv6-Opts') + names.count('AH'), 32, names) + self.assertEqual(messages, []) + + def test_a_chain_of_33_stops_at_the_limit(self) -> None: + from pcapkit.const.reg.transtype import TransType + + for carrier in ETHERTYPE: + for mixed in (False, True): + with self.subTest(carrier=carrier, mixed=mixed): + payload = chain(33, mixed=mixed) + parsed, messages = self.dissect(frame(carrier, 33, mixed=mixed)) + self.assertNotIn('TCP', parsed) + names = str(parsed.protochain).split(':') + self.assertEqual(names.count('IPv6-Opts') + names.count('AH'), 32, names) + + layer = raw_of(parsed) + # the 33rd header on, kept as it was captured + cut = len(payload) - len(SEGMENT) - len(options(6)) + self.assertEqual(bytes(layer.data), payload[cut:]) + self.assertEqual(layer.info.protocol, TransType.IPv6_Opts) + + self.assertEqual(len(messages), 1, messages) + self.assertIn('limit of 32 headers', messages[0]) + self.assertEqual(layer.info.error, messages[0]) + + def test_where_a_chain_stops_depends_on_neither_its_length_nor_the_recursion_limit(self) -> None: + for carrier in ETHERTYPE: + with self.subTest(carrier=carrier): + want = self.summary(frame(carrier, 33)) + for count in (33, 34, 100, 1000): + for margin in (None, 550): + with self.subTest(count=count, margin=margin): + data = frame(carrier, count) + if margin is None: + got = self.summary(data) + else: + with _Depth(margin): + got = self.summary(data) + # the remainder is longer by the headers past the 33rd + self.assertEqual(got[:2] + got[3:], want[:2] + want[3:]) + self.assertEqual((got[2] or 0) - (want[2] or 0), (count - 33) * len(options(6))) + + def test_a_chain_of_32_fits_a_lowered_recursion_limit(self) -> None: + for carrier in ETHERTYPE: + with self.subTest(carrier=carrier), _Depth(550): + parsed, messages = self.dissect(frame(carrier, 32)) + self.assertIn('TCP', parsed) + self.assertEqual(messages, []) + + def test_layers_between_two_chains_start_a_new_count(self) -> None: + # Two chains of 20, with IP-in-IP between them: 40 headers in all, but + # never more than 20 in a row. + for carrier, tunnel in (('ipv4', 4), ('ipv6', 41)): + with self.subTest(carrier=carrier): + outer = chain(20, last=tunnel)[:-len(SEGMENT)] + data = wire.ethernet(ip(carrier, outer + ip(carrier, chain(20))), ETHERTYPE[carrier]) + parsed, messages = self.dissect(data) + self.assertIn('TCP', parsed) + self.assertEqual(str(parsed.protochain).split(':').count('IPv6-Opts'), 40) + self.assertEqual(messages, []) + + def test_a_cut_frame_rebuilds_byte_exact(self) -> None: + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.link.ethernet import Ethernet + + for carrier, klass in (('ipv4', IPv4), ('ipv6', IPv6)): + for count in (32, 33, 1000): + with self.subTest(carrier=carrier, count=count), warnings.catch_warnings(): + warnings.simplefilter('ignore') + data = frame(carrier, count) + parsed = Ethernet(data) + self.assertEqual(parsed.data, data) + self.assertEqual(Ethernet.from_data(parsed.info.to_dict()).data, data) + packet = klass(data[14:]) + self.assertEqual(klass.from_data(packet.info.to_dict()).data, data[14:]) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/toolkit/test_pypcapfile_ext_chain_limit_1604_unit.py b/tests/toolkit/test_pypcapfile_ext_chain_limit_1604_unit.py new file mode 100644 index 0000000000..c30c519cfc --- /dev/null +++ b/tests/toolkit/test_pypcapfile_ext_chain_limit_1604_unit.py @@ -0,0 +1,149 @@ +""":mod:`pcapkit.toolkit.pypcapfile` stops a chain of IPv6 extension headers where the default engine does. + +GitHub issue #1604: the default engine dissects at most +:data:`~pcapkit.protocols.internet.internet.EXTENSION_HEADER_LIMIT` extension +headers in a row and leaves the rest raw, so the TCP behind a longer chain is +not found. :func:`~pcapkit.toolkit.pypcapfile._past_extension_headers` (inside +IPv4) and :func:`~pcapkit.toolkit.pypcapfile._upper_layer` (over IPv6) walk such +a chain in a loop, and stop at the same depth, so the TCP the toolkit reads and +the TCP it warns of leaving out are the TCP the default engine finds. + +These tests use stand-ins for `pypcapfile`_'s decoders, so run without it; the +comparison of the two engines over a capture is in +:mod:`tests.foundation.engines.test_ext_chain_limit_agreement_1604_unit`. + +.. _pypcapfile: https://github.com/kisom/pypcapfile + +""" +from __future__ import annotations + +import unittest +from unittest import mock + +from tests._support import reimport_once_per_class +from tests.foundation import _roundtrip as wire +from tests.toolkit import test_pypcapfile_unit as base + +#: Chain lengths either side of the limit, and one far past it. +AT_LIMIT, PAST_LIMIT = (32,), (33, 1000) + + +def chain(count: int, last: int, *, mixed: bool = False) -> bytes: + """``count`` extension headers, the last naming ``last``. + + Destination Options headers only, or alternately Destination Options and + AH when ``mixed``; the first is always Destination Options (protocol 60). + + """ + kinds = ['ah' if mixed and index % 2 else 'opts' for index in range(count)] + codes = [51 if kind == 'ah' else 60 for kind in kinds] + [last] + return b''.join(base.ah(nxt, 1) if kind == 'ah' else base.options_header(nxt) + for kind, nxt in zip(kinds, codes[1:])) + + +@unittest.skipUnless(base.HAS_RUNTIME, 'runtime dependencies not installed') +class PyPCAPFileExtensionHeaderChainLimitTests(unittest.TestCase): + """C.f. #1604.""" + + decline = base.PyPCAPFileTCPOverIPv6Tests.decline + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_transport_reads_tcp_behind_a_chain_of_32_inside_ipv4(self) -> None: + from pcapkit.toolkit.pypcapfile import _transport + + segment = base.make_tcp(b'data') + for count in AT_LIMIT: + for mixed in (False, True): + with self.subTest(count=count, mixed=mixed): + layer = base.FakeIP(chain(count, 6, mixed=mixed) + segment, p=60) + self.assertEqual(_transport(layer), segment) + + def test_no_tcp_behind_a_longer_chain_inside_ipv4(self) -> None: + from pcapkit.toolkit import pypcapfile + + segment = base.make_tcp(b'data') + for count in PAST_LIMIT: + for mixed in (False, True): + with self.subTest(count=count, mixed=mixed): + layer = base.FakeIP(chain(count, 6, mixed=mixed) + segment, p=60) + with mock.patch.object(pypcapfile, '_default_extension_header', + wraps=pypcapfile._default_extension_header) as parse: + self.assertIsNone(pypcapfile._transport(layer)) + # the walk stops at the limit, rather than reading the rest + self.assertLessEqual(parse.call_count, 32) + self.assertEqual(self.decline(base.make_packet(base.FakeEthernet(layer))), []) + + def test_tcp_over_ipv6_behind_a_chain_of_32_warns(self) -> None: + for count in AT_LIMIT: + for mixed in (False, True): + with self.subTest(count=count, mixed=mixed): + packet = wire.ipv6(chain(count, 6, mixed=mixed) + base.TCP_SEGMENT, nxt=60) + messages = self.decline(base.make_packet(base.ipv6_frame(packet))) + self.assertEqual(len(messages), 1, messages) + self.assertIn('TCP over IPv6', messages[0]) + + def test_tcp_over_ipv6_behind_a_longer_chain_does_not_warn(self) -> None: + # The default engine finds no TCP there, so none is left out. + for count in PAST_LIMIT: + for mixed in (False, True): + with self.subTest(count=count, mixed=mixed): + packet = wire.ipv6(chain(count, 6, mixed=mixed) + base.TCP_SEGMENT, nxt=60) + self.assertEqual(self.decline(base.make_packet(base.ipv6_frame(packet))), []) + + def test_a_tunnel_behind_the_chain_is_entered_only_within_the_limit(self) -> None: + # C.f. #1581: the TCP in a 4in4 tunnel is read from the inner IPv4, but + # only if the chain in front of the tunnel is within the limit. + from pcapkit.toolkit.pypcapfile import _innermost, _transport + + tcp4 = wire.ipv4(base.TCP_SEGMENT, proto=6) + for count, entered in ((32, True), (33, False), (1000, False)): + with self.subTest(count=count), \ + mock.patch.dict('sys.modules', base.fake_ip_decoder()) as modules: + IP = modules['pcapfile.protocols.network.ip'].IP + packet = base.make_packet(base.tunnel(chain(count, 4) + tcp4, 60)) + ipv4, upper = _innermost(packet) + self.assertIsNone(upper) + self.assertEqual(IP.calls, [(tcp4, 0)] if entered else []) + self.assertEqual(_transport(ipv4), base.TCP_SEGMENT if entered else None) + if not entered: # no TCP, and none left out either + self.assertEqual(self.decline(packet), []) + + def test_a_new_chain_inside_a_tunnel_is_counted_afresh(self) -> None: + # 20 headers, a tunnel, then 20 more: 40 in the packet, but never more + # than 20 in a row, so the TCP is found. 33 behind the tunnel are not + # read past, wherever the count starts. + tcp4 = base.TCP_SEGMENT + for count, found in ((20, True), (33, False)): + with self.subTest(tunnel='6in6', count=count), \ + mock.patch.dict('sys.modules', base.fake_ip_decoder()): + inner = wire.ipv6(chain(count, 6) + base.TCP_SEGMENT, nxt=60) + packet = wire.ipv6(chain(20, 41) + inner, nxt=60) + messages = self.decline(base.make_packet(base.ipv6_frame(packet))) + self.assertEqual(len(messages), int(found), messages) + if found: + self.assertTrue(messages[0].startswith('Frame 1: TCP over IPv6 is left out'), + messages) + with self.subTest(tunnel='4in4', count=count), \ + mock.patch.dict('sys.modules', base.fake_ip_decoder()): + from pcapkit.toolkit.pypcapfile import _innermost, _transport + + inner = wire.ipv4(chain(count, 6) + tcp4, proto=60) + ipv4, _ = _innermost(base.make_packet(base.tunnel(chain(20, 4) + inner, 60))) + self.assertEqual(_transport(ipv4), tcp4 if found else None) + + def test_ipv4_in_ipv6_is_found_only_within_the_limit(self) -> None: + from pcapkit.toolkit.pypcapfile import _ipv4_in_ipv6 + + inner = base.udp_fragment() + for count, found in ((32, True), (33, False), (1000, False)): + with self.subTest(count=count), \ + mock.patch.dict('sys.modules', base.fake_ip_decoder()) as modules: + IP = modules['pcapfile.protocols.network.ip'].IP + frame = base.ipv6_frame(wire.ipv6(chain(count, 4) + inner, nxt=60)) + self.assertEqual(isinstance(_ipv4_in_ipv6(base.make_packet(frame)), IP), found) + + +if __name__ == '__main__': + unittest.main()