Source code for pcapkit.protocols.link.loopback.protocol

# -*- coding: utf-8 -*-
# mypy: disable-error-code=dict-item
"""BSD Loopback Encapsulation
================================

.. module:: pcapkit.protocols.link.loopback.protocol

:mod:`pcapkit.protocols.link.loopback.protocol` contains
:class:`~pcapkit.protocols.link.loopback.protocol.Loopback`
only, which implements the extractor for the BSD loopback
encapsulation [*]_ of ``LINKTYPE_NULL`` and ``LINKTYPE_LOOP``,
whose structure is described as below:

.. table::

   ====== ===== =================== ===================================
   Octets Bits  Name                Description
   ====== ===== =================== ===================================
   0          0 ``loopback.family`` Address Family (Internet Layer)
   ====== ===== =================== ===================================

.. [*] https://www.tcpdump.org/linktypes/LINKTYPE_NULL.html

"""
import collections
import sys
from typing import TYPE_CHECKING

from pcapkit.const.reg.linktype import LinkType as Enum_LinkType
from pcapkit.corekit.module import ModuleDescriptor
from pcapkit.protocols.link.link import Link
from pcapkit.protocols.link.loopback.data import Loopback as Data_Loopback
from pcapkit.protocols.link.loopback.schema import Loopback as Schema_Loopback
from pcapkit.utilities.exceptions import ProtocolError
from pcapkit.utilities.warnings import RegistryWarning, warn

if TYPE_CHECKING:
    from typing import Any, DefaultDict, Optional, Type

    from typing_extensions import Literal

    from pcapkit.protocols.protocol import ProtocolBase
    from pcapkit.protocols.schema import Schema

__all__ = ['Loopback', 'family_byteorder']


