From ab543c46f3084f20035c80bc9c723b5be889cabf Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Sat, 10 Oct 2026 02:54:48 -0400 Subject: [PATCH] fix(protocols): dissect LINKTYPE_RAW by version nibble; add the BSD loopback layer for LINKTYPE_NULL/LOOP (#1502, #1574) - Frame and PCAPNG dispatch LINKTYPE_RAW (101) data as IPV4 or IPV6 by the version nibble of its first octet (_dispatch_linktype), unless RAW itself is registered; any other version stays Raw. - New Loopback link layer (protocols/link, schema/link, data/link): a 4-octet address family, big-endian under LINKTYPE_LOOP (108) and inferred under LINKTYPE_NULL (0) (family_byteorder). The byte order is kept in info and read back by from_data, so a rebuild returns the same info. 2 goes to IPv4 and 24/28/30 to IPv6 through its own __proto__. Registered for NULL and LOOP in both tables, exported, and documented. - dpkt engine: RAW by version nibble; NULL/LOOP through a dpkt Loopback, cached per base/IP/IP6 class and link type, that reads the family as pcapkit does and names its ip/ip6 payload. - Delete Gap(1502): all four labels now agree, many_interfaces.pcapng's two NULL frames included (36 datagrams in default, scapy and dpkt). - Dispatch generator pins the new entries and walks Loopback.__proto__. - New tests for RAW, Loopback (7 families x 2 byte orders x 3 link-type hints, info-exact from_data, PCAP and PCAP-NG) and the dpkt engine. Closes #1502 Closes #1574 --- docs/source/ext.rst | 2 + docs/source/pcapkit/protocols/index.rst | 3 +- docs/source/pcapkit/protocols/link/index.rst | 1 + .../pcapkit/protocols/link/loopback.rst | 65 ++++ docs/source/pcapkit/protocols/misc/pcap.rst | 1 + docs/source/pcapkit/protocols/misc/pcapng.rst | 1 + examples/generators/dispatch.py | 43 ++- pcapkit/__init__.py | 2 +- pcapkit/all.py | 2 +- pcapkit/foundation/engines/dpkt.py | 72 ++++- pcapkit/protocols/__init__.py | 2 +- pcapkit/protocols/data/__init__.py | 3 + pcapkit/protocols/data/link/__init__.py | 6 + pcapkit/protocols/data/link/loopback.py | 25 ++ pcapkit/protocols/link/__init__.py | 3 +- pcapkit/protocols/link/link.py | 1 + pcapkit/protocols/link/loopback.py | 286 ++++++++++++++++++ pcapkit/protocols/misc/pcap/frame.py | 51 ++++ pcapkit/protocols/misc/pcapng.py | 47 +++ pcapkit/protocols/schema/__init__.py | 1 + pcapkit/protocols/schema/link/__init__.py | 2 + pcapkit/protocols/schema/link/loopback.py | 33 ++ .../test_dpkt_pcapng_interfaces_unit.py | 2 +- .../test_dpkt_raw_null_linktypes_unit.py | 145 +++++++++ .../engines/test_engine_agreement_runtime.py | 7 - .../engines/test_runtime_engines.py | 2 +- .../link/test_link_edge_roundtrip_unit.py | 21 +- tests/protocols/link/test_loopback_unit.py | 268 ++++++++++++++++ .../test_raw_linktype_dispatch_1502_unit.py | 112 +++++++ .../protocols/test_dispatch_registry_unit.py | 16 +- .../test_short_read_roundtrip_unit.py | 2 + 31 files changed, 1195 insertions(+), 32 deletions(-) create mode 100644 docs/source/pcapkit/protocols/link/loopback.rst create mode 100644 pcapkit/protocols/data/link/loopback.py create mode 100644 pcapkit/protocols/link/loopback.py create mode 100644 pcapkit/protocols/schema/link/loopback.py create mode 100644 tests/foundation/engines/test_dpkt_raw_null_linktypes_unit.py create mode 100644 tests/protocols/link/test_loopback_unit.py create mode 100644 tests/protocols/misc/test_raw_linktype_dispatch_1502_unit.py diff --git a/docs/source/ext.rst b/docs/source/ext.rst index 7071a51771..51b3e57526 100644 --- a/docs/source/ext.rst +++ b/docs/source/ext.rst @@ -30,6 +30,8 @@ The protocol classes :mod:`pcapkit` ships: + + +-----------------------+-------------------------------------------------------------+ | | L2TP Family | :class:`pcapkit.protocols.link.l2tpv2.L2TPv2` | + +----------------+-----------------------+-------------------------------------------------------------+ +| | :class:`pcapkit.protocols.link.loopback.Loopback` | ++ +----------------+-----------------------+-------------------------------------------------------------+ | | | :class:`pcapkit.protocols.link.vlan.VLAN` | + + +-----------------------+-------------------------------------------------------------+ | | VLAN Family | :class:`pcapkit.protocols.link.c_tag.C_Tag` | diff --git a/docs/source/pcapkit/protocols/index.rst b/docs/source/pcapkit/protocols/index.rst index 130d05b192..e5550e756c 100644 --- a/docs/source/pcapkit/protocols/index.rst +++ b/docs/source/pcapkit/protocols/index.rst @@ -27,7 +27,7 @@ from given information. The class hierarchy of :mod:`pcapkit.protocols`: A{{ProtocolMeta}} -.->|metaclass| B(ProtocolBase) subgraph link [Link Layer] - Link --> Ethernet & L2TP & VLAN & ARP + Link --> Ethernet & L2TP & Loopback & VLAN & ARP subgraph arp [ARP Family] ARP --> InARP @@ -108,6 +108,7 @@ from given information. The class hierarchy of :mod:`pcapkit.protocols`: click Ethernet "/pcapkit/protocols/link/ethernet.html#pcapkit.protocols.link.ethernet.Ethernet" click L2TP "/pcapkit/protocols/link/l2tp.html#pcapkit.protocols.link.l2tp.L2TP" click L2TPv2 "/pcapkit/protocols/link/l2tpv2.html#pcapkit.protocols.link.l2tpv2.L2TPv2" + click Loopback "/pcapkit/protocols/link/loopback.html#pcapkit.protocols.link.loopback.Loopback" click VLAN "/pcapkit/protocols/link/vlan.html#pcapkit.protocols.link.vlan.VLAN" click C_Tag "/pcapkit/protocols/link/c_tag.html#pcapkit.protocols.link.c_tag.C_Tag" click S_Tag "/pcapkit/protocols/link/s_tag.html#pcapkit.protocols.link.s_tag.S_Tag" diff --git a/docs/source/pcapkit/protocols/link/index.rst b/docs/source/pcapkit/protocols/link/index.rst index ca697e1b4a..d7862d6d51 100644 --- a/docs/source/pcapkit/protocols/link/index.rst +++ b/docs/source/pcapkit/protocols/link/index.rst @@ -16,6 +16,7 @@ link layer, with detailed implementation and methods. arp l2tp l2tpv2 + loopback vlan c_tag s_tag diff --git a/docs/source/pcapkit/protocols/link/loopback.rst b/docs/source/pcapkit/protocols/link/loopback.rst new file mode 100644 index 0000000000..a489d007c4 --- /dev/null +++ b/docs/source/pcapkit/protocols/link/loopback.rst @@ -0,0 +1,65 @@ +BSD Loopback Encapsulation +========================== + +.. module:: pcapkit.protocols.link.loopback + +:mod:`pcapkit.protocols.link.loopback` contains +:class:`~pcapkit.protocols.link.loopback.Loopback` +only, which implements extractor for the BSD loopback +encapsulation [*]_ of ``LINKTYPE_NULL`` and ``LINKTYPE_LOOP``, +whose structure is described as below: + +.. table:: + + ====== ===== =================== =================================== + Octets Bits Name Description + ====== ===== =================== =================================== + 0 0 ``loopback.family`` Address Family (Internet Layer) + ====== ===== =================== =================================== + +The address family is in the byte order of the capturing host for +``LINKTYPE_NULL``, which no capture format records, so it is inferred from the +octets; for ``LINKTYPE_LOOP`` it is always in network byte order. Either way it +is kept as ``loopback.byteorder``. + +.. autoclass:: pcapkit.protocols.link.loopback.Loopback + :no-members: + :show-inheritance: + + .. autoproperty:: name + .. autoproperty:: length + .. autoproperty:: protocol + + .. automethod:: read + .. automethod:: make + + .. automethod:: _make_data + + .. automethod:: __index__ + + .. autoattribute:: __proto__ + :no-value: + +.. autofunction:: pcapkit.protocols.link.loopback.family_byteorder + +Header Schemas +-------------- + +.. module:: pcapkit.protocols.schema.link.loopback + +.. autoclass:: pcapkit.protocols.schema.link.loopback.Loopback + :members: + :show-inheritance: + +Data Models +----------- + +.. module:: pcapkit.protocols.data.link.loopback + +.. autoclass:: pcapkit.protocols.data.link.loopback.Loopback + :members: + :show-inheritance: + +.. rubric:: Footnotes + +.. [*] https://www.tcpdump.org/linktypes/LINKTYPE_NULL.html diff --git a/docs/source/pcapkit/protocols/misc/pcap.rst b/docs/source/pcapkit/protocols/misc/pcap.rst index 362132d662..5c964daab4 100644 --- a/docs/source/pcapkit/protocols/misc/pcap.rst +++ b/docs/source/pcapkit/protocols/misc/pcap.rst @@ -114,6 +114,7 @@ whose structure is described as below: .. automethod:: make .. automethod:: _make_data + .. automethod:: _dispatch_linktype .. autoattribute:: __proto__ :no-value: diff --git a/docs/source/pcapkit/protocols/misc/pcapng.rst b/docs/source/pcapkit/protocols/misc/pcapng.rst index b03f833903..a1559bb13f 100644 --- a/docs/source/pcapkit/protocols/misc/pcapng.rst +++ b/docs/source/pcapkit/protocols/misc/pcapng.rst @@ -38,6 +38,7 @@ which implements extractor for PCAP-NG file format [*]_. .. automethod:: index .. automethod:: _decode_next_layer + .. automethod:: _dispatch_linktype .. automethod:: _get_payload .. automethod:: _make_data diff --git a/examples/generators/dispatch.py b/examples/generators/dispatch.py index 06f78794c3..3fcae09be3 100644 --- a/examples/generators/dispatch.py +++ b/examples/generators/dispatch.py @@ -1,5 +1,5 @@ # -*- coding: utf-8 -*- -"""Enumerate the seven ``__proto__`` dispatch registries, and probe every entry. +"""Enumerate the eight ``__proto__`` dispatch registries, and probe every entry. :file:`pcapkit/foundation/registry/protocols.py` documents the registries that decide *which* :class:`~pcapkit.protocols.protocol.Protocol` subclass parses the @@ -274,6 +274,12 @@ def _ethernet(ethertype: 'int', payload: 'bytes') -> 'bytes': return bytes(Ethernet(type=ethertype, payload=payload)) +def _loopback(family: 'int', payload: 'bytes', byteorder: 'str' = 'little') -> 'bytes': + """A BSD loopback header carrying ``payload``.""" + from pcapkit.protocols.link.loopback import Loopback + return bytes(Loopback(family=family, byteorder=byteorder, payload=payload)) + + ############################################################################### # Link -- EtherType. ############################################################################### @@ -508,7 +514,8 @@ def _linktype_payload(code: 'Any') -> 'tuple[bytes, bool]': Returns: The record's octets, and whether they still need an Ethernet wrapper (true for :attr:`~pcapkit.const.reg.linktype.LinkType.ETHERNET`, false - for the two raw-IP link types, which carry no link layer at all). + for the two raw-IP link types, which carry no link layer at all, and + for the two loopback link types, whose header is built here). """ from pcapkit.const.reg.linktype import LinkType @@ -519,6 +526,10 @@ def _linktype_payload(code: 'Any') -> 'tuple[bytes, bool]': return _ip4(6, _tcp(9999)), False if code == LinkType.IPV6: return _ip6(6, _tcp(9999)), False + if code == LinkType.NULL: # in the capturing host's byte order (#1574) + return _loopback(2, _ip4(6, _tcp(9999))), False + if code == LinkType.LOOP: # in network byte order (#1574) + return _loopback(2, _ip4(6, _tcp(9999)), byteorder='big'), False raise LookupError(f'linktype: no payload builder for {code!r}') # pragma: no cover @@ -539,6 +550,21 @@ def _pcapng_build(code: 'Any') -> 'Any': return _extract_first(_pcapng_bytes(code, record), '.pcapng') +############################################################################### +# Loopback -- address family. +############################################################################### + + +def _loopback_registry() -> 'Any': + from pcapkit.protocols.link.loopback import Loopback + return Loopback.__proto__ + + +def _loopback_build(family: 'int') -> 'Any': + payload = _ip4(6, _tcp(9999)) if family == 2 else _ip6(6, _tcp(9999)) + return _extract_first(_pcap_bytes((_loopback(family, payload),), linktype=0), '.pcap') + + ############################################################################### # The families, the cases, and the probe. ############################################################################### @@ -571,7 +597,7 @@ def _internet_enum() -> 'Any': #: Every family this module exercises, one per ``__proto__`` table named in -#: GitHub issue #496. +#: GitHub issue #496, and the loopback table #1574 added. FAMILIES = ( Family('link', _link_registry, _link_build, _link_enum), Family('internet', _internet_registry, _internet_build, _internet_enum), @@ -580,6 +606,7 @@ def _internet_enum() -> 'Any': Family('sctp', _sctp_registry, _sctp_build), Family('pcap-frame', _pcap_frame_registry, _pcap_frame_build), Family('pcapng-frame', _pcapng_registry, _pcapng_build), + Family('loopback', _loopback_registry, _loopback_build), ) #: The families keyed by label, for a caller -- :func:`probe` -- that wants @@ -665,14 +692,24 @@ def _internet_enum() -> 'Any': ('pcapkit.protocols.application.ngap', 'NGAP'), # -- Frame.__proto__ (pcap LinkType) --------------------------------------- + 'pcap-frame/NULL': ('pcapkit.protocols.link', 'Loopback'), 'pcap-frame/ETHERNET': ('pcapkit.protocols.link', 'Ethernet'), + 'pcap-frame/LOOP': ('pcapkit.protocols.link', 'Loopback'), 'pcap-frame/IPV4': ('pcapkit.protocols.internet', 'IPv4'), 'pcap-frame/IPV6': ('pcapkit.protocols.internet', 'IPv6'), # -- PCAPNG.__proto__ (LinkType) -------------------------------------------- + 'pcapng-frame/NULL': ('pcapkit.protocols.link', 'Loopback'), 'pcapng-frame/ETHERNET': ('pcapkit.protocols.link', 'Ethernet'), + 'pcapng-frame/LOOP': ('pcapkit.protocols.link', 'Loopback'), 'pcapng-frame/IPV4': ('pcapkit.protocols.internet', 'IPv4'), 'pcapng-frame/IPV6': ('pcapkit.protocols.internet', 'IPv6'), + + # -- Loopback.__proto__ (address family) ------------------------------------ + 'loopback/2': ('pcapkit.protocols.internet.ipv4', 'IPv4'), + 'loopback/24': ('pcapkit.protocols.internet.ipv6', 'IPv6'), + 'loopback/28': ('pcapkit.protocols.internet.ipv6', 'IPv6'), + 'loopback/30': ('pcapkit.protocols.internet.ipv6', 'IPv6'), } diff --git a/pcapkit/__init__.py b/pcapkit/__init__.py index 6aff15d1d7..83f7ed3aef 100644 --- a/pcapkit/__init__.py +++ b/pcapkit/__init__.py @@ -113,7 +113,7 @@ 'Raw', # Raw Packet 'ARP', 'C_Tag', 'Ethernet', 'InARP', 'L2TP', # Link Layer - 'L2TPv2', 'S_Tag', 'VLAN', + 'L2TPv2', 'Loopback', 'S_Tag', 'VLAN', 'AH', 'ESP', 'IP', 'IPsec', 'IPv4', 'IPv6', 'IPX', # Internet Layer 'HIP', 'HOPOPT', 'IPv6_Frag', 'IPv6_Ext', 'IPv6_Opts', 'IPv6_Route', 'MH', diff --git a/pcapkit/all.py b/pcapkit/all.py index fe77f29d9a..8575a83ea1 100644 --- a/pcapkit/all.py +++ b/pcapkit/all.py @@ -120,7 +120,7 @@ 'NoPayload', # No Payload 'Raw', # Raw Packet 'ARP', 'C_Tag', 'Ethernet', 'InARP', 'L2TP', 'L2TPv2', - 'S_Tag', 'VLAN', # Link Layer + 'Loopback', 'S_Tag', 'VLAN', # Link Layer 'AH', 'ESP', 'IP', 'IPsec', 'IPv4', 'IPv6', 'IPX', # Internet Layer 'HIP', 'HOPOPT', 'IPv6_Ext', 'IPv6_Frag', 'IPv6_Opts', 'IPv6_Route', 'MH', # IPv6 Extension Header diff --git a/pcapkit/foundation/engines/dpkt.py b/pcapkit/foundation/engines/dpkt.py index 4a05b42a79..e49b1ac33e 100644 --- a/pcapkit/foundation/engines/dpkt.py +++ b/pcapkit/foundation/engines/dpkt.py @@ -11,6 +11,7 @@ """ import decimal +import functools from typing import TYPE_CHECKING, cast from pcapkit.const.pcapng.block_type import BlockType as Enum_BlockType @@ -298,6 +299,58 @@ def __iter__(self) -> 'Iterator[tuple[float, bytes]]': return _Reader +@functools.lru_cache(maxsize=None) +def _loopback_protocol(base: 'Type[DPKTPacket]', ip: 'Type[DPKTPacket]', ip6: 'Type[DPKTPacket]', + linktype: 'int') -> 'Type[DPKTPacket]': + """A :class:`dpkt.loopback.Loopback` that reads as the default engine's. + + It reads the address family in the byte order + :func:`~pcapkit.protocols.link.loopback.family_byteorder` gives for + ``linktype`` -- network byte order for ``LINKTYPE_LOOP``, inferred for + ``LINKTYPE_NULL`` -- and dispatches on the same families as + :class:`~pcapkit.protocols.link.loopback.Loopback`. Unlike `DPKT`_'s own, it + names an IPv4 or IPv6 payload ``ip`` or ``ip6``, as + :class:`dpkt.ethernet.Ethernet` does and as :mod:`pcapkit.toolkit.dpkt` + looks for it, and keeps a payload of any other family as raw octets rather + than parsing it as Ethernet (:issue:`1574`). + + Args: + base: :class:`dpkt.loopback.Loopback` itself. + ip: :class:`dpkt.ip.IP`, the payload of family 2. + ip6: :class:`dpkt.ip6.IP6`, the payload of families 24, 28 and 30. + linktype: ``LINKTYPE_NULL`` or ``LINKTYPE_LOOP``. + + Returns: + The packet type, built once per argument tuple rather than once per + packet. Every class it is built from is part of that key, so a reloaded + `DPKT`_ module gets a class of its own instead of one holding stale + classes. + + .. _DPKT: https://dpkt.readthedocs.io + + """ + import dpkt # isort:skip + + from pcapkit.protocols.link.loopback import family_byteorder # isort:skip + + layers = {2: ip, 24: ip6, 28: ip6, 30: ip6} + + class Loopback(base): + """BSD loopback encapsulation.""" + + def unpack(ext, buf: 'bytes') -> 'None': + dpkt.dpkt.Packet.unpack(ext, buf) + family = buf[:ext.__hdr_len__] + ext.family = int.from_bytes(family, family_byteorder(family, linktype)) + layer = layers.get(ext.family) + if layer is not None: + payload = layer(buf[ext.__hdr_len__:]) + ext.data = payload + setattr(ext, layer.__name__.lower(), payload) + + return Loopback + + class DPKT(EngineBase['DPKTPacket']): """DPKT engine support. @@ -434,7 +487,7 @@ def read_frame(self) -> 'DPKTPacket': # interface this packet was captured on, so ask only after reading it timestamp, pkt = cast('tuple[float, bytes]', next(reader)) linktype = Enum_LinkType.get(reader.datalink()) - protocol = self._get_protocol(linktype) + protocol = self._get_protocol(linktype, pkt) try: packet = protocol(pkt) # type: DPKTPacket except Exception as exc: # pylint: disable=broad-except @@ -511,11 +564,19 @@ def read_frame(self) -> 'DPKTPacket': # Utilities. ########################################################################## - def _get_protocol(self, linktype: 'Optional[Enum_LinkType]' = None) -> 'Type[DPKTPacket]': + def _get_protocol(self, linktype: 'Optional[Enum_LinkType]' = None, + data: 'bytes' = b'') -> 'Type[DPKTPacket]': """Returns the protocol for parsing the current packet. Args: linktype: Link type code. + data: Octets of the current packet. ``LINKTYPE_RAW`` (101) carries + a bare IPv4 or IPv6 datagram, and only the version nibble of + its first octet tells which (:issue:`1502`). + + Notes: + ``LINKTYPE_NULL`` (0) and ``LINKTYPE_LOOP`` (108) are parsed with + :func:`_loopback_protocol` (:issue:`1574`). """ dpkt = self._expkg @@ -523,13 +584,16 @@ def _get_protocol(self, linktype: 'Optional[Enum_LinkType]' = None) -> 'Type[DPK if linktype is None: linktype = Enum_LinkType.get(reader.datalink()) + version = data[0] >> 4 if data else None if linktype == Enum_LinkType.ETHERNET: pkg = dpkt.ethernet.Ethernet - elif linktype.value == Enum_LinkType.IPV4: + elif linktype.value == Enum_LinkType.IPV4 or (linktype.value == Enum_LinkType.RAW and version == 4): pkg = dpkt.ip.IP - elif linktype.value == Enum_LinkType.IPV6: + elif linktype.value == Enum_LinkType.IPV6 or (linktype.value == Enum_LinkType.RAW and version == 6): pkg = dpkt.ip6.IP6 + elif linktype.value in (Enum_LinkType.NULL, Enum_LinkType.LOOP): + pkg = _loopback_protocol(dpkt.loopback.Loopback, dpkt.ip.IP, dpkt.ip6.IP6, linktype.value) else: warn('unrecognised link layer protocol; all analysis functions ignored', DPKTWarning, stacklevel=stacklevel()) diff --git a/pcapkit/protocols/__init__.py b/pcapkit/protocols/__init__.py index 1798841cc9..8a54371f26 100644 --- a/pcapkit/protocols/__init__.py +++ b/pcapkit/protocols/__init__.py @@ -50,7 +50,7 @@ 'Raw', # Link Layer - 'ARP', 'C_Tag', 'Ethernet', 'InARP', 'L2TP', 'L2TPv2', 'S_Tag', 'VLAN', + 'ARP', 'C_Tag', 'Ethernet', 'InARP', 'L2TP', 'L2TPv2', 'Loopback', 'S_Tag', 'VLAN', # Internet Layer 'AH', 'ESP', 'IP', 'IPsec', 'IPv4', 'IPv6', 'IPX', diff --git a/pcapkit/protocols/data/__init__.py b/pcapkit/protocols/data/__init__.py index b3f9c7eac8..6be59ffe86 100644 --- a/pcapkit/protocols/data/__init__.py +++ b/pcapkit/protocols/data/__init__.py @@ -61,6 +61,9 @@ # Ethernet Protocol 'Ethernet', + # BSD Loopback Encapsulation + 'Loopback', + # 802.1Q/802.1ad VLAN Tag Types 'VLAN', 'VLAN_TCI', diff --git a/pcapkit/protocols/data/link/__init__.py b/pcapkit/protocols/data/link/__init__.py index 08c4fb79d5..8d82710f4e 100644 --- a/pcapkit/protocols/data/link/__init__.py +++ b/pcapkit/protocols/data/link/__init__.py @@ -9,6 +9,9 @@ # Ethernet Protocol from pcapkit.protocols.data.link.ethernet import Ethernet +# BSD Loopback Encapsulation +from pcapkit.protocols.data.link.loopback import Loopback + # 802.1Q/802.1ad VLAN Tag Types from pcapkit.protocols.data.link.vlan import TCI as VLAN_TCI from pcapkit.protocols.data.link.vlan import VLAN @@ -20,6 +23,9 @@ # Ethernet Protocol 'Ethernet', + # BSD Loopback Encapsulation + 'Loopback', + # 802.1Q/802.1ad VLAN Tag Types 'VLAN', 'VLAN_TCI', ] diff --git a/pcapkit/protocols/data/link/loopback.py b/pcapkit/protocols/data/link/loopback.py new file mode 100644 index 0000000000..1728c7b3dd --- /dev/null +++ b/pcapkit/protocols/data/link/loopback.py @@ -0,0 +1,25 @@ +# -*- coding: utf-8 -*- +"""data models for BSD loopback encapsulation""" + +from typing import TYPE_CHECKING + +from pcapkit.corekit.infoclass import info_final +from pcapkit.protocols.data.protocol import Protocol + +if TYPE_CHECKING: + from typing_extensions import Literal + +__all__ = ['Loopback'] + + +@info_final +class Loopback(Protocol): + """Data model for BSD loopback encapsulation.""" + + #: Address family (internet layer). + family: 'int' + #: Byte order the address family was written in. + byteorder: 'Literal["big", "little"]' + + if TYPE_CHECKING: + def __init__(self, family: 'int', byteorder: 'Literal["big", "little"]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements diff --git a/pcapkit/protocols/link/__init__.py b/pcapkit/protocols/link/__init__.py index 905135c742..b1391e6a1e 100644 --- a/pcapkit/protocols/link/__init__.py +++ b/pcapkit/protocols/link/__init__.py @@ -17,6 +17,7 @@ # Link Layer Protocols from pcapkit.protocols.link.arp import ARP, InARP from pcapkit.protocols.link.ethernet import Ethernet +from pcapkit.protocols.link.loopback import Loopback # VLAN Tag Family from pcapkit.protocols.link.vlan import VLAN @@ -35,5 +36,5 @@ 'LINKTYPE', # Link Layer Protocols - 'ARP', 'C_Tag', 'Ethernet', 'InARP', 'L2TP', 'L2TPv2', 'S_Tag', 'VLAN', + 'ARP', 'C_Tag', 'Ethernet', 'InARP', 'L2TP', 'L2TPv2', 'Loopback', 'S_Tag', 'VLAN', ] diff --git a/pcapkit/protocols/link/link.py b/pcapkit/protocols/link/link.py index 2010f550d0..2ed84de434 100644 --- a/pcapkit/protocols/link/link.py +++ b/pcapkit/protocols/link/link.py @@ -11,6 +11,7 @@ :class:`~pcapkit.protocols.link.arp.ARP`/:class:`~pcapkit.protocols.link.arp.InARP`, :class:`~pcapkit.protocols.link.ethernet.Ethernet`, :class:`~pcapkit.protocols.link.l2tp.L2TP`, +:class:`~pcapkit.protocols.link.loopback.Loopback`, :class:`~pcapkit.protocols.link.vlan.VLAN` and etc. diff --git a/pcapkit/protocols/link/loopback.py b/pcapkit/protocols/link/loopback.py new file mode 100644 index 0000000000..88c9961f9d --- /dev/null +++ b/pcapkit/protocols/link/loopback.py @@ -0,0 +1,286 @@ +# -*- coding: utf-8 -*- +# mypy: disable-error-code=dict-item +"""BSD Loopback Encapsulation +================================ + +.. module:: pcapkit.protocols.link.loopback + +:mod:`pcapkit.protocols.link.loopback` contains +:class:`~pcapkit.protocols.link.loopback.Loopback` +only, which implements extractor for the BSD loopback +encapsulation [*]_ of ``LINKTYPE_NULL`` and ``LINKTYPE_LOOP``, +whose structure is described as below: + +.. table:: + + ====== ===== =================== =================================== + Octets Bits Name Description + ====== ===== =================== =================================== + 0 0 ``loopback.family`` Address Family (Internet Layer) + ====== ===== =================== =================================== + +.. [*] https://www.tcpdump.org/linktypes/LINKTYPE_NULL.html + +""" +import collections +import sys +from typing import TYPE_CHECKING + +from pcapkit.const.reg.linktype import LinkType as Enum_LinkType +from pcapkit.corekit.module import ModuleDescriptor +from pcapkit.protocols.data.link.loopback import Loopback as Data_Loopback +from pcapkit.protocols.link.link import Link +from pcapkit.protocols.schema.link.loopback import Loopback as Schema_Loopback +from pcapkit.utilities.exceptions import ProtocolError + +if TYPE_CHECKING: + from typing import Any, DefaultDict, Optional, Type + + from typing_extensions import Literal + + from pcapkit.protocols.protocol import ProtocolBase + from pcapkit.protocols.schema.schema import Schema + +__all__ = ['Loopback', 'family_byteorder'] + + +def family_byteorder(family: 'bytes', linktype: 'Optional[int]' = None) -> 'Literal["big", "little"]': + """Byte order a loopback header's address family was written in. + + Args: + family: The 4 octets of the address family. + linktype: Link type the frame was captured under; :data:`None` reads as + ``LINKTYPE_NULL``. + + Returns: + ``'big'`` for ``LINKTYPE_LOOP`` (108), whose family is in network byte + order. Otherwise, for ``LINKTYPE_NULL`` (0), ``'big'`` if the octets + read as a smaller number big-endian than little-endian, else + ``'little'``. + + Note: + A ``LINKTYPE_NULL`` family is in the byte order of the host that + captured the packet, which the capture file does not record. Every + address family is below ``0x10000``, so its zero octets lie at the high + end, and the smaller of the two readings is the one in the writer's byte + order. A family that reads the same both ways, such as 0, packs to the + same octets either way. A ``LINKTYPE_LOOP`` family is never inferred, + so one written little-endian reads as a family no table holds. + + """ + if linktype == Enum_LinkType.LOOP: + return 'big' + if int.from_bytes(family, 'big') < int.from_bytes(family, 'little'): + return 'big' + return 'little' + + +class Loopback(Link[Data_Loopback, Schema_Loopback], + schema=Schema_Loopback, data=Data_Loopback): + """This class implements BSD Loopback Encapsulation. + + The header is one 4-octet address family, in the byte order of the host + that captured the packet for ``LINKTYPE_NULL`` and in network byte order + for ``LINKTYPE_LOOP``. Which link type a frame was captured under is the + ``alias`` it was dispatched with; for ``LINKTYPE_NULL``, which a frame + parsed on its own is read as, the byte order is read from the octets + themselves (see :func:`family_byteorder`). Either way it is kept in + :attr:`info`, and :meth:`from_data ` + reads the family back in that byte order rather than inferring it again. + + The address families are those of the BSD socket API, whose ``AF_INET6`` + differs between the BSDs, so each of its values is read as IPv6. This class + supports parsing of the following protocols, which are registered in the + :attr:`self.__proto__ ` + attribute: + + .. list-table:: + :header-rows: 1 + + * - Index + - Protocol + * - 2 + - :class:`pcapkit.protocols.internet.ipv4.IPv4` + * - 24 + - :class:`pcapkit.protocols.internet.ipv6.IPv6` + * - 28 + - :class:`pcapkit.protocols.internet.ipv6.IPv6` + * - 30 + - :class:`pcapkit.protocols.internet.ipv6.IPv6` + + Any other family is kept as :class:`~pcapkit.protocols.misc.raw.Raw`. + + """ + + ########################################################################## + # Defaults. + ########################################################################## + + #: DefaultDict[int, ModuleDescriptor[ProtocolBase] | ~typing.Type[ProtocolBase]]: Protocol index + #: mapping for decoding next layer, + #: c.f. :meth:`self._decode_next_layer ` + #: & :meth:`self._import_next_layer `. + #: Keyed by address family rather than by EtherType, so this class keeps a + #: registry of its own instead of sharing :attr:`Link.__proto__ + #: `. + __proto__ = collections.defaultdict( + lambda: ModuleDescriptor('pcapkit.protocols.misc.raw', 'Raw'), + { + 2: ModuleDescriptor('pcapkit.protocols.internet.ipv4', 'IPv4'), + 24: ModuleDescriptor('pcapkit.protocols.internet.ipv6', 'IPv6'), + 28: ModuleDescriptor('pcapkit.protocols.internet.ipv6', 'IPv6'), + 30: ModuleDescriptor('pcapkit.protocols.internet.ipv6', 'IPv6'), + }, + ) # type: DefaultDict[int, ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]] + + ########################################################################## + # Properties. + ########################################################################## + + @property + def name(self) -> 'Literal["BSD Loopback Encapsulation"]': + """Name of current protocol.""" + return 'BSD Loopback Encapsulation' + + @property + def length(self) -> 'Literal[4]': + """Header length of current protocol.""" + return 4 + + @property + def protocol(self) -> 'int': # type: ignore[override] + """Address family of next layer protocol.""" + return self._info.family + + ########################################################################## + # Methods. + ########################################################################## + + def read(self, length: 'Optional[int]' = None, *, alias: 'Optional[int]' = None, + byteorder: 'Optional[Literal["big", "little"]]' = None, + **kwargs: 'Any') -> 'Data_Loopback': # pylint: disable=unused-argument + """Read BSD Loopback Encapsulation. + + Structure of the BSD loopback header: + + .. code-block:: text + + 0 1 2 3 + 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | Address Family | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + + Args: + length: Length of packet data. + alias: Link type the frame was dispatched under, as + :class:`~pcapkit.protocols.misc.pcap.frame.Frame` and + :class:`~pcapkit.protocols.misc.pcapng.PCAPNG` pass it; + ``LINKTYPE_LOOP`` reads the family in network byte order, and + anything else as ``LINKTYPE_NULL``. + byteorder: Byte order to read the family in, ahead of ``alias``. + :meth:`make` is handed the one :attr:`info` holds by + :meth:`from_data `, + and it reaches here too, so a rebuild reads the family as it was + parsed rather than inferring it again. + **kwargs: Arbitrary keyword arguments. + + Returns: + Parsed packet data. + + Raises: + ProtocolError: If ``byteorder`` is neither ``'big'`` nor + ``'little'``. + + """ + if length is None: + length = len(self) + schema = self.__header__ + + if byteorder is None: + _byte = family_byteorder(schema.family, alias) + elif byteorder in ('big', 'little'): + _byte = byteorder + else: + raise ProtocolError(f'Loopback: invalid byte order: {byteorder!r}') + _family = int.from_bytes(schema.family, _byte) + + loopback = Data_Loopback( + family=_family, + byteorder=_byte, + ) + return self._decode_next_layer(loopback, _family, length - self.length) + + def make(self, + family: 'int' = 2, + byteorder: 'Literal["big", "little"]' = sys.byteorder, + payload: 'bytes | ProtocolBase | Schema' = b'', + **kwargs: 'Any') -> 'Schema_Loopback': + """Make (construct) packet data. + + Args: + family: Address family, e.g. 2 for IPv4. + byteorder: Byte order to write the address family in, by default + this host's, as a capture of its own loopback interface does. + payload: Payload data. + **kwargs: Arbitrary keyword arguments. + + Returns: + Constructed packet data. + + Raises: + ProtocolError: If ``byteorder`` is neither ``'big'`` nor + ``'little'``, or ``family`` does not fit in 4 octets. + + """ + if byteorder not in ('big', 'little'): + raise ProtocolError(f'Loopback: invalid byte order: {byteorder!r}') + if not 0 <= family <= 0xFFFF_FFFF: + raise ProtocolError(f'Loopback: invalid address family: {family!r}') + + return Schema_Loopback( + family=family.to_bytes(4, byteorder), + payload=payload, + ) + + ########################################################################## + # Data models. + ########################################################################## + + def __length_hint__(self) -> 'Literal[4]': + """Return an estimated length for the object.""" + return 4 + + @classmethod + def __index__(cls) -> 'Enum_LinkType': # pylint: disable=invalid-index-returned + """Numeral registry index of the protocol. + + Returns: + Numeral registry index of the protocol in `tcpdump`_ link-layer + header types. ``LINKTYPE_LOOP`` (108) is dispatched here as well. + + .. _tcpdump: https://www.tcpdump.org/linktypes.html + + """ + return Enum_LinkType.NULL # type: ignore[return-value] + + ########################################################################## + # Utilities. + ########################################################################## + + @classmethod + def _make_data(cls, data: 'Data_Loopback') -> 'dict[str, Any]': # type: ignore[override] + """Create key-value pairs from ``data`` for protocol construction. + + Args: + data: protocol data + + Returns: + Key-value pairs for protocol construction. + + """ + return { + 'family': data.family, + 'byteorder': data.byteorder, + 'payload': cls._make_payload(data), + } diff --git a/pcapkit/protocols/misc/pcap/frame.py b/pcapkit/protocols/misc/pcap/frame.py index a35b793035..c3a6de617e 100644 --- a/pcapkit/protocols/misc/pcap/frame.py +++ b/pcapkit/protocols/misc/pcap/frame.py @@ -69,13 +69,21 @@ class Frame(ProtocolBase[Data_Frame, Schema_Frame], * - Index - Protocol + * - :attr:`pcapkit.const.reg.linktype.LinkType.NULL` + - :class:`pcapkit.protocols.link.loopback.Loopback` * - :attr:`pcapkit.const.reg.linktype.LinkType.ETHERNET` - :class:`pcapkit.protocols.link.ethernet.Ethernet` + * - :attr:`pcapkit.const.reg.linktype.LinkType.LOOP` + - :class:`pcapkit.protocols.link.loopback.Loopback` * - :attr:`pcapkit.const.reg.linktype.LinkType.IPV4` - :class:`pcapkit.protocols.internet.ipv4.IPv4` * - :attr:`pcapkit.const.reg.linktype.LinkType.IPV6` - :class:`pcapkit.protocols.internet.ipv6.IPv6` + :attr:`pcapkit.const.reg.linktype.LinkType.RAW` has no entry of its own: + its data is dispatched as ``IPV4`` or ``IPV6`` by the version nibble of + its first octet, see :meth:`_dispatch_linktype`. + """ ########################################################################## @@ -90,7 +98,9 @@ class Frame(ProtocolBase[Data_Frame, Schema_Frame], __proto__ = collections.defaultdict( lambda: ModuleDescriptor('pcapkit.protocols.misc.raw', 'Raw'), { + Enum_LinkType.NULL: ModuleDescriptor('pcapkit.protocols.link', 'Loopback'), Enum_LinkType.ETHERNET: ModuleDescriptor('pcapkit.protocols.link', 'Ethernet'), + Enum_LinkType.LOOP: ModuleDescriptor('pcapkit.protocols.link', 'Loopback'), Enum_LinkType.IPV4: ModuleDescriptor('pcapkit.protocols.internet', 'IPv4'), Enum_LinkType.IPV6: ModuleDescriptor('pcapkit.protocols.internet', 'IPv6'), }, @@ -595,7 +605,11 @@ def _decode_next_layer(self, dict_: 'Data_Frame', proto: 'Optional[int]' = None, We also added a new key ``protocols`` to ``dict_`` to store the protocol chain of the current packet (frame). + ``proto`` is dispatched as :meth:`_dispatch_linktype` says, so + ``LINKTYPE_RAW`` data is read as IPv4 or IPv6 (:issue:`1502`). + """ + proto = self._dispatch_linktype(proto, length) next_ = cast('ProtocolBase', self._import_next_layer(proto, length, packet=packet)) # type: ignore[misc,call-arg,redundant-cast] info, chain = next_.info, next_.protochain @@ -613,3 +627,40 @@ def _decode_next_layer(self, dict_: 'Data_Frame', proto: 'Optional[int]' = None, self._next = next_ # pylint: disable=attribute-defined-outside-init self._protos = chain # pylint: disable=attribute-defined-outside-init return dict_ + + def _dispatch_linktype(self, proto: 'Optional[int]', length: 'Optional[int]' = None) -> 'Optional[int]': + """Link type a frame's data is dispatched under. + + Arguments: + proto: link type of the capture + length: valid (*non-padding*) length + + Returns: + For :attr:`LinkType.RAW ` + data whose first octet carries IP version 4 or 6, + :attr:`LinkType.IPV4 ` or + :attr:`LinkType.IPV6 `; + otherwise ``proto``. + + Note: + ``LINKTYPE_RAW`` (101) carries a bare IPv4 or IPv6 datagram, and + only the version nibble of its first octet tells which, so it has no + entry of its own in :attr:`__proto__` (:issue:`1502`). Dispatched as + ``IPV4`` or ``IPV6``, it reaches whatever those link types are + registered against. An entry registered under ``RAW`` itself is used + as it stands, and data of any other version is kept as + :class:`~pcapkit.protocols.misc.raw.Raw`. + :meth:`PCAPNG._dispatch_linktype + ` does the + same for PCAP-NG. + + """ + if proto != Enum_LinkType.RAW or Enum_LinkType.RAW in self.__proto__: + return proto + data = self._get_payload()[:length] + version = data[0] >> 4 if data else None + if version == 4: + return Enum_LinkType.IPV4 + if version == 6: + return Enum_LinkType.IPV6 + return proto diff --git a/pcapkit/protocols/misc/pcapng.py b/pcapkit/protocols/misc/pcapng.py index 357f610587..463366d830 100644 --- a/pcapkit/protocols/misc/pcapng.py +++ b/pcapkit/protocols/misc/pcapng.py @@ -429,13 +429,21 @@ class PCAPNG(ProtocolBase[Data_PCAPNG, Schema_PCAPNG], * - Index - Protocol + * - :attr:`pcapkit.const.reg.linktype.LinkType.NULL` + - :class:`pcapkit.protocols.link.loopback.Loopback` * - :attr:`pcapkit.const.reg.linktype.LinkType.ETHERNET` - :class:`pcapkit.protocols.link.ethernet.Ethernet` + * - :attr:`pcapkit.const.reg.linktype.LinkType.LOOP` + - :class:`pcapkit.protocols.link.loopback.Loopback` * - :attr:`pcapkit.const.reg.linktype.LinkType.IPV4` - :class:`pcapkit.protocols.internet.ipv4.IPv4` * - :attr:`pcapkit.const.reg.linktype.LinkType.IPV6` - :class:`pcapkit.protocols.internet.ipv6.IPv6` + :attr:`pcapkit.const.reg.linktype.LinkType.RAW` has no entry of its own: + its data is dispatched as ``IPV4`` or ``IPV6`` by the version nibble of + its first octet, see :meth:`_dispatch_linktype`. + The class currently supports parsing of the following block types, which are registered in the :attr:`self.__block__ ` attribute: @@ -699,7 +707,9 @@ class PCAPNG(ProtocolBase[Data_PCAPNG, Schema_PCAPNG], __proto__ = collections.defaultdict( lambda: ModuleDescriptor('pcapkit.protocols.misc.raw', 'Raw'), { + Enum_LinkType.NULL: ModuleDescriptor('pcapkit.protocols.link', 'Loopback'), Enum_LinkType.ETHERNET: ModuleDescriptor('pcapkit.protocols.link', 'Ethernet'), + Enum_LinkType.LOOP: ModuleDescriptor('pcapkit.protocols.link', 'Loopback'), Enum_LinkType.IPV4: ModuleDescriptor('pcapkit.protocols.internet', 'IPv4'), Enum_LinkType.IPV6: ModuleDescriptor('pcapkit.protocols.internet', 'IPv6'), }, @@ -2081,7 +2091,11 @@ def _decode_next_layer(self, dict_: 'Data_PCAPNG', proto: 'Optional[int]' = None We also added a new key ``protocols`` to ``dict_`` to store the protocol chain of the current packet (frame). + ``proto`` is dispatched as :meth:`_dispatch_linktype` says, so + ``LINKTYPE_RAW`` data is read as IPv4 or IPv6 (:issue:`1502`). + """ + proto = self._dispatch_linktype(proto, length) next_ = cast('ProtocolBase', self._import_next_layer(proto, length, packet=packet)) # type: ignore[misc,call-arg,redundant-cast] info, chain = next_.info, next_.protochain @@ -2100,6 +2114,39 @@ def _decode_next_layer(self, dict_: 'Data_PCAPNG', proto: 'Optional[int]' = None self._protos = chain # pylint: disable=attribute-defined-outside-init return dict_ + def _dispatch_linktype(self, proto: 'Optional[int]', length: 'Optional[int]' = None) -> 'Optional[int]': + """Link type a packet block's data is dispatched under. + + Arguments: + proto: link type of the interface the block was captured on + length: valid (*non-padding*) length + + Returns: + For :attr:`LinkType.RAW ` + data whose first octet carries IP version 4 or 6, + :attr:`LinkType.IPV4 ` or + :attr:`LinkType.IPV6 `; + otherwise ``proto``. + + Note: + As :meth:`Frame._dispatch_linktype + ` does + for PCAP: ``LINKTYPE_RAW`` (101) data is IPv4 or IPv6 by its version + nibble, an entry registered under ``RAW`` itself is used as it + stands, and data of any other version is kept as + :class:`~pcapkit.protocols.misc.raw.Raw` (:issue:`1502`). + + """ + if proto != Enum_LinkType.RAW or Enum_LinkType.RAW in self.__proto__: + return proto + data = self._get_payload()[:length] + version = data[0] >> 4 if data else None + if version == 4: + return Enum_LinkType.IPV4 + if version == 6: + return Enum_LinkType.IPV6 + return proto + def _read_block_unknown(self, schema: 'Schema_UnknownBlock', *, header: 'Schema_PCAPNG') -> 'Data_UnknownBlock': """Read unknown PCAP-NG block. diff --git a/pcapkit/protocols/schema/__init__.py b/pcapkit/protocols/schema/__init__.py index b3c4d58651..3bb94232b8 100644 --- a/pcapkit/protocols/schema/__init__.py +++ b/pcapkit/protocols/schema/__init__.py @@ -28,6 +28,7 @@ 'ARP', 'Ethernet', 'L2TP', + 'Loopback', 'VLAN', 'VLAN_TCI', # Internet Layer Protocols diff --git a/pcapkit/protocols/schema/link/__init__.py b/pcapkit/protocols/schema/link/__init__.py index 6f38d5aeae..4f8104ccd4 100644 --- a/pcapkit/protocols/schema/link/__init__.py +++ b/pcapkit/protocols/schema/link/__init__.py @@ -4,6 +4,7 @@ from pcapkit.protocols.schema.link.arp import ARP from pcapkit.protocols.schema.link.ethernet import Ethernet from pcapkit.protocols.schema.link.l2tp import L2TP +from pcapkit.protocols.schema.link.loopback import Loopback from pcapkit.protocols.schema.link.vlan import TCI as VLAN_TCI from pcapkit.protocols.schema.link.vlan import VLAN @@ -11,5 +12,6 @@ 'ARP', 'Ethernet', 'L2TP', + 'Loopback', 'VLAN', 'VLAN_TCI', ] diff --git a/pcapkit/protocols/schema/link/loopback.py b/pcapkit/protocols/schema/link/loopback.py new file mode 100644 index 0000000000..312be62314 --- /dev/null +++ b/pcapkit/protocols/schema/link/loopback.py @@ -0,0 +1,33 @@ +# -*- coding: utf-8 -*- +# mypy: disable-error-code=assignment +"""header schema for BSD loopback encapsulation""" + +from typing import TYPE_CHECKING + +from pcapkit.corekit.fields.misc import PayloadField +from pcapkit.corekit.fields.strings import BytesField +from pcapkit.protocols.schema.schema import Schema, schema_final + +__all__ = ['Loopback'] + +if TYPE_CHECKING: + from pcapkit.protocols.protocol import ProtocolBase + + +@schema_final +class Loopback(Schema): + """Header schema for BSD loopback encapsulation. + + The address family is kept as its 4 octets, since their byte order depends + on the link type the frame was captured under; see + :func:`~pcapkit.protocols.link.loopback.family_byteorder`. + + """ + + #: Address family. + family: 'bytes' = BytesField(length=4) + #: Payload. + payload: 'bytes' = PayloadField(length=lambda pkt: pkt['__length__']) + + if TYPE_CHECKING: + def __init__(self, family: 'bytes', payload: 'bytes | ProtocolBase | Schema') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements diff --git a/tests/foundation/engines/test_dpkt_pcapng_interfaces_unit.py b/tests/foundation/engines/test_dpkt_pcapng_interfaces_unit.py index bc9f1e7b4d..7fad892c1d 100644 --- a/tests/foundation/engines/test_dpkt_pcapng_interfaces_unit.py +++ b/tests/foundation/engines/test_dpkt_pcapng_interfaces_unit.py @@ -79,7 +79,7 @@ def _spb(order: str, data: bytes, original_len: 'int | None' = None) -> bytes: #: Packets of :func:`_capture`: (link type, parsed-as, timestamp), in file order. EXPECTED = [ (1, 'Ethernet', TICKS / 10**6), # section 1, interface 0 - (101, 'RawPacket', TICKS / 10**9), # section 1, interface 1, if_tsresol=9 + (101, 'IP', TICKS / 10**9), # section 1, interface 1, if_tsresol=9; #1502 (1, 'Ethernet', TICKS / 10**6), # section 1, interface 0 again (228, 'IP', TICKS / 2**10 + 100), # section 2, interface 0, 2^-10 + offset ] diff --git a/tests/foundation/engines/test_dpkt_raw_null_linktypes_unit.py b/tests/foundation/engines/test_dpkt_raw_null_linktypes_unit.py new file mode 100644 index 0000000000..3f927e4337 --- /dev/null +++ b/tests/foundation/engines/test_dpkt_raw_null_linktypes_unit.py @@ -0,0 +1,145 @@ +# -*- coding: utf-8 -*- +"""The DPKT engine reads raw-IP and BSD loopback frames as the default engine does. + +GitHub issues #1502 and #1574. :meth:`DPKT._get_protocol +` knew Ethernet, IPv4 and +IPv6 only, so a ``LINKTYPE_RAW`` (101), ``LINKTYPE_NULL`` (0) or +``LINKTYPE_LOOP`` (108) frame was kept as raw data with a warning. Raw IP is now +read as IPv4 or IPv6 by its version nibble, and both loopback link types by the +address family, as :class:`~pcapkit.protocols.link.loopback.Loopback` reads +them, with the network layer where :mod:`pcapkit.toolkit.dpkt` looks for it. + +Each case writes a capture built here to a temporary directory. Everything from +:mod:`pcapkit` is imported inside each test, after +:func:`~tests._support.reimport_once_per_class`. + +""" + +import importlib.util +import os +import tempfile +import unittest +import warnings + +from tests._support import reimport_once_per_class +from tests.protocols.link.test_loopback_unit import IPV4, IPV6, _pcap + +HAS_DPKT = importlib.util.find_spec('dpkt') is not None + +#: ``(link type, records, dpkt class name per record)``. +CAPTURES = ( + (101, [IPV4, IPV6, b'\x50' + IPV4[1:]], ['IP', 'IP6', 'RawPacket']), + (0, [b'\x02\x00\x00\x00' + IPV4, b'\x00\x00\x00\x02' + IPV4, b'\x1e\x00\x00\x00' + IPV6, + b'\x07\x00\x00\x00' + IPV4], ['Loopback'] * 4), + # LINKTYPE_LOOP is network byte order, so a little-endian family 2 is not IPv4 + (108, [b'\x00\x00\x00\x02' + IPV4, b'\x00\x00\x00\x18' + IPV6, b'\x02\x00\x00\x00' + IPV4], + ['Loopback'] * 3), +) + + +@unittest.skipUnless(HAS_DPKT, 'dpkt is not installed') +class TestDPKTRawAndLoopback(unittest.TestCase): + """Pin the DPKT engine's raw-IP and loopback link types against the default engine's.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + self._tmp = tempfile.TemporaryDirectory() # pylint: disable=consider-using-with + self.addCleanup(self._tmp.cleanup) + + def _extract(self, linktype: int, records: 'list[bytes]', engine: str): # type: ignore[no-untyped-def] + import pcapkit + + path = os.path.join(self._tmp.name, f'{linktype}.pcap') + with open(path, 'wb') as file: + file.write(_pcap(linktype, records)) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + return pcapkit.extract(fin=path, engine=engine, nofile=True, store=True, + reassembly=True, ip=True) + + def test_frames_parse_into_their_network_layer(self) -> None: + import dpkt + + from pcapkit.toolkit.dpkt import ipv4_reassembly + + for linktype, records, names in CAPTURES: + with self.subTest(linktype=linktype): + extractor = self._extract(linktype, records, 'dpkt') + self.assertEqual([type(frame).__name__ for frame in extractor.frame], names) + for frame, record in zip(extractor.frame, records): + loopback = type(frame).__name__ == 'Loopback' + network = frame.data if loopback else frame + if isinstance(network, (dpkt.ip.IP, dpkt.ip6.IP6)): + if loopback: # named as dpkt.ethernet.Ethernet names it + self.assertIs(getattr(frame, type(network).__name__.lower()), network) + if isinstance(network, dpkt.ip.IP): + self.assertIsNotNone(ipv4_reassembly(frame, 0.0, count=1)) + else: # any other family or version is kept as octets + self.assertEqual(network if loopback else frame.data, + record[4:] if loopback else record) + + def test_reassembly_inputs_match_the_default_engine(self) -> None: + for linktype, records, _ in CAPTURES: + with self.subTest(linktype=linktype): + default = self._extract(linktype, records, 'default') + dpkt = self._extract(linktype, records, 'dpkt') + self.assertEqual(len(dpkt.reassembly.ipv4), len(default.reassembly.ipv4)) + self.assertEqual([datagram.packet for datagram in dpkt.reassembly.ipv4], + [datagram.packet for datagram in default.reassembly.ipv4]) + + def test_get_protocol_by_link_type(self) -> None: + import dpkt + + from pcapkit.const.reg.linktype import LinkType + from pcapkit.foundation.engines.dpkt import DPKT + + engine = DPKT.__new__(DPKT) + engine._expkg = dpkt + engine._extmp = None + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + self.assertIs(engine._get_protocol(LinkType.RAW, IPV4), dpkt.ip.IP) + self.assertIs(engine._get_protocol(LinkType.RAW, IPV6), dpkt.ip6.IP6) + self.assertEqual(engine._get_protocol(LinkType.RAW, b'\x50').__name__, 'RawPacket') + self.assertEqual(engine._get_protocol(LinkType.RAW).__name__, 'RawPacket') + for code in (LinkType.NULL, LinkType.LOOP): + with self.subTest(code=code): + self.assertTrue(issubclass(engine._get_protocol(code), dpkt.loopback.Loopback)) + # built once, not once per packet + self.assertIs(engine._get_protocol(code), engine._get_protocol(code)) + self.assertIsNot(engine._get_protocol(LinkType.NULL), engine._get_protocol(LinkType.LOOP)) + + def test_loopback_class_is_keyed_by_every_class_it_holds(self) -> None: + """A new ``dpkt.ip.IP`` -- e.g. after a reload -- gets a class that parses into it.""" + import dpkt + + from pcapkit.const.reg.linktype import LinkType + from pcapkit.foundation.engines.dpkt import _loopback_protocol + + base, ip, ip6 = dpkt.loopback.Loopback, dpkt.ip.IP, dpkt.ip6.IP6 + for layer, record in ((ip, b'\x02\x00\x00\x00' + IPV4), (ip6, b'\x1e\x00\x00\x00' + IPV6)): + other = type(layer.__name__, (layer,), {}) + args = (base, other, ip6) if layer is ip else (base, ip, other) + with self.subTest(layer=layer.__name__): + cls = _loopback_protocol(*args, LinkType.NULL.value) + self.assertIsNot(cls, _loopback_protocol(base, ip, ip6, LinkType.NULL.value)) + self.assertIs(type(cls(record).data), other) + + def test_loop_family_is_network_byte_order(self) -> None: + import dpkt + + from pcapkit.const.reg.linktype import LinkType + from pcapkit.foundation.engines.dpkt import DPKT + + engine = DPKT.__new__(DPKT) + engine._expkg = dpkt + engine._extmp = None + little = b'\x02\x00\x00\x00' + IPV4 + null = engine._get_protocol(LinkType.NULL)(little) + loop = engine._get_protocol(LinkType.LOOP)(little) + self.assertEqual((null.family, type(null.data)), (2, dpkt.ip.IP)) + self.assertEqual((loop.family, loop.data), (0x0200_0000, IPV4)) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/foundation/engines/test_engine_agreement_runtime.py b/tests/foundation/engines/test_engine_agreement_runtime.py index f13affd332..74586500fd 100644 --- a/tests/foundation/engines/test_engine_agreement_runtime.py +++ b/tests/foundation/engines/test_engine_agreement_runtime.py @@ -365,13 +365,6 @@ class TestEngineAgreement(harness.RoundTripBase): 'Extractor swaps the format with a FormatWarning ' '(pcapkit/foundation/extraction.py:1439-1443)', 'FLOWS', 'file of ', tuple(f'{capture}/pypcapfile/flows' for capture in TRACED_IPV4_ONLY_PCAP)), - Gap(1502, 'the default engine does not dissect LINKTYPE_RAW (101) frames, which scapy reads ' - 'as IPv4: neither Frame.__proto__ nor PCAPNG.__proto__ registers LinkType.RAW ' - '(pcapkit/protocols/misc/pcap/frame.py:90, pcapkit/protocols/misc/pcapng.py:699), ' - 'so those frames are Raw and never reach reassembly', - ('INPUTS', 'DATAGRAMS'), ('extra, default has none', 'datagrams != '), - ('test.pcapng/scapy/ipv4', 'test.pcapng/scapy/datagrams', - 'many_interfaces.pcapng/scapy/ipv4', 'many_interfaces.pcapng/scapy/datagrams')), ) def setUp(self) -> None: diff --git a/tests/foundation/engines/test_runtime_engines.py b/tests/foundation/engines/test_runtime_engines.py index 19a26d451f..a989cd67e4 100644 --- a/tests/foundation/engines/test_runtime_engines.py +++ b/tests/foundation/engines/test_runtime_engines.py @@ -341,7 +341,7 @@ def __init__(self, data: bytes) -> None: self.assertEqual(engine._get_protocol(LinkType.ETHERNET).__name__, 'Ethernet') self.assertEqual(engine._get_protocol(LinkType.IPV4).__name__, 'IP') self.assertEqual(engine._get_protocol(LinkType.IPV6).__name__, 'IP6') - raw_cls = engine._get_protocol(LinkType.NULL) + raw_cls = engine._get_protocol(LinkType.FDDI) raw = raw_cls(b'raw') self.assertEqual(bytes(raw), b'raw') self.assertEqual(len(raw), 3) diff --git a/tests/protocols/link/test_link_edge_roundtrip_unit.py b/tests/protocols/link/test_link_edge_roundtrip_unit.py index 2d06ae9b82..9de6fd1471 100644 --- a/tests/protocols/link/test_link_edge_roundtrip_unit.py +++ b/tests/protocols/link/test_link_edge_roundtrip_unit.py @@ -1,10 +1,10 @@ # -*- coding: utf-8 -*- """Link-layer edge cases round-trip byte for byte. C.f. #1202. -Ethernet, the 802.1Q/802.1ad tags, ARP and RARP, L2TPv2 and OSPF, at their -minimum and maximum lengths, with reserved bits set, with zero-length -addresses and offset pads, and with unassigned codes. The harness and the -meaning of each table are in :mod:`tests.protocols._edge_roundtrip`. +Ethernet, the 802.1Q/802.1ad tags, ARP and RARP, L2TPv2, the BSD loopback +and OSPF, at their minimum and maximum lengths, with reserved bits set, with +zero-length addresses and offset pads, and with unassigned codes. The harness +and the meaning of each table are in :mod:`tests.protocols._edge_roundtrip`. Every case builds its own octets in memory and reads no capture. @@ -24,6 +24,7 @@ ARP = 'pcapkit.protocols.link.arp:ARP' RARP = 'pcapkit.protocols.application.rarp:RARP' L2TPV2 = 'pcapkit.protocols.link.l2tpv2:L2TPv2' +LOOPBACK = 'pcapkit.protocols.link.loopback:Loopback' OSPF = 'pcapkit.protocols.application.ospf:OSPF' #: Destination and source MAC. @@ -86,6 +87,14 @@ def ospf(type_: int = 1, plen: 'int | None' = None, auth: int = 0, Case('l2tpv2/length-short-of-the-header', L2TPV2, b'\x40\x02' + b'\x00\x02' + b'\x00\x01\x00\x02' + b'pp'), Case('l2tpv2/length-past-the-data', L2TPV2, b'\x40\x02' + b'\x00\x40' + b'\x00\x01\x00\x02' + b'pp'), Case('l2tpv2/version-3', L2TPV2, b'\x00\x03' + b'\x00\x01\x00\x02'), + # -- BSD loopback (#1574) ---------------------------------------------- + Case('loopback/header-only-ipv4-little', LOOPBACK, b'\x02\x00\x00\x00'), + Case('loopback/header-only-ipv6-big', LOOPBACK, b'\x00\x00\x00\x1e'), + Case('loopback/family-0', LOOPBACK, bytes(4) + b'xyz'), + Case('loopback/family-osi-7', LOOPBACK, b'\x07\x00\x00\x00' + b'xyz'), + Case('loopback/family-ffffffff', LOOPBACK, b'\xff' * 4 + b'xyz'), + Case('loopback/both-halves-set', LOOPBACK, b'\x02\x00\x02\x00' + b'xyz'), + Case('loopback/ipv4-short-of-a-header', LOOPBACK, b'\x00\x00\x00\x02' + b'\x45\x00'), # -- OSPF -------------------------------------------------------------- Case('ospf/hello-header-only', OSPF, ospf()), Case('ospf/type-0-unassigned', OSPF, ospf(type_=0)), @@ -111,6 +120,10 @@ def ospf(type_: int = 1, plen: 'int | None' = None, auth: int = 0, {'type': 1, 'length_flag': True, 'ns': 1, 'nr': 2, 'offset': 0, 'priority': True}), MakeCase('l2tpv2/make/offset-pad', L2TPV2, {'offset': 3, 'padding': b'\x01\x02\x03', 'payload': b'pp'}), MakeCase('l2tpv2/make/reserved-bits', L2TPV2, {'reserved': 0x34f0}), + MakeCase('loopback/make/default', LOOPBACK, {}), + MakeCase('loopback/make/ipv6-darwin-big', LOOPBACK, {'family': 30, 'byteorder': 'big', 'payload': b'xy'}), + MakeCase('loopback/make/unassigned-family-little', LOOPBACK, + {'family': 0xffff, 'byteorder': 'little', 'payload': b'xy'}), MakeCase('ospf/make/default', OSPF, {}), MakeCase('ospf/make/unassigned-codes', OSPF, {'type': 200, 'auth_type': 9}), ) diff --git a/tests/protocols/link/test_loopback_unit.py b/tests/protocols/link/test_loopback_unit.py new file mode 100644 index 0000000000..c90fc58a31 --- /dev/null +++ b/tests/protocols/link/test_loopback_unit.py @@ -0,0 +1,268 @@ +# -*- coding: utf-8 -*- +"""GitHub issue #1574: the BSD loopback encapsulation is dissected. + +A frame on a ``LINKTYPE_NULL`` (0) or ``LINKTYPE_LOOP`` (108) interface was +kept whole as :class:`~pcapkit.protocols.misc.raw.Raw`, because neither link +type had a dissector: frames 3 and 4 of ``many_interfaces.pcapng`` read ``NULL`` +where scapy reads IPv4. :class:`~pcapkit.protocols.link.loopback.Loopback` +reads the 4-octet address family -- in the capturing host's byte order for +``NULL``, in network byte order for ``LOOP`` -- and dispatches 2 to IPv4 and 24, +28 and 30 to IPv6. Both link types are registered against it for PCAP and +PCAP-NG. + +Every layer is rebuilt byte-exactly in both byte orders: parse then rebuild, +make then parse, and ``from_data(info.to_dict())``. Each case builds its own +octets, and captures are written to a temporary directory. Everything from +:mod:`pcapkit` is imported inside each test, after +:func:`~tests._support.reimport_once_per_class`. + +""" + +import os +import struct +import sys +import tempfile +import unittest +import warnings + +from tests._support import reimport_once_per_class + +#: An IPv4 header (DF clear, protocol UDP) and an 8-octet UDP header. +IPV4 = (bytes.fromhex('4500001c00010000401100000a0000010a000002') + + bytes.fromhex('3039003500080000')) +#: An IPv6 header (next header UDP) and an 8-octet UDP header. +IPV6 = (bytes.fromhex('6000000000081140') + bytes(15) + b'\x01' + bytes(15) + b'\x02' + + bytes.fromhex('3039003500080000')) + +#: ``(family, payload, next layer class name)`` for every registered family. +FAMILIES = ((2, IPV4, 'IPv4'), (24, IPV6, 'IPv6'), (28, IPV6, 'IPv6'), (30, IPV6, 'IPv6')) + + +def _header(family: int, byteorder: str) -> bytes: + return family.to_bytes(4, byteorder) # type: ignore[arg-type] + + +def _pcap(linktype: int, records: 'list[bytes]') -> bytes: + data = struct.pack(' bytes: + body += bytes(-len(body) % 4) + length = len(body) + 12 + return struct.pack(' bytes: + data = (_block(0x0A0D0D0A, struct.pack(' None: + reimport_once_per_class(self) + + def _assert_rebuilds(self, layer, **kwargs) -> None: # type: ignore[no-untyped-def] + """``layer`` rebuilds byte-exactly from its info, and from ``info.to_dict()``.""" + for data in (layer.info, layer.info.to_dict()): + with self.subTest(layer=type(layer).__name__, data=type(data).__name__): + self.assertEqual(type(layer).from_data(data, **kwargs).data.hex(), layer.data.hex()) + + def test_each_family_in_both_byte_orders(self) -> None: + from pcapkit.protocols.link.loopback import Loopback + + for family, payload, name in FAMILIES: + for byteorder in ('little', 'big'): + with self.subTest(family=family, byteorder=byteorder): + raw = _header(family, byteorder) + payload + parsed = Loopback(raw, len(raw)) + self.assertEqual(parsed.info.family, family) + self.assertEqual(parsed.info.byteorder, byteorder) + self.assertEqual(parsed.protocol, family) + self.assertEqual((parsed.name, parsed.length), ('BSD Loopback Encapsulation', 4)) + self.assertEqual(type(parsed.payload).__name__, name) + self.assertEqual(parsed.protochain.chain, f'Loopback:{name}:UDP') + self.assertEqual(parsed.data, raw) + self._assert_rebuilds(parsed) + + def test_other_family_is_kept_raw(self) -> None: + from pcapkit.protocols.link.loopback import Loopback + from pcapkit.protocols.misc.raw import Raw + + for family, byteorder in ((7, 'little'), (23, 'big'), (0, 'little'), (0x0102, 'big')): + with self.subTest(family=family, byteorder=byteorder): + raw = _header(family, byteorder) + IPV4 + parsed = Loopback(raw, len(raw)) + self.assertEqual(parsed.info.family, family) + self.assertIsInstance(parsed.payload, Raw) + self.assertEqual(parsed.data, raw) + self._assert_rebuilds(parsed) + + def test_make_then_parse(self) -> None: + from pcapkit.protocols.link.loopback import Loopback + + for family, payload, name in FAMILIES: + for byteorder in ('little', 'big'): + with self.subTest(family=family, byteorder=byteorder): + made = Loopback(family=family, byteorder=byteorder, payload=payload).data + self.assertEqual(made, _header(family, byteorder) + payload) + parsed = Loopback(made, len(made)) + self.assertEqual((parsed.info.family, parsed.info.byteorder), (family, byteorder)) + self.assertEqual(type(parsed.payload).__name__, name) + self._assert_rebuilds(parsed) + self.assertEqual(Loopback().data, _header(2, sys.byteorder)) + + def test_make_rejects_what_does_not_fit(self) -> None: + from pcapkit.protocols.link.loopback import Loopback + from pcapkit.utilities.exceptions import ProtocolError + + for kwargs in ({'byteorder': 'native'}, {'family': -1}, {'family': 0x1_0000_0000}): + with self.subTest(**kwargs), self.assertRaises(ProtocolError): + Loopback(**kwargs) + + def test_registered_for_null_and_loop(self) -> None: + from pcapkit.const.reg.linktype import LinkType + from pcapkit.corekit.module import ModuleDescriptor + from pcapkit.protocols.link.loopback import Loopback + from pcapkit.protocols.misc.pcap.frame import Frame + from pcapkit.protocols.misc.pcapng import PCAPNG + + for cls in (Frame, PCAPNG): + for code in (LinkType.NULL, LinkType.LOOP): + with self.subTest(table=cls.__name__, code=code): + self.assertEqual(cls.__proto__[code], ModuleDescriptor('pcapkit.protocols.link', 'Loopback')) + self.assertIs(cls._lookup_next_layer(cls.__proto__, code), Loopback) + self.assertEqual(Loopback.__index__(), LinkType.NULL) + + def _extract(self, octets: bytes, suffix: str): # type: ignore[no-untyped-def] + import pcapkit + + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, f'loopback{suffix}') + with open(path, 'wb') as file: + file.write(octets) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + return pcapkit.extract(fin=path, nofile=True, store=True, reassembly=True, ip=True) + + def _assert_capture(self, linktype: int, records: 'list[tuple[bytes, str]]', ipv4: int) -> 'list': + """Extract ``records`` under ``linktype`` in both formats; check chains and rebuilds.""" + from pcapkit.protocols.misc.null import NoPayload + from pcapkit.protocols.misc.pcap.frame import Frame + + frames = [] + for build, suffix in ((_pcap, '.pcap'), (_pcapng, '.pcapng')): + with self.subTest(linktype=linktype, format=suffix): + extractor = self._extract(build(linktype, [record for record, _ in records]), suffix) + self.assertEqual([frame.protochain.chain for frame in extractor.frame], + [chain for _, chain in records]) + self.assertEqual(len(extractor.reassembly.ipv4), ipv4) + for frame in extractor.frame: + if isinstance(frame, Frame): + self._assert_rebuilds(frame, num=frame._fnum, header=frame._ghdr) + else: + self._assert_rebuilds(frame, num=frame._fnum, sct=frame._sect, ctx=frame._ctx) + layer = frame.payload + while not isinstance(layer, NoPayload): + self._assert_rebuilds(layer) + layer = layer.payload + frames.extend(extractor.frame) + return frames + + def test_null_captures_dissect_in_either_byte_order(self) -> None: + self._assert_capture(0, [ + (_header(2, 'little') + IPV4, 'Loopback:IPv4:UDP'), + (_header(30, 'little') + IPV6, 'Loopback:IPv6:UDP'), + (_header(2, 'big') + IPV4, 'Loopback:IPv4:UDP'), + (_header(24, 'big') + IPV6, 'Loopback:IPv6:UDP'), + (_header(7, 'little') + IPV4, 'Loopback:Raw'), + ], ipv4=2) + + def test_loop_captures_dissect_in_network_byte_order(self) -> None: + self._assert_capture(108, [ + (_header(2, 'big') + IPV4, 'Loopback:IPv4:UDP'), + (_header(24, 'big') + IPV6, 'Loopback:IPv6:UDP'), + (_header(30, 'big') + IPV6, 'Loopback:IPv6:UDP'), + (_header(7, 'big') + IPV4, 'Loopback:Raw'), + ], ipv4=1) + + def test_little_endian_loop_family_is_not_ipv4(self) -> None: + """``LINKTYPE_LOOP`` is network byte order: ``02 00 00 00`` is family 0x02000000.""" + from pcapkit.const.reg.linktype import LinkType + from pcapkit.protocols.link.loopback import Loopback + from pcapkit.protocols.misc.raw import Raw + + raw = _header(2, 'little') + IPV4 + frames = self._assert_capture(108, [(raw, 'Loopback:Raw')], ipv4=0) + parsed = Loopback(raw, len(raw), alias=LinkType.LOOP) + for layer in [frame.payload for frame in frames] + [parsed]: + with self.subTest(layer=layer): + self.assertEqual((layer.info.family, layer.info.byteorder), (0x0200_0000, 'big')) + self.assertIsInstance(layer.payload, Raw) + self._assert_rebuilds_info(parsed) + + def _assert_rebuilds_info(self, layer) -> None: # type: ignore[no-untyped-def] + """``from_data`` of ``layer``'s info, or of its dict, yields that same info and octets.""" + for data in (layer.info, layer.info.to_dict()): + with self.subTest(data=type(data).__name__): + rebuilt = type(layer).from_data(data) + self.assertEqual(rebuilt.info, layer.info) + self.assertEqual(rebuilt.data.hex(), layer.data.hex()) + + def test_rebuild_keeps_the_parsed_info(self) -> None: + """Every family, in either byte order, under every link-type hint, rebuilds to its own info.""" + from pcapkit.const.reg.linktype import LinkType + from pcapkit.protocols.link.loopback import Loopback + + for family in (0, 2, 7, 23, 24, 28, 30): + payload = IPV6 if family in (24, 28, 30) else IPV4 + for byteorder in ('little', 'big'): + raw = _header(family, byteorder) + payload + for hint in ({}, {'alias': LinkType.NULL}, {'alias': LinkType.LOOP}): + with self.subTest(family=family, byteorder=byteorder, **hint): + parsed = Loopback(raw, len(raw), **hint) + if hint.get('alias') == LinkType.LOOP: + expected = (int.from_bytes(raw[:4], 'big'), 'big') + else: # inferred; a family of 0 reads the same either way + expected = (family, 'little' if family == 0 else byteorder) + self.assertEqual((parsed.info.family, parsed.info.byteorder), expected) + self.assertEqual(parsed.data, raw) + self._assert_rebuilds_info(parsed) + + def test_read_rejects_an_unknown_byteorder(self) -> None: + from pcapkit.protocols.link.loopback import Loopback + from pcapkit.utilities.exceptions import ProtocolError + + raw = _header(2, 'little') + IPV4 + with self.assertRaises(ProtocolError): + Loopback(raw, len(raw), byteorder='native') + + def test_family_byteorder(self) -> None: + from pcapkit.const.reg.linktype import LinkType + from pcapkit.protocols.link.loopback import family_byteorder + + for octets, null in ((b'\x02\x00\x00\x00', 'little'), (b'\x00\x00\x00\x02', 'big'), + (bytes(4), 'little'), (b'\x02\x00\x02\x00', 'little')): + with self.subTest(octets=octets.hex()): + self.assertEqual(family_byteorder(octets), null) + self.assertEqual(family_byteorder(octets, LinkType.NULL), null) + self.assertEqual(family_byteorder(octets, LinkType.LOOP), 'big') + + def test_short_header_is_kept_raw(self) -> None: + from pcapkit.protocols.misc.raw import Raw + + extractor = self._extract(_pcap(0, [b'\x02\x00\x00']), '.pcap') + frame = extractor.frame[0] + self.assertIsInstance(frame.payload, Raw) + self._assert_rebuilds(frame, num=frame._fnum, header=frame._ghdr) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/misc/test_raw_linktype_dispatch_1502_unit.py b/tests/protocols/misc/test_raw_linktype_dispatch_1502_unit.py new file mode 100644 index 0000000000..5696970fcf --- /dev/null +++ b/tests/protocols/misc/test_raw_linktype_dispatch_1502_unit.py @@ -0,0 +1,112 @@ +# -*- coding: utf-8 -*- +"""GitHub issue #1502: ``LINKTYPE_RAW`` (101) is dissected as IPv4 or IPv6. + +Neither :attr:`Frame.__proto__ ` +nor :attr:`PCAPNG.__proto__ ` +had an entry for ``LinkType.RAW``, so a raw-IP frame was kept as +:class:`~pcapkit.protocols.misc.raw.Raw` and never reached reassembly. It +carries a bare IPv4 or IPv6 datagram that only the version nibble tells apart, +so both now dispatch it as ``IPV4`` or ``IPV6`` by that nibble, unless ``RAW`` +itself is registered. + +Every layer is rebuilt byte-exactly, from ``info`` and from ``info.to_dict()``. +Each capture is built here and written to a temporary directory. Everything from +:mod:`pcapkit` is imported inside each test, after +:func:`~tests._support.reimport_once_per_class`. + +""" + +import os +import tempfile +import unittest +import warnings +from unittest import mock + +from tests._support import reimport_once_per_class +from tests.protocols.link.test_loopback_unit import IPV4, IPV6, _pcap, _pcapng + +#: Records of a raw-IP capture, and the protocol chain each one reads as. +RECORDS = ((IPV4, 'IPv4:UDP'), (IPV6, 'IPv6:UDP'), (b'\x50' + IPV4[1:], 'RAW')) +FORMATS = ((_pcap, '.pcap'), (_pcapng, '.pcapng')) + + +class TestRawLinktypeDispatch(unittest.TestCase): + """Pin how both capture formats dispatch a raw-IP frame.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _extract(self, octets: bytes, suffix: str): # type: ignore[no-untyped-def] + import pcapkit + + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, f'raw{suffix}') + with open(path, 'wb') as file: + file.write(octets) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + return pcapkit.extract(fin=path, nofile=True, store=True, reassembly=True, ip=True) + + @staticmethod + def _keywords(frame): # type: ignore[no-untyped-def] + from pcapkit.protocols.misc.pcap.frame import Frame + + if isinstance(frame, Frame): + return {'num': frame._fnum, 'header': frame._ghdr} + return {'num': frame._fnum, 'sct': frame._sect, 'ctx': frame._ctx} + + def test_version_nibble_selects_the_layer(self) -> None: + from pcapkit.protocols.misc.null import NoPayload + + for build, suffix in FORMATS: + with self.subTest(format=suffix): + extractor = self._extract(build(101, [record for record, _ in RECORDS]), suffix) + self.assertEqual([frame.protochain.chain for frame in extractor.frame], + [chain for _, chain in RECORDS]) + self.assertEqual(len(extractor.reassembly.ipv4), 1) + for frame in extractor.frame: + layer, keywords = frame, self._keywords(frame) + while not isinstance(layer, NoPayload): + for data in (layer.info, layer.info.to_dict()): + with self.subTest(frame=frame.info.number, layer=type(layer).__name__, + data=type(data).__name__): + self.assertEqual(type(layer).from_data(data, **keywords).data.hex(), + layer.data.hex()) + layer, keywords = layer.payload, {} + + def test_dispatch_reaches_what_ipv4_is_registered_against(self) -> None: + from pcapkit.const.reg.linktype import LinkType + from pcapkit.protocols.misc.pcap.frame import Frame + from pcapkit.protocols.misc.pcapng import PCAPNG + from pcapkit.protocols.misc.raw import Raw + + marker = type('W1502Marker', (Raw,), {'__module__': __name__}) + for (build, suffix), cls in zip(FORMATS, (Frame, PCAPNG)): + with self.subTest(format=suffix), mock.patch.dict(cls.__proto__, {LinkType.IPV4: marker}): + extractor = self._extract(build(101, [IPV4]), suffix) + self.assertIsInstance(extractor.frame[0].payload, marker) + + def test_an_entry_registered_under_raw_is_used_as_it_stands(self) -> None: + from pcapkit.const.reg.linktype import LinkType + from pcapkit.protocols.misc.pcap.frame import Frame + from pcapkit.protocols.misc.pcapng import PCAPNG + from pcapkit.protocols.misc.raw import Raw + + marker = type('W1502RawMarker', (Raw,), {'__module__': __name__}) + for (build, suffix), cls in zip(FORMATS, (Frame, PCAPNG)): + with self.subTest(format=suffix), mock.patch.dict(cls.__proto__, {LinkType.RAW: marker}): + extractor = self._extract(build(101, [IPV4, IPV6]), suffix) + self.assertEqual([type(frame.payload) for frame in extractor.frame], [marker, marker]) + + def test_raw_is_not_registered(self) -> None: + from pcapkit.const.reg.linktype import LinkType + from pcapkit.protocols.misc.pcap.frame import Frame + from pcapkit.protocols.misc.pcapng import PCAPNG + + for cls in (Frame, PCAPNG): + with self.subTest(table=cls.__name__): + self.assertNotIn(LinkType.RAW, cls.__proto__) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/test_dispatch_registry_unit.py b/tests/protocols/test_dispatch_registry_unit.py index c60167b5f5..c6d8d9e535 100644 --- a/tests/protocols/test_dispatch_registry_unit.py +++ b/tests/protocols/test_dispatch_registry_unit.py @@ -212,7 +212,7 @@ def test_registry_currently_matches_pinned_target(self) -> None: ) def test_cases_cover_every_table_named_in_the_issue(self) -> None: - """The enumeration finds all 39 entries the issue counted, across all seven tables. + """The enumeration finds all 47 entries, across the seven tables of #496 and #1574's loopback table. A guard on the shape of the result rather than on any one case: if the registries grow or shrink without this module noticing, the per-family @@ -224,18 +224,20 @@ def test_cases_cover_every_table_named_in_the_issue(self) -> None: # new entry, Internet.__proto__[TransType.Shim6], registered at # IPv6_Ext where it previously had none at all (defaulted to # Raw). Diffed against origin/main: no other table changed shape. - self.assertEqual(len(cases), 39, - 'expected exactly 39 entries across the seven __proto__ ' - 'tables named in GitHub issue #496; a different count ' - 'means a registry changed shape and this module was not ' - 'updated to match') + # #1574: 39 -> 47 -- LinkType.NULL and LinkType.LOOP in both link-type + # tables, and the four address families of the new Loopback.__proto__. + self.assertEqual(len(cases), 47, + 'expected exactly 47 entries across the seven __proto__ ' + 'tables named in GitHub issue #496 and the loopback table ' + 'of #1574; a different count means a registry changed ' + 'shape and this module was not updated to match') by_family = {} # type: dict[str, int] for case in cases: by_family[case.family] = by_family.get(case.family, 0) + 1 self.assertEqual(by_family, { 'link': 7, 'internet': 17, 'tcp': 4, 'udp': 3, 'sctp': 2, - 'pcap-frame': 3, 'pcapng-frame': 3, + 'pcap-frame': 5, 'pcapng-frame': 5, 'loopback': 4, }) def test_dispatch_reaches_target_or_is_a_recorded_degrade(self) -> None: diff --git a/tests/protocols/test_short_read_roundtrip_unit.py b/tests/protocols/test_short_read_roundtrip_unit.py index 99a0e102ae..f508820d55 100644 --- a/tests/protocols/test_short_read_roundtrip_unit.py +++ b/tests/protocols/test_short_read_roundtrip_unit.py @@ -46,6 +46,8 @@ def _ipv6(code: int, body: bytes) -> bytes: ('arp', 'link.arp:ARP', {}, bytes.fromhex('000186dd06100001') + bytes(range(44))), ('rarp', 'application.rarp:RARP', {}, bytes.fromhex('0001080006040003') + bytes(range(20))), ('l2tpv2', 'link.l2tpv2:L2TPv2', {}, bytes.fromhex('fff2000e00000000000000000000')), + ('loopback-null', 'link.loopback:Loopback', {}, bytes.fromhex('020000004500')), + ('loopback-loop', 'link.loopback:Loopback', {'alias': 108}, bytes.fromhex('000000024500')), ('ospf', 'application.ospf:OSPF', {}, bytes.fromhex('020100180102030400000000abcd0000') + bytes(8)), ('ipx', 'internet.ipx:IPX', {}, bytes.fromhex('ffff001e0000000000000101010101010451000000000202020202020452')), ('ipv4', 'internet.ipv4:IPv4', {}, bytes.fromhex('450000141234000040fd7b83c0000201c6336401')),