|
| 1 | +# -*- coding: utf-8 -*- |
| 2 | +"""A FRAD header rebuilds as captured and quietly at every cut, or is refused. |
| 3 | +
|
| 4 | +GitHub issue #2710: a :class:`~pcapkit.protocols.link.frad.FRAD` header the data ends |
| 5 | +before its NLPID reads that octet as the ``0x00`` pad, so its data model holds an NLPID |
| 6 | +of zero. :meth:`FRAD.make <pcapkit.protocols.link.frad.FRAD.make>` then left out the |
| 7 | +``ext_nlpid`` the schema reads after the pad, and the rebuild logged ``NoDefaultValue`` |
| 8 | +at ``CRITICAL``. It writes the pad and an NLPID of zero instead, which the schema's own |
| 9 | +short read cuts back. |
| 10 | +
|
| 11 | +The data model keeps no EA bits of the Q.922 address, so a header whose EA bits are not |
| 12 | +the 2-octet form, ``0`` then ``1``, rebuilt other octets, cut or whole. |
| 13 | +:meth:`FRAD.read <pcapkit.protocols.link.frad.FRAD.read>` refuses it, which a |
| 14 | +:class:`~pcapkit.protocols.link.linux_sll.LinuxSLL` frame keeps as |
| 15 | +:class:`~pcapkit.protocols.misc.raw.Raw`, quietly. An EA bit the data never held is not |
| 16 | +checked. |
| 17 | +
|
| 18 | +The frames are built here; this module is unit-tier and reads no capture file. |
| 19 | +
|
| 20 | +""" |
| 21 | +from __future__ import annotations |
| 22 | + |
| 23 | +import contextlib |
| 24 | +import logging |
| 25 | +import struct |
| 26 | +import unittest |
| 27 | +import warnings |
| 28 | +from typing import TYPE_CHECKING |
| 29 | + |
| 30 | +from tests._support import reimport_once_per_class |
| 31 | + |
| 32 | +if TYPE_CHECKING: |
| 33 | + from typing import Any, Callable, Iterator |
| 34 | + |
| 35 | +#: Test tier (#1958): hand-picked parse -> rebuild round trips. |
| 36 | +__tier__ = 'ALL' |
| 37 | + |
| 38 | +#: An IPv4 header (protocol UDP) and an 8-octet UDP header. |
| 39 | +IPV4 = (bytes.fromhex('4500001c00010000401100000a0000010a000002') |
| 40 | + + bytes.fromhex('3039003500080000')) |
| 41 | +#: A minimal ARP request (RFC 826), 28 octets. |
| 42 | +ARP = bytes.fromhex('00010800060400010011223344550a0000010000000000000a000002') |
| 43 | + |
| 44 | +#: Each frame cut at every octet: the RFC 2427 address of DLCI 50, UI control, then a |
| 45 | +#: direct NLPID, a SNAP header, or the pad and an NLPID of zero. |
| 46 | +FRAMES = { |
| 47 | + 'direct': bytes.fromhex('0c21' '03' 'cc') + IPV4, |
| 48 | + 'SNAP': bytes.fromhex('0c21' '03' '00' '80' '000000' '0806') + ARP, |
| 49 | + 'pad, zero NLPID': bytes.fromhex('0c21' '03' '00' '00') + b'\xaa\xbb', |
| 50 | +} |
| 51 | + |
| 52 | +#: Addresses of DLCI 50 whose EA bits are not ``0`` then ``1``. |
| 53 | +OTHER_FORMS = { |
| 54 | + 'EA 1 then 1': bytes.fromhex('0d21'), |
| 55 | + 'EA 0 then 0': bytes.fromhex('0c20'), |
| 56 | + 'EA 1 then 0': bytes.fromhex('0d20'), |
| 57 | +} |
| 58 | + |
| 59 | +#: A Linux cooked-capture source address, padded to its 8 octets. |
| 60 | +LINK_ADDR = bytes.fromhex('001122334455') + bytes(2) |
| 61 | + |
| 62 | + |
| 63 | +class _Logged(logging.Handler): |
| 64 | + """Collect what the ``pcapkit`` logger reports.""" |
| 65 | + |
| 66 | + def __init__(self) -> None: |
| 67 | + super().__init__(logging.DEBUG) |
| 68 | + self.messages = [] # type: list[str] |
| 69 | + |
| 70 | + def emit(self, record: 'logging.LogRecord') -> None: |
| 71 | + self.messages.append(f'logged: {record.getMessage()}') |
| 72 | + |
| 73 | + |
| 74 | +@contextlib.contextmanager |
| 75 | +def _reported() -> 'Iterator[list[str]]': |
| 76 | + """Collect what the block reports, on the :mod:`warnings` channel and the logger.""" |
| 77 | + handler = _Logged() |
| 78 | + logger = logging.getLogger('pcapkit') |
| 79 | + logger.addHandler(handler) |
| 80 | + try: |
| 81 | + with warnings.catch_warnings(record=True) as caught: |
| 82 | + warnings.simplefilter('always') |
| 83 | + yield handler.messages |
| 84 | + finally: |
| 85 | + logger.removeHandler(handler) |
| 86 | + handler.messages.extend(f'warned: {item.message}' for item in caught) |
| 87 | + |
| 88 | + |
| 89 | +class FRADCutRebuildTests(unittest.TestCase): |
| 90 | + """A FRAD header rebuilds as captured, or the address form is refused.""" |
| 91 | + |
| 92 | + def setUp(self) -> None: |
| 93 | + reimport_once_per_class(self) |
| 94 | + from pcapkit.const.linux.arphrd import ARPHRD |
| 95 | + from pcapkit.protocols.link.frad import FRAD |
| 96 | + from pcapkit.protocols.link.linux_sll1 import LinuxSLL1 |
| 97 | + from pcapkit.protocols.link.linux_sll2 import LinuxSLL2 |
| 98 | + from pcapkit.utilities.exceptions import ProtocolError |
| 99 | + |
| 100 | + self.FRAD = FRAD |
| 101 | + self.ProtocolError = ProtocolError |
| 102 | + self.cooked = { |
| 103 | + 'LinuxSLL1': (LinuxSLL1, struct.pack('>HHH', 0, ARPHRD.FRAD, 6) + LINK_ADDR |
| 104 | + + struct.pack('>H', 0)), |
| 105 | + 'LinuxSLL2': (LinuxSLL2, struct.pack('>HHIHBB', 0, 0, 1, ARPHRD.FRAD, 0, 6) + LINK_ADDR), |
| 106 | + } # type: dict[str, tuple[Any, bytes]] |
| 107 | + |
| 108 | + def parse(self, build: 'Callable[[], Any]') -> 'tuple[Any, list[str]]': |
| 109 | + """What ``build`` returns, and what it reported.""" |
| 110 | + with _reported() as messages: |
| 111 | + layer = build() |
| 112 | + return layer, messages |
| 113 | + |
| 114 | + def assertRebuildsQuietly(self, layer: 'Any') -> 'None': # pylint: disable=invalid-name |
| 115 | + """``layer`` rebuilt from its data model and its dict: its octets, and nothing reported.""" |
| 116 | + for path, data in (('model', layer.info), ('dict', layer.info.to_dict())): |
| 117 | + with self.subTest(path=path): |
| 118 | + rebuilt, messages = self.parse(lambda: type(layer).from_data(data)) # pylint: disable=cell-var-from-loop |
| 119 | + self.assertEqual((rebuilt.data.hex(), messages), (layer.data.hex(), [])) |
| 120 | + |
| 121 | + def test_every_cut_rebuilds_as_captured(self) -> None: |
| 122 | + for label, frame in FRAMES.items(): |
| 123 | + for count in range(1, len(frame) + 1): |
| 124 | + with self.subTest(frame=label, captured=count): |
| 125 | + with warnings.catch_warnings(): |
| 126 | + warnings.simplefilter('ignore') |
| 127 | + layer = self.FRAD(frame[:count], count) |
| 128 | + self.assertEqual(layer.data, frame[:count]) |
| 129 | + self.assertRebuildsQuietly(layer) |
| 130 | + |
| 131 | + def test_a_cut_before_the_nlpid_reads_the_pad(self) -> None: |
| 132 | + for count in (1, 2, 3): |
| 133 | + with self.subTest(captured=count): |
| 134 | + with warnings.catch_warnings(): |
| 135 | + warnings.simplefilter('ignore') |
| 136 | + layer = self.FRAD(FRAMES['direct'][:count], count) |
| 137 | + self.assertIsNotNone(layer.info.get('__short_read__')) |
| 138 | + self.assertEqual((layer.info.nlpid, layer.info.len), (0, 5)) |
| 139 | + |
| 140 | + def test_make_writes_the_pad_and_a_zero_nlpid(self) -> None: |
| 141 | + made, messages = self.parse(lambda: self.FRAD(dlci=50, nlpid=0)) |
| 142 | + self.assertEqual((made.data.hex(), messages), ('0c21030000', [])) |
| 143 | + |
| 144 | + def test_another_address_form_is_refused(self) -> None: |
| 145 | + from pcapkit.utilities.exceptions import CaughtByCaller |
| 146 | + |
| 147 | + for label, addr in OTHER_FORMS.items(): |
| 148 | + frame = addr + FRAMES['direct'][2:] |
| 149 | + for count in range(2, len(frame) + 1): |
| 150 | + with self.subTest(address=label, captured=count), CaughtByCaller(self.ProtocolError), \ |
| 151 | + self.assertRaisesRegex(self.ProtocolError, r'^FRAD: not a 2-octet Q\.922 address: EA bits'): |
| 152 | + self.FRAD(frame[:count], count) |
| 153 | + |
| 154 | + def test_an_ea_bit_cut_off_is_not_checked(self) -> None: |
| 155 | + # One octet holds the first EA bit alone: 0 reads, 1 is refused. |
| 156 | + from pcapkit.utilities.exceptions import CaughtByCaller |
| 157 | + |
| 158 | + with warnings.catch_warnings(): |
| 159 | + warnings.simplefilter('ignore') |
| 160 | + layer = self.FRAD(b'\x0c', 1) |
| 161 | + self.assertEqual(layer.info.get('__short_read__'), ('addr', 1)) |
| 162 | + self.assertRebuildsQuietly(layer) |
| 163 | + with CaughtByCaller(self.ProtocolError), self.assertRaises(self.ProtocolError): |
| 164 | + self.FRAD(b'\x0d', 1) |
| 165 | + |
| 166 | + def test_linux_sll_keeps_another_address_form_raw(self) -> None: |
| 167 | + for name, (cls, head) in self.cooked.items(): |
| 168 | + for label, addr in OTHER_FORMS.items(): |
| 169 | + raw = head + addr + FRAMES['direct'][2:] |
| 170 | + with self.subTest(cooked=name, address=label): |
| 171 | + frame, messages = self.parse(lambda: cls(raw, len(raw))) # pylint: disable=cell-var-from-loop |
| 172 | + self.assertEqual(messages, []) |
| 173 | + self.assertEqual(type(frame.payload).__name__, 'Raw') |
| 174 | + self.assertRegex(frame.payload.info.error, r'^FRAD: not a 2-octet Q\.922 address: ') |
| 175 | + self.assertEqual(frame.data, raw) |
| 176 | + self.assertRebuildsQuietly(frame) |
| 177 | + |
| 178 | + def test_linux_sll_dispatches_the_2_octet_form(self) -> None: |
| 179 | + # The control: the RFC 2427 address is still FRAD behind LinuxSLL. |
| 180 | + for name, (cls, head) in self.cooked.items(): |
| 181 | + raw = head + FRAMES['direct'] |
| 182 | + with self.subTest(cooked=name): |
| 183 | + frame, messages = self.parse(lambda: cls(raw, len(raw))) # pylint: disable=cell-var-from-loop |
| 184 | + self.assertEqual(messages, []) |
| 185 | + self.assertIsInstance(frame.payload, self.FRAD) |
| 186 | + self.assertRebuildsQuietly(frame) |
| 187 | + |
| 188 | + |
| 189 | +if __name__ == '__main__': |
| 190 | + unittest.main() |
0 commit comments