protocols: stop next-layer dispatch writing to shared registries, and fix two transport defects

docs/source/pcapkit/protocols/protocol.rst

@@ -61,6 +61,7 @@ utility arguments and methods of specified protocols.
    .. automethod:: _make_index
    .. automethod:: _make_payload
 
+   .. automethod:: _lookup_next_layer
    .. automethod:: _decode_next_layer
    .. automethod:: _import_next_layer
 

docs/source/pcapkit/protocols/transport/transport.rst

@@ -19,6 +19,7 @@ which is a base class for transport layer protocols, eg.
    .. automethod:: register
    .. automethod:: analyze
 
+   .. automethod:: _make_port
    .. automethod:: _decode_next_layer
 
    .. autoattribute:: __layer__

pcapkit/protocols/internet/internet.py

@@ -245,10 +245,7 @@ def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *,  #
         elif self._sigterm:
             from pcapkit.protocols.misc.raw import Raw as protocol  # isort: skip # pylint: disable=import-outside-toplevel
         else:
-            protocol = self.__proto__[proto]  # type: ignore[assignment]
-            if isinstance(protocol, ModuleDescriptor):
-                protocol = protocol.klass  # type: ignore[unreachable]
-                self.__proto__[proto] = protocol  # update mapping upon import
+            protocol = self._lookup_next_layer(self.__proto__, proto)
 
         next_ = protocol(file_, length, version=version, extension=extension,  # type: ignore[abstract]
                          alias=proto, packet=packet, layer=self._exlayer, protocol=self._exproto,

pcapkit/protocols/internet/ipv6.py

@@ -31,7 +31,6 @@
 
 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.multidict import OrderedMultiDict
 from pcapkit.corekit.protochain import ProtoChain
 from pcapkit.protocols.data.internet.ipv6 import IPv6 as Data_IPv6
@@ -406,10 +405,7 @@ def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *,  #
             from pcapkit.protocols.misc.raw import \
                 Raw as protocol  # isort: skip # pylint: disable=import-outside-toplevel
         else:
-            protocol = self.__proto__[proto]  # type: ignore[assignment]
-            if isinstance(protocol, ModuleDescriptor):
-                protocol = protocol.klass  # type: ignore[unreachable]
-                self.__proto__[proto] = protocol  # update mapping upon import
+            protocol = self._lookup_next_layer(self.__proto__, proto)
 
         next_ = protocol(file_, length, version=version, extension=extension,  # type: ignore[abstract]
                          alias=proto, packet=packet, layer=self._exlayer, protocol=self._exproto,

pcapkit/protocols/protocol.py

@@ -46,7 +46,7 @@
 
 if TYPE_CHECKING:
     from enum import IntEnum as StdlibEnum
-    from typing import IO, Any, DefaultDict, Optional, Type
+    from typing import IO, Any, Callable, DefaultDict, Optional, Type
 
     from aenum import IntEnum as AenumEnum
     from typing_extensions import Literal, Self
@@ -412,10 +412,7 @@ def analyze(cls, proto: 'int', payload: 'bytes', **kwargs: 'Any') -> 'ProtocolBa
             instance.
 
         """
-        protocol = cls.__proto__[proto]
-        if isinstance(protocol, ModuleDescriptor):
-            protocol = protocol.klass
-            cls.__proto__[proto] = protocol  # update mapping upon import
+        protocol = cls._lookup_next_layer(cls.__proto__, proto)
 
         payload_io = io.BytesIO(payload)
         try:
@@ -1181,6 +1178,11 @@ def _make_index(cls, name: 'str | int | StdlibEnum | AenumEnum', default: 'Optio
                             namespace = cast('dict[int, str]', namespace)
                         index = {v: k for k, v in namespace.items()}[name]
                 else:
+                    # Caught by the handler immediately below and converted, so
+                    # this never escapes -- it is a jump to the shared "name is
+                    # not in namespace" path, not a stdlib exception leaking out
+                    # of the library. A pcapkit exception here would log at
+                    # CRITICAL for something that is handled two lines later.
                     raise KeyError(name)
             except KeyError as error:
                 if default is None:
@@ -1235,6 +1237,50 @@ def _make_payload(cls, data: 'Data') -> 'ProtocolBase':
 
         return proto.from_data(data[name])
 
+    @staticmethod
+    def _lookup_next_layer(registry: 'DefaultDict[int, ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]]',
+                           proto: 'int') -> 'Type[ProtocolBase]':
+        """Look up the protocol class registered for a next layer code.
+
+        Arguments:
+            registry: next layer protocol registry, i.e. :attr:`self.__proto__
+                <ProtocolBase.__proto__>`. Passed in rather than read from the
+                class, so that a caller reaching the registry through an
+                instance keeps doing so.
+            proto: next layer protocol index
+
+        Returns:
+            The class registered for ``proto``, or the fallback ``registry``
+            declares -- normally :class:`~pcapkit.protocols.misc.raw.Raw` -- when
+            ``proto`` is not registered.
+
+        Important:
+            ``registry`` is a :class:`collections.defaultdict`, so indexing it
+            with an unregistered code would *insert* that code. It is
+            class-level -- shared by every instance in the process -- so parsing
+            a single packet with an unregistered code would grow it, and make
+            :meth:`self.register <ProtocolBase.register>` afterwards report that
+            code as already registered. The fallback is therefore read from the
+            default factory rather than through a lookup that records it.
+
+            Resolving a :class:`~pcapkit.corekit.module.ModuleDescriptor` is
+            still written back, since that is memoisation of an import for a
+            code that *is* registered rather than a new entry.
+
+        """
+        if proto in registry:
+            protocol = registry[proto]
+            if isinstance(protocol, ModuleDescriptor):
+                protocol = protocol.klass
+                registry[proto] = protocol  # update mapping upon import
+            return protocol
+
+        fallback = cast('Callable[[], ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]]',
+                        registry.default_factory)()
+        if isinstance(fallback, ModuleDescriptor):
+            return fallback.klass
+        return fallback
+
     def _decode_next_layer(self, dict_: '_PT', proto: 'int', length: 'Optional[int]' = None, *,
                            packet: 'Optional[dict[str, Any]]' = None) -> '_PT':
         r"""Decode next layer protocol.
@@ -1298,10 +1344,7 @@ def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *,
         elif self._sigterm:
             from pcapkit.protocols.misc.raw import Raw as protocol  # isort: skip # pylint: disable=import-outside-toplevel
         else:
-            protocol = self.__proto__[proto]  # type: ignore[assignment]
-            if isinstance(protocol, ModuleDescriptor):
-                protocol = protocol.klass  # type: ignore[unreachable]
-                self.__proto__[proto] = protocol  # update mapping upon import
+            protocol = self._lookup_next_layer(self.__proto__, proto)
 
         next_ = protocol(file_, length, alias=proto, packet=packet,
                          layer=self._exlayer, protocol=self._exproto,

pcapkit/protocols/schema/link/ethernet.py

@@ -8,7 +8,6 @@
 from pcapkit.corekit.fields.misc import PayloadField
 from pcapkit.corekit.fields.numbers import EnumField
 from pcapkit.corekit.fields.strings import BytesField
-from pcapkit.corekit.module import ModuleDescriptor
 from pcapkit.protocols.schema.schema import Schema, schema_final
 
 __all__ = ['Ethernet']
@@ -20,14 +19,35 @@
 
 
 def callback_payload(self: 'PayloadField', packet: 'dict[str, Any]') -> 'None':
-    """Callback function for :attr:`Ethernet.payload`."""
+    """Callback function for :attr:`Ethernet.payload`.
+
+    Args:
+        self: Payload field to resolve.
+        packet: Packet data, whose ``type`` names the next layer.
+
+    Returns:
+        :obj:`None`; the resolved class is assigned to ``self.protocol``.
+
+    Important:
+        The lookup goes through :meth:`ProtocolBase._lookup_next_layer
+        <pcapkit.protocols.protocol.ProtocolBase._lookup_next_layer>` rather than
+        subscripting the registry. :attr:`Ethernet.__proto__
+        <pcapkit.protocols.link.link.Link.__proto__>` *is*
+        :attr:`Link.__proto__ <pcapkit.protocols.link.link.Link.__proto__>`, a
+        class-level :class:`collections.defaultdict`, so reading a missing key
+        inserted it -- every unregistered EtherType a parse saw was recorded as
+        though somebody had registered it, and
+        :meth:`Link.register <pcapkit.protocols.link.link.Link.register>`
+        afterwards reported it as an overwrite. The helper reads the fallback
+        without recording it, and memoises a
+        :class:`~pcapkit.corekit.module.ModuleDescriptor` for a code that really
+        is registered, which this did not.
+
+    """
     from pcapkit.protocols.link.ethernet import Ethernet  # pylint: disable=import-outside-toplevel
 
-    type_ = packet['type']
-    protocol = Ethernet.__proto__[type_]
-    if isinstance(protocol, ModuleDescriptor):
-        protocol = protocol.klass
-    self.protocol = protocol
+    self.protocol = Ethernet._lookup_next_layer(  # pylint: disable=protected-access
+        Ethernet.__proto__, packet['type'])
 
 
 @schema_final

pcapkit/protocols/transport/sctp.py

@@ -28,6 +28,7 @@
 import struct
 from typing import TYPE_CHECKING, cast
 
+from pcapkit.const.reg.apptype import TransportProtocol as Enum_TransportProtocol
 from pcapkit.const.reg.transtype import TransType as Enum_TransType
 from pcapkit.const.sctp.cause_code import CauseCode as Enum_CauseCode
 from pcapkit.const.sctp.chunk import Chunk as Enum_Chunk
@@ -155,7 +156,6 @@
 from pcapkit.protocols.schema.transport.sctp import \
     UserInitiatedAbortCause as Schema_UserInitiatedAbortCause
 from pcapkit.protocols.transport.transport import Transport
-from pcapkit.utilities.decorators import beholder
 from pcapkit.utilities.exceptions import ProtocolError, RegistryError
 from pcapkit.utilities.warnings import RegistryWarning, warn
 
@@ -578,8 +578,8 @@ def make(self,
             chunks_value = []
 
         schema = Schema_SCTP(
-            srcport=srcport,
-            dstport=dstport,
+            srcport=self._make_port(srcport, Enum_TransportProtocol.sctp),
+            dstport=self._make_port(dstport, Enum_TransportProtocol.sctp),
             vtag=vtag,
             chksum=b'\x00\x00\x00\x00' if chksum is None else chksum,
             chunks=chunks_value,
@@ -812,66 +812,17 @@ def _decode_next_layer(self, dict_: 'Data_SCTP', proto: 'Optional[int]' = None,
             <pcapkit.protocols.internet.internet.Internet._import_next_layer>`
             does for an unregistered transport type. Resolving it to
             :class:`~pcapkit.protocols.misc.raw.Raw` is
-            :meth:`self._import_next_layer <SCTP._import_next_layer>`'s job.
+            :meth:`ProtocolBase._import_next_layer
+            <pcapkit.protocols.protocol.ProtocolBase._import_next_layer>`'s job,
+            which looks the PPID up through
+            :meth:`ProtocolBase._lookup_next_layer
+            <pcapkit.protocols.protocol.ProtocolBase._lookup_next_layer>` and so
+            leaves :attr:`self.__proto__ <SCTP.__proto__>` untouched.
 
         """
         return ProtocolBase._decode_next_layer(  # pylint: disable=protected-access
             self, dict_, proto, length, packet=packet)  # type: ignore[arg-type,return-value]
 
-    @beholder  # type: ignore[arg-type]
-    def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *,
-                           packet: 'Optional[dict[str, Any]]' = None) -> 'Protocol':
-        """Import next layer extractor.
-
-        Arguments:
-            proto: payload protocol identifier (PPID) of the DATA chunk carrying
-                the payload, or :obj:`None` if the packet carries no user data
-            length: valid (*non-padding*) length
-            packet: packet info (passed from :meth:`self.unpack <pcapkit.protocols.protocol.Protocol.unpack>`)
-
-        Returns:
-            Instance of next layer.
-
-        Important:
-            This overrides :meth:`ProtocolBase._import_next_layer
-            <pcapkit.protocols.protocol.ProtocolBase._import_next_layer>` for one
-            reason only: to look the PPID up **without** mutating
-            :attr:`self.__proto__ <SCTP.__proto__>`.
-
-            The registry is a :class:`collections.defaultdict`, so the base
-            implementation's ``self.__proto__[proto]`` *inserts* any key it is
-            handed. Every packet with an unregistered PPID would therefore grow
-            a registry that is class-level -- shared by every :class:`SCTP`
-            instance in the process -- and make
-            :meth:`self.register <SCTP.register>` report that PPID as already
-            registered. The fallback itself is unchanged: an unregistered PPID
-            still resolves to :class:`~pcapkit.protocols.misc.raw.Raw`, which is
-            what :attr:`self.__proto__ <SCTP.__proto__>` declares as its default.
-
-        """
-        if TYPE_CHECKING:
-            protocol: 'Type[Protocol]'
-
-        file_ = self._get_payload()
-        if length is None:
-            length = len(file_)
-
-        if length == 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
-        elif proto in self.__proto__:
-            protocol = self.__proto__[proto]  # type: ignore[assignment]
-            if isinstance(protocol, ModuleDescriptor):
-                protocol = protocol.klass  # type: ignore[unreachable]
-                self.__proto__[proto] = protocol  # update mapping upon import
-        else:
-            from pcapkit.protocols.misc.raw import Raw as protocol  # isort: skip # pylint: disable=import-outside-toplevel
-
-        next_ = protocol(file_, length, alias=proto, packet=packet,  # type: ignore[abstract]
-                         layer=self._exlayer, protocol=self._exproto)
-        return next_
-
     def _read_sctp_chunks(self) -> 'Chunks':
         """Read SCTP chunk list.
 

pcapkit/protocols/transport/tcp.py

@@ -44,6 +44,7 @@
 import math
 from typing import TYPE_CHECKING, cast
 
+from pcapkit.const.reg.apptype import TransportProtocol as Enum_TransportProtocol
 from pcapkit.const.reg.transtype import TransType
 from pcapkit.const.tcp.checksum import Checksum as Enum_Checksum
 from pcapkit.const.tcp.flags import Flags as Enum_Flags
@@ -549,8 +550,8 @@ def make(self,
         self._flags = _flag
 
         return Schema_TCP(
-            srcport=srcport,
-            dstport=dstport,
+            srcport=self._make_port(srcport, Enum_TransportProtocol.tcp),
+            dstport=self._make_port(dstport, Enum_TransportProtocol.tcp),
             seq=seq_no,
             ack=ack_no,
             offset={