fix(link): keep a captured FRAD pad octet through a whole-frame rebuild (#2716)

docs/source/pcapkit/protocols/link/frad.rst

@@ -18,13 +18,12 @@ LAPF address onward with no trailing FCS, is described as below:
    ====== ==== ================= ==========================================
    0      0    ``frad.addr``     Q.922 Address (2-octet form)
    2      16   ``frad.control``  Control (``0x03``, UI, for data traffic)
-   3      24   ``frad.nlpid``    NLPID, or the ``0x00`` pad ahead of a SNAP
-                                  header
+   3      24   ``frad.nlpid``    NLPID, or the ``0x00`` pad ahead of one
    ====== ==== ================= ==========================================
 
-When ``nlpid`` is the ``0x00`` pad marker, a further NLPID octet (``0x80``
-for SNAP), a 3-octet OUI and a 2-octet PID follow, kept as ``frad.ext_nlpid``,
-``frad.oui`` and ``frad.pid``.
+When ``nlpid`` is the ``0x00`` pad, the NLPID follows it as ``frad.ext_nlpid``.
+An NLPID of ``0x80``, with or without the pad, is followed by a SNAP header, a
+3-octet OUI and a 2-octet PID, kept as ``frad.oui`` and ``frad.pid``.
 
 .. autoclass:: pcapkit.protocols.link.frad.FRAD
    :no-members:

pcapkit/protocols/data/link/frad.py

@@ -31,15 +31,17 @@ class FRAD(Protocol):
     #: NLPID selecting the encapsulated protocol, with any ``0x00`` pad
     #: already resolved to the real value that followed it.
     nlpid: 'int'
+    #: Whether a ``0x00`` pad octet precedes the NLPID (:rfc:`2427#section-3`).
+    pad: 'bool'
     #: SNAP OUI, present only when :attr:`nlpid` is ``0x80``.
     snap_oui: 'Optional[bytes]'
     #: SNAP PID, present only when :attr:`nlpid` is ``0x80``.
     snap_pid: 'Optional[int]'
-    #: Header length, in octets -- 4 for a direct NLPID, 5 with the pad byte,
-    #: or 10 with the SNAP header besides.
+    #: Header length, in octets -- 4 for a direct NLPID, one more with the pad
+    #: octet, and 5 more with a SNAP header.
     len: 'int'
 
     if TYPE_CHECKING:
         def __init__(self, dlci: 'int', cr: 'int', fecn: 'int', becn: 'int', de: 'int',  # pylint: disable=unused-argument,super-init-not-called
-                     control: 'int', nlpid: 'int', snap_oui: 'Optional[bytes]',  # pylint: disable=unused-argument
+                     control: 'int', nlpid: 'int', pad: 'bool', snap_oui: 'Optional[bytes]',  # pylint: disable=unused-argument
                      snap_pid: 'Optional[int]', len: 'int') -> 'None': ...  # pylint: disable=unused-argument,redefined-builtin

pcapkit/protocols/link/frad.py

@@ -15,9 +15,9 @@
 :attr:`LinuxSLL.__proto__ <pcapkit.protocols.link.linux_sll.LinuxSLL.__proto__>`. Its
 structure is the Q.922 LAPF address onward, with no trailing FCS: a 2-octet
 ``frad.addr``, a 1-octet ``frad.control`` (``0x03``, UI, for ordinary data
-traffic), then a 1-octet ``frad.nlpid`` -- or, when SNAP-encapsulated, the
-``0x00`` pad RFC 2427 writes in its place, followed by the real NLPID
-(``0x80``) and a 5-octet SNAP header (OUI and PID). See
+traffic), then a 1-octet ``frad.nlpid``, which a ``0x00`` pad octet may
+precede. RFC 2427 writes the pad ahead of a SNAP header: the NLPID ``0x80``,
+then a 5-octet OUI and PID. See
 :meth:`FRAD.read <pcapkit.protocols.link.frad.FRAD.read>` for the full byte
 diagram and :mod:`pcapkit.protocols.schema.link.frad` for the field-by-field
 schema.
