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()