# -*- coding: utf-8 -*-
"""ARP - Address Resolution Protocol
=======================================
.. module:: pcapkit.protocols.link.arp.protocol
:mod:`pcapkit.protocols.link.arp.protocol` contains
:class:`~pcapkit.protocols.link.arp.protocol.ARP` only,
which implements an extractor for Address Resolution
Protocol (ARP) [*]_, whose structure is described as
below:
.. table::
====== ==== ============= =======================
Octets Bits Name Description
====== ==== ============= =======================
0 0 ``arp.htype`` Hardware Type
------ ---- ------------- -----------------------
2 16 ``arp.ptype`` Protocol Type
------ ---- ------------- -----------------------
4 32 ``arp.hlen`` Hardware Address Length
------ ---- ------------- -----------------------
5 40 ``arp.plen`` Protocol Address Length
------ ---- ------------- -----------------------
6 48 ``arp.oper`` Operation
------ ---- ------------- -----------------------
8 64 ``arp.sha`` Sender Hardware Address
------ ---- ------------- -----------------------
14 112 ``arp.spa`` Sender Protocol Address
------ ---- ------------- -----------------------
18 144 ``arp.tha`` Target Hardware Address
------ ---- ------------- -----------------------
24 192 ``arp.tpa`` Target Protocol Address
====== ==== ============= =======================
.. [*] http://en.wikipedia.org/wiki/Address_Resolution_Protocol
.. versionremoved:: 1.6.0
``pcapkit.protocols.link.arp.InARP``, with no alias. The class moved to
:mod:`pcapkit.protocols.link.inarp.protocol`, since a module holds one protocol
class, named after it (:issue:`2529`).
"""
import ipaddress
from typing import TYPE_CHECKING, cast
from pcapkit.const.arp.hardware import Hardware as Enum_Hardware
from pcapkit.const.arp.operation import Operation as Enum_Operation
from pcapkit.const.reg.ethertype import EtherType as Enum_EtherType
from pcapkit.corekit.schema.ipaddress import parse_ip_address
from pcapkit.protocols.link.arp.data import ARP as Data_ARP
from pcapkit.protocols.link.arp.data import Address as Data_Address
from pcapkit.protocols.link.arp.data import Type as Data_Type
from pcapkit.protocols.link.arp.schema import ARP as Schema_ARP
from pcapkit.protocols.link.link import Link
from pcapkit.utilities.compat import cached_property
from pcapkit.utilities.exceptions import FieldValueError, ProtocolError
if TYPE_CHECKING:
from ipaddress import IPv4Address, IPv6Address
from typing import Any, Optional
from typing_extensions import Literal
from pcapkit.protocols.protocol import ProtocolBase
from pcapkit.protocols.schema import Schema
__all__ = ['ARP']
[docs]
class ARP(Link[Data_ARP, Schema_ARP],
schema=Schema_ARP, data=Data_ARP):
"""This class implements all protocols in ARP family.
- Address Resolution Protocol (:class:`~pcapkit.protocols.link.arp.protocol.ARP`) [:rfc:`826`]
- Reverse Address Resolution Protocol (:class:`~pcapkit.protocols.application.rarp.protocol.RARP`) [:rfc:`903`]
- Dynamic Reverse Address Resolution Protocol (:class:`~pcapkit.protocols.application.drarp.protocol.DRARP`)
[:rfc:`1931`]
- Inverse Address Resolution Protocol (:class:`~pcapkit.protocols.link.inarp.protocol.InARP`) [:rfc:`2390`]
.. versionchanged:: 1.5.0
:meth:`make` refuses a :obj:`bool` protocol address (:issue:`540`), packs
a non-IP protocol address from its *hex* text rather than as ASCII, and
checks every address against ``hlen`` and ``plen`` (:issue:`1139`). An
Ethernet ``hlen`` other than 6 rebuilds (:issue:`1205`), and
:meth:`read` takes an IPv4 or IPv6 protocol address only when both
``ptype`` and the address length agree (:issue:`1206`).
.. versionchanged:: 1.6.0
``htype``, ``ptype`` and ``oper`` of :meth:`make` each take a member or
an :class:`int` code, resolved with the enumeration's ``get``; a name, a
:obj:`bool`, or a code the field cannot carry raises
:exc:`~pcapkit.utilities.exceptions.ProtocolError`, rather than
resolving against an enumeration inferred from the caller's frame or
writing a different value (:issue:`2594`).
.. versionremoved:: 1.6.0
The ``_default``, ``_namespace`` and ``_reversed`` arguments that went
with each of ``htype``, ``ptype`` and ``oper`` of :meth:`make`, such as
``htype_default``; resolve a name with
:meth:`~pcapkit.protocols.protocol.Protocol.resolve_index` first
(:issue:`2594`).
"""
#: Name of corresponding protocol.
_name: 'Literal["Address Resolution Protocol", "Inverse Address Resolution Protocol", "Reverse Address Resolution Protocol", "Dynamic Reverse Address Resolution Protocol"]' # pylint: disable=line-too-long
#: Acronym of corresponding protocol.
_acnm: 'Literal["ARP", "InARP", "RARP", "DRARP"]'
##########################################################################
# Properties.
##########################################################################
@property
def name(self) -> 'Literal["Dynamic Reverse Address Resolution Protocol", "Inverse Address Resolution Protocol", "Reverse Address Resolution Protocol", "Address Resolution Protocol"]': # pylint: disable=line-too-long
"""Name of current protocol."""
return self._name
@property
def alias(self) -> 'Literal["ARP", "InARP", "RARP", "DRARP"]':
"""Acronym of corresponding protocol."""
return self._acnm
@property
def length(self) -> 'int':
"""Header length of current protocol."""
return self._info.len
[docs]
@cached_property
def src(self) -> 'Data_Address':
"""Sender hardware & protocol address."""
return Data_Address(self._info.sha, self._info.spa)
[docs]
@cached_property
def dst(self) -> 'Data_Address':
"""Target hardware & protocol address."""
return Data_Address(self._info.tha, self._info.tpa)
[docs]
@cached_property
def type(self) -> 'Data_Type':
"""Hardware & protocol type."""
return Data_Type(self._info.htype, self._info.ptype)
##########################################################################
# Methods.
##########################################################################
[docs]
@classmethod
def id(cls) -> 'tuple[Literal["ARP"]]':
"""Index ID of the protocol."""
return ('ARP',)
[docs]
def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_ARP': # pylint: disable=unused-argument
r"""Read Address Resolution Protocol.
Data structure of ARP Request header [:rfc:`826`]:
.. 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
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Hdr Type | Proto Type |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Hdr Len | Proto Len | Operation |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
\ Sender Hdr Addr \
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
\ Sender Proto Addr \
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
\ Target Hdr Addr \
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
\ Target Proto Addr \
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Args:
length: Length of packet data.
**kwargs: Arbitrary keyword arguments.
Returns:
Parsed packet data.
"""
if length is None:
length = self.__len__()
schema = self.__header__
_hwty = schema.htype
_ptty = schema.ptype
_hlen = schema.hlen
_plen = schema.plen
_oper = schema.oper
_shwa = self._read_addr_resolve(schema.sha, _hwty)
_spta = self._read_proto_resolve(schema.spa, _ptty)
_thwa = self._read_addr_resolve(schema.tha, _hwty)
_tpta = self._read_proto_resolve(schema.tpa, _ptty)
if _oper in (5, 6, 7):
self._acnm = 'DRARP'
self._name = 'Dynamic Reverse Address Resolution Protocol'
elif _oper in (8, 9):
self._acnm = 'InARP'
self._name = 'Inverse Address Resolution Protocol'
elif _oper in (3, 4):
self._acnm = 'RARP'
self._name = 'Reverse Address Resolution Protocol'
else:
self._acnm = 'ARP'
self._name = 'Address Resolution Protocol'
arp = Data_ARP(
htype=_hwty,
ptype=_ptty,
hlen=_hlen,
plen=_plen,
oper=_oper,
sha=_shwa,
spa=_spta,
tha=_thwa,
tpa=_tpta,
len=8 + _hlen * 2 + _plen * 2,
)
return self._decode_next_layer(arp, -1, length - arp.len)
[docs]
def make(self, *,
htype: 'Enum_Hardware | int' = Enum_Hardware.Ethernet,
ptype: 'Enum_EtherType | int' = Enum_EtherType.Internet_Protocol_version_4,
hlen: 'Optional[int]' = None,
plen: 'Optional[int]' = None,
oper: 'Enum_Operation | int' = Enum_Operation.REQUEST,
sha: 'str | bytes | bytearray' = '00:00:00:00:00:00',
spa: 'IPv4Address | IPv6Address | str | bytes | bytearray' = '0.0.0.0', # nosec: B104
tha: 'str | bytes | bytearray' = '00:00:00:00:00:00',
tpa: 'IPv4Address | IPv6Address | str | bytes | bytearray' = '0.0.0.0', # nosec: B104
payload: 'bytes | ProtocolBase | Schema' = b'',
**kwargs: 'Any') -> 'Schema_ARP':
"""Make (construct) packet data.
Args:
htype: Hardware type.
ptype: Protocol type.
hlen: Hardware address length; if not given, ``6`` when ``htype``
is Ethernet, otherwise derived from the packed ``sha``.
plen: Protocol address length; derived from the packed ``spa`` if
not given.
oper: Operation.
sha: Sender hardware address.
spa: Sender protocol address.
tha: Target hardware address.
tpa: Target protocol address.
payload: Payload.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed packet data.
Raises:
FieldValueError: If ``spa`` or ``tpa`` is rejected by
:meth:`_make_proto_resolve`, e.g. a :obj:`bool`.
ProtocolError: If ``htype``, ``ptype`` or ``oper`` is a name, a
:obj:`bool`, or a member or code its field cannot carry; if
``sha`` or ``tha`` is rejected by
:meth:`_make_addr_resolve`, or ``spa`` or ``tpa`` by
:meth:`_make_proto_resolve`; if a packed hardware address is
not ``hlen`` octets long (with ``hlen`` not given and
``htype`` Ethernet, not 6 octets long), or a packed protocol
address is not ``plen`` octets long.
"""
# NOTE: prefixed with the class name, for the reason
# ``_make_proto_resolve`` gives: ``alias`` is unset before ``read()``.
_htype = self._make_index(htype, Enum_Hardware, bits=16, name='hardware type',
alias=self.__class__.__name__)
_ptype = self._make_index(ptype, Enum_EtherType, bits=16, name='protocol type',
alias=self.__class__.__name__)
_oper = self._make_index(oper, Enum_Operation, bits=16, name='operation',
alias=self.__class__.__name__)
_sha = self._make_addr_resolve(sha, _htype)
_spa = self._make_proto_resolve(spa, _ptype, plen)
_tha = self._make_addr_resolve(tha, _htype)
_tpa = self._make_proto_resolve(tpa, _ptype, plen)
# an Ethernet MAC is 6 octets unless the caller says otherwise
if hlen is None:
_hlen = 6 if _htype == Enum_Hardware.Ethernet else len(_sha)
else:
_hlen = hlen
_plen = len(_spa) if plen is None else plen
for name, addr, size in (('sha', _sha, _hlen), ('tha', _tha, _hlen),
('spa', _spa, _plen), ('tpa', _tpa, _plen)):
if len(addr) != size:
raise ProtocolError(f'{self.__class__.__name__}: [{name}] address is '
f'{len(addr)} octets, but its length field is {size}')
return Schema_ARP(
htype=_htype,
ptype=_ptype,
hlen=_hlen,
plen=_plen,
oper=_oper,
sha=_sha,
spa=_spa,
tha=_tha,
tpa=_tpa,
payload=payload,
)
##########################################################################
# Data models.
##########################################################################
def __length_hint__(self) -> 'Literal[28]':
"""Return an estimated length for the object."""
return 28
[docs]
@classmethod
def __index__(cls) -> 'Enum_EtherType': # pylint: disable=invalid-index-returned
"""Numeral registry index of the protocol.
Returns:
Numeral registry index of the protocol in `IANA`_.
.. _IANA: https://www.iana.org/assignments/ieee-802-numbers/ieee-802-numbers.xhtml
"""
return Enum_EtherType.Address_Resolution_Protocol # type: ignore[return-value]
##########################################################################
# Utilities.
##########################################################################
[docs]
@classmethod
def _make_data(cls, data: 'Data_ARP') -> '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 {
'htype': data.htype,
'ptype': data.ptype,
'hlen': data.hlen,
'plen': data.plen,
'oper': data.oper,
'sha': data.sha,
'spa': data.spa,
'tha': data.tha,
'tpa': data.tpa,
'payload': cls._make_payload(data),
}
[docs]
def _read_addr_resolve(self, addr: 'bytes', htype: 'int') -> 'str':
"""Resolve hardware address according to protocol.
Arguments:
addr: Hardware address.
htype: Hardware type.
Returns:
Hardware address. If ``htype`` is ``1``, i.e. MAC address,
returns ``:`` separated *hex* encoded MAC address.
"""
if htype == Enum_Hardware.Ethernet: # Ethernet
return self._read_mac_addr(addr)
return addr.hex()
[docs]
def _make_addr_resolve(self, addr: 'str | bytes | bytearray', htype: 'int') -> 'bytes':
"""Resolve hardware address according to protocol.
Arguments:
addr: Hardware address.
htype: Hardware type.
Returns:
Hardware address packed as raw octets. If ``htype`` is ``1``, i.e.
MAC address, the ``:``- or ``-``-separated *hex* encoded MAC address
is packed by :meth:`~pcapkit.protocols.protocol.Protocol._make_mac_addr`,
one octet per group, of any number since ``hlen`` is not fixed at 6
(an empty string packs as no octets); otherwise, a :class:`str` is
taken as the *hex* encoded address that :meth:`_read_addr_resolve`
returns, and :class:`bytes`/:class:`bytearray` are returned as
:class:`bytes`.
Raises:
ProtocolError: If ``addr`` is not a :class:`str`, :class:`bytes`
or :class:`bytearray` (a :obj:`bool` included), or if
``htype`` is ``1`` and ``addr`` is not a well-formed MAC
address, or if ``htype`` is not ``1`` and ``addr`` is a
:class:`str` that is not *hex* encoded.
Notes:
The messages are prefixed with :attr:`self.__class__.__name__
<type.__name__>` rather than :attr:`self.alias
<pcapkit.protocols.protocol.Protocol.alias>`, for the reason
:meth:`_make_proto_resolve` gives.
"""
if htype == Enum_Hardware.Ethernet:
return self._make_mac_addr(addr, size=None, alias=self.__class__.__name__)
if not isinstance(addr, (str, bytes, bytearray)):
raise ProtocolError(f'{self.__class__.__name__}: invalid hardware address: must be a str, '
f'bytes or bytearray, not {addr!r}')
if isinstance(addr, str):
try:
return bytes.fromhex(addr.replace(':', '').replace('-', ''))
except ValueError:
raise ProtocolError(f'{self.__class__.__name__}: invalid hardware address: '
f'{addr!r}') from None
return bytes(addr)
[docs]
def _read_proto_resolve(self, addr: 'bytes', ptype: 'int') -> 'str | IPv4Address | IPv6Address':
"""Resolve protocol address according to protocol.
Arguments:
addr: Protocol address.
ptype: Protocol type.
Returns:
Protocol address. If ``ptype`` is ``0x0800`` (IPv4) and ``addr`` is
4 octets, an :class:`~ipaddress.IPv4Address`; if ``0x86dd`` (IPv6)
and 16 octets, an :class:`~ipaddress.IPv6Address`; otherwise, the
address as a hex :class:`str`.
"""
if ptype == Enum_EtherType.Internet_Protocol_version_4 and len(addr) == 4: # IPv4
return ipaddress.IPv4Address(addr)
if ptype == Enum_EtherType.Internet_Protocol_version_6 and len(addr) == 16: # IPv6
return ipaddress.IPv6Address(addr)
return addr.hex()
[docs]
def _make_proto_resolve(self, addr: 'IPv4Address | IPv6Address | str | bytes', ptype: 'int',
plen: 'Optional[int]' = None) -> 'bytes':
"""Resolve protocol address according to protocol.
Arguments:
addr: Protocol address.
ptype: Protocol type.
plen: Protocol address length, if given.
Returns:
Packed protocol address. If ``ptype`` is ``0x0800`` (IPv4) or
``0x86dd`` (IPv6), and ``plen`` is either not given or 4 or 16
respectively, the address is validated and packed to 4 or 16
octets; otherwise, a :class:`str` is taken as the *hex* encoded
address that :meth:`_read_proto_resolve` returns, an
:class:`~ipaddress.IPv4Address`/:class:`~ipaddress.IPv6Address` is
packed, and :class:`bytes`/:class:`bytearray` are returned as
:class:`bytes`.
Raises:
FieldValueError: If ``addr`` is a :obj:`bool`, whatever ``ptype``
is, or is packed as an IPv4 or IPv6 address and is not a
valid address or not of that version (see
:func:`~pcapkit.corekit.schema.ipaddress.parse_ip_address`).
ProtocolError: If the address is not packed as IPv4 or IPv6, and
``addr`` is either a :class:`str` that is not *hex* encoded or
none of :class:`str`, an IP address, :class:`bytes` and
:class:`bytearray`.
Notes:
Through :func:`~pcapkit.corekit.schema.ipaddress.parse_ip_address` rather than
:class:`~ipaddress.IPv4Address`/:class:`~ipaddress.IPv6Address`
directly, because :obj:`bool` is an :class:`int` subclass that
either constructor accepts without complaint: ``addr=True`` would
pack as ``00000001`` (IPv4) or ``::1`` (IPv6) with no exception and
no warning.
The messages are prefixed with :attr:`self.__class__.__name__
<type.__name__>` rather than :attr:`self.alias
<pcapkit.protocols.protocol.Protocol.alias>`, even though the
latter is this project's usual choice: :attr:`alias` here reads
``self._acnm``, which :meth:`read` only assigns once the operation
code is known, and this method runs from :meth:`make` -- before
:meth:`read` has ever executed on a construction-only instance.
Using :attr:`alias` would turn every call into an
:exc:`AttributeError`, bool or not.
"""
# NOTE: Checked before the ``ptype`` dispatch, so a ``bool`` raises ``FieldValueError``
# on the non-IP branches too, not their generic ``ProtocolError`` (#2166).
if isinstance(addr, bool):
raise FieldValueError(
f'{self.__class__.__name__}: invalid protocol address: must not be a bool, '
f'not {addr!r} -- pass int({addr!r}) if the numeric value is what is wanted')
if ptype == Enum_EtherType.Internet_Protocol_version_4 and plen in (None, 4):
return cast('IPv4Address', parse_ip_address(
addr, f'{self.__class__.__name__}: invalid protocol address', version=4)).packed
if ptype == Enum_EtherType.Internet_Protocol_version_6 and plen in (None, 16):
return cast('IPv6Address', parse_ip_address(
addr, f'{self.__class__.__name__}: invalid protocol address', version=6)).packed
if isinstance(addr, str):
try:
return bytes.fromhex(addr.replace(':', '').replace('-', ''))
except ValueError:
raise ProtocolError(f'{self.__class__.__name__}: invalid protocol address: '
f'{addr!r}') from None
if isinstance(addr, (ipaddress.IPv4Address, ipaddress.IPv6Address)):
return addr.packed
if isinstance(addr, (bytes, bytearray)):
return bytes(addr)
raise ProtocolError(f'{self.__class__.__name__}: invalid protocol address: {addr!r}')