@@ -54,8 +54,8 @@ class FRAD(Link[Data_FRAD, Schema_FRAD],
           schema=Schema_FRAD, data=Data_FRAD):
     """This class implements Frame Relay Access Device (FRAD) frames.
 
-    A frame whose NLPID (after resolving the ``0x00`` pad RFC 2427 writes
-    ahead of a SNAP header) is ``0xCC`` or ``0x8E`` dispatches to
+    A frame whose NLPID (after resolving any ``0x00`` pad ahead of it) is
+    ``0xCC`` or ``0x8E`` dispatches to
     :class:`~pcapkit.protocols.internet.ipv4.IPv4` or
     :class:`~pcapkit.protocols.internet.ipv6.IPv6` respectively, registered
     in the :attr:`self.__proto__ <pcapkit.protocols.link.frad.FRAD.__proto__>`
@@ -145,7 +145,10 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_FRAD':
 
         A SNAP-encapsulated frame inserts a ``0x00`` pad and a ``0x80`` NLPID
         in place of the single NLPID octet above, followed by a 3-octet OUI
-        and a 2-octet PID (:rfc:`2427#section-4.1`).
+        and a 2-octet PID (:rfc:`2427#section-4.1`). The pad may precede any
+        NLPID, or be left out ahead of ``0x80`` (:rfc:`2427#section-3`), and
+        :attr:`~pcapkit.protocols.data.link.frad.FRAD.pad` records whether
+        it is there.
 
         Args:
             length: Length of packet data.
@@ -175,7 +178,8 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_FRAD':
         _dlci = (addr['dlci_hi'] << 4) | addr['dlci_lo']
 
         _raw_nlpid = schema.nlpid
-        if not _raw_nlpid:
+        _pad = not _raw_nlpid
+        if _pad:
             _ext = schema.ext_nlpid
             _nlpid = _ext if isinstance(_ext, int) else 0
             _hdr_len = 5
@@ -201,6 +205,7 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_FRAD':
             de=addr['de'],
             control=schema.control,
             nlpid=_nlpid,
+            pad=_pad,
             snap_oui=_snap_oui,
             snap_pid=_snap_pid,
             len=_hdr_len,
@@ -215,6 +220,7 @@ def make(self,
              de: 'int' = 0,
              control: 'int' = 0x03,
              nlpid: 'int' = 0xCC,
+             pad: 'Optional[bool]' = None,
              snap_oui: 'Optional[bytes]' = None,
              snap_pid: 'Optional[int]' = None,
              payload: 'bytes | ProtocolBase | Schema' = b'',
@@ -230,6 +236,10 @@ def make(self,
             control: Control field; ``0x03`` (UI) for ordinary data traffic.
             nlpid: NLPID selecting the encapsulated protocol; ``0xCC`` for
                 IPv4, ``0x8E`` for IPv6, ``0x80`` for a SNAP header.
+            pad: Whether a ``0x00`` pad octet precedes ``nlpid``
+                (:rfc:`2427#section-3`). When :data:`None`, there is one ahead
+                of a SNAP header alone, as :rfc:`2427#section-4.1` writes it,
+                and ahead of an ``nlpid`` of ``0``, which reads as the pad.
             snap_oui: SNAP OUI; required, and only meaningful, when ``nlpid``
                 is ``0x80``.
             snap_pid: SNAP PID; required, and only meaningful, when ``nlpid``
@@ -245,9 +255,10 @@ def make(self,
                 ``fecn``, ``becn`` or ``de`` is not ``0`` or ``1``; if
                 ``control`` or ``nlpid`` does not fit in 8 bits; if ``nlpid``
                 is ``0x80`` and ``snap_oui`` is not exactly 3 octets or
-                ``snap_pid`` does not fit in 16 bits; or if either
+                ``snap_pid`` does not fit in 16 bits; if either
                 ``snap_oui`` or ``snap_pid`` is given while ``nlpid`` is not
-                ``0x80``.
+                ``0x80``; or if ``pad`` is :data:`False` with an ``nlpid`` of
+                ``0``.
 
         Note:
             ``nlpid`` is not refused for being ``0`` -- the value of the one
@@ -273,22 +284,25 @@ def make(self,
         control = self._make_uint(control, bits=8, name='control')
         nlpid = self._make_uint(nlpid, bits=8, name='nlpid')
 
+        if pad is None:
+            pad = nlpid in (0x00, 0x80)
+        elif not pad and not nlpid:
+            raise ProtocolError(f'{self.alias}: an nlpid of 0 reads as the pad, so pad cannot be False')
+
         if nlpid == 0x80:
             if not isinstance(snap_oui, (bytes, bytearray)) or len(snap_oui) != 3:
                 raise ProtocolError(f'{self.alias}: invalid snap_oui: {snap_oui!r}, '
                                     f'must be exactly 3 octets when nlpid is 0x80')
-            _schema_nlpid, _ext_nlpid = 0, 0x80
             _oui, _pid = bytes(snap_oui), self._make_uint(snap_pid, bits=16, name='snap_pid')
         else:
             if snap_oui is not None or snap_pid is not None:
                 raise ProtocolError(f'{self.alias}: snap_oui and snap_pid are only meaningful '
                                     f'when nlpid is 0x80, not {nlpid!r}')
-            _schema_nlpid, _ext_nlpid = nlpid, None
             _oui, _pid = None, None
-            if not nlpid:
-                # NOTE: A zero NLPID is the pad, which the schema follows with
-                # ``ext_nlpid``, so that is written too (#2710).
-                _ext_nlpid = 0
+
+        # NOTE: The pad is a zero ``nlpid`` octet, which the schema follows with
+        # ``ext_nlpid``: a zero NLPID is written that way too (#2710, #2716).
+        _schema_nlpid, _ext_nlpid = (0, nlpid) if pad else (nlpid, None)
 
         return Schema_FRAD(
             addr={
@@ -318,7 +332,7 @@ def register(cls, code: 'int', protocol: 'ModuleDescriptor[ProtocolBase] | Type[
             should be ``{protocol.module}.{protocol.name}``.
 
         Arguments:
-            code: NLPID, after resolving the ``0x00`` pad ahead of a SNAP header
+            code: NLPID, after resolving any ``0x00`` pad ahead of it
             protocol: module descriptor or a
                 :class:`~pcapkit.protocols.protocol.ProtocolBase` subclass
 
@@ -405,6 +419,7 @@ def _make_data(cls, data: 'Data_FRAD') -> 'dict[str, Any]':  # type: ignore[over
             'de': data.de,
             'control': data.control,
             'nlpid': data.nlpid,
+            'pad': data.pad,
             'snap_oui': data.snap_oui,
             'snap_pid': data.snap_pid,
             'payload': cls._make_payload(data),

pcapkit/protocols/schema/link/frad.py

@@ -51,7 +51,7 @@ def _effective_nlpid(packet: 'dict[str, Any]') -> 'int':
 
     Returns:
         :attr:`~FRAD.ext_nlpid` when :attr:`~FRAD.nlpid` is the ``0x00`` pad
-        marker :rfc:`2427#section-3` writes ahead of a SNAP header; otherwise
+        :rfc:`2427#section-3` allows ahead of any NLPID; otherwise
         :attr:`~FRAD.nlpid` itself.
 
     """
