diff --git a/pcapkit/protocols/link/linux_sll.py b/pcapkit/protocols/link/linux_sll.py index b2f7875492..ee7d8b5c0a 100644 --- a/pcapkit/protocols/link/linux_sll.py +++ b/pcapkit/protocols/link/linux_sll.py @@ -238,8 +238,9 @@ def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *, # ` registers for ``arphrd_type`` when it has a key for it, and otherwise the class :attr:`Link.__proto__ ` registers for ``proto`` - as an EtherType. A :class:`~pcapkit.protocols.misc.raw.Raw` fallback - records ``proto`` either way. + as an EtherType. Where parsing stops at this layer, either path + keeps the payload :class:`~pcapkit.protocols.misc.raw.Raw`, and a + :class:`~pcapkit.protocols.misc.raw.Raw` records ``proto`` either way. Note: :attr:`self.__proto__ ` holds the @@ -261,13 +262,11 @@ def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *, # if not min(length, len(file_)): from pcapkit.protocols.misc.null import \ NoPayload as protocol # isort: skip # pylint: disable=import-outside-toplevel - # NOTE: ahead of ``_sigterm``, so a registered ``arphrd_type`` is parsed - # even where parsing stops at this layer (#2510). - elif arphrd_type in self.__proto__: - protocol = self._lookup_next_layer(self.__proto__, arphrd_type) elif self._sigterm: from pcapkit.protocols.misc.raw import \ Raw as protocol # isort: skip # pylint: disable=import-outside-toplevel + elif arphrd_type in self.__proto__: + protocol = self._lookup_next_layer(self.__proto__, arphrd_type) else: protocol = self._lookup_next_layer(Link.__proto__, proto) diff --git a/tests/protocols/link/test_linux_sll_arphrd_registry_unit.py b/tests/protocols/link/test_linux_sll_arphrd_registry_unit.py index 7b88aab280..b8515af517 100644 --- a/tests/protocols/link/test_linux_sll_arphrd_registry_unit.py +++ b/tests/protocols/link/test_linux_sll_arphrd_registry_unit.py @@ -4,6 +4,7 @@ Covers :attr:`LinuxSLL.__proto__ ` as shipped, dispatch through it and through the EtherType path from both versions, +a parse stopping at either version on either path (#2532), the ``__index__`` of the classes it reaches, and the registrars of it and of :class:`~pcapkit.protocols.link.frad.FRAD`'s and :class:`~pcapkit.protocols.link.netlink.Netlink`'s registries. @@ -138,25 +139,73 @@ def test_an_empty_payload_is_no_payload_on_either_path(self) -> None: raw = header(0x0800, arphrd) self.assertIsInstance(version(raw, len(raw)).payload, NoPayload) - def test_a_stop_at_this_layer_still_parses_a_registered_arphrd_type(self) -> None: - """The order the NOTE in ``LinuxSLL._import_next_layer`` records. + def _stops(self, version): # type: ignore[no-untyped-def] + """Each spelling of a stop at ``version``: its layer, family and own name or class.""" + from pcapkit.protocols.link.linux_sll import LinuxSLL + + return ({'layer': 'Link'}, {'layer': 'link'}, {'protocol': 'LinuxSLL'}, + {'protocol': version.__name__}, {'protocol': LinuxSLL}, {'protocol': version}) - Stopping at the link layer or at the version keeps an EtherType payload - :class:`~pcapkit.protocols.misc.raw.Raw`, while a registered - ``arphrd_type`` is still dispatched. - """ + def test_a_stop_at_this_layer_keeps_a_registered_arphrd_type_raw(self) -> None: + """The registry path stops as the EtherType path does (#2532).""" from pcapkit.const.linux.arphrd import ARPHRD + from pcapkit.protocols.misc.raw import Raw + + for version, header, alias in self._versions(): + for name, (payload, _) in CASES.items(): + for stop in self._stops(version): + with self.subTest(version=version.__name__, arphrd=name, stop=stop): + raw = header(0x0800, getattr(ARPHRD, name)) + payload + parsed = version(raw, len(raw), **stop) + self.assertEqual(parsed.protochain.chain, f'{alias}:Raw') + self.assertIsInstance(parsed.payload, Raw) + self.assertEqual(parsed.payload.data, payload) + self.assertEqual(parsed.payload.info.protocol, 0x0800) + self.assertEqual(parsed.data, raw) + self.assertEqual(version.from_data(parsed.info).data, raw) + + # Without a length, the whole payload is the next layer's. + again = parsed._import_next_layer(parsed.protocol) + self.assertIsInstance(again, Raw) + self.assertEqual(again.data, payload) + + def test_a_stop_at_this_layer_keeps_an_ethertype_raw(self) -> None: + """The Raw is named after the EtherType it would have dispatched on.""" + from pcapkit.const.reg.ethertype import EtherType + from pcapkit.protocols.misc.raw import Raw for version, header, alias in self._versions(): - for stop in ({'layer': 'Link'}, {'protocol': version.__name__}): + for stop in self._stops(version): with self.subTest(version=version.__name__, stop=stop): raw = header(0x0800, 1) + IPV4 - self.assertEqual(version(raw, len(raw), **stop).protochain.chain, + parsed = version(raw, len(raw), **stop) + self.assertEqual(parsed.protochain.chain, f'{alias}:Internet_Protocol_version_4') + self.assertIsInstance(parsed.payload, Raw) + self.assertEqual(parsed.payload.data, IPV4) + self.assertIs(parsed.payload.info.protocol, + EtherType.Internet_Protocol_version_4) + self.assertEqual(parsed.data, raw) + self.assertEqual(version.from_data(parsed.info).data, raw) + + def test_a_stop_at_another_protocol_still_dispatches_here(self) -> None: + """The other version's name and a protocol further in do not stop this layer.""" + from pcapkit.const.linux.arphrd import ARPHRD + + frad = CASES['FRAD'][0] + for version, header, alias in self._versions(): + other = 'LinuxSLL2' if version.__name__ == 'LinuxSLL1' else 'LinuxSLL1' + for stop, chain in (({'protocol': other}, 'FRAD:IPv4:UDP'), + ({'protocol': 'FRAD'}, 'FRAD:Raw')): + with self.subTest(version=version.__name__, stop=stop): + raw = header(0x0800, ARPHRD.FRAD) + frad + parsed = version(raw, len(raw), **stop) + self.assertEqual(parsed.protochain.chain, f'{alias}:{chain}') + self.assertEqual(version.from_data(parsed.info).data, raw) - raw = header(0x0800, ARPHRD.NETLINK) + CASES['NETLINK'][0] + raw = header(0x0800, 1) + IPV4 self.assertEqual(version(raw, len(raw), **stop).protochain.chain, - f'{alias}:Netlink:Raw') + f'{alias}:IPv4:UDP') def test_index_is_the_arphrd_type_that_dispatches_here(self) -> None: from pcapkit.const.linux.arphrd import ARPHRD diff --git a/tests/protocols/link/test_linux_sll_selector_unit.py b/tests/protocols/link/test_linux_sll_selector_unit.py index 35a6604d74..e914308221 100644 --- a/tests/protocols/link/test_linux_sll_selector_unit.py +++ b/tests/protocols/link/test_linux_sll_selector_unit.py @@ -21,16 +21,20 @@ + bytes.fromhex('3039003500080000')) -def _header_v1(protocol_type: int = 0x0800) -> bytes: +#: A FRAD frame, Q.922 address and UI control, carrying :data:`IPV4` by NLPID ``0xCC``. +FRAD = bytes.fromhex('0c2103cc') + IPV4 + + +def _header_v1(protocol_type: int = 0x0800, arphrd_type: int = 1) -> bytes: """Pack a minimal ``LINUX_SLL`` (v1) header.""" link_addr = bytes.fromhex('001122334455').ljust(8, b'\x00') - return struct.pack('>HHH', 0, 1, 6) + link_addr + struct.pack('>H', protocol_type) + return struct.pack('>HHH', 0, arphrd_type, 6) + link_addr + struct.pack('>H', protocol_type) -def _header_v2(protocol_type: int = 0x0800) -> bytes: +def _header_v2(protocol_type: int = 0x0800, arphrd_type: int = 1) -> bytes: """Pack a minimal ``LINUX_SLL2`` header.""" link_addr = bytes.fromhex('001122334455').ljust(8, b'\x00') - return struct.pack('>HHIHBB', protocol_type, 0, 1, 1, 0, 6) + link_addr + return struct.pack('>HHIHBB', protocol_type, 0, 1, arphrd_type, 0, 6) + link_addr def _pcap(records: 'list[bytes]', linktype: int) -> bytes: @@ -156,6 +160,20 @@ def test_extract_version_name_stops_only_that_version(self) -> None: self.assertEqual(_extract(_header_v1() + IPV4, LinkType.LINUX_SLL, 'LinuxSLL2'), 'LINUX_SLL1:IPv4:UDP') + def test_extract_stops_a_registered_arphrd_type_alike(self) -> None: + """A FRAD frame stops where an EtherType one does (#2532).""" + from pcapkit.const.linux.arphrd import ARPHRD + from pcapkit.const.reg.linktype import LinkType + + for linktype, header, alias, own, other in ( + (LinkType.LINUX_SLL, _header_v1, 'LINUX_SLL1', 'LinuxSLL1', 'LinuxSLL2'), + (LinkType.LINUX_SLL2, _header_v2, 'LINUX_SLL2', 'LinuxSLL2', 'LinuxSLL1')): + record = header(arphrd_type=ARPHRD.FRAD) + FRAD + for protocol in ('LinuxSLL', own): + with self.subTest(alias=alias, protocol=protocol): + self.assertEqual(_extract(record, linktype, protocol), f'{alias}:Raw') + self.assertEqual(_extract(record, linktype, other), f'{alias}:FRAD:IPv4:UDP') + if __name__ == '__main__': unittest.main()