fix(protocols): PCAPNG.register restores the shipped descriptor; a zero snaplen drops SPB padding (#1504, #1558)

pcapkit/foundation/engines/pcapng.py

@@ -407,9 +407,12 @@ def _get_snaplen(self) -> 'int':
         argument when parsing a Simple Packet Block (SPB).
 
         Notes:
-            If there is no interface, return ``0xFFFF_FFFF_FFFF_FFFF``.
+            If there is no interface, or interface 0's snaplen is zero, which
+            means no limit, return ``0xFFFF_FFFF_FFFF_FFFF``. A zero passed
+            through reads as a limit of no octets, and the block's padding
+            then becomes packet data (:issue:`1558`).
 
         """
-        if self._ctx.interfaces:
+        if self._ctx.interfaces and self._ctx.interfaces[0].snaplen:
             return self._ctx.interfaces[0].snaplen
         return 0xFFFF_FFFF_FFFF_FFFF

pcapkit/protocols/misc/pcapng.py

@@ -180,8 +180,7 @@
 from pcapkit.protocols.schema.schema import Schema
 from pcapkit.utilities.compat import StrEnum, localcontext
 from pcapkit.utilities.exceptions import (BaseError, EnumValueError, FormatError, ProtocolError,
-                                          RegistryError, StreamEOFError, UnsupportedCall,
-                                          stacklevel)
+                                          StreamEOFError, UnsupportedCall, stacklevel)
 from pcapkit.utilities.warnings import (AttributeWarning, DeprecatedFormatWarning, ProtocolWarning,
                                         RegistryWarning, warn)
 
@@ -1015,24 +1014,20 @@ def register(cls, code: 'Enum_LinkType', protocol: 'ModuleDescriptor[ProtocolBas
                 class it names; handing back the stored entry, or a descriptor
                 equal to it, is a silent no-op that keeps it. This registry is separate from the PCAP one, so
                 :func:`~pcapkit.foundation.registry.protocols.register_linktype`
-                writing to both cannot make either warn about the other.
+                writing to both cannot make either warn about the other. See
+                :meth:`ProtocolBase.register
+                <pcapkit.protocols.protocol.ProtocolBase.register>` for the
+                shipped descriptor a restore puts back (:issue:`1504`).
 
         """
         incumbent = cls.__proto__.get(code)
-        if incumbent is not None and (incumbent is protocol or (
-                isinstance(incumbent, ModuleDescriptor) and isinstance(protocol, ModuleDescriptor)
-                and incumbent == protocol)):
+        entry = cls._next_layer_entry(code, protocol)
+        if entry is None:
             return
-        if isinstance(protocol, ModuleDescriptor):
-            protocol = protocol.klass
-        if not isinstance(protocol, type):
-            raise RegistryError(f'protocol must be a class, not {protocol!r}')
-        if not issubclass(protocol, ProtocolBase):
-            raise RegistryError(f'protocol must be a Protocol subclass, not {protocol!r}')
-        if incumbent is not None and incumbent is not protocol:
+        if incumbent is not None and incumbent is not entry:
             warn(f'protocol {code} already registered, overwriting '
-                 f'{incumbent!r} with {protocol!r}', RegistryWarning)
-        cls.__proto__[code] = protocol
+                 f'{incumbent!r} with {entry!r}', RegistryWarning)
+        cls.__proto__[code] = entry
 
     @classmethod
     def register_block(cls, code: 'Enum_BlockType', meth: 'str | tuple[BlockParser, BlockConstructor]') -> 'None':
