From e864fd8012c5e9d1c7c7b55ae8480879184044d9 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Mon, 2 Nov 2026 21:34:43 -0500 Subject: [PATCH] fix(link): keep a captured FRAD pad octet through a whole-frame rebuild (#2716) RFC 2427 section 3 allows the one zero pad octet ahead of any NLPID, and FRAD reads one wherever it is, but the data model did not record it, so make() wrote the pad ahead of a SNAP header alone. A pad ahead of a direct NLPID (0c21 03 00 cc) was lost on a rebuild, and one was invented ahead of a SNAP header captured without it (0c21 03 80 ...). - The FRAD data model records pad, and _make_data passes it on. - make() takes pad. When it is None, the pad goes 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, so a direct build is unchanged. pad=False with an NLPID of 0 raises ProtocolError. - The FRAD prose stops calling the pad SNAP's alone. test_frad_pad_rebuild_2716_unit fails without the pcapkit/ change, but for the control that a direct build is unchanged. Shipped captures parse identically. The link, internet, project and root protocols modules, both schema round trips, and the #2710 and short-read sweeps pass. Closes #2716 --- docs/source/pcapkit/protocols/link/frad.rst | 9 +- pcapkit/protocols/data/link/frad.py | 8 +- pcapkit/protocols/link/frad.py | 47 +++-- pcapkit/protocols/schema/link/frad.py | 6 +- .../link/test_frad_pad_rebuild_2716_unit.py | 183 ++++++++++++++++++ 5 files changed, 226 insertions(+), 27 deletions(-) create mode 100644 tests/protocols/link/test_frad_pad_rebuild_2716_unit.py diff --git a/docs/source/pcapkit/protocols/link/frad.rst b/docs/source/pcapkit/protocols/link/frad.rst index 4900e94398..92638ea319 100644 --- a/docs/source/pcapkit/protocols/link/frad.rst +++ b/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: diff --git a/pcapkit/protocols/data/link/frad.py b/pcapkit/protocols/data/link/frad.py index 6af1092de4..876e748858 100644 --- a/pcapkit/protocols/data/link/frad.py +++ b/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 diff --git a/pcapkit/protocols/link/frad.py b/pcapkit/protocols/link/frad.py index a1b84fb6c0..1b43c9d30b 100644 --- a/pcapkit/protocols/link/frad.py +++ b/pcapkit/protocols/link/frad.py @@ -15,9 +15,9 @@ :attr:`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 ` 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__ ` @@ -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), diff --git a/pcapkit/protocols/schema/link/frad.py b/pcapkit/protocols/schema/link/frad.py index 9c798a18dc..22d8796d83 100644 --- a/pcapkit/protocols/schema/link/frad.py +++ b/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'], diff --git a/tests/protocols/link/test_frad_pad_rebuild_2716_unit.py b/tests/protocols/link/test_frad_pad_rebuild_2716_unit.py new file mode 100644 index 0000000000..b289294f50 --- /dev/null +++ b/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 ` 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()