[docs] def family_byteorder(family: 'bytes', linktype: 'Optional[int]' = None) -> 'Literal["big", "little"]': """Byte order a loopback header's address family was written in. Args: family: The 4 octets of the address family. linktype: Link type the frame was captured under; :data:`None` reads as ``LINKTYPE_NULL``. Returns: ``'big'`` for ``LINKTYPE_LOOP`` (108), whose family is in network byte order. Otherwise, for ``LINKTYPE_NULL`` (0), ``'big'`` if the octets read as a smaller number big-endian than little-endian, else ``'little'``. Note: A ``LINKTYPE_NULL`` family is in the byte order of the host that captured the packet, which the capture file does not record. Every address family is below ``0x10000``, so its zero octets lie at the high end, and the smaller of the two readings is the one in the writer's byte order. A family that reads the same both ways, such as 0, packs to the same octets either way. A ``LINKTYPE_LOOP`` family is never inferred, so one written little-endian reads as a family no table holds. """ if linktype == Enum_LinkType.LOOP: return 'big' if int.from_bytes(family, 'big') < int.from_bytes(family, 'little'): return 'big' return 'little'
[docs] class Loopback(Link[Data_Loopback, Schema_Loopback], schema=Schema_Loopback, data=Data_Loopback): """This class implements BSD Loopback Encapsulation. The header is one 4-octet address family, in the byte order of the host that captured the packet for ``LINKTYPE_NULL`` and in network byte order for ``LINKTYPE_LOOP``. Which link type a frame was captured under is the ``alias`` it was dispatched with; for ``LINKTYPE_NULL``, which a frame parsed on its own is read as, the byte order is read from the octets themselves (see :func:`family_byteorder`). Either way it is kept in :attr:`~pcapkit.protocols.protocol.Protocol.info`, and :meth:`from_data <pcapkit.protocols.link.link.Link.from_data>` reads the family back in that byte order rather than inferring it again. The address families are those of the BSD socket API, whose ``AF_INET6`` differs between the BSDs, so each of its values is read as IPv6. This class supports parsing of the following protocols, which are registered in the :attr:`self.__proto__ <pcapkit.protocols.link.loopback.protocol.Loopback.__proto__>` attribute: .. list-table:: :header-rows: 1 * - Index - Protocol * - 2 - :class:`pcapkit.protocols.internet.ipv4.protocol.IPv4` * - 24 - :class:`pcapkit.protocols.internet.ipv6.protocol.IPv6` * - 28 - :class:`pcapkit.protocols.internet.ipv6.protocol.IPv6` * - 30 - :class:`pcapkit.protocols.internet.ipv6.protocol.IPv6` Any other family is kept as :class:`~pcapkit.protocols.misc.raw.protocol.Raw`. """ ########################################################################## # Defaults. ########################################################################## #: Protocol index mapping for decoding next layer, see #: :meth:`self._decode_next_layer #: <pcapkit.protocols.protocol.Protocol._decode_next_layer>` & #: :meth:`self._import_next_layer #: <pcapkit.protocols.protocol.Protocol._import_next_layer>`. Keyed by #: address family rather than by EtherType, so this class keeps a registry #: of its own instead of sharing :attr:`Link.__proto__ #: <pcapkit.protocols.link.link.Link.__proto__>`; an entry is registered with #: :func:`~pcapkit.foundation.registry.protocols.link.register_loopback`. __proto__ = collections.defaultdict( lambda: ModuleDescriptor('pcapkit.protocols.misc.raw.protocol', 'Raw'), { 2: ModuleDescriptor('pcapkit.protocols.internet.ipv4.protocol', 'IPv4'), 24: ModuleDescriptor('pcapkit.protocols.internet.ipv6.protocol', 'IPv6'), 28: ModuleDescriptor('pcapkit.protocols.internet.ipv6.protocol', 'IPv6'), 30: ModuleDescriptor('pcapkit.protocols.internet.ipv6.protocol', 'IPv6'), }, ) # type: DefaultDict[int, ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]] ########################################################################## # Properties. ########################################################################## @property def name(self) -> 'Literal["BSD Loopback Encapsulation"]': """Name of current protocol.""" return 'BSD Loopback Encapsulation' @property def length(self) -> 'Literal[4]': """Header length of current protocol.""" return 4 @property def protocol(self) -> 'int': # type: ignore[override] """Address family of next layer protocol.""" return self._info.family ########################################################################## # Methods. ##########################################################################
[docs] def read(self, length: 'Optional[int]' = None, *, alias: 'Optional[int]' = None, byteorder: 'Optional[Literal["big", "little"]]' = None, **kwargs: 'Any') -> 'Data_Loopback': # pylint: disable=unused-argument """Read BSD Loopback Encapsulation. Structure of the BSD loopback header: .. code-block:: text 0 1 2 3 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ | Address Family | +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ Args: length: Length of packet data. alias: Link type the frame was dispatched under, as :class:`~pcapkit.protocols.misc.pcap.frame.protocol.Frame` and :class:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG` pass it; ``LINKTYPE_LOOP`` reads the family in network byte order, and anything else as ``LINKTYPE_NULL``. byteorder: Byte order to read the family in, ahead of ``alias``. :meth:`make` is handed the one :attr:`~pcapkit.protocols.protocol.Protocol.info` holds by :meth:`from_data <pcapkit.protocols.link.link.Link.from_data>`, and it reaches here too, so a rebuild reads the family as it was parsed rather than inferring it again. **kwargs: Arbitrary keyword arguments. Returns: Parsed packet data. Raises: ProtocolError: If ``byteorder`` is neither ``'big'`` nor ``'little'``. """ if length is None: length = len(self) schema = self.__header__ if byteorder is None: _byte = family_byteorder(schema.family, alias) elif byteorder in ('big', 'little'): _byte = byteorder else: raise ProtocolError(f'Loopback: invalid byte order: {byteorder!r}') _family = int.from_bytes(schema.family, _byte) loopback = Data_Loopback( family=_family, byteorder=_byte, ) return self._decode_next_layer(loopback, _family, length - self.length)
[docs] def make(self, family: 'int' = 2, byteorder: 'Literal["big", "little"]' = sys.byteorder, payload: 'bytes | ProtocolBase | Schema' = b'', **kwargs: 'Any') -> 'Schema_Loopback': """Make (construct) packet data. Args: family: Address family, e.g. 2 for IPv4. byteorder: Byte order to write the address family in, by default this host's, as a capture of its own loopback interface does. payload: Payload data. **kwargs: Arbitrary keyword arguments. Returns: Constructed packet data. Raises: ProtocolError: If ``byteorder`` is neither ``'big'`` nor ``'little'``, or ``family`` is not an :obj:`int` (a :obj:`bool` included, :issue:`2289`) or does not fit in 4 octets. """ if byteorder not in ('big', 'little'): raise ProtocolError(f'Loopback: invalid byte order: {byteorder!r}') # NOTE: ``_make_uint`` refuses a ``bool``, which would otherwise pack # as address family 1 (#2289). family = self._make_uint(family, bits=32, name='address family') return Schema_Loopback( family=self._make_pack(family, size=4, lilendian=byteorder == 'little'), payload=payload, )
[docs] @classmethod def register(cls, code: 'int', protocol: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]') -> 'None': r"""Register a new protocol class for an address family. Notes: The fully qualified class name of the new protocol class should be ``{protocol.module}.{protocol.name}``. Arguments: code: BSD address family protocol: module descriptor or a :class:`~pcapkit.protocols.protocol.ProtocolBase` subclass Important: This overrides :meth:`Link.register <pcapkit.protocols.link.link.Link.register>` because :attr:`self.__proto__ <Loopback.__proto__>` is keyed by address family rather than by EtherType (:issue:`2511`). Raises: pcapkit.utilities.exceptions.RegistryError: If ``protocol`` is not a class, or not a :class:`~pcapkit.protocols.protocol.ProtocolBase` subclass. Warns: pcapkit.utilities.warnings.RegistryWarning: If this address family is already registered and the incumbent differs from the replacement; the message names both. See :meth:`Protocol.register <pcapkit.protocols.protocol.Protocol.register>` for the guard, and for the shipped descriptor a restore puts back (:issue:`1504`). """ incumbent = cls.__proto__.get(code) entry = cls._next_layer_entry(code, protocol) if entry is None: return if incumbent is not None and incumbent is not entry: warn(f'address family {code} already registered, overwriting ' f'{incumbent!r} with {entry!r}', RegistryWarning) cls.__proto__[code] = entry
########################################################################## # Data models. ########################################################################## def __length_hint__(self) -> 'Literal[4]': """Return an estimated length for the object.""" return 4
[docs] @classmethod def __index__(cls) -> 'Enum_LinkType': # pylint: disable=invalid-index-returned """Numeral registry index of the protocol. Returns: Numeral registry index of the protocol in `tcpdump`_ link-layer header types. ``LINKTYPE_LOOP`` (108) is dispatched here as well. .. _tcpdump: https://www.tcpdump.org/linktypes.html """ return Enum_LinkType.NULL # type: ignore[return-value]
########################################################################## # Utilities. ##########################################################################
[docs] @classmethod def _make_data(cls, data: 'Data_Loopback') -> 'dict[str, Any]': # type: ignore[override] """Create key-value pairs from ``data`` for protocol construction. Args: data: protocol data Returns: Key-value pairs for protocol construction. """ return { 'family': data.family, 'byteorder': data.byteorder, 'payload': cls._make_payload(data), }