@@ -92,9 +92,9 @@ class FRAD(Schema):
     })
     #: Control field; ``0x03`` (UI) for ordinary data traffic.
     control: 'int' = UInt8Field()
-    #: NLPID, or the ``0x00`` pad marker ahead of a SNAP header.
+    #: NLPID, or the ``0x00`` pad octet ahead of one.
     nlpid: 'int' = UInt8Field()
-    #: NLPID proper, present only when :attr:`nlpid` is the pad marker.
+    #: NLPID proper, present only when :attr:`nlpid` is the pad.
     ext_nlpid: 'Optional[int]' = ConditionalField(
         UInt8Field(),
         lambda packet: not packet['nlpid'],

tests/protocols/link/test_frad_pad_rebuild_2716_unit.py

@@ -0,0 +1,183 @@
+# -*- coding: utf-8 -*-
+"""A FRAD pad octet is kept, and never invented, through a rebuild.
+
+GitHub issue #2716: :rfc:`2427#section-3` allows a single ``0x00`` pad octet ahead of
+any NLPID, and :class:`~pcapkit.protocols.link.frad.FRAD` reads one ahead of every
+NLPID, but its data model did not record whether one was there. So
+:meth:`FRAD.make <pcapkit.protocols.link.frad.FRAD.make>` wrote the pad ahead of a
+SNAP header alone: a pad before a direct NLPID was lost on a rebuild, and one was
+invented ahead of a SNAP header captured without it. The data model's ``pad``
+records it, and ``make()`` takes it, writing the pad ahead of a SNAP header alone
+when it is not given.
+
+The frames are built here; this module is unit-tier and reads no capture file.
+
+"""
+from __future__ import annotations
+
+import contextlib
+import logging
+import struct
+import unittest
+import warnings
+from typing import TYPE_CHECKING
+
+from tests._support import reimport_once_per_class
+
+if TYPE_CHECKING:
+    from typing import Any, Callable, Iterator
+
+#: Test tier (#1958): hand-picked parse -> rebuild round trips.
+__tier__ = 'ALL'
+
+#: An IPv4 header (protocol UDP) and an 8-octet UDP header.
+IPV4 = (bytes.fromhex('4500001c00010000401100000a0000010a000002')
+        + bytes.fromhex('3039003500080000'))
+#: A minimal ARP request (RFC 826), 28 octets.
+ARP = bytes.fromhex('00010800060400010011223344550a0000010000000000000a000002')
+
+#: Each frame: the RFC 2427 address of DLCI 50 and UI control, then the NLPID with
+#: or without the pad; its ``pad``, ``nlpid`` and header length; and its protocol chain.
+FRAMES = {
+    'pad, IPv4': (bytes.fromhex('0c21' '03' '00' 'cc') + IPV4, (True, 0xCC, 5), 'FRAD:IPv4:UDP'),
+    'SNAP without the pad': (bytes.fromhex('0c21' '03' '80' '000000' '0806') + ARP, (False, 0x80, 9),
+                             'FRAD:ARP'),
+    # the forms the maker writes when ``pad`` is not given, as controls
+    'IPv4': (bytes.fromhex('0c21' '03' 'cc') + IPV4, (False, 0xCC, 4), 'FRAD:IPv4:UDP'),
+    'pad, SNAP': (bytes.fromhex('0c21' '03' '00' '80' '000000' '0806') + ARP, (True, 0x80, 10), 'FRAD:ARP'),
+}  # type: dict[str, tuple[bytes, tuple[bool, int, int], str]]
+
+#: A Linux cooked-capture source address, padded to its 8 octets.
+LINK_ADDR = bytes.fromhex('001122334455') + bytes(2)
+
+
+class _Logged(logging.Handler):
+    """Collect what the ``pcapkit`` logger reports."""
+
+    def __init__(self) -> None:
+        super().__init__(logging.DEBUG)
+        self.messages = []  # type: list[str]
+
+    def emit(self, record: 'logging.LogRecord') -> None:
+        self.messages.append(f'logged: {record.getMessage()}')
+
+
+@contextlib.contextmanager
+def _reported() -> 'Iterator[list[str]]':
+    """Collect what the block reports, on the :mod:`warnings` channel and the logger."""
+    handler = _Logged()
+    logger = logging.getLogger('pcapkit')
+    logger.addHandler(handler)
+    try:
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter('always')
+            yield handler.messages
+    finally:
+        logger.removeHandler(handler)
+    handler.messages.extend(f'warned: {item.message}' for item in caught)
+
+
+class FRADPadRebuildTests(unittest.TestCase):
+    """A FRAD frame rebuilds with the pad octet it was captured with, and no other."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+        from pcapkit.const.linux.arphrd import ARPHRD
+        from pcapkit.protocols.link.frad import FRAD
+        from pcapkit.protocols.link.linux_sll1 import LinuxSLL1
+        from pcapkit.protocols.link.linux_sll2 import LinuxSLL2
+        from pcapkit.utilities.exceptions import ProtocolError
+
+        self.FRAD = FRAD
+        self.ProtocolError = ProtocolError
+        self.cooked = {
+            'LinuxSLL1': (LinuxSLL1, struct.pack('>HHH', 0, ARPHRD.FRAD, 6) + LINK_ADDR
+                          + struct.pack('>H', 0)),
+            'LinuxSLL2': (LinuxSLL2, struct.pack('>HHIHBB', 0, 0, 1, ARPHRD.FRAD, 0, 6) + LINK_ADDR),
+        }  # type: dict[str, tuple[Any, bytes]]
+
+    def parse(self, build: 'Callable[[], Any]') -> 'tuple[Any, list[str]]':
+        """What ``build`` returns, and what it reported."""
+        with _reported() as messages:
+            layer = build()
+        return layer, messages
+
+    def assertRebuildsQuietly(self, layer: 'Any') -> 'None':  # pylint: disable=invalid-name
+        """``layer`` rebuilt from its data model and its dict: its octets, and nothing reported."""
+        for path, data in (('model', layer.info), ('dict', layer.info.to_dict())):
+            with self.subTest(path=path):
+                rebuilt, messages = self.parse(lambda: type(layer).from_data(data))  # pylint: disable=cell-var-from-loop
+                self.assertEqual((rebuilt.data.hex(), messages), (layer.data.hex(), []))
+
+    def test_a_whole_frame_reads_its_pad(self) -> None:
+        for label, (frame, (pad, nlpid, length), chain) in FRAMES.items():
+            with self.subTest(frame=label):
+                layer, messages = self.parse(lambda: self.FRAD(frame, len(frame)))  # pylint: disable=cell-var-from-loop
+                self.assertEqual(messages, [])
+                self.assertEqual((layer.info.pad, layer.info.nlpid, layer.info.len), (pad, nlpid, length))
+                self.assertEqual(str(layer.protochain), chain)
+
+    def test_a_whole_frame_rebuilds_as_captured(self) -> None:
+        for label, (frame, _, _) in FRAMES.items():
+            with self.subTest(frame=label):
+                layer = self.FRAD(frame, len(frame))
+                self.assertRebuildsQuietly(layer)
+
+    def test_every_cut_rebuilds_as_captured(self) -> None:
+        for label, (frame, _, _) in FRAMES.items():
+            for count in range(1, len(frame)):
+                with self.subTest(frame=label, captured=count):
+                    with warnings.catch_warnings():
+                        warnings.simplefilter('ignore')
+                        layer = self.FRAD(frame[:count], count)
+                    self.assertEqual(layer.data, frame[:count])
+                    self.assertRebuildsQuietly(layer)
+
+    def test_linux_sll_rebuilds_the_pad_as_captured(self) -> None:
+        for name, (cls, head) in self.cooked.items():
+            for label, (frame, (pad, _, _), chain) in FRAMES.items():
+                raw = head + frame
+                with self.subTest(cooked=name, frame=label):
+                    sll, messages = self.parse(lambda: cls(raw, len(raw)))  # pylint: disable=cell-var-from-loop
+                    self.assertEqual(messages, [])
+                    self.assertEqual(str(sll.protochain).split(':', 1)[1], chain)
+                    self.assertIs(sll.payload.info.pad, pad)
+                    self.assertRebuildsQuietly(sll)
+
+    def test_make_writes_the_pad_ahead_of_snap_alone_by_default(self) -> None:
+        # A direct build is as it was: the pad ahead of a SNAP header, and an
+        # NLPID of 0 written as the pad it reads as.
+        cases = {
+            'IPv4': ({'nlpid': 0xCC}, '0c2103cc'),
+            'SNAP': ({'nlpid': 0x80, 'snap_oui': bytes(3), 'snap_pid': 0x0806}, '0c2103008000000008' '06'),
+            'zero NLPID': ({'nlpid': 0}, '0c21030000'),
+        }
+        for label, (kwargs, octets) in cases.items():
+            with self.subTest(case=label):
+                made, messages = self.parse(lambda: self.FRAD(dlci=50, **kwargs))  # pylint: disable=cell-var-from-loop
+                self.assertEqual((made.data.hex(), messages), (octets, []))
+
+    def test_make_takes_the_pad(self) -> None:
+        cases = {
+            'pad, IPv4': ({'nlpid': 0xCC, 'pad': True}, '0c2103' '00cc'),
+            'IPv4 without the pad': ({'nlpid': 0xCC, 'pad': False}, '0c2103' 'cc'),
+            'SNAP without the pad': ({'nlpid': 0x80, 'pad': False, 'snap_oui': bytes(3), 'snap_pid': 0x0806},
+                                     '0c2103' '80' '000000' '0806'),
+            'pad, zero NLPID': ({'nlpid': 0, 'pad': True}, '0c2103' '0000'),
+        }
+        for label, (kwargs, octets) in cases.items():
+            with self.subTest(case=label):
+                made, messages = self.parse(lambda: self.FRAD(dlci=50, **kwargs))  # pylint: disable=cell-var-from-loop
+                self.assertEqual((made.data.hex(), messages), (octets, []))
+                self.assertIs(self.FRAD(made.data, len(made.data)).info.pad, kwargs['pad'])
+
+    def test_make_refuses_a_zero_nlpid_without_the_pad(self) -> None:
+        from pcapkit.utilities.exceptions import CaughtByCaller
+
+        with CaughtByCaller(self.ProtocolError), self.assertRaisesRegex(
+                self.ProtocolError, r'^FRAD: an nlpid of 0 reads as the pad, so pad cannot be False$'):
+            self.FRAD(nlpid=0, pad=False)
+
+
+if __name__ == '__main__':
+    unittest.main()