@@ -1203,6 +1198,11 @@ def unpack(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_PCAPN
                 interfaces = getattr(self._ctx, 'interfaces', None)
                 if interfaces and interfaces[0].snaplen:
                     packet.setdefault('snaplen', interfaces[0].snaplen)
+            # NOTE: A zero snaplen handed in, e.g. interface 0's passed through,
+            # means no limit as well; kept, it reads as a limit of no octets and
+            # the block's padding becomes packet data (:issue:`1558`).
+            if 'snaplen' in packet and not packet['snaplen']:
+                del packet['snaplen']
             self.__header__ = cast('Schema_PCAPNG', self.__schema__.unpack(self._file, length, packet))  # type: ignore[call-arg,misc]
 
         data = self.read(length, **kwargs)

pcapkit/protocols/protocol.py

@@ -945,8 +945,9 @@ def _next_layer_entry(cls, code: 'Any', protocol: 'ModuleDescriptor[ProtocolBase
         :meth:`register` and the :class:`~pcapkit.protocols.link.link.Link`,
         :class:`~pcapkit.protocols.internet.internet.Internet`,
         :class:`~pcapkit.protocols.transport.transport.Transport`,
-        :class:`~pcapkit.protocols.transport.sctp.SCTP` and
-        :class:`~pcapkit.protocols.misc.pcap.frame.Frame` overrides share this;
+        :class:`~pcapkit.protocols.transport.sctp.SCTP`,
+        :class:`~pcapkit.protocols.misc.pcap.frame.Frame` and
+        :class:`~pcapkit.protocols.misc.pcapng.PCAPNG` overrides share this;
         each still warns and stores for itself.
 
         Arguments:

tests/foundation/engines/test_pcapng_spb_zero_snaplen_1558_unit.py

@@ -0,0 +1,114 @@
+# -*- coding: utf-8 -*-
+"""GitHub issue #1558: a Simple Packet Block keeps its padding when the snaplen is 0.
+
+An SPB's packet data is ``min(original_len, snaplen)`` octets, with interface
+0's snaplen, and a snaplen of 0 means no limit. The default engine handed
+interface 0's snaplen to the parser as ``__packet__['snaplen']`` as it stood, so
+a 0 read as a limit of no octets, and the block's padding was read as packet
+data: 57 data octets came out as 60. The engine now hands over no limit for a
+zero snaplen, and :class:`~pcapkit.protocols.misc.pcapng.PCAPNG` reads a zero
+snaplen handed in as no limit too.
+
+Each case builds its capture here. Everything from :mod:`pcapkit` is imported
+inside each test, after :func:`~tests._support.reimport_once_per_class`.
+
+"""
+
+import os
+import struct
+import tempfile
+import unittest
+import warnings
+
+from tests._support import reimport_once_per_class
+
+#: 57 data octets, so the block pads them with 3 zero octets.
+DATA = bytes(range(1, 58))
+
+
+def _block(kind: int, body: bytes) -> bytes:
+    body += bytes(-len(body) % 4)
+    length = len(body) + 12
+    return struct.pack('<II', kind, length) + body + struct.pack('<I', length)
+
+
+def _spb() -> bytes:
+    return _block(3, struct.pack('<I', len(DATA)) + DATA)
+
+
+def _capture(snaplen: int) -> bytes:
+    """One SHB, one Ethernet IDB with ``snaplen``, and one SPB of :data:`DATA`."""
+    return b''.join([
+        _block(0x0A0D0D0A, struct.pack('<IHHq', 0x1A2B3C4D, 1, 0, -1)),
+        _block(1, struct.pack('<HHI', 1, 0, snaplen)),
+        _spb(),
+    ])
+
+
+class TestSPBZeroSnaplen(unittest.TestCase):
+    """A zero snaplen clips nothing, and the padding stays padding."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+        self._tmp = tempfile.TemporaryDirectory()  # pylint: disable=consider-using-with
+        self.addCleanup(self._tmp.cleanup)
+
+    def _extract(self, snaplen: int):  # type: ignore[no-untyped-def]
+        import pcapkit
+
+        path = os.path.join(self._tmp.name, f'snaplen-{snaplen}.pcapng')
+        with open(path, 'wb') as file:
+            file.write(_capture(snaplen))
+        with warnings.catch_warnings():
+            warnings.simplefilter('ignore')
+            return pcapkit.extract(fin=path, engine='default', nofile=True, store=True)
+
+    def test_engine_reads_the_original_length(self) -> None:
+        for snaplen in (0, 65535):
+            with self.subTest(snaplen=snaplen):
+                extractor = self._extract(snaplen)
+                self.assertEqual(extractor.length, 1)
+                info = extractor.frame[0].info
+                self.assertEqual(info.captured_len, len(DATA))
+                self.assertEqual(bytes(info.packet), DATA)
+
+    def test_engine_hands_over_no_limit_for_a_zero_snaplen(self) -> None:
+        engine = self._extract(0)._exeng
+        self.assertEqual(engine._ctx.interfaces[0].snaplen, 0)
+        self.assertEqual(engine._get_snaplen(), 0xFFFF_FFFF_FFFF_FFFF)
+
+    def test_engine_frame_rebuilds_byte_exactly(self) -> None:
+        from pcapkit.protocols.misc.pcapng import PCAPNG
+
+        extractor = self._extract(0)
+        frame, context = extractor.frame[0], extractor._exeng._ctx
+        for data in (frame.info, frame.info.to_dict()):
+            with self.subTest(data=type(data).__name__):
+                rebuilt = PCAPNG.from_data(data, num=1, sct=1, ctx=context)
+                self.assertEqual(rebuilt.data.hex(), _spb().hex())
+
+    def test_parser_reads_a_zero_snaplen_handed_in_as_no_limit(self) -> None:
+        from pcapkit.const.pcapng.block_type import BlockType
+        from pcapkit.foundation.engines.pcapng import Context
+        from pcapkit.protocols.misc.pcapng import PCAPNG
+
+        octets = _spb()
+        for snaplen in (0, 65535):
+            with self.subTest(interface_snaplen=snaplen):
+                section = PCAPNG(num=0, sct=1, ctx=None, type=BlockType.Section_Header_Block, block={})
+                context = Context(section.info)
+                interface = PCAPNG(num=1, sct=1, ctx=context, type=BlockType.Interface_Description_Block,
+                                   block={'linktype': 1, 'snaplen': snaplen})
+                context.interfaces.append(interface.info)
+                with warnings.catch_warnings():
+                    warnings.simplefilter('ignore')
+                    parsed = PCAPNG(octets, len(octets), num=2, sct=1, ctx=context,
+                                    __packet__={'snaplen': 0})
+                self.assertEqual(parsed.info.captured_len, len(DATA))
+                self.assertEqual(bytes(parsed.info.packet), DATA)
+                rebuilt = PCAPNG.from_data(parsed.info, num=2, sct=1, ctx=context)
+                self.assertEqual(rebuilt.data.hex(), octets.hex())
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/foundation/registry/test_registry_symmetry_unit.py

@@ -404,17 +404,7 @@ class TestRegistrySymmetry(harness.RoundTripBase):
     """Override then restore leaves every registry exactly as shipped."""
 
     STATUSES = ('RAISED', 'CHANGED')
-    KNOWN_FAILURES = (
-        Gap(1504, 'restoring a shipped PCAPNG.__proto__ entry over an override stores the class '
-               'it names, not the shipped ModuleDescriptor: PCAPNG.register resolves a '
-               'descriptor unequal to the incumbent (pcapkit/protocols/misc/pcapng.py:1027) '
-               'instead of sharing ProtocolBase._next_layer_entry, as the other layers do',
-            'CHANGED', ("PCAPNG.__proto__[<LinkType.ETHERNET: 1>] "
-                        "ModuleDescriptor(module='pcapkit.protocols.",),
-            tuple(f'{name}/{variant}/dispatch'
-                  for name in ('register_linktype', 'register_pcapng')
-                  for variant in ('override', 'override-same-name', 'restore-by-name'))),
-    )
+    KNOWN_FAILURES = ()  # type: tuple[Gap, ...]
 
     def setUp(self) -> None:
         super().setUp()

tests/protocols/test_register_restores_shipped_descriptor_1504_unit.py

@@ -2,18 +2,15 @@
 """GitHub issue #1504: restoring a shipped dispatch entry puts back its descriptor.
 
 :meth:`ProtocolBase.register <pcapkit.protocols.protocol.ProtocolBase.register>`
-and its ``Link``, ``Internet``, ``Transport`` (``TCP``, ``UDP``), ``SCTP`` and
-``Frame`` overrides resolved a :class:`~pcapkit.corekit.module.ModuleDescriptor`
-argument to its class before storing it. The #1364 guard returns early only when
-the argument equals the *incumbent*, so after an override, handing back the
-shipped descriptor stored the class it names: the round trip did not close, and
-the restore imported the module the entry was meant to defer. A descriptor equal
-to the one a code shipped with is now matched before it is resolved, and the
-shipped object itself is stored.
-
-``PCAPNG.register`` has the same defect and is not covered here; it stays
-recorded under ``Gap(1504, ...)`` in
-:file:`tests/foundation/registry/test_registry_symmetry_unit.py`.
+and its ``Link``, ``Internet``, ``Transport`` (``TCP``, ``UDP``), ``SCTP``,
+``Frame`` and ``PCAPNG`` overrides resolved a
+:class:`~pcapkit.corekit.module.ModuleDescriptor` argument to its class before
+storing it. The #1364 guard returns early only when the argument equals the
+*incumbent*, so after an override, handing back the shipped descriptor stored
+the class it names: the round trip did not close, and the restore imported the
+module the entry was meant to defer. A descriptor equal to the one a code
+shipped with is now matched before it is resolved, and the shipped object itself
+is stored.
 
 Every registry is process-wide, so each case runs under
 :func:`unittest.mock.patch.dict`. Classes are imported inside each test, after
@@ -46,6 +43,7 @@
      'pcapkit.const.sctp.payload_protocol_identifier:PayloadProtocolIdentifier.'
      'PayloadProtocolIdentifier_3GPP_NG_Application_Protocol'),
     ('pcapkit.protocols.misc.pcap.frame', 'Frame', 'pcapkit.const.reg.linktype:LinkType.ETHERNET'),
+    ('pcapkit.protocols.misc.pcapng', 'PCAPNG', 'pcapkit.const.reg.linktype:LinkType.ETHERNET'),
 )
 
 #: A code no site ships an entry under.