fix(protocols): bound the IPv6 extension-header chain at 32 headers (#1604) (#1608)

Author: JarryShaw

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

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
+#: <pcapkit.protocols.internet.ipv6.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
+                <pcapkit.protocols.internet.ipv6.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,24 +408,23 @@ 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:
             file_ = payload
         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)

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
+            # <pcapkit.protocols.internet.internet.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,30 +700,27 @@ 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
+            <pcapkit.protocols.internet.internet.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
+            <pcapkit.protocols.internet.internet.Internet._import_next_layer>`
+            (:issue:`1604`).
 
+        """
         if payload is None:
             file_ = self.__header__.get_payload()
         else:
             file_ = payload
         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

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