diff --git a/docs/source/contributing/pep.rst b/docs/source/contributing/pep.rst index 53a9fa4ab0..f97e882953 100644 --- a/docs/source/contributing/pep.rst +++ b/docs/source/contributing/pep.rst @@ -40,7 +40,7 @@ step-by-step guide for adding a protocol, and is worth reading before starting one. Grouped by each TCP/IP layer and ordered by protocol name alphabetically, the ones outstanding are, -* Link Layer: DSL, EAPOL, FDDI, ISDN, LINUX_SLL, LINUX_SLL2, NDP, PPP +* Link Layer: DSL, EAPOL, FDDI, ISDN, NDP, PPP * Internet Layer: ECN, ICMP, ICMPv6, IGMP, Shim6 * Transport Layer: DCCP, RSVP * Application Layer: BGP, DHCP, DHCPv6, DNS, IMAP, LDAP, MQTT, NNTP, @@ -49,10 +49,10 @@ the ones outstanding are, Several of these come in pairs that want a shared abstract base rather than two independent implementations, in the way -:class:`~pcapkit.protocols.internet.ip.IP` already covers its family: ICMP with -ICMPv6, and ``LINUX_SLL`` with ``LINUX_SLL2``. **Done** for TLS/SSL with -DTLS: :class:`~pcapkit.protocols.application.ssl.SSL` is the shared -record-layer base of :class:`~pcapkit.protocols.application.tls.TLS` and +:class:`~pcapkit.protocols.internet.ip.IP` already covers its family: ICMP +with ICMPv6, and TLS/SSL with DTLS. **Done** for TLS/SSL with DTLS: +:class:`~pcapkit.protocols.application.ssl.SSL` is the shared record-layer base +of :class:`~pcapkit.protocols.application.tls.TLS` and :class:`~pcapkit.protocols.application.dtls.DTLS` -- see the `DTLS`_ section below. :class:`~pcapkit.protocols.link.vlan.VLAN` is the closer precedent for a pair @@ -66,8 +66,7 @@ whose *layout* is identical -- it holds the whole of the tag, and ``NotImplemented`` folder. **Those files are not a roadmap.** They are scratch reminders, so a stub's presence does not mean a protocol is planned or claimed, and its absence does not mean the protocol is unwanted: - ``LINUX_SLL``, ``LINUX_SLL2`` and ``QUIC`` have no stub and are on the - list above, + ``QUIC`` has no stub and is on the list above, while **NGAP** has none either and is implemented. Take this page as the record and ignore the folder. @@ -363,9 +362,7 @@ Follow-ups to the bindings above: instead of guessing, since a guess path is wrong on non-HTTP input. Beyond those, the gaps most likely to be met in a real capture are ICMP (1), -ICMPv6 (58) and IGMP (2) on the internet layer, each of which has a stub; -and ``LINUX_SLL`` and ``LINUX_SLL2`` at the link layer, which every -``tcpdump -i any`` capture uses and which have no stub. +ICMPv6 (58) and IGMP (2) on the internet layer, each of which has a stub. PCAPNG Support -------------- @@ -1084,12 +1081,10 @@ post-wave-1 consistency sweep described under `Release Plan — 1.5.0 in Two Steps`_ above. **Wave 2 — the protocol pairs that want a shared abstract base**, per `More -Protocols, More!!!`_ above: **ICMP with ICMPv6**, **TLS/SSL with DTLS**, and -``LINUX_SLL`` with ``LINUX_SLL2``, plus **IGMP** and **DCCP** on their own. These are -grouped because the shared-base question is the design work and doing either half of -a pair alone would answer it twice. ``LINUX_SLL`` matters out of proportion to its -size: every ``tcpdump -i any`` capture uses it, and neither variant has even a -stub. **DCCP** is the transport-layer implementation, tracked by :issue:`2489`, and +Protocols, More!!!`_ above: **ICMP with ICMPv6** and **TLS/SSL with DTLS**, +plus **IGMP** and **DCCP** on their own. These are grouped because the shared-base +question is the design work and doing either half of a pair alone would answer it +twice. **DCCP** is the transport-layer implementation, tracked by :issue:`2489`, and blocked on :issue:`2479` for its pseudo-header checksum. **Done within wave 2 — TLS/SSL with DTLS.** The `DTLS`_ section above lists @@ -1097,17 +1092,24 @@ what is decoded (the shared :class:`~pcapkit.protocols.application.ssl.SSL` base, TLS, DTLS, and handshake message framing) and what is not (DTLS handshake-fragment reassembly, :issue:`2440`). -**Done ahead of wave 2 — handlers for the** ``DLT_NULL`` **and** ``DLT_RAW`` -**link types.** Both are genuine DLTs that a capture can legitimately be rooted -in — ``DLT_NULL`` (0) is BSD's loopback encapsulation, ``DLT_RAW`` (101) an IP -packet with no link header at all. +**Done ahead of wave 2 — handlers for the** ``DLT_NULL``, ``DLT_RAW``, +``DLT_LINUX_SLL`` **and** ``DLT_LINUX_SLL2`` **link types.** Each is a +genuine DLT that a capture can legitimately be rooted in — ``DLT_NULL`` (0) +is BSD's loopback encapsulation, ``DLT_RAW`` (101) an IP packet with no link +header at all, and ``DLT_LINUX_SLL`` (113) with ``DLT_LINUX_SLL2`` (276) the +Linux "cooked" capture header every ``tcpdump -i any`` capture uses. :class:`~pcapkit.protocols.link.loopback.Loopback` dissects ``DLT_NULL`` and -``DLT_LOOP`` (:issue:`1574`), and the frame dispatcher reads ``DLT_RAW`` as IPv4 -or IPv6 by its version nibble (:issue:`1502`), so of the link types this -section names only ``LINUX_SLL`` and ``LINUX_SLL2`` remain enumerated but -unparsed. Neither ``NULL`` nor ``RAW`` is an honest stand-in for "unknown link -type", which is why the toolkit raises for an unresolvable link-layer name -instead of defaulting to :attr:`LinkType.NULL `. +``DLT_LOOP`` (:issue:`1574`), the frame dispatcher reads ``DLT_RAW`` as IPv4 +or IPv6 by its version nibble (:issue:`1502`), and +:class:`~pcapkit.protocols.link.linux_sll1.LinuxSLL1` with +:class:`~pcapkit.protocols.link.linux_sll2.LinuxSLL2` dissect the Linux +cooked-capture variants (:issue:`2438`). The ``arphrd_type`` values +neither header treats as carrying an EtherType (``NETLINK``, ``IPGRE``, ``IP6GRE``, +``IEEE80211_RADIOTAP``, ``FRAD``) have no dissector of their own, tracked +as :issue:`2460`. Neither ``NULL`` nor ``RAW`` is an honest stand-in for "unknown +link type", which is why the toolkit raises for an unresolvable link-layer +name instead of defaulting to :attr:`LinkType.NULL +`. **Wave 3 — the remaining protocols** from the same list, taken three or four at a time. QUIC, DSL, FDDI and ISDN come last: QUIC because it is really HTTP/3 over a diff --git a/docs/source/ext.rst b/docs/source/ext.rst index 151dfd046f..a9d385ca47 100644 --- a/docs/source/ext.rst +++ b/docs/source/ext.rst @@ -30,6 +30,12 @@ The protocol classes :mod:`pcapkit` ships: + + +-----------------------+-------------------------------------------------------------+ | | L2TP Family | :class:`pcapkit.protocols.link.l2tpv2.L2TPv2` | + +----------------+-----------------------+-------------------------------------------------------------+ +| | | :class:`pcapkit.protocols.link.linux_sll.LinuxSLL` | ++ + +-----------------------+-------------------------------------------------------------+ +| | LINUX_SLL | :class:`pcapkit.protocols.link.linux_sll1.LinuxSLL1` | ++ + Family +-----------------------+-------------------------------------------------------------+ +| | | :class:`pcapkit.protocols.link.linux_sll2.LinuxSLL2` | ++ +----------------+-----------------------+-------------------------------------------------------------+ | | :class:`pcapkit.protocols.link.loopback.Loopback` | + +----------------+-----------------------+-------------------------------------------------------------+ | | | :class:`pcapkit.protocols.link.vlan.VLAN` | diff --git a/docs/source/pcapkit/protocols/index.rst b/docs/source/pcapkit/protocols/index.rst index 280c830988..4c4fdbb691 100644 --- a/docs/source/pcapkit/protocols/index.rst +++ b/docs/source/pcapkit/protocols/index.rst @@ -63,7 +63,7 @@ from given information. The class hierarchy of :mod:`pcapkit.protocols`: end subgraph link [Link Layer] - Link --> Ethernet & L2TP & Loopback & VLAN & ARP + Link --> Ethernet & L2TP & LinuxSLL & Loopback & VLAN & ARP subgraph arp [ARP Family] ARP --> InARP @@ -76,6 +76,10 @@ from given information. The class hierarchy of :mod:`pcapkit.protocols`: subgraph l2tp [L2TP Family] L2TP --> L2TPv2 end + + subgraph sll [LinuxSLL Family] + LinuxSLL --> LinuxSLL1 & LinuxSLL2 + end end subgraph internet [Internet Layer] @@ -117,6 +121,9 @@ from given information. The class hierarchy of :mod:`pcapkit.protocols`: click Ethernet "link/ethernet.html#pcapkit.protocols.link.ethernet.Ethernet" click L2TP "link/l2tp.html#pcapkit.protocols.link.l2tp.L2TP" click L2TPv2 "link/l2tpv2.html#pcapkit.protocols.link.l2tpv2.L2TPv2" + click LinuxSLL "link/linux_sll.html#pcapkit.protocols.link.linux_sll.LinuxSLL" + click LinuxSLL1 "link/linux_sll1.html#pcapkit.protocols.link.linux_sll1.LinuxSLL1" + click LinuxSLL2 "link/linux_sll2.html#pcapkit.protocols.link.linux_sll2.LinuxSLL2" click Loopback "link/loopback.html#pcapkit.protocols.link.loopback.Loopback" click VLAN "link/vlan.html#pcapkit.protocols.link.vlan.VLAN" click C_Tag "link/c_tag.html#pcapkit.protocols.link.c_tag.C_Tag" diff --git a/docs/source/pcapkit/protocols/link/index.rst b/docs/source/pcapkit/protocols/link/index.rst index af786477c7..f0b47982e2 100644 --- a/docs/source/pcapkit/protocols/link/index.rst +++ b/docs/source/pcapkit/protocols/link/index.rst @@ -26,6 +26,9 @@ link layer, with detailed implementation and methods. arp l2tp l2tpv2 + linux_sll + linux_sll1 + linux_sll2 loopback vlan c_tag diff --git a/docs/source/pcapkit/protocols/link/linux_sll.rst b/docs/source/pcapkit/protocols/link/linux_sll.rst new file mode 100644 index 0000000000..c7651bc1f7 --- /dev/null +++ b/docs/source/pcapkit/protocols/link/linux_sll.rst @@ -0,0 +1,64 @@ +LinuxSLL - Linux "Cooked" Capture Encapsulation Family +====================================================== + +.. module:: pcapkit.protocols.link.linux_sll + +:mod:`pcapkit.protocols.link.linux_sll` contains +:class:`~pcapkit.protocols.link.linux_sll.LinuxSLL` only, the base class of the +Linux "cooked" capture encapsulation family: + +.. list-table:: + :header-rows: 1 + + * - ``LinkType`` + - Class + * - ``LINUX_SLL`` (113) + - :class:`~pcapkit.protocols.link.linux_sll1.LinuxSLL1` + * - ``LINUX_SLL2`` (276) + - :class:`~pcapkit.protocols.link.linux_sll2.LinuxSLL2` + +The versions differ in field order and width, so each parses its own header. +The base holds what they share: the next-layer dispatch, and link-layer address +validation. It has no registry index, so its ``__index__`` raises. + +.. autoclass:: pcapkit.protocols.link.linux_sll.LinuxSLL + :no-members: + :show-inheritance: + + .. automethod:: id + .. automethod:: __index__ + .. automethod:: _import_next_layer + +Auxiliary Functions +------------------- + +.. autofunction:: pcapkit.protocols.link.linux_sll._resolve_next_layer_proto +.. autofunction:: pcapkit.protocols.link.linux_sll._validate_link_addr + +Auxiliary Constants +------------------- + +.. autodata:: pcapkit.protocols.link.linux_sll._ARPHRD_NOT_ETHERTYPE +.. autodata:: pcapkit.protocols.link.linux_sll._ETHERTYPE_FIELD + :no-value: + +Hardware Types +-------------- + +The ``arphrd_type`` values in ``_ARPHRD_NOT_ETHERTYPE`` are named by the enumeration below. + +.. autoclass:: pcapkit.const.linux.arphrd.ARPHRD + :members: + :show-inheritance: + +.. _linux-sll-labels: + +Selecting a Version +------------------- + +The family name selects either version, and each version's own name selects that +version only, as :class:`~pcapkit.protocols.internet.ipv4.IPv4` and +:class:`~pcapkit.protocols.internet.ipv6.IPv6` do under +:class:`~pcapkit.protocols.internet.ip.IP`. ``'LinuxSLL'`` is in both versions; +``'LinuxSLL1' in`` a v2 layer is :data:`False`. String ``==`` does not expand a +key, so ``v1 == 'LinuxSLL'`` is :data:`False` while ``'LinuxSLL' in v1`` is :data:`True`. diff --git a/docs/source/pcapkit/protocols/link/linux_sll1.rst b/docs/source/pcapkit/protocols/link/linux_sll1.rst new file mode 100644 index 0000000000..1be95ee3cf --- /dev/null +++ b/docs/source/pcapkit/protocols/link/linux_sll1.rst @@ -0,0 +1,75 @@ +Linux "Cooked" Capture Encapsulation +===================================== + +.. module:: pcapkit.protocols.link.linux_sll1 + +:mod:`pcapkit.protocols.link.linux_sll1` contains +:class:`~pcapkit.protocols.link.linux_sll1.LinuxSLL1` +only, which implements the extractor for the Linux "cooked" capture +encapsulation [*]_ of ``LINKTYPE_LINUX_SLL`` -- the header every +``tcpdump -i any`` capture on Linux uses -- whose structure is described as +below: + +.. table:: + + ====== ===== ============================== ================================ + Octets Bits Name Description + ====== ===== ============================== ================================ + 0 0 ``linux_sll.packet_type`` Packet Type + ------ ----- ------------------------------ -------------------------------- + 2 16 ``linux_sll.arphrd_type`` ARP Hardware Type + ------ ----- ------------------------------ -------------------------------- + 4 32 ``linux_sll.link_addr_length`` Link-Layer Address Length + ------ ----- ------------------------------ -------------------------------- + 6 48 ``linux_sll.link_addr`` Link-Layer Address (padded) + ------ ----- ------------------------------ -------------------------------- + 14 112 ``linux_sll.protocol_type`` Protocol (Internet Layer) + ====== ===== ============================== ================================ + +Next-layer dispatch is shared with +:class:`~pcapkit.protocols.link.linux_sll2.LinuxSLL2`; see +:mod:`~pcapkit.protocols.link.linux_sll`. Its +:meth:`~pcapkit.protocols.link.linux_sll1.LinuxSLL1.id` is +``('LinuxSLL1',)``, and its +:attr:`~pcapkit.protocols.link.linux_sll1.LinuxSLL1.alias` is ``'LINUX_SLL1'``; +see :ref:`linux-sll-labels`. + +.. autoclass:: pcapkit.protocols.link.linux_sll1.LinuxSLL1 + :no-members: + :show-inheritance: + + .. autoproperty:: name + .. autoproperty:: alias + .. autoproperty:: info_name + .. autoproperty:: length + .. autoproperty:: protocol + + .. automethod:: id + .. automethod:: read + .. automethod:: make + + .. automethod:: _make_data + + .. automethod:: __index__ + +Header Schemas +-------------- + +.. module:: pcapkit.protocols.schema.link.linux_sll1 + +.. autoclass:: pcapkit.protocols.schema.link.linux_sll1.LinuxSLL1 + :members: + :show-inheritance: + +Data Models +----------- + +.. module:: pcapkit.protocols.data.link.linux_sll1 + +.. autoclass:: pcapkit.protocols.data.link.linux_sll1.LinuxSLL1 + :members: + :show-inheritance: + +.. rubric:: Footnotes + +.. [*] https://www.tcpdump.org/linktypes/LINKTYPE_LINUX_SLL.html diff --git a/docs/source/pcapkit/protocols/link/linux_sll2.rst b/docs/source/pcapkit/protocols/link/linux_sll2.rst new file mode 100644 index 0000000000..8d0dac66f3 --- /dev/null +++ b/docs/source/pcapkit/protocols/link/linux_sll2.rst @@ -0,0 +1,79 @@ +Linux "Cooked" Capture Encapsulation v2 +========================================= + +.. module:: pcapkit.protocols.link.linux_sll2 + +:mod:`pcapkit.protocols.link.linux_sll2` contains +:class:`~pcapkit.protocols.link.linux_sll2.LinuxSLL2` +only, which implements the extractor for the Linux "cooked" capture +encapsulation v2 [*]_ of ``LINKTYPE_LINUX_SLL2``, whose structure is described +as below: + +.. table:: + + ====== ===== ================================ ================================ + Octets Bits Name Description + ====== ===== ================================ ================================ + 0 0 ``linux_sll2.protocol_type`` Protocol (Internet Layer) + ------ ----- -------------------------------- -------------------------------- + 2 16 ``linux_sll2.reserved`` Reserved + ------ ----- -------------------------------- -------------------------------- + 4 32 ``linux_sll2.interface_index`` Interface Index + ------ ----- -------------------------------- -------------------------------- + 8 64 ``linux_sll2.arphrd_type`` ARP Hardware Type + ------ ----- -------------------------------- -------------------------------- + 10 80 ``linux_sll2.packet_type`` Packet Type + ------ ----- -------------------------------- -------------------------------- + 11 88 ``linux_sll2.link_addr_length`` Link-Layer Address Length + ------ ----- -------------------------------- -------------------------------- + 12 96 ``linux_sll2.link_addr`` Link-Layer Address (padded) + ====== ===== ================================ ================================ + +Unlike v1, ``protocol_type`` comes first, ``interface_index`` is new, and +``packet_type`` and ``link_addr_length`` are 1 octet rather than 2. See +:class:`~pcapkit.protocols.link.linux_sll1.LinuxSLL1` for v1; next-layer +dispatch is shared, see :mod:`~pcapkit.protocols.link.linux_sll`. Its +:meth:`~pcapkit.protocols.link.linux_sll2.LinuxSLL2.id` is +``('LinuxSLL2',)``, and its +:attr:`~pcapkit.protocols.link.linux_sll2.LinuxSLL2.alias` is ``'LINUX_SLL2'``; +see :ref:`linux-sll-labels`. + +.. autoclass:: pcapkit.protocols.link.linux_sll2.LinuxSLL2 + :no-members: + :show-inheritance: + + .. autoproperty:: name + .. autoproperty:: alias + .. autoproperty:: info_name + .. autoproperty:: length + .. autoproperty:: protocol + + .. automethod:: id + .. automethod:: read + .. automethod:: make + + .. automethod:: _make_data + + .. automethod:: __index__ + +Header Schemas +-------------- + +.. module:: pcapkit.protocols.schema.link.linux_sll2 + +.. autoclass:: pcapkit.protocols.schema.link.linux_sll2.LinuxSLL2 + :members: + :show-inheritance: + +Data Models +----------- + +.. module:: pcapkit.protocols.data.link.linux_sll2 + +.. autoclass:: pcapkit.protocols.data.link.linux_sll2.LinuxSLL2 + :members: + :show-inheritance: + +.. rubric:: Footnotes + +.. [*] https://www.tcpdump.org/linktypes/LINKTYPE_LINUX_SLL2.html diff --git a/examples/generators/dispatch.py b/examples/generators/dispatch.py index cd379258d8..13f8856900 100644 --- a/examples/generators/dispatch.py +++ b/examples/generators/dispatch.py @@ -278,6 +278,18 @@ def _loopback(family: 'int', payload: 'bytes', byteorder: 'str' = 'little') -> ' return bytes(Loopback(family=family, byteorder=byteorder, payload=payload)) +def _linux_sll(payload: 'bytes') -> 'bytes': + """A ``LINUX_SLL`` header carrying ``payload``.""" + from pcapkit.protocols.link.linux_sll1 import LinuxSLL1 + return bytes(LinuxSLL1(payload=payload)) + + +def _linux_sll2(payload: 'bytes') -> 'bytes': + """A ``LINUX_SLL2`` header carrying ``payload``.""" + from pcapkit.protocols.link.linux_sll2 import LinuxSLL2 + return bytes(LinuxSLL2(payload=payload)) + + ############################################################################### # Link -- EtherType. ############################################################################### @@ -536,6 +548,10 @@ def _linktype_payload(code: 'Any') -> 'tuple[bytes, bool]': 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 + if code == LinkType.LINUX_SLL: + return _linux_sll(_ip4(6, _tcp(9999))), False + if code == LinkType.LINUX_SLL2: + return _linux_sll2(_ip4(6, _tcp(9999))), False raise LookupError(f'linktype: no payload builder for {code!r}') # pragma: no cover @@ -697,6 +713,8 @@ def _link_enum() -> 'Any': 'pcap-frame/LOOP': ('pcapkit.protocols.link', 'Loopback'), 'pcap-frame/IPV4': ('pcapkit.protocols.internet', 'IPv4'), 'pcap-frame/IPV6': ('pcapkit.protocols.internet', 'IPv6'), + 'pcap-frame/LINUX_SLL': ('pcapkit.protocols.link', 'LinuxSLL1'), + 'pcap-frame/LINUX_SLL2': ('pcapkit.protocols.link', 'LinuxSLL2'), # -- PCAPNG.__proto__ (LinkType) -------------------------------------------- 'pcapng-frame/NULL': ('pcapkit.protocols.link', 'Loopback'), @@ -704,6 +722,8 @@ def _link_enum() -> 'Any': 'pcapng-frame/LOOP': ('pcapkit.protocols.link', 'Loopback'), 'pcapng-frame/IPV4': ('pcapkit.protocols.internet', 'IPv4'), 'pcapng-frame/IPV6': ('pcapkit.protocols.internet', 'IPv6'), + 'pcapng-frame/LINUX_SLL': ('pcapkit.protocols.link', 'LinuxSLL1'), + 'pcapng-frame/LINUX_SLL2': ('pcapkit.protocols.link', 'LinuxSLL2'), # -- Loopback.__proto__ (address family) ------------------------------------ 'loopback/2': ('pcapkit.protocols.internet.ipv4', 'IPv4'), diff --git a/pcapkit/__init__.py b/pcapkit/__init__.py index ffb7913ba2..b60c1e8e61 100644 --- a/pcapkit/__init__.py +++ b/pcapkit/__init__.py @@ -112,7 +112,7 @@ 'Raw', # Raw Packet 'ARP', 'C_Tag', 'Ethernet', 'InARP', 'L2TP', # Link Layer - 'L2TPv2', 'Loopback', 'S_Tag', 'VLAN', + 'L2TPv2', 'LinuxSLL', 'LinuxSLL1', 'LinuxSLL2', '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 c195938cec..a5a782c8b3 100644 --- a/pcapkit/all.py +++ b/pcapkit/all.py @@ -119,8 +119,8 @@ 'PCAPNG', # PCAPNG Format 'NoPayload', # No Payload 'Raw', # Raw Packet - 'ARP', 'C_Tag', 'Ethernet', 'InARP', 'L2TP', 'L2TPv2', - 'Loopback', 'S_Tag', 'VLAN', # Link Layer + 'ARP', 'C_Tag', 'Ethernet', 'InARP', 'L2TP', 'L2TPv2', 'LinuxSLL', 'LinuxSLL1', + 'LinuxSLL2', '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/const/linux/__init__.py b/pcapkit/const/linux/__init__.py new file mode 100644 index 0000000000..bbdd28f49c --- /dev/null +++ b/pcapkit/const/linux/__init__.py @@ -0,0 +1,19 @@ +# -*- coding: utf-8 -*- +# pylint: disable=unused-import +""":mod:`pcapkit.const.linux` Constant Enumerations +===================================================== + +This module contains the constant enumerations of the Linux-specific +encapsulations, e.g. :class:`~pcapkit.protocols.link.linux_sll.LinuxSLL`. +Available enumerations include: + +.. list-table:: + + * - :class:`ARPHRD ` + - Linux ARP Hardware Types + +""" + +from pcapkit.const.linux.arphrd import ARPHRD + +__all__ = ['ARPHRD'] diff --git a/pcapkit/const/linux/arphrd.py b/pcapkit/const/linux/arphrd.py new file mode 100644 index 0000000000..f449b55333 --- /dev/null +++ b/pcapkit/const/linux/arphrd.py @@ -0,0 +1,78 @@ +# -*- coding: utf-8 -*- +"""Linux ARP Hardware Types +============================== + +.. module:: pcapkit.const.linux.arphrd + +This module contains the constant enumeration for the Linux ``ARPHRD_*`` +hardware types of ````, the ``arphrd_type`` of the Linux +"cooked" capture encapsulations. Only the values that pcapkit names are listed. +The rest of the numbering is IANA's, see +:class:`~pcapkit.const.arp.hardware.Hardware`, which differs for most +non-Ethernet types. + +""" + +from typing import TYPE_CHECKING + +from aenum import IntEnum + +from pcapkit.corekit.enum import EnumRegistry +from pcapkit.utilities.exceptions import EnumValueError + +if TYPE_CHECKING: + from typing import Any + +__all__ = ['ARPHRD'] + + +class ARPHRD(EnumRegistry, IntEnum): + """[ARPHRD] Linux ARP hardware types [````]""" + + #: NET/ROM pseudo. + NETROM = 0 + + #: Frame Relay Access Device + FRAD = 770 + + #: GRE over IP + IPGRE = 778 + + #: IEEE 802.11 with radiotap header + IEEE80211_RADIOTAP = 803 + + #: GRE over IPv6 + IP6GRE = 823 + + #: Kernel Netlink socket + NETLINK = 824 + + @classmethod + def _validate_value(cls, value: 'Any') -> 'None': + """Reject ``value`` unless it is an ``arphrd_type`` (a 16-bit value). + + Overrides :meth:`~pcapkit.corekit.enum.EnumLookup._validate_value` + so that :meth:`~pcapkit.corekit.enum.EnumRegistry.register` refuses + the values :meth:`_missing_` refuses. + + Raises: + EnumValueError: If ``value`` is not an integer in ``0..0xFFFF``. + + """ + if not (isinstance(value, int) and 0x0000 <= value <= 0xFFFF): + raise EnumValueError('%r is not a valid %s' % (value, cls.__name__)) + + @classmethod + def _missing_(cls, value: 'int') -> 'ARPHRD': + """Resolve an unnamed ``arphrd_type`` to an unregistered member. + + Args: + value: Value to get enum item. + + Raises: + ValueError: If ``value`` is not a 16-bit integer. + + """ + if not (isinstance(value, int) and 0x0000 <= value <= 0xFFFF): + raise ValueError('%r is not a valid %s' % (value, cls.__name__)) + return cls._unregistered_member(value, 'Unassigned') diff --git a/pcapkit/protocols/__init__.py b/pcapkit/protocols/__init__.py index b2665e3e6d..225b11131b 100644 --- a/pcapkit/protocols/__init__.py +++ b/pcapkit/protocols/__init__.py @@ -50,7 +50,8 @@ 'Raw', # Link Layer - 'ARP', 'C_Tag', 'Ethernet', 'InARP', 'L2TP', 'L2TPv2', 'Loopback', 'S_Tag', 'VLAN', + 'ARP', 'C_Tag', 'Ethernet', 'InARP', 'L2TP', 'L2TPv2', 'LinuxSLL', 'LinuxSLL1', + 'LinuxSLL2', '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 9c700e139b..eb6880ee91 100644 --- a/pcapkit/protocols/data/__init__.py +++ b/pcapkit/protocols/data/__init__.py @@ -65,6 +65,9 @@ # Layer Two Tunnelling Protocol 'L2TP', 'L2TP_Flags', + # Linux "Cooked" Capture Encapsulation + 'LinuxSLL1', 'LinuxSLL2', + # BSD Loopback Encapsulation 'Loopback', diff --git a/pcapkit/protocols/data/link/__init__.py b/pcapkit/protocols/data/link/__init__.py index 00584b66e0..cae293e6cc 100644 --- a/pcapkit/protocols/data/link/__init__.py +++ b/pcapkit/protocols/data/link/__init__.py @@ -13,6 +13,10 @@ from pcapkit.protocols.data.link.l2tp import L2TP from pcapkit.protocols.data.link.l2tp import Flags as L2TP_Flags +# Linux "Cooked" Capture Encapsulation +from pcapkit.protocols.data.link.linux_sll1 import LinuxSLL1 +from pcapkit.protocols.data.link.linux_sll2 import LinuxSLL2 + # BSD Loopback Encapsulation from pcapkit.protocols.data.link.loopback import Loopback @@ -30,6 +34,9 @@ # Layer Two Tunnelling Protocol 'L2TP', 'L2TP_Flags', + # Linux "Cooked" Capture Encapsulation + 'LinuxSLL1', 'LinuxSLL2', + # BSD Loopback Encapsulation 'Loopback', diff --git a/pcapkit/protocols/data/link/linux_sll1.py b/pcapkit/protocols/data/link/linux_sll1.py new file mode 100644 index 0000000000..bdc32cb5fc --- /dev/null +++ b/pcapkit/protocols/data/link/linux_sll1.py @@ -0,0 +1,41 @@ +# -*- coding: utf-8 -*- +"""data models for Linux "cooked" capture encapsulation (``LINUX_SLL``, v1)""" + +from typing import TYPE_CHECKING + +from pcapkit.corekit.infoclass import info_final +from pcapkit.protocols.data.protocol import Protocol + +if TYPE_CHECKING: + from pcapkit.const.arp.hardware import Hardware + +__all__ = ['LinuxSLL1'] + + +@info_final +class LinuxSLL1(Protocol): + """Data model for Linux "cooked" capture encapsulation (``LINUX_SLL``, v1). + + ``protocol_type`` is an :class:`~pcapkit.const.reg.ethertype.EtherType` + for every :attr:`arphrd_type` except a few -- see + :class:`~pcapkit.protocols.link.linux_sll1.LinuxSLL1`; it is kept as its + raw wire value here regardless. + + """ + + #: Packet type (direction), where ``0`` is to us, ``1`` broadcast, ``2`` + #: multicast, ``3`` sent to another host promiscuously and ``4`` sent by us. + packet_type: 'int' + #: ARP hardware type of the interface the packet was seen on. + arphrd_type: 'Hardware' + #: Link-layer address length actually in use; may exceed 8. + link_addr_length: 'int' + #: Link-layer address, padded to 8 octets as captured; only the first + #: :attr:`link_addr_length` of them are meaningful. + link_addr: 'bytes' + #: Protocol (internet layer), kept as its raw wire value. + protocol_type: 'int' + + if TYPE_CHECKING: + def __init__(self, packet_type: 'int', arphrd_type: 'Hardware', link_addr_length: 'int', # pylint: disable=unused-argument,super-init-not-called + link_addr: 'bytes', protocol_type: 'int') -> 'None': ... # pylint: disable=unused-argument,multiple-statements diff --git a/pcapkit/protocols/data/link/linux_sll2.py b/pcapkit/protocols/data/link/linux_sll2.py new file mode 100644 index 0000000000..0f5e175b60 --- /dev/null +++ b/pcapkit/protocols/data/link/linux_sll2.py @@ -0,0 +1,47 @@ +# -*- coding: utf-8 -*- +"""data models for Linux "cooked" capture encapsulation v2 (``LINUX_SLL2``)""" + +from typing import TYPE_CHECKING + +from pcapkit.corekit.infoclass import info_final +from pcapkit.protocols.data.protocol import Protocol + +if TYPE_CHECKING: + from pcapkit.const.arp.hardware import Hardware + +__all__ = ['LinuxSLL2'] + + +@info_final +class LinuxSLL2(Protocol): + """Data model for Linux "cooked" capture encapsulation v2 (``LINUX_SLL2``). + + ``protocol_type`` is an :class:`~pcapkit.const.reg.ethertype.EtherType` + for every :attr:`arphrd_type` except a few -- see + :class:`~pcapkit.protocols.link.linux_sll2.LinuxSLL2`; it is kept as its + raw wire value here regardless. + + """ + + #: Protocol (internet layer), kept as its raw wire value. + protocol_type: 'int' + #: Reserved, must be zero. + reserved: 'int' + #: Index, in the capturing host's own numbering, of the interface the + #: packet was seen on. + interface_index: 'int' + #: ARP hardware type of the interface the packet was seen on. + arphrd_type: 'Hardware' + #: Packet type (direction), where ``0`` is to us, ``1`` broadcast, ``2`` + #: multicast, ``3`` sent to another host promiscuously and ``4`` sent by us. + packet_type: 'int' + #: Link-layer address length actually in use; may exceed 8. + link_addr_length: 'int' + #: Link-layer address, padded to 8 octets as captured; only the first + #: :attr:`link_addr_length` of them are meaningful. + link_addr: 'bytes' + + if TYPE_CHECKING: + def __init__(self, protocol_type: 'int', reserved: 'int', interface_index: 'int', # pylint: disable=unused-argument,super-init-not-called + arphrd_type: 'Hardware', packet_type: 'int', link_addr_length: 'int', # pylint: disable=unused-argument + link_addr: 'bytes') -> 'None': ... # pylint: disable=unused-argument,multiple-statements diff --git a/pcapkit/protocols/link/__init__.py b/pcapkit/protocols/link/__init__.py index ecb7a0f604..0a35e6a5db 100644 --- a/pcapkit/protocols/link/__init__.py +++ b/pcapkit/protocols/link/__init__.py @@ -19,6 +19,11 @@ from pcapkit.protocols.link.ethernet import Ethernet from pcapkit.protocols.link.loopback import Loopback +# Linux "Cooked" Capture Encapsulation Family +from pcapkit.protocols.link.linux_sll import LinuxSLL +from pcapkit.protocols.link.linux_sll1 import LinuxSLL1 +from pcapkit.protocols.link.linux_sll2 import LinuxSLL2 + # VLAN Tag Family from pcapkit.protocols.link.vlan import VLAN from pcapkit.protocols.link.c_tag import C_Tag @@ -36,5 +41,6 @@ 'LINKTYPE', # Link Layer Protocols - 'ARP', 'C_Tag', 'Ethernet', 'InARP', 'L2TP', 'L2TPv2', 'Loopback', 'S_Tag', 'VLAN', + 'ARP', 'C_Tag', 'Ethernet', 'InARP', 'L2TP', 'L2TPv2', 'LinuxSLL', 'LinuxSLL1', + 'LinuxSLL2', 'Loopback', 'S_Tag', 'VLAN', ] diff --git a/pcapkit/protocols/link/linux_sll.py b/pcapkit/protocols/link/linux_sll.py new file mode 100644 index 0000000000..bc5209d8fc --- /dev/null +++ b/pcapkit/protocols/link/linux_sll.py @@ -0,0 +1,168 @@ +# -*- coding: utf-8 -*- +"""LINUX_SLL - Linux "Cooked" Capture Encapsulation Family +============================================================== + +.. module:: pcapkit.protocols.link.linux_sll + +:mod:`pcapkit.protocols.link.linux_sll` contains +:class:`~pcapkit.protocols.link.linux_sll.LinuxSLL` only, the base class of +:class:`~pcapkit.protocols.link.linux_sll1.LinuxSLL1` and +:class:`~pcapkit.protocols.link.linux_sll2.LinuxSLL2`. + +""" +from typing import TYPE_CHECKING, Generic + +from pcapkit.const.linux.arphrd import ARPHRD +from pcapkit.const.reg.ethertype import EtherType as Enum_EtherType +from pcapkit.corekit.decorators import beholder +from pcapkit.corekit.fields.numbers import EnumField +from pcapkit.protocols.link.link import Link +from pcapkit.protocols.protocol import _PT, _ST +from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall + +if TYPE_CHECKING: + from typing import Any, NoReturn, Optional + + from typing_extensions import Literal + + from pcapkit.protocols.protocol import ProtocolBase + +__all__ = ['LinuxSLL'] + +#: Resolves ``protocol_type`` exactly as +#: :class:`~pcapkit.protocols.schema.link.ethernet.Ethernet` resolves its ``type``. +_ETHERTYPE_FIELD = EnumField(length=2, namespace=Enum_EtherType) + +#: ``arphrd_type`` values whose ``protocol_type`` is not an EtherType, per +#: tcpdump.org's LINUX_SLL and LINUX_SLL2 pages. +_ARPHRD_NOT_ETHERTYPE = frozenset({ + ARPHRD.FRAD, + ARPHRD.IPGRE, + ARPHRD.IEEE80211_RADIOTAP, + ARPHRD.IP6GRE, + ARPHRD.NETLINK, +}) + + +def _resolve_next_layer_proto(arphrd_type: 'int', + protocol_type: 'int') -> 'Enum_EtherType | int': + """Resolve the next-layer dispatch key of either version. + + Args: + arphrd_type: ARP hardware type of the interface the packet was seen on. + protocol_type: Protocol (internet layer), as parsed off the wire. + + Returns: + ``protocol_type`` itself if ``arphrd_type`` is in + :data:`_ARPHRD_NOT_ETHERTYPE`; otherwise the value + :class:`~pcapkit.protocols.link.ethernet.Ethernet` stores for the same + ``type``. + + """ + if int(arphrd_type) in _ARPHRD_NOT_ETHERTYPE: + return protocol_type + return _ETHERTYPE_FIELD.post_process(protocol_type, {}) + + +def _validate_link_addr(alias: 'str', link_addr: 'bytes | bytearray', + link_addr_length: 'Optional[int]', *, + max_length: 'int') -> 'tuple[bytes, int]': + """Validate and pad the link-layer address for ``make()`` of either version. + + Args: + alias: Acronym of the calling protocol, for the error message. + link_addr: Link-layer address, at most 8 octets. + link_addr_length: Length in use, by default ``len(link_addr)``; may + exceed 8, e.g. 20 for InfiniBand, of which the header keeps 8. + max_length: Maximum of the wire field holding the length. + + Returns: + ``link_addr`` padded with ``NUL`` octets to 8, and the length. + + Raises: + ProtocolError: If ``link_addr`` exceeds 8 octets, or the length is + outside ``0`` to ``max_length``. + + """ + _addr = bytes(link_addr) + if len(_addr) > 8: + raise ProtocolError(f'{alias}: invalid link-layer address: {link_addr!r}, ' + f'must fit in 8 octets') + if link_addr_length is None: + link_addr_length = len(_addr) + elif not 0 <= link_addr_length <= max_length: + raise ProtocolError(f'{alias}: invalid link-layer address length: {link_addr_length!r}, ' + f'must be between 0 and {max_length}') + return _addr.ljust(8, b'\x00'), link_addr_length + + +class LinuxSLL(Link[_PT, _ST], Generic[_PT, _ST]): # pylint: disable=abstract-method + """Base class of the Linux "cooked" capture encapsulation family. + + The versions differ in field order and width, so this base holds only the + next-layer dispatch they share. See :ref:`linux-sll-labels` for the labels + that select a version. + + """ + + ########################################################################## + # Methods. + ########################################################################## + + @classmethod + def id(cls) -> 'tuple[Literal["LinuxSLL1"], Literal["LinuxSLL2"]]': + """Index ID of the protocol. + + Returns: + Index ID of the protocol: the direct members, each by its own name. + + """ + return ('LinuxSLL1', 'LinuxSLL2') + + ########################################################################## + # Data models. + ########################################################################## + + @classmethod + def __index__(cls) -> 'NoReturn': # pylint: disable=invalid-index-returned + """Numeral registry index of the protocol. + + Raises: + UnsupportedCall: The family base has no registry entry; nothing + dispatches to it, only to its members. + + """ + raise UnsupportedCall(f'{cls.__name__!r} object cannot be interpreted as an integer') + + ########################################################################## + # Utilities. + ########################################################################## + + @beholder # type: ignore[arg-type] + def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *, # type: ignore[misc] + packet: 'Optional[dict[str, Any]]' = None) -> 'ProtocolBase': + """Import next layer extractor. + + Arguments: + proto: next layer protocol index + length: valid (*non-padding*) length + packet: packet info (passed from :meth:`self.unpack + `) + + Returns: + Instance of next layer: :class:`~pcapkit.protocols.misc.raw.Raw`, + recording ``proto``, when ``arphrd_type`` is in + :data:`_ARPHRD_NOT_ETHERTYPE`. + + """ + file_ = self._get_payload() + if length is None: + length = len(file_) + arphrd_type = int(self.__header__.arphrd_type) # type: ignore[attr-defined] + if not min(length, len(file_)) or arphrd_type not in _ARPHRD_NOT_ETHERTYPE: + return super()._import_next_layer(proto, length, packet=packet) # type: ignore[misc,call-arg] + + from pcapkit.protocols.misc.raw import Raw # pylint: disable=import-outside-toplevel # isort: skip + return self._parse_next_layer(Raw, file_, length, alias=proto, packet=packet, + layer=self._exlayer, protocol=self._exproto, + __context__=self._exctx) diff --git a/pcapkit/protocols/link/linux_sll1.py b/pcapkit/protocols/link/linux_sll1.py new file mode 100644 index 0000000000..aa41ab1fb1 --- /dev/null +++ b/pcapkit/protocols/link/linux_sll1.py @@ -0,0 +1,240 @@ +# -*- coding: utf-8 -*- +# mypy: disable-error-code=dict-item +"""Linux "Cooked" Capture Encapsulation +========================================== + +.. module:: pcapkit.protocols.link.linux_sll1 + +:mod:`pcapkit.protocols.link.linux_sll1` contains +:class:`~pcapkit.protocols.link.linux_sll1.LinuxSLL1` only: the Linux "cooked" +capture encapsulation [*]_ of ``LINKTYPE_LINUX_SLL``, the header of every +``tcpdump -i any`` capture on Linux. Its structure: + +.. table:: + + ====== ===== ============================== ================================ + Octets Bits Name Description + ====== ===== ============================== ================================ + 0 0 ``linux_sll.packet_type`` Packet Type + ------ ----- ------------------------------ -------------------------------- + 2 16 ``linux_sll.arphrd_type`` ARP Hardware Type + ------ ----- ------------------------------ -------------------------------- + 4 32 ``linux_sll.link_addr_length`` Link-Layer Address Length + ------ ----- ------------------------------ -------------------------------- + 6 48 ``linux_sll.link_addr`` Link-Layer Address (padded) + ------ ----- ------------------------------ -------------------------------- + 14 112 ``linux_sll.protocol_type`` Protocol (Internet Layer) + ====== ===== ============================== ================================ + +.. [*] https://www.tcpdump.org/linktypes/LINKTYPE_LINUX_SLL.html + +""" +from typing import TYPE_CHECKING + +from pcapkit.const.arp.hardware import Hardware as Enum_Hardware +from pcapkit.const.reg.ethertype import EtherType as Enum_EtherType +from pcapkit.const.reg.linktype import LinkType as Enum_LinkType +from pcapkit.protocols.data.link.linux_sll1 import LinuxSLL1 as Data_LinuxSLL1 +from pcapkit.protocols.link.linux_sll import (LinuxSLL, _resolve_next_layer_proto, + _validate_link_addr) +from pcapkit.protocols.schema.link.linux_sll1 import LinuxSLL1 as Schema_LinuxSLL1 + +if TYPE_CHECKING: + from typing import Any, Optional + + from typing_extensions import Literal + + from pcapkit.protocols.protocol import ProtocolBase + from pcapkit.protocols.schema.schema import Schema + +__all__ = ['LinuxSLL1'] + + +class LinuxSLL1(LinuxSLL[Data_LinuxSLL1, Schema_LinuxSLL1], + schema=Schema_LinuxSLL1, data=Data_LinuxSLL1): + """Linux "cooked" capture encapsulation v1 (``LINUX_SLL``). + + ``protocol_type`` is dispatched as an EtherType, unless ``arphrd_type`` is + in :data:`~pcapkit.protocols.link.linux_sll._ARPHRD_NOT_ETHERTYPE`; in that + case, and when no dissector is registered, the payload is + :class:`~pcapkit.protocols.misc.raw.Raw`. + + """ + + ########################################################################## + # Properties. + ########################################################################## + + @property + def name(self) -> 'Literal["Linux cooked capture encapsulation"]': + """Name of current protocol.""" + return 'Linux cooked capture encapsulation' + + @property + def alias(self) -> 'Literal["LINUX_SLL1"]': + """Acronym of corresponding protocol.""" + return 'LINUX_SLL1' + + #: NOTE: ``linux_sll``, not the default ``linuxsll``; see + #: :class:`~pcapkit.protocols.link.linux_sll2.LinuxSLL2`. + @property + def info_name(self) -> 'Literal["linux_sll"]': + """Key name of the :attr:`~pcapkit.protocols.protocol.Protocol.info` dict.""" + return 'linux_sll' + + @property + def length(self) -> 'Literal[16]': + """Header length of current protocol.""" + return 16 + + @property + def protocol(self) -> 'int': # type: ignore[override] + """Protocol (internet layer) of next layer protocol.""" + return self._info.protocol_type + + ########################################################################## + # Methods. + ########################################################################## + + @classmethod + def id(cls) -> 'tuple[Literal["LinuxSLL1"]]': # type: ignore[override] + """Index ID of the protocol. + + Returns: + Index ID of the protocol, this version's own name. + + """ + return ('LinuxSLL1',) + + def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_LinuxSLL1': # pylint: disable=unused-argument + """Read Linux "cooked" capture encapsulation. + + Structure of the ``LINUX_SLL`` 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 + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | Packet Type | ARPHRD Type | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | Link-Layer Addr Length | Link-Layer Address... | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | ...Link-Layer Address (padded) | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | Protocol Type | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + + Args: + length: Length of packet data. + **kwargs: Arbitrary keyword arguments. + + Returns: + Parsed packet data. + + """ + if length is None: + length = len(self) + schema = self.__header__ + + linux_sll = Data_LinuxSLL1( + packet_type=schema.packet_type, + arphrd_type=schema.arphrd_type, + link_addr_length=schema.link_addr_length, + link_addr=schema.link_addr, + protocol_type=schema.protocol_type, + ) + + _proto = _resolve_next_layer_proto(schema.arphrd_type, schema.protocol_type) + return self._decode_next_layer(linux_sll, _proto, length - self.length) + + def make(self, + packet_type: 'int' = 0, + arphrd_type: 'Enum_Hardware | int' = Enum_Hardware.Ethernet, + link_addr: 'bytes | bytearray' = b'', + link_addr_length: 'Optional[int]' = None, + protocol_type: 'Enum_EtherType | int' = Enum_EtherType.Internet_Protocol_version_4, + payload: 'bytes | ProtocolBase | Schema' = b'', + **kwargs: 'Any') -> 'Schema_LinuxSLL1': + """Make (construct) packet data. + + Args: + packet_type: Packet type (direction). + arphrd_type: ARP hardware type of the interface the packet was + seen on. + link_addr: Link-layer address, at most 8 octets; padded with + ``NUL`` octets to 8 if shorter. + link_addr_length: Length of the address actually in use, by + default ``len(link_addr)``; may exceed 8, up to ``0xFFFF``. + protocol_type: Protocol (internet layer); an EtherType unless + ``arphrd_type`` is in the exceptions of :class:`LinuxSLL1`. + payload: Payload data. + **kwargs: Arbitrary keyword arguments. + + Returns: + Constructed packet data. + + Raises: + ProtocolError: If ``link_addr`` does not fit in 8 octets, or + ``link_addr_length`` is given outside ``0`` to ``0xFFFF``. + + """ + _addr, link_addr_length = _validate_link_addr( + self.alias, link_addr, link_addr_length, max_length=0xFFFF) + + _arphrd = self._make_index(arphrd_type, pack=False) + _proto = self._make_index(protocol_type, pack=False) + + return Schema_LinuxSLL1( + packet_type=packet_type, + arphrd_type=_arphrd, + link_addr_length=link_addr_length, + link_addr=_addr, + protocol_type=_proto, + payload=payload, + ) + + ########################################################################## + # Data models. + ########################################################################## + + def __length_hint__(self) -> 'Literal[16]': + """Return an estimated length for the object.""" + return 16 + + @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. + + .. _tcpdump: https://www.tcpdump.org/linktypes.html + + """ + return Enum_LinkType.LINUX_SLL # type: ignore[return-value] + + ########################################################################## + # Utilities. + ########################################################################## + + @classmethod + def _make_data(cls, data: 'Data_LinuxSLL1') -> '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 { + 'packet_type': data.packet_type, + 'arphrd_type': data.arphrd_type, + 'link_addr': data.link_addr, + 'link_addr_length': data.link_addr_length, + 'protocol_type': data.protocol_type, + 'payload': cls._make_payload(data), + } diff --git a/pcapkit/protocols/link/linux_sll2.py b/pcapkit/protocols/link/linux_sll2.py new file mode 100644 index 0000000000..1a604e227d --- /dev/null +++ b/pcapkit/protocols/link/linux_sll2.py @@ -0,0 +1,251 @@ +# -*- coding: utf-8 -*- +# mypy: disable-error-code=dict-item +"""Linux "Cooked" Capture Encapsulation v2 +============================================= + +.. module:: pcapkit.protocols.link.linux_sll2 + +:mod:`pcapkit.protocols.link.linux_sll2` contains +:class:`~pcapkit.protocols.link.linux_sll2.LinuxSLL2` only: the Linux "cooked" +capture encapsulation v2 [*]_ of ``LINKTYPE_LINUX_SLL2``. Its structure: + +.. table:: + + ====== ===== ================================ ================================ + Octets Bits Name Description + ====== ===== ================================ ================================ + 0 0 ``linux_sll2.protocol_type`` Protocol (Internet Layer) + ------ ----- -------------------------------- -------------------------------- + 2 16 ``linux_sll2.reserved`` Reserved + ------ ----- -------------------------------- -------------------------------- + 4 32 ``linux_sll2.interface_index`` Interface Index + ------ ----- -------------------------------- -------------------------------- + 8 64 ``linux_sll2.arphrd_type`` ARP Hardware Type + ------ ----- -------------------------------- -------------------------------- + 10 80 ``linux_sll2.packet_type`` Packet Type + ------ ----- -------------------------------- -------------------------------- + 11 88 ``linux_sll2.link_addr_length`` Link-Layer Address Length + ------ ----- -------------------------------- -------------------------------- + 12 96 ``linux_sll2.link_addr`` Link-Layer Address (padded) + ====== ===== ================================ ================================ + +.. [*] https://www.tcpdump.org/linktypes/LINKTYPE_LINUX_SLL2.html + +""" +from typing import TYPE_CHECKING + +from pcapkit.const.arp.hardware import Hardware as Enum_Hardware +from pcapkit.const.reg.ethertype import EtherType as Enum_EtherType +from pcapkit.const.reg.linktype import LinkType as Enum_LinkType +from pcapkit.protocols.data.link.linux_sll2 import LinuxSLL2 as Data_LinuxSLL2 +from pcapkit.protocols.link.linux_sll import (LinuxSLL, _resolve_next_layer_proto, + _validate_link_addr) +from pcapkit.protocols.schema.link.linux_sll2 import LinuxSLL2 as Schema_LinuxSLL2 + +if TYPE_CHECKING: + from typing import Any, Optional + + from typing_extensions import Literal + + from pcapkit.protocols.protocol import ProtocolBase + from pcapkit.protocols.schema.schema import Schema + +__all__ = ['LinuxSLL2'] + + +class LinuxSLL2(LinuxSLL[Data_LinuxSLL2, Schema_LinuxSLL2], + schema=Schema_LinuxSLL2, data=Data_LinuxSLL2): + """Linux "cooked" capture encapsulation v2 (``LINUX_SLL2``). + + Next-layer dispatch is that of :class:`~pcapkit.protocols.link.linux_sll1.LinuxSLL1`. + + """ + + ########################################################################## + # Properties. + ########################################################################## + + @property + def name(self) -> 'Literal["Linux cooked capture encapsulation v2"]': + """Name of current protocol.""" + return 'Linux cooked capture encapsulation v2' + + @property + def alias(self) -> 'Literal["LINUX_SLL2"]': + """Acronym of corresponding protocol.""" + return 'LINUX_SLL2' + + #: NOTE: see :attr:`LinuxSLL1.info_name + #: `. + @property + def info_name(self) -> 'Literal["linux_sll2"]': + """Key name of the :attr:`~pcapkit.protocols.protocol.Protocol.info` dict.""" + return 'linux_sll2' + + @property + def length(self) -> 'Literal[20]': + """Header length of current protocol.""" + return 20 + + @property + def protocol(self) -> 'int': # type: ignore[override] + """Protocol (internet layer) of next layer protocol.""" + return self._info.protocol_type + + ########################################################################## + # Methods. + ########################################################################## + + @classmethod + def id(cls) -> 'tuple[Literal["LinuxSLL2"]]': # type: ignore[override] + """Index ID of the protocol. + + Returns: + Index ID of the protocol, this version's own name. + + """ + return ('LinuxSLL2',) + + def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_LinuxSLL2': # pylint: disable=unused-argument + """Read Linux "cooked" capture encapsulation v2. + + Structure of the ``LINUX_SLL2`` 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 + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | Protocol Type | Reserved | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | Interface Index | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | ARPHRD Type |Packet Type|Link-Addr Length| + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | Link-Layer Address (padded) | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + + Args: + length: Length of packet data. + **kwargs: Arbitrary keyword arguments. + + Returns: + Parsed packet data. + + """ + if length is None: + length = len(self) + schema = self.__header__ + + linux_sll2 = Data_LinuxSLL2( + protocol_type=schema.protocol_type, + reserved=schema.reserved, + interface_index=schema.interface_index, + arphrd_type=schema.arphrd_type, + packet_type=schema.packet_type, + link_addr_length=schema.link_addr_length, + link_addr=schema.link_addr, + ) + + _proto = _resolve_next_layer_proto(schema.arphrd_type, schema.protocol_type) + return self._decode_next_layer(linux_sll2, _proto, length - self.length) + + def make(self, + protocol_type: 'Enum_EtherType | int' = Enum_EtherType.Internet_Protocol_version_4, + reserved: 'int' = 0, + interface_index: 'int' = 0, + arphrd_type: 'Enum_Hardware | int' = Enum_Hardware.Ethernet, + packet_type: 'int' = 0, + link_addr: 'bytes | bytearray' = b'', + link_addr_length: 'Optional[int]' = None, + payload: 'bytes | ProtocolBase | Schema' = b'', + **kwargs: 'Any') -> 'Schema_LinuxSLL2': + """Make (construct) packet data. + + Args: + protocol_type: Protocol (internet layer); an EtherType unless + ``arphrd_type`` is in the exceptions of :class:`LinuxSLL2`. + reserved: Reserved, must be zero. + interface_index: Index, in the capturing host's own numbering, of + the interface the packet was seen on. + arphrd_type: ARP hardware type of the interface the packet was + seen on. + packet_type: Packet type (direction). + link_addr: Link-layer address, at most 8 octets; padded with + ``NUL`` octets to 8 if shorter. + link_addr_length: Length of the address actually in use, by + default ``len(link_addr)``; may exceed 8, up to ``0xFF``. + payload: Payload data. + **kwargs: Arbitrary keyword arguments. + + Returns: + Constructed packet data. + + Raises: + ProtocolError: If ``link_addr`` does not fit in 8 octets, or + ``link_addr_length`` is given outside ``0`` to ``0xFF``. + + """ + _addr, link_addr_length = _validate_link_addr( + self.alias, link_addr, link_addr_length, max_length=0xFF) + + _arphrd = self._make_index(arphrd_type, pack=False) + _proto = self._make_index(protocol_type, pack=False) + + return Schema_LinuxSLL2( + protocol_type=_proto, + reserved=reserved, + interface_index=interface_index, + arphrd_type=_arphrd, + packet_type=packet_type, + link_addr_length=link_addr_length, + link_addr=_addr, + payload=payload, + ) + + ########################################################################## + # Data models. + ########################################################################## + + def __length_hint__(self) -> 'Literal[20]': + """Return an estimated length for the object.""" + return 20 + + @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. + + .. _tcpdump: https://www.tcpdump.org/linktypes.html + + """ + return Enum_LinkType.LINUX_SLL2 # type: ignore[return-value] + + ########################################################################## + # Utilities. + ########################################################################## + + @classmethod + def _make_data(cls, data: 'Data_LinuxSLL2') -> '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 { + 'protocol_type': data.protocol_type, + 'reserved': data.reserved, + 'interface_index': data.interface_index, + 'arphrd_type': data.arphrd_type, + 'packet_type': data.packet_type, + 'link_addr': data.link_addr, + 'link_addr_length': data.link_addr_length, + 'payload': cls._make_payload(data), + } diff --git a/pcapkit/protocols/misc/pcap/frame.py b/pcapkit/protocols/misc/pcap/frame.py index 3f429dd082..64093feb54 100644 --- a/pcapkit/protocols/misc/pcap/frame.py +++ b/pcapkit/protocols/misc/pcap/frame.py @@ -85,10 +85,14 @@ class Frame(ProtocolBase[Data_Frame, Schema_Frame], - :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.LINUX_SLL` + - :class:`pcapkit.protocols.link.linux_sll1.LinuxSLL1` * - :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.LINUX_SLL2` + - :class:`pcapkit.protocols.link.linux_sll2.LinuxSLL2` :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 @@ -108,11 +112,13 @@ 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'), + 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.LINUX_SLL: ModuleDescriptor('pcapkit.protocols.link', 'LinuxSLL1'), + Enum_LinkType.IPV4: ModuleDescriptor('pcapkit.protocols.internet', 'IPv4'), + Enum_LinkType.IPV6: ModuleDescriptor('pcapkit.protocols.internet', 'IPv6'), + Enum_LinkType.LINUX_SLL2: ModuleDescriptor('pcapkit.protocols.link', 'LinuxSLL2'), }, ) # type: DefaultDict[Enum_LinkType | int, ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]] diff --git a/pcapkit/protocols/misc/pcapng.py b/pcapkit/protocols/misc/pcapng.py index 282f5d09cf..8f24d206bb 100644 --- a/pcapkit/protocols/misc/pcapng.py +++ b/pcapkit/protocols/misc/pcapng.py @@ -484,10 +484,14 @@ class PCAPNG(ProtocolBase[Data_PCAPNG, Schema_PCAPNG], - :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.LINUX_SLL` + - :class:`pcapkit.protocols.link.linux_sll1.LinuxSLL1` * - :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.LINUX_SLL2` + - :class:`pcapkit.protocols.link.linux_sll2.LinuxSLL2` :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 @@ -767,11 +771,13 @@ 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'), + 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.LINUX_SLL: ModuleDescriptor('pcapkit.protocols.link', 'LinuxSLL1'), + Enum_LinkType.IPV4: ModuleDescriptor('pcapkit.protocols.internet', 'IPv4'), + Enum_LinkType.IPV6: ModuleDescriptor('pcapkit.protocols.internet', 'IPv6'), + Enum_LinkType.LINUX_SLL2: ModuleDescriptor('pcapkit.protocols.link', 'LinuxSLL2'), }, ) # type: DefaultDict[Enum_LinkType | int, ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]] diff --git a/pcapkit/protocols/schema/__init__.py b/pcapkit/protocols/schema/__init__.py index 7dffb2cae6..e51985a216 100644 --- a/pcapkit/protocols/schema/__init__.py +++ b/pcapkit/protocols/schema/__init__.py @@ -28,6 +28,7 @@ 'ARP', 'Ethernet', 'L2TP', + 'LinuxSLL1', 'LinuxSLL2', 'Loopback', 'VLAN', 'VLAN_TCI', diff --git a/pcapkit/protocols/schema/link/__init__.py b/pcapkit/protocols/schema/link/__init__.py index 4f8104ccd4..b9eae9ffab 100644 --- a/pcapkit/protocols/schema/link/__init__.py +++ b/pcapkit/protocols/schema/link/__init__.py @@ -4,6 +4,8 @@ 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.linux_sll1 import LinuxSLL1 +from pcapkit.protocols.schema.link.linux_sll2 import LinuxSLL2 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 @@ -12,6 +14,7 @@ 'ARP', 'Ethernet', 'L2TP', + 'LinuxSLL1', 'LinuxSLL2', 'Loopback', 'VLAN', 'VLAN_TCI', ] diff --git a/pcapkit/protocols/schema/link/linux_sll1.py b/pcapkit/protocols/schema/link/linux_sll1.py new file mode 100644 index 0000000000..10ce9dcd9d --- /dev/null +++ b/pcapkit/protocols/schema/link/linux_sll1.py @@ -0,0 +1,46 @@ +# -*- coding: utf-8 -*- +# mypy: disable-error-code=assignment +"""header schema for Linux "cooked" capture encapsulation (``LINUX_SLL``, v1)""" + +from typing import TYPE_CHECKING + +from pcapkit.const.arp.hardware import Hardware as Enum_Hardware +from pcapkit.corekit.fields.misc import PayloadField +from pcapkit.corekit.fields.numbers import EnumField, UInt16Field +from pcapkit.corekit.fields.strings import BytesField +from pcapkit.protocols.schema.schema import Schema, schema_final + +__all__ = ['LinuxSLL1'] + +if TYPE_CHECKING: + from pcapkit.protocols.protocol import ProtocolBase + + +@schema_final +class LinuxSLL1(Schema): + """Header schema for Linux "cooked" capture encapsulation (``LINUX_SLL``, v1). + + ``protocol_type`` is kept as its raw 2 octets here; whether it is an + :class:`~pcapkit.const.reg.ethertype.EtherType` depends on ``arphrd_type``, + see :class:`~pcapkit.protocols.link.linux_sll1.LinuxSLL1`. + + """ + + #: Packet type (direction). + packet_type: 'int' = UInt16Field() + #: ARP hardware type of the interface the packet was seen on. + arphrd_type: 'Enum_Hardware' = EnumField(length=2, namespace=Enum_Hardware) + #: Link-layer address length actually in use; may exceed 8. + link_addr_length: 'int' = UInt16Field() + #: Link-layer address, padded to 8 octets; only the first + #: ``min(link_addr_length, 8)`` of them are meaningful. + link_addr: 'bytes' = BytesField(length=8) + #: Protocol (internet layer). + protocol_type: 'int' = UInt16Field() + #: Payload. + payload: 'bytes' = PayloadField(length=lambda pkt: pkt['__length__']) + + if TYPE_CHECKING: + def __init__(self, packet_type: 'int', arphrd_type: 'int', link_addr_length: 'int', # pylint: disable=unused-argument,super-init-not-called + link_addr: 'bytes', protocol_type: 'int', # pylint: disable=unused-argument + payload: 'bytes | ProtocolBase | Schema') -> 'None': ... # pylint: disable=unused-argument,multiple-statements diff --git a/pcapkit/protocols/schema/link/linux_sll2.py b/pcapkit/protocols/schema/link/linux_sll2.py new file mode 100644 index 0000000000..45c094ea55 --- /dev/null +++ b/pcapkit/protocols/schema/link/linux_sll2.py @@ -0,0 +1,52 @@ +# -*- coding: utf-8 -*- +# mypy: disable-error-code=assignment +"""header schema for Linux "cooked" capture encapsulation v2 (``LINUX_SLL2``)""" + +from typing import TYPE_CHECKING + +from pcapkit.const.arp.hardware import Hardware as Enum_Hardware +from pcapkit.corekit.fields.misc import PayloadField +from pcapkit.corekit.fields.numbers import EnumField, UInt8Field, UInt16Field, UInt32Field +from pcapkit.corekit.fields.strings import BytesField +from pcapkit.protocols.schema.schema import Schema, schema_final + +__all__ = ['LinuxSLL2'] + +if TYPE_CHECKING: + from pcapkit.protocols.protocol import ProtocolBase + + +@schema_final +class LinuxSLL2(Schema): + """Header schema for Linux "cooked" capture encapsulation v2 (``LINUX_SLL2``). + + The field order and widths differ from v1 -- ``protocol_type`` moves to + the front, ``interface_index`` is new, and ``packet_type`` / + ``link_addr_length`` shrink from 2 octets to 1 -- see + :class:`~pcapkit.protocols.schema.link.linux_sll1.LinuxSLL1` for v1. + + """ + + #: Protocol (internet layer). + protocol_type: 'int' = UInt16Field() + #: Reserved, must be zero. + reserved: 'int' = UInt16Field() + #: Index, in the capturing host's own numbering, of the interface the + #: packet was seen on. + interface_index: 'int' = UInt32Field() + #: ARP hardware type of the interface the packet was seen on. + arphrd_type: 'Enum_Hardware' = EnumField(length=2, namespace=Enum_Hardware) + #: Packet type (direction). + packet_type: 'int' = UInt8Field() + #: Link-layer address length actually in use; may exceed 8. + link_addr_length: 'int' = UInt8Field() + #: Link-layer address, padded to 8 octets; only the first + #: ``min(link_addr_length, 8)`` of them are meaningful. + link_addr: 'bytes' = BytesField(length=8) + #: Payload. + payload: 'bytes' = PayloadField(length=lambda pkt: pkt['__length__']) + + if TYPE_CHECKING: + def __init__(self, protocol_type: 'int', reserved: 'int', interface_index: 'int', # pylint: disable=unused-argument,super-init-not-called + arphrd_type: 'int', packet_type: 'int', link_addr_length: 'int', # pylint: disable=unused-argument + link_addr: 'bytes', payload: 'bytes | ProtocolBase | Schema') -> 'None': ... # pylint: disable=unused-argument,multiple-statements diff --git a/tests/const/test_const_enum_builtin_parity.py b/tests/const/test_const_enum_builtin_parity.py index 8b87715511..a3940913ee 100644 --- a/tests/const/test_const_enum_builtin_parity.py +++ b/tests/const/test_const_enum_builtin_parity.py @@ -377,11 +377,11 @@ def test_the_sweep_size_is_pinned(self) -> None: # tests.const.test_const_enum_get's narrower (IntEnum, IntFlag)-only # sweep. The int count includes GitHub issue #880's ngap.ProcedureCode # and ngap.ProtocolIE, which are on EnumRegistry. - self.assertEqual(len(ints), 116) + self.assertEqual(len(ints), 117) self.assertEqual(len(flags), 6) self.assertEqual(len(strs), 10) - self.assertEqual(len(self.enums), 132) - self.assertEqual(len({obj.__module__ for obj in self.enums}), 126) + self.assertEqual(len(self.enums), 133) + self.assertEqual(len({obj.__module__ for obj in self.enums}), 127) def test_every_registry_rejects_a_negative_value(self) -> None: """The registry-wide form of GitHub issue #647. diff --git a/tests/const/test_const_enum_get.py b/tests/const/test_const_enum_get.py index 8a936a873d..836475a478 100644 --- a/tests/const/test_const_enum_get.py +++ b/tests/const/test_const_enum_get.py @@ -297,7 +297,7 @@ def test_the_sweep_size_is_pinned(self) -> None: # pcapkit.protocols.application.ngap; neither needs an exception-set # entry, since both take the base's own get(key, default). names = {_qualname(obj) for obj in self.enums} - self.assertEqual(len(self.enums), 122) + self.assertEqual(len(self.enums), 123) for expected in (EXPECTED_TO_RESOLVE_ANYTHING, EXPECTED_WITHOUT_AN_INTEGER_DEFAULT, EXPECTED_WITHOUT_A_CACHEABLE_FALLBACK): self.assertTrue(expected.issubset(names), @@ -337,7 +337,7 @@ def test_every_integer_path_consults_the_default(self) -> None: # registry excused above: ``EXPECTED_TO_RESOLVE_ANYTHING`` and # ``EXPECTED_WITHOUT_A_CACHEABLE_FALLBACK``, with # ``EXPECTED_WITHOUT_AN_INTEGER_DEFAULT`` empty. - self.assertEqual(covered, 119) + self.assertEqual(covered, 120) def test_the_always_resolving_registries_have_nothing_to_fall_back_to(self) -> None: """The two registries excused from the sweep, and why. diff --git a/tests/const/test_const_enum_lookup.py b/tests/const/test_const_enum_lookup.py index 749f95e6e1..8b4094b51f 100644 --- a/tests/const/test_const_enum_lookup.py +++ b/tests/const/test_const_enum_lookup.py @@ -200,7 +200,7 @@ def test_every_registry_enum_was_discovered(self) -> None: # ``_iter_const_int_flags``'s, and so are GitHub issue #880's # ngap.ProcedureCode and ngap.ProtocolIE, which are on EnumRegistry. names = {f'{obj.__module__}.{obj.__qualname__}' for obj in self.enums} - self.assertEqual(len(self.enums), 116) + self.assertEqual(len(self.enums), 117) self.assertTrue(EXPECTED_TO_REJECT_ZERO.issubset(names), f'expected reject-list entries missing from the sweep: ' f'{EXPECTED_TO_REJECT_ZERO - names}') diff --git a/tests/const/test_const_registry_protocol.py b/tests/const/test_const_registry_protocol.py index c8af9846a5..6ee574f08f 100644 --- a/tests/const/test_const_registry_protocol.py +++ b/tests/const/test_const_registry_protocol.py @@ -846,8 +846,8 @@ def test_every_generated_const_module_is_accounted_for(self) -> None: generated = [path for path in all_files if path not in converted_relpaths and path not in EXCLUDED_FROM_INTENUM_CENSUS] - self.assertEqual(len(all_files), 126) - self.assertEqual(len(generated), 109) + self.assertEqual(len(all_files), 127) + self.assertEqual(len(generated), 110) for relpath in generated: with self.subTest(module=relpath): diff --git a/tests/project/test_rfc_constants_one_home_1739_unit.py b/tests/project/test_rfc_constants_one_home_1739_unit.py index dbbf8353c0..8eec6094e3 100644 --- a/tests/project/test_rfc_constants_one_home_1739_unit.py +++ b/tests/project/test_rfc_constants_one_home_1739_unit.py @@ -219,6 +219,10 @@ 'the IP version number, not the Next Header offset', ('pcapkit/protocols/schema/internet/ipv6.py', 'ptr + 6'): 'the end of a Jumbo Payload option (2 octets of header, 4 of value)', + ('pcapkit/protocols/link/linux_sll1.py', 'return 16'): + 'the fixed length of the LINUX_SLL (v1) header, not the PCAP record header', + ('pcapkit/protocols/link/linux_sll2.py', 'return 20'): + 'the fixed length of the LINUX_SLL2 header, not the IPv4 or TCP minimum header', ('pcapkit/protocols/misc/pcap/frame.py', 'version == 6'): 'the IP version number, not the decimal places of a resolution', ('pcapkit/protocols/misc/pcapng.py', 'version == 6'): @@ -397,6 +401,9 @@ 'pcapkit/protocols/link/loopback.py': ((24, 28), 'the BSD loopback address families of IPv6 ' '(NetBSD and OpenBSD 24, FreeBSD 28)'), 'pcapkit/protocols/link/l2tpv2.py': ((6, 12, 16), 'L2TPv2 header flags, widths and length'), + 'pcapkit/protocols/link/linux_sll1.py': ((65535,), "the max value of LINUX_SLL's own " + '2-octet link_addr_length field, not an IP total ' + 'length'), 'pcapkit/protocols/schema/link/ethernet.py': ((6,), 'a MAC address is 6 octets'), 'pcapkit/protocols/schema/link/l2tp.py': ((6, 12), 'an L2TP header field offset, and the ' 'version bit-field'), diff --git a/tests/protocols/link/test_linux_sll1_unit.py b/tests/protocols/link/test_linux_sll1_unit.py new file mode 100644 index 0000000000..54813ffd0e --- /dev/null +++ b/tests/protocols/link/test_linux_sll1_unit.py @@ -0,0 +1,339 @@ +# -*- coding: utf-8 -*- +"""GitHub issue #2438: the Linux "cooked" capture encapsulation v1 (``LINUX_SLL``). + +Covers the 16-octet header, the ``arphrd_type``-gated next-layer rule (see +:data:`~pcapkit.protocols.link.linux_sll._ARPHRD_NOT_ETHERTYPE`), link-address +validation, and byte-exact round trips. + +Each test imports :mod:`pcapkit` itself, after +:func:`~tests._support.reimport_once_per_class`. + +""" + +import struct +import unittest + +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')) + + +def _header(packet_type: int, arphrd_type: int, link_addr: bytes, link_addr_length: int, + protocol_type: int) -> bytes: + """Pack a ``LINUX_SLL`` (v1) header: see the module docstring for the layout.""" + padded = link_addr.ljust(8, b'\x00')[:8] + return struct.pack('>HHH', packet_type, arphrd_type, link_addr_length) + padded \ + + struct.pack('>H', protocol_type) + + +def _pcap(records: 'list[bytes]') -> bytes: + from pcapkit.const.reg.linktype import LinkType + + data = struct.pack(' bytes: + """A minimal header carrying ``protocol_type`` on ``arphrd_type``.""" + return _header(0, arphrd_type, b'', 0, protocol_type) + + +class TestLinuxSLL1(unittest.TestCase): + """Pin the ``LINUX_SLL`` (v1) layer, alone and under full extraction.""" + + def setUp(self) -> 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_ethernet_arphrd_dispatches_by_ethertype(self) -> None: + from pcapkit.const.arp.hardware import Hardware + from pcapkit.protocols.link.linux_sll1 import LinuxSLL1 + + raw = _header(0, Hardware.Ethernet, bytes.fromhex('001122334455'), 6, 0x0800) + IPV4 + parsed = LinuxSLL1(raw, len(raw)) + self.assertEqual(parsed.info.packet_type, 0) + self.assertEqual(parsed.info.arphrd_type, Hardware.Ethernet) + self.assertEqual(parsed.info.link_addr_length, 6) + self.assertEqual(parsed.info.link_addr, bytes.fromhex('0011223344550000')) + self.assertEqual(parsed.info.protocol_type, 0x0800) + self.assertEqual(parsed.protocol, 0x0800) + self.assertEqual((parsed.name, parsed.length), ('Linux cooked capture encapsulation', 16)) + self.assertEqual(type(parsed.payload).__name__, 'IPv4') + self.assertEqual(parsed.protochain.chain, 'LINUX_SLL1:IPv4:UDP') + self.assertEqual(parsed.data, raw) + self._assert_rebuilds(parsed) + + def test_the_non_ethertype_arphrds_are_kept_raw(self) -> None: + """``NETLINK``, ``IPGRE``, ``IP6GRE``, ``IEEE80211_RADIOTAP`` and + ``FRAD``, per tcpdump.org's LINUX_SLL page, never have ``protocol_type`` looked up as an + EtherType, even though 0x0800 is IPv4's. + """ + from pcapkit.const.linux.arphrd import ARPHRD + from pcapkit.protocols.link.linux_sll1 import LinuxSLL1 + from pcapkit.protocols.misc.raw import Raw + + for arphrd in (ARPHRD.NETLINK, ARPHRD.IPGRE, ARPHRD.IP6GRE, + ARPHRD.IEEE80211_RADIOTAP, ARPHRD.FRAD): + with self.subTest(arphrd=arphrd): + # protocol_type 0x0800 is IPv4's EtherType, but these + # hardware types never carry an EtherType in protocol_type. + raw = _header(2, arphrd, b'', 0, 0x0800) + IPV4 + parsed = LinuxSLL1(raw, len(raw)) + self.assertEqual(parsed.info.arphrd_type, arphrd) + self.assertEqual(parsed.info.protocol_type, 0x0800) + self.assertIsInstance(parsed.payload, Raw) + self.assertIs(type(parsed.payload.info.protocol), int) + self.assertEqual(parsed.payload.info.protocol, 0x0800) + self.assertEqual(parsed.protochain.chain, 'LINUX_SLL1:Raw') + self.assertEqual(parsed.data, raw) + self._assert_rebuilds(parsed) + + def test_non_ethernet_non_excluded_arphrds_still_dispatch_by_ethertype(self) -> None: + """Loopback, PPP and SLIP/CSLIP are not + :attr:`~pcapkit.const.arp.hardware.Hardware.Ethernet`, and have no + member in :class:`~pcapkit.const.arp.hardware.Hardware` at all, but + are not among the exceptions either -- so ``protocol_type`` is + still an EtherType for them. + """ + from pcapkit.protocols.link.linux_sll1 import LinuxSLL1 + + #: Linux's own ```` numbering: loopback (772), PPP + #: (512), SLIP (256) and CSLIP (257). 256/257 coincide in *value*, + #: not meaning, with :class:`~pcapkit.const.arp.hardware.Hardware`'s + #: own ``HW_EXP2``/``AEthernet`` members. + for arphrd, name in ((772, 'loopback'), (512, 'PPP'), (256, 'SLIP'), (257, 'CSLIP')): + with self.subTest(arphrd=arphrd, name=name): + raw = _header(0, arphrd, b'', 0, 0x0800) + IPV4 + parsed = LinuxSLL1(raw, len(raw)) + self.assertEqual(parsed.info.arphrd_type, arphrd) + self.assertEqual(parsed.info.protocol_type, 0x0800) + self.assertEqual(type(parsed.payload).__name__, 'IPv4') + self.assertEqual(parsed.protochain.chain, 'LINUX_SLL1:IPv4:UDP') + self.assertEqual(parsed.data, raw) + self._assert_rebuilds(parsed) + + def test_unregistered_ethertype_is_kept_raw_without_breaking_the_header(self) -> None: + """An unregistered ``protocol_type`` on an Ethernet-semantics + ``arphrd_type`` must not blow up the header parse -- it should + resolve to :class:`~pcapkit.protocols.misc.raw.Raw`, exactly as + :class:`~pcapkit.protocols.link.ethernet.Ethernet` does for the same + unregistered EtherType. + """ + from pcapkit.const.arp.hardware import Hardware + from pcapkit.protocols.link.linux_sll1 import LinuxSLL1 + from pcapkit.protocols.misc.raw import Raw + + raw = _header(0, Hardware.Ethernet, b'', 0, 0x1234) + IPV4 + parsed = LinuxSLL1(raw, len(raw)) + self.assertEqual(parsed.info.protocol_type, 0x1234) + self.assertIsInstance(parsed.payload, Raw) + self.assertEqual(parsed.protochain.chain, 'LINUX_SLL1:') + self.assertEqual(parsed.data, raw) + self._assert_rebuilds(parsed) + + ########################################################################## + # Raw fallback and registry overrides match Ethernet's. + ########################################################################## + + @staticmethod + def _ethernet(ether_type: int): # type: ignore[no-untyped-def] + """A bare Ethernet frame carrying ``ether_type``, for comparison.""" + from pcapkit.protocols.link.ethernet import Ethernet + + raw = bytes.fromhex('ffffffffffff001122334455') + struct.pack('>H', ether_type) + IPV4 + return Ethernet(raw, len(raw)) + + def test_known_ethertype_without_a_dissector_matches_ethernet(self) -> None: + """0x88CC (LLDP) is an assigned EtherType with no dissector: the + :class:`~pcapkit.protocols.misc.raw.Raw` fallback records the + EtherType member, as a bare Ethernet frame's does, not a bare int. + """ + from pcapkit.const.arp.hardware import Hardware + from pcapkit.const.reg.ethertype import EtherType + from pcapkit.protocols.link.linux_sll1 import LinuxSLL1 + from pcapkit.protocols.misc.raw import Raw + + raw = _sll(0x88CC, Hardware.Ethernet) + IPV4 + parsed = LinuxSLL1(raw, len(raw)) + self.assertIsInstance(parsed.payload, Raw) + self.assertIs(parsed.payload.info.protocol, + EtherType.IEEE_Std_802_1AB_Link_Layer_Discovery_Protocol) + self.assertIs(self._ethernet(0x88CC).payload.info.protocol, parsed.payload.info.protocol) + self._assert_rebuilds(parsed) + + def test_unassigned_ethertype_raw_fallback_records_the_value(self) -> None: + """0x1234 is unassigned in EtherType: the + :class:`~pcapkit.protocols.misc.raw.Raw` fallback records the + ```` pseudo-member a bare Ethernet frame's does. + """ + from pcapkit.const.arp.hardware import Hardware + from pcapkit.protocols.link.linux_sll1 import LinuxSLL1 + from pcapkit.protocols.misc.raw import Raw + + raw = _sll(0x1234, Hardware.Ethernet) + IPV4 + parsed = LinuxSLL1(raw, len(raw)) + self.assertIsInstance(parsed.payload, Raw) + ethernet = self._ethernet(0x1234) + self.assertEqual(repr(parsed.payload.info.protocol), '<.: 4660>') + self.assertEqual(repr(parsed.payload.info.protocol), repr(ethernet.payload.info.protocol)) + self.assertEqual(ethernet.protochain.chain, 'Ethernet:') + self.assertEqual(parsed.protochain.chain, 'LINUX_SLL1:') + self._assert_rebuilds(parsed) + + def test_registered_override_is_reached_as_from_ethernet(self) -> None: + """``Link.register(0x1234, IPv4)`` makes an unassigned EtherType reach + IPv4 from an SLL frame, as it does from a bare Ethernet frame -- but + not on a hardware type whose ``protocol_type`` is no EtherType. + """ + import warnings + + from pcapkit.const.arp.hardware import Hardware + from pcapkit.const.linux.arphrd import ARPHRD + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.link.link import Link + from pcapkit.protocols.link.linux_sll1 import LinuxSLL1 + from pcapkit.protocols.misc.raw import Raw + + registry = Link.__dict__['__proto__'] + had_original = 0x1234 in registry + original = registry.get(0x1234) + try: + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + Link.register(0x1234, IPv4) # type: ignore[arg-type] + self.assertIsInstance(self._ethernet(0x1234).payload, IPv4) + + raw = _sll(0x1234, Hardware.Ethernet) + IPV4 + parsed = LinuxSLL1(raw, len(raw)) + self.assertIsInstance(parsed.payload, IPv4) + self.assertEqual(parsed.protochain.chain, 'LINUX_SLL1:IPv4:UDP') + self._assert_rebuilds(parsed) + + raw = _sll(0x1234, ARPHRD.NETLINK) + IPV4 + parsed = LinuxSLL1(raw, len(raw)) + self.assertIsInstance(parsed.payload, Raw) + finally: + if had_original: + registry[0x1234] = original + else: + registry.pop(0x1234, None) + + def test_padding_octets_round_trip_even_when_non_zero(self) -> None: + """The padding beyond ``link_addr_length`` is kept verbatim, not re-zeroed.""" + from pcapkit.const.arp.hardware import Hardware + from pcapkit.protocols.link.linux_sll1 import LinuxSLL1 + + raw = _header(0, Hardware.Ethernet, bytes.fromhex('0011223344'), 5, 0x0800) + IPV4 + # Overwrite the padding tail (octets 5-7 of link_addr) with non-zero garbage. + raw = bytearray(raw) + raw[11:14] = b'\xde\xad\xbe'[:3] + raw = bytes(raw) + + parsed = LinuxSLL1(raw, len(raw)) + self.assertEqual(parsed.info.link_addr_length, 5) + self.assertEqual(parsed.info.link_addr, bytes.fromhex('0011223344') + b'\xde\xad\xbe') + self.assertEqual(parsed.data, raw) + self._assert_rebuilds(parsed) + + def test_make_then_parse(self) -> None: + from pcapkit.const.arp.hardware import Hardware + from pcapkit.const.reg.ethertype import EtherType + from pcapkit.protocols.link.linux_sll1 import LinuxSLL1 + + made = LinuxSLL1(packet_type=4, arphrd_type=Hardware.Ethernet, + link_addr=bytes.fromhex('001122334455'), + protocol_type=EtherType.Internet_Protocol_version_4, payload=IPV4).data + self.assertEqual(made, _header(4, 1, bytes.fromhex('001122334455'), 6, 0x0800) + IPV4) + + parsed = LinuxSLL1(made, len(made)) + self.assertEqual(parsed.info.packet_type, 4) + self.assertEqual(type(parsed.payload).__name__, 'IPv4') + self._assert_rebuilds(parsed) + + def test_make_defaults(self) -> None: + from pcapkit.protocols.link.linux_sll1 import LinuxSLL1 + + self.assertEqual(LinuxSLL1().data, _header(0, 1, b'', 0, 0x0800)) + + def test_make_rejects_what_does_not_fit(self) -> None: + from pcapkit.protocols.link.linux_sll1 import LinuxSLL1 + from pcapkit.utilities.exceptions import ProtocolError + + with self.assertRaises(ProtocolError): + LinuxSLL1(link_addr=bytes(9)) + # link_addr_length is a 2-octet field (0 to 0xFFFF). + with self.assertRaises(ProtocolError): + LinuxSLL1(link_addr_length=0x10000) + with self.assertRaises(ProtocolError): + LinuxSLL1(link_addr_length=-1) + + def test_link_addr_length_over_8_round_trips(self) -> None: + """A ``link_addr_length`` over 8 parses and rebuilds.""" + from pcapkit.const.arp.hardware import Hardware + from pcapkit.protocols.link.linux_sll1 import LinuxSLL1 + + raw = _header(0, Hardware.Ethernet, bytes.fromhex('0011223344556677'), 20, 0x0800) + IPV4 + parsed = LinuxSLL1(raw, len(raw)) + self.assertEqual(parsed.info.link_addr_length, 20) + self.assertEqual(parsed.data, raw) + self._assert_rebuilds(parsed) + + # make() accepts the same length directly, not only via a round trip. + made = LinuxSLL1(arphrd_type=Hardware.Ethernet, + link_addr=bytes.fromhex('0011223344556677'), + link_addr_length=20, protocol_type=0x0800, payload=IPV4).data + self.assertEqual(made, raw) + + def test_registered_against_frame_and_pcapng(self) -> None: + from pcapkit.const.reg.linktype import LinkType + from pcapkit.corekit.module import ModuleDescriptor + from pcapkit.protocols.link.linux_sll1 import LinuxSLL1 + 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.assertEqual(cls.__proto__[LinkType.LINUX_SLL], + ModuleDescriptor('pcapkit.protocols.link', 'LinuxSLL1')) + self.assertIs(cls._lookup_next_layer(cls.__proto__, LinkType.LINUX_SLL), LinuxSLL1) + self.assertEqual(LinuxSLL1.__index__(), LinkType.LINUX_SLL) + + def test_full_extraction_dissects_an_any_capture(self) -> None: + import os + import tempfile + import warnings + + import pcapkit + from pcapkit.const.linux.arphrd import ARPHRD + from pcapkit.protocols.misc.pcap.frame import Frame + + raw = (_header(0, 1, bytes.fromhex('001122334455'), 6, 0x0800) + IPV4, + # loopback (772): not Ethernet, not one of the exceptions. + _header(2, 772, b'', 0, 0x0800) + IPV4, + # NETLINK (824): one of the exceptions, stays Raw. + _header(2, ARPHRD.NETLINK, b'', 0, 0x0800) + IPV4) + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, 'any.pcap') + with open(path, 'wb') as file: + file.write(_pcap(list(raw))) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + extractor = pcapkit.extract(fin=path, nofile=True, store=True) + + self.assertEqual([frame.protochain.chain for frame in extractor.frame], + ['LINUX_SLL1:IPv4:UDP', 'LINUX_SLL1:IPv4:UDP', 'LINUX_SLL1:Raw']) + for frame in extractor.frame: + self._assert_rebuilds(frame, num=frame._fnum, header=frame._ghdr) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/link/test_linux_sll2_unit.py b/tests/protocols/link/test_linux_sll2_unit.py new file mode 100644 index 0000000000..619dd93277 --- /dev/null +++ b/tests/protocols/link/test_linux_sll2_unit.py @@ -0,0 +1,327 @@ +# -*- coding: utf-8 -*- +"""GitHub issue #2438: the Linux "cooked" capture encapsulation v2 (``LINUX_SLL2``). + +Covers the 20-octet header, the same ``arphrd_type``-gated next-layer rule as +v1 (see :data:`~pcapkit.protocols.link.linux_sll._ARPHRD_NOT_ETHERTYPE`), +link-address validation, and byte-exact round trips. + +Each test imports :mod:`pcapkit` itself, after +:func:`~tests._support.reimport_once_per_class`. + +""" + +import struct +import unittest + +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')) + + +def _header(protocol_type: int, reserved: int, interface_index: int, arphrd_type: int, + packet_type: int, link_addr: bytes, link_addr_length: int) -> bytes: + """Pack a ``LINUX_SLL2`` header: see the module docstring for the layout.""" + padded = link_addr.ljust(8, b'\x00')[:8] + return struct.pack('>HHIHBB', protocol_type, reserved, interface_index, arphrd_type, + packet_type, link_addr_length) + padded + + +def _pcap(records: 'list[bytes]') -> bytes: + from pcapkit.const.reg.linktype import LinkType + + data = struct.pack(' bytes: + """A minimal header carrying ``protocol_type`` on ``arphrd_type``.""" + return _header(protocol_type, 0, 1, arphrd_type, 0, b'', 0) + + +class TestLinuxSLL2(unittest.TestCase): + """Pin the ``LINUX_SLL2`` layer, alone and under full extraction.""" + + def setUp(self) -> 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_ethernet_arphrd_dispatches_by_ethertype(self) -> None: + from pcapkit.const.arp.hardware import Hardware + from pcapkit.protocols.link.linux_sll2 import LinuxSLL2 + + raw = _header(0x0800, 0, 7, Hardware.Ethernet, 0, bytes.fromhex('001122334455'), 6) + IPV4 + parsed = LinuxSLL2(raw, len(raw)) + self.assertEqual(parsed.info.protocol_type, 0x0800) + self.assertEqual(parsed.info.reserved, 0) + self.assertEqual(parsed.info.interface_index, 7) + self.assertEqual(parsed.info.arphrd_type, Hardware.Ethernet) + self.assertEqual(parsed.info.packet_type, 0) + self.assertEqual(parsed.info.link_addr_length, 6) + self.assertEqual(parsed.info.link_addr, bytes.fromhex('0011223344550000')) + self.assertEqual(parsed.protocol, 0x0800) + self.assertEqual((parsed.name, parsed.length), ('Linux cooked capture encapsulation v2', 20)) + self.assertEqual(type(parsed.payload).__name__, 'IPv4') + self.assertEqual(parsed.protochain.chain, 'LINUX_SLL2:IPv4:UDP') + self.assertEqual(parsed.data, raw) + self._assert_rebuilds(parsed) + + def test_the_non_ethertype_arphrds_are_kept_raw(self) -> None: + """``NETLINK``, ``IPGRE``, ``IP6GRE``, ``IEEE80211_RADIOTAP`` and + ``FRAD``, per tcpdump.org's LINUX_SLL2 page, are kept raw. + """ + from pcapkit.const.linux.arphrd import ARPHRD + from pcapkit.protocols.link.linux_sll2 import LinuxSLL2 + from pcapkit.protocols.misc.raw import Raw + + for arphrd in (ARPHRD.NETLINK, ARPHRD.IPGRE, ARPHRD.IP6GRE, + ARPHRD.IEEE80211_RADIOTAP, ARPHRD.FRAD): + with self.subTest(arphrd=arphrd): + raw = _header(0x0800, 0, 0, arphrd, 2, b'', 0) + IPV4 + parsed = LinuxSLL2(raw, len(raw)) + self.assertEqual(parsed.info.arphrd_type, arphrd) + self.assertEqual(parsed.info.protocol_type, 0x0800) + self.assertIsInstance(parsed.payload, Raw) + self.assertIs(type(parsed.payload.info.protocol), int) + self.assertEqual(parsed.payload.info.protocol, 0x0800) + self.assertEqual(parsed.protochain.chain, 'LINUX_SLL2:Raw') + self.assertEqual(parsed.data, raw) + self._assert_rebuilds(parsed) + + def test_non_ethernet_non_excluded_arphrds_still_dispatch_by_ethertype(self) -> None: + """Loopback, PPP and SLIP/CSLIP are not among the exceptions, so + ``protocol_type`` is still an EtherType for them. + """ + from pcapkit.protocols.link.linux_sll2 import LinuxSLL2 + + for arphrd, name in ((772, 'loopback'), (512, 'PPP'), (256, 'SLIP'), (257, 'CSLIP')): + with self.subTest(arphrd=arphrd, name=name): + raw = _header(0x0800, 0, 0, arphrd, 0, b'', 0) + IPV4 + parsed = LinuxSLL2(raw, len(raw)) + self.assertEqual(parsed.info.arphrd_type, arphrd) + self.assertEqual(parsed.info.protocol_type, 0x0800) + self.assertEqual(type(parsed.payload).__name__, 'IPv4') + self.assertEqual(parsed.protochain.chain, 'LINUX_SLL2:IPv4:UDP') + self.assertEqual(parsed.data, raw) + self._assert_rebuilds(parsed) + + def test_unregistered_ethertype_is_kept_raw_without_breaking_the_header(self) -> None: + """An unregistered ``protocol_type`` must not blow up the header + parse, exactly as v1 (see the sibling test on ``LinuxSLL1``). + """ + from pcapkit.const.arp.hardware import Hardware + from pcapkit.protocols.link.linux_sll2 import LinuxSLL2 + from pcapkit.protocols.misc.raw import Raw + + raw = _header(0x1234, 0, 0, Hardware.Ethernet, 0, b'', 0) + IPV4 + parsed = LinuxSLL2(raw, len(raw)) + self.assertEqual(parsed.info.protocol_type, 0x1234) + self.assertIsInstance(parsed.payload, Raw) + self.assertEqual(parsed.protochain.chain, 'LINUX_SLL2:') + self.assertEqual(parsed.data, raw) + self._assert_rebuilds(parsed) + + ########################################################################## + # Raw fallback and registry overrides match Ethernet's. + ########################################################################## + + @staticmethod + def _ethernet(ether_type: int): # type: ignore[no-untyped-def] + """A bare Ethernet frame carrying ``ether_type``, for comparison.""" + from pcapkit.protocols.link.ethernet import Ethernet + + raw = bytes.fromhex('ffffffffffff001122334455') + struct.pack('>H', ether_type) + IPV4 + return Ethernet(raw, len(raw)) + + def test_known_ethertype_without_a_dissector_matches_ethernet(self) -> None: + """0x88CC (LLDP) is an assigned EtherType with no dissector: the + :class:`~pcapkit.protocols.misc.raw.Raw` fallback records the + EtherType member, as a bare Ethernet frame's does, not a bare int. + """ + from pcapkit.const.arp.hardware import Hardware + from pcapkit.const.reg.ethertype import EtherType + from pcapkit.protocols.link.linux_sll2 import LinuxSLL2 + from pcapkit.protocols.misc.raw import Raw + + raw = _sll(0x88CC, Hardware.Ethernet) + IPV4 + parsed = LinuxSLL2(raw, len(raw)) + self.assertIsInstance(parsed.payload, Raw) + self.assertIs(parsed.payload.info.protocol, + EtherType.IEEE_Std_802_1AB_Link_Layer_Discovery_Protocol) + self.assertIs(self._ethernet(0x88CC).payload.info.protocol, parsed.payload.info.protocol) + self._assert_rebuilds(parsed) + + def test_unassigned_ethertype_raw_fallback_records_the_value(self) -> None: + """0x1234 is unassigned in EtherType: the + :class:`~pcapkit.protocols.misc.raw.Raw` fallback records the + ```` pseudo-member a bare Ethernet frame's does. + """ + from pcapkit.const.arp.hardware import Hardware + from pcapkit.protocols.link.linux_sll2 import LinuxSLL2 + from pcapkit.protocols.misc.raw import Raw + + raw = _sll(0x1234, Hardware.Ethernet) + IPV4 + parsed = LinuxSLL2(raw, len(raw)) + self.assertIsInstance(parsed.payload, Raw) + ethernet = self._ethernet(0x1234) + self.assertEqual(repr(parsed.payload.info.protocol), '<.: 4660>') + self.assertEqual(repr(parsed.payload.info.protocol), repr(ethernet.payload.info.protocol)) + self.assertEqual(ethernet.protochain.chain, 'Ethernet:') + self.assertEqual(parsed.protochain.chain, 'LINUX_SLL2:') + self._assert_rebuilds(parsed) + + def test_registered_override_is_reached_as_from_ethernet(self) -> None: + """``Link.register(0x1234, IPv4)`` makes an unassigned EtherType reach + IPv4 from an SLL frame, as it does from a bare Ethernet frame -- but + not on a hardware type whose ``protocol_type`` is no EtherType. + """ + import warnings + + from pcapkit.const.arp.hardware import Hardware + from pcapkit.const.linux.arphrd import ARPHRD + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.link.link import Link + from pcapkit.protocols.link.linux_sll2 import LinuxSLL2 + from pcapkit.protocols.misc.raw import Raw + + registry = Link.__dict__['__proto__'] + had_original = 0x1234 in registry + original = registry.get(0x1234) + try: + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + Link.register(0x1234, IPv4) # type: ignore[arg-type] + self.assertIsInstance(self._ethernet(0x1234).payload, IPv4) + + raw = _sll(0x1234, Hardware.Ethernet) + IPV4 + parsed = LinuxSLL2(raw, len(raw)) + self.assertIsInstance(parsed.payload, IPv4) + self.assertEqual(parsed.protochain.chain, 'LINUX_SLL2:IPv4:UDP') + self._assert_rebuilds(parsed) + + raw = _sll(0x1234, ARPHRD.NETLINK) + IPV4 + parsed = LinuxSLL2(raw, len(raw)) + self.assertIsInstance(parsed.payload, Raw) + finally: + if had_original: + registry[0x1234] = original + else: + registry.pop(0x1234, None) + + def test_reserved_and_padding_round_trip_even_when_non_zero(self) -> None: + """``reserved`` and the padding beyond ``link_addr_length`` round-trip verbatim.""" + from pcapkit.const.arp.hardware import Hardware + from pcapkit.protocols.link.linux_sll2 import LinuxSLL2 + + raw = bytearray(_header(0x0800, 0, 0, Hardware.Ethernet, 0, + bytes.fromhex('0011223344'), 5) + IPV4) + raw[2:4] = b'\xbe\xef' # reserved, meant to be zero + raw[17:20] = b'\xde\xad\xbe' # padding tail of link_addr + raw = bytes(raw) + + parsed = LinuxSLL2(raw, len(raw)) + self.assertEqual(parsed.info.reserved, 0xbeef) + self.assertEqual(parsed.info.link_addr, bytes.fromhex('0011223344') + b'\xde\xad\xbe') + self.assertEqual(parsed.data, raw) + self._assert_rebuilds(parsed) + + def test_make_then_parse(self) -> None: + from pcapkit.const.arp.hardware import Hardware + from pcapkit.const.reg.ethertype import EtherType + from pcapkit.protocols.link.linux_sll2 import LinuxSLL2 + + made = LinuxSLL2(protocol_type=EtherType.Internet_Protocol_version_4, interface_index=3, + arphrd_type=Hardware.Ethernet, packet_type=4, + link_addr=bytes.fromhex('001122334455'), payload=IPV4).data + self.assertEqual(made, _header(0x0800, 0, 3, 1, 4, bytes.fromhex('001122334455'), 6) + IPV4) + + parsed = LinuxSLL2(made, len(made)) + self.assertEqual(parsed.info.packet_type, 4) + self.assertEqual(type(parsed.payload).__name__, 'IPv4') + self._assert_rebuilds(parsed) + + def test_make_defaults(self) -> None: + from pcapkit.protocols.link.linux_sll2 import LinuxSLL2 + + self.assertEqual(LinuxSLL2().data, _header(0x0800, 0, 0, 1, 0, b'', 0)) + + def test_make_rejects_what_does_not_fit(self) -> None: + from pcapkit.protocols.link.linux_sll2 import LinuxSLL2 + from pcapkit.utilities.exceptions import ProtocolError + + with self.assertRaises(ProtocolError): + LinuxSLL2(link_addr=bytes(9)) + # link_addr_length is a 1-octet field (0 to 0xFF). + with self.assertRaises(ProtocolError): + LinuxSLL2(link_addr_length=0x100) + with self.assertRaises(ProtocolError): + LinuxSLL2(link_addr_length=-1) + + def test_link_addr_length_over_8_round_trips(self) -> None: + """A ``link_addr_length`` over 8 parses and rebuilds.""" + from pcapkit.const.arp.hardware import Hardware + from pcapkit.protocols.link.linux_sll2 import LinuxSLL2 + + raw = _header(0x0800, 0, 0, Hardware.Ethernet, 0, + bytes.fromhex('0011223344556677'), 20) + IPV4 + parsed = LinuxSLL2(raw, len(raw)) + self.assertEqual(parsed.info.link_addr_length, 20) + self.assertEqual(parsed.data, raw) + self._assert_rebuilds(parsed) + + made = LinuxSLL2(arphrd_type=Hardware.Ethernet, + link_addr=bytes.fromhex('0011223344556677'), + link_addr_length=20, payload=IPV4).data + self.assertEqual(made, raw) + + def test_registered_against_frame_and_pcapng(self) -> None: + from pcapkit.const.reg.linktype import LinkType + from pcapkit.corekit.module import ModuleDescriptor + from pcapkit.protocols.link.linux_sll2 import LinuxSLL2 + 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.assertEqual(cls.__proto__[LinkType.LINUX_SLL2], + ModuleDescriptor('pcapkit.protocols.link', 'LinuxSLL2')) + self.assertIs(cls._lookup_next_layer(cls.__proto__, LinkType.LINUX_SLL2), LinuxSLL2) + self.assertEqual(LinuxSLL2.__index__(), LinkType.LINUX_SLL2) + + def test_full_extraction_dissects_an_any_capture(self) -> None: + import os + import tempfile + import warnings + + import pcapkit + from pcapkit.const.linux.arphrd import ARPHRD + + raw = (_header(0x0800, 0, 1, 1, 0, bytes.fromhex('001122334455'), 6) + IPV4, + # loopback (772): not Ethernet, not one of the exceptions. + _header(0x0800, 0, 1, 772, 2, b'', 0) + IPV4, + # NETLINK (824): one of the exceptions, stays Raw. + _header(0x0800, 0, 1, ARPHRD.NETLINK, 2, b'', 0) + IPV4) + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, 'any2.pcap') + with open(path, 'wb') as file: + file.write(_pcap(list(raw))) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + extractor = pcapkit.extract(fin=path, nofile=True, store=True) + + self.assertEqual([frame.protochain.chain for frame in extractor.frame], + ['LINUX_SLL2:IPv4:UDP', 'LINUX_SLL2:IPv4:UDP', 'LINUX_SLL2:Raw']) + for frame in extractor.frame: + self._assert_rebuilds(frame, num=frame._fnum, header=frame._ghdr) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/link/test_linux_sll_selector_unit.py b/tests/protocols/link/test_linux_sll_selector_unit.py new file mode 100644 index 0000000000..35a6604d74 --- /dev/null +++ b/tests/protocols/link/test_linux_sll_selector_unit.py @@ -0,0 +1,161 @@ +# -*- coding: utf-8 -*- +"""GitHub issue #2438: the ``LinuxSLL`` family selector. + +The versions share a family name, not a wire format, so they follow +:class:`~pcapkit.protocols.internet.ip.IP`: the family name selects either +version and each version's own name selects that version only. See +:ref:`linux-sll-labels`. + +""" + +import os +import struct +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')) + + +def _header_v1(protocol_type: int = 0x0800) -> 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) + + +def _header_v2(protocol_type: int = 0x0800) -> 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 + + +def _pcap(records: 'list[bytes]', linktype: int) -> bytes: + data = struct.pack(' 'str': + """Build a one-frame capture and return its ``protochain.chain`` after extraction.""" + import pcapkit + + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, 'x.pcap') + with open(path, 'wb') as file: + file.write(_pcap([record], linktype)) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + kwargs = {} if protocol is None else {'protocol': protocol} + extractor = pcapkit.extract(fin=path, nofile=True, store=True, **kwargs) + return extractor.frame[0].protochain.chain + + +class TestLinuxSLLSelector(unittest.TestCase): + """Pin :meth:`id` and :attr:`alias`, and the selectors they drive.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_id_and_alias_values(self) -> None: + from pcapkit.protocols.link.linux_sll import LinuxSLL + from pcapkit.protocols.link.linux_sll1 import LinuxSLL1 + from pcapkit.protocols.link.linux_sll2 import LinuxSLL2 + + self.assertEqual(LinuxSLL.id(), ('LinuxSLL1', 'LinuxSLL2')) + self.assertEqual(LinuxSLL1.id(), ('LinuxSLL1',)) + self.assertEqual(LinuxSLL2.id(), ('LinuxSLL2',)) + + self.assertEqual(LinuxSLL1(payload=b'').alias, 'LINUX_SLL1') + self.assertEqual(LinuxSLL2(payload=b'').alias, 'LINUX_SLL2') + + def test_family_base_has_no_registry_index(self) -> None: + from pcapkit.protocols.link.linux_sll import LinuxSLL + from pcapkit.utilities.exceptions import UnsupportedCall + + with self.assertRaises(UnsupportedCall): + LinuxSLL.__index__() + + def test_registered_in_the_name_registry(self) -> None: + from pcapkit.protocols import __proto__ as registry + from pcapkit.protocols.link.linux_sll import LinuxSLL + + self.assertIs(registry.get('LINUXSLL'), LinuxSLL) + + def test_family_name_selects_either_version(self) -> None: + from pcapkit.protocols.link.linux_sll import LinuxSLL + from pcapkit.protocols.link.linux_sll1 import LinuxSLL1 + from pcapkit.protocols.link.linux_sll2 import LinuxSLL2 + + v1 = LinuxSLL1(payload=b'') + v2 = LinuxSLL2(payload=b'') + + self.assertIn('LinuxSLL', v1) + self.assertIn('LinuxSLL', v2) + self.assertIn(LinuxSLL, v1) + self.assertIn(LinuxSLL, v2) + + def test_version_name_selects_only_that_version(self) -> None: + from pcapkit.protocols.link.linux_sll1 import LinuxSLL1 + from pcapkit.protocols.link.linux_sll2 import LinuxSLL2 + + v1 = LinuxSLL1(payload=b'') + v2 = LinuxSLL2(payload=b'') + + self.assertIn('LinuxSLL1', v1) + self.assertNotIn('LinuxSLL1', v2) + self.assertIn('LinuxSLL2', v2) + self.assertNotIn('LinuxSLL2', v1) + + def test_class_selector_is_version_narrow(self) -> None: + from pcapkit.protocols.link.linux_sll1 import LinuxSLL1 + from pcapkit.protocols.link.linux_sll2 import LinuxSLL2 + + v1 = LinuxSLL1(payload=b'') + v2 = LinuxSLL2(payload=b'') + + self.assertIn(LinuxSLL1, v1) + self.assertNotIn(LinuxSLL1, v2) + self.assertIn(LinuxSLL2, v2) + self.assertNotIn(LinuxSLL2, v1) + + def test_string_eq_is_narrower_than_in_for_the_family_name(self) -> None: + """String ``==`` compares against ``(cls, *id())`` only, so a family name is not ``==``.""" + from pcapkit.protocols.link.linux_sll1 import LinuxSLL1 + + v1 = LinuxSLL1(payload=b'') + + self.assertTrue(v1 == 'LinuxSLL1') + self.assertFalse(v1 == 'LinuxSLL') + self.assertIn('LinuxSLL', v1) + + def test_extract_family_name_stops_at_either_version(self) -> None: + from pcapkit.const.reg.linktype import LinkType + + # A stopped layer's Raw payload is named after the EtherType it would + # have dispatched on, as a stopped Ethernet frame's is. + self.assertEqual(_extract(_header_v1() + IPV4, LinkType.LINUX_SLL, 'LinuxSLL'), + 'LINUX_SLL1:Internet_Protocol_version_4') + self.assertEqual(_extract(_header_v2() + IPV4, LinkType.LINUX_SLL2, 'LinuxSLL'), + 'LINUX_SLL2:Internet_Protocol_version_4') + + def test_extract_version_name_stops_only_that_version(self) -> None: + from pcapkit.const.reg.linktype import LinkType + + self.assertEqual(_extract(_header_v1() + IPV4, LinkType.LINUX_SLL, 'LinuxSLL1'), + 'LINUX_SLL1:Internet_Protocol_version_4') + self.assertEqual(_extract(_header_v2() + IPV4, LinkType.LINUX_SLL2, 'LinuxSLL1'), + 'LINUX_SLL2:IPv4:UDP') + + self.assertEqual(_extract(_header_v2() + IPV4, LinkType.LINUX_SLL2, 'LinuxSLL2'), + 'LINUX_SLL2:Internet_Protocol_version_4') + self.assertEqual(_extract(_header_v1() + IPV4, LinkType.LINUX_SLL, 'LinuxSLL2'), + 'LINUX_SLL1:IPv4:UDP') + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/test_dispatch_registry_unit.py b/tests/protocols/test_dispatch_registry_unit.py index e39e279b04..43ca51e9f7 100644 --- a/tests/protocols/test_dispatch_registry_unit.py +++ b/tests/protocols/test_dispatch_registry_unit.py @@ -225,10 +225,12 @@ def test_cases_cover_every_table_named_in_the_issue(self) -> None: cases = self.dispatch.cases() # The tables of #496, including #904's Internet.__proto__[TransType.Shim6] # registered at IPv6_Ext rather than defaulting to Raw, plus #1574's - # LinkType.NULL and LinkType.LOOP in both link-type tables and the four - # address families of Loopback.__proto__. - self.assertEqual(len(cases), 49, - 'expected exactly 49 entries across the seven __proto__ ' + # LinkType.NULL and LinkType.LOOP in both link-type tables, the four + # address families of Loopback.__proto__, and #2438's + # Frame.__proto__[LinkType.LINUX_SLL]/[LinkType.LINUX_SLL2] in both the + # pcap and pcapng frame tables. + self.assertEqual(len(cases), 53, + 'expected exactly 53 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') @@ -238,7 +240,7 @@ def test_cases_cover_every_table_named_in_the_issue(self) -> None: by_family[case.family] = by_family.get(case.family, 0) + 1 self.assertEqual(by_family, { 'link': 7, 'internet': 17, 'tcp': 5, 'udp': 4, 'sctp': 2, - 'pcap-frame': 5, 'pcapng-frame': 5, 'loopback': 4, + 'pcap-frame': 7, 'pcapng-frame': 7, 'loopback': 4, }) def test_dispatch_reaches_target_or_is_a_recorded_degrade(self) -> None: