diff --git a/pcapkit/protocols/application/rarp.py b/pcapkit/protocols/application/rarp.py index 49dc37dfc4..2be5bcde67 100644 --- a/pcapkit/protocols/application/rarp.py +++ b/pcapkit/protocols/application/rarp.py @@ -94,7 +94,7 @@ def __index__(cls) -> 'Enum_EtherType': # pylint: disable=invalid-index-returne return Enum_EtherType.Reverse_Address_Resolution_Protocol # type: ignore[return-value] -class DRARP(RARP): +class DRARP(RARP, schema=Schema_ARP, data=Data_ARP): """This class implements Dynamic Reverse Address Resolution Protocol. Inherits RARP's bases unchanged, so ``layer == 'Application'`` here too -- diff --git a/pcapkit/protocols/link/arp.py b/pcapkit/protocols/link/arp.py index 077177c829..3be2821700 100644 --- a/pcapkit/protocols/link/arp.py +++ b/pcapkit/protocols/link/arp.py @@ -72,8 +72,8 @@ # check Python version py38 = ((version_info := sys.version_info).major >= 3 and version_info.minor >= 8) -# Ethernet address pattern -PAT_MAC_ADDR = re.compile(rb'(?i)(?:[0-9a-f]{2}[:-]){5}[0-9a-f]{2}') +# Ethernet address pattern, of any number of octets (``hlen`` is not fixed at 6) +PAT_MAC_ADDR = re.compile(rb'(?i)(?:[0-9a-f]{2}(?:[:-][0-9a-f]{2})*)?') class ARP(Link[Data_ARP, Schema_ARP], @@ -238,8 +238,8 @@ def make(self, *, ptype_default: Default value of protocol type. ptype_namespace: Namespace of protocol type. ptype_reversed: Reversed flag of protocol type. - hlen: Hardware address length; derived from the packed ``sha`` if - not given. + 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. @@ -258,7 +258,9 @@ def make(self, *, Raises: ProtocolError: If a packed hardware address is not ``hlen`` octets - long, or a packed protocol address is not ``plen`` octets long. + long (with ``hlen`` not given and ``htype`` Ethernet, not 6 + octets long), or a packed protocol address is not ``plen`` + octets long. """ _htype = self._make_index(htype, htype_default, namespace=htype_namespace, @@ -269,11 +271,15 @@ def make(self, *, reversed=oper_reversed, pack=False) _sha = self._make_addr_resolve(sha, _htype) - _spa = self._make_proto_resolve(spa, _ptype) + _spa = self._make_proto_resolve(spa, _ptype, plen) _tha = self._make_addr_resolve(tha, _htype) - _tpa = self._make_proto_resolve(tpa, _ptype) + _tpa = self._make_proto_resolve(tpa, _ptype, plen) - _hlen = len(_sha) if hlen is None else hlen + # 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)): @@ -371,16 +377,16 @@ def _read_proto_resolve(self, addr: 'bytes', ptype: 'int') -> 'str | IPv4Address ptype: Protocol type. Returns: - Protocol address. If ``ptype`` is ``0x0800`` (IPv4), an - :class:`~ipaddress.IPv4Address`; if ``0x86dd`` (IPv6), an - :class:`~ipaddress.IPv6Address`; otherwise, the address as a hex - :data:`str`. + 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 :data:`str`. """ - if ptype == Enum_EtherType.Internet_Protocol_version_4: # IPv4 - return ipaddress.ip_address(addr) - if ptype == Enum_EtherType.Internet_Protocol_version_6: # IPv6 - return ipaddress.ip_address(addr) + 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() def _make_addr_resolve(self, addr: 'str | bytes', htype: 'int') -> 'bytes': @@ -393,7 +399,8 @@ def _make_addr_resolve(self, addr: 'str | bytes', htype: 'int') -> 'bytes': Returns: Hardware address packed as raw octets. If ``htype`` is ``1``, i.e. MAC address, the ``:``- or ``-``-separated *hex* encoded MAC address - is validated and packed as 6 octets; otherwise, a :data:`str` is + is validated and packed, one octet per group (an empty string packs + as no octets); otherwise, a :data:`str` is taken as the *hex* encoded address that :meth:`_read_addr_resolve` returns, and :obj:`bytes` are returned as given. @@ -415,16 +422,19 @@ def _make_addr_resolve(self, addr: 'str | bytes', htype: 'int') -> 'bytes': raise ProtocolError(f'Invalid hardware address: {addr!r}') from None return addr - def _make_proto_resolve(self, addr: 'IPv4Address | IPv6Address | str | bytes', ptype: 'int') -> 'bytes': + 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), the address is validated and packed to 4 or 16 + ``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 :data:`str` is taken as the *hex* encoded address that :meth:`_read_proto_resolve` returns, an :class:`~ipaddress.IPv4Address`/:class:`~ipaddress.IPv6Address` is @@ -455,10 +465,10 @@ def _make_proto_resolve(self, addr: 'IPv4Address | IPv6Address | str | bytes', p :exc:`AttributeError`, bool or not. """ - if ptype == Enum_EtherType.Internet_Protocol_version_4: + 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: + 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 @@ -473,7 +483,7 @@ def _make_proto_resolve(self, addr: 'IPv4Address | IPv6Address | str | bytes', p return addr -class InARP(ARP): +class InARP(ARP, schema=Schema_ARP, data=Data_ARP): """This class implements Inverse Address Resolution Protocol.""" ########################################################################## diff --git a/tests/protocols/link/test_arp_roundtrip_unit.py b/tests/protocols/link/test_arp_roundtrip_unit.py new file mode 100644 index 0000000000..938dcdfb7d --- /dev/null +++ b/tests/protocols/link/test_arp_roundtrip_unit.py @@ -0,0 +1,174 @@ +# -*- coding: utf-8 -*- +"""ARP, InARP, RARP and DRARP rebuild every header they parse byte for byte. + +GitHub issues: + +* #1204: :class:`~pcapkit.protocols.link.arp.InARP` and + :class:`~pcapkit.protocols.application.rarp.DRARP` did not pass + ``schema=``/``data=``, so they fell back to the ``Raw`` schema and could not + parse any input. +* #1205: an Ethernet hardware address whose ``hlen`` is not 6 was read as + colon-joined hex, but :meth:`ARP._make_addr_resolve + ` accepted only a + 6-octet MAC. +* #1206: the protocol address was decoded as IPv4 or IPv6 by its length alone, + ignoring ``ptype``/``plen``, so a mismatched pair either decoded as the other + version or raised a bare :exc:`ValueError`. + +Every case builds its own octets in memory and reads no capture. + +The classes are imported inside each test, after +:func:`~tests._support.reimport_once_per_class`, not at module load (c.f. +``tests/protocols/link/test_ethernet_mac_roundtrip_unit.py``). + +""" + +import importlib +import unittest + +from tests._support import reimport_once_per_class + +#: ``(module, class)`` for every ARP-family class. +CLASSES = ( + ('pcapkit.protocols.link.arp', 'ARP'), + ('pcapkit.protocols.link.arp', 'InARP'), + ('pcapkit.protocols.application.rarp', 'RARP'), + ('pcapkit.protocols.application.rarp', 'DRARP'), +) + + +def _header(ptype: str, hlen: int, plen: int, oper: str) -> bytes: + """Ethernet-htype header with distinct, non-zero addresses.""" + return bytes.fromhex('0001' + ptype + f'{hlen:02x}{plen:02x}' + oper + + '01' * hlen + 'a1' * plen + '02' * hlen + 'b2' * plen) + + +class TestARPRoundTrip(unittest.TestCase): + """Parse, then ``from_data``, reproduces the original octets.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _cls(self, module: str, name: str) -> type: + return getattr(importlib.import_module(module), name) + + def assertRoundTrips(self, cls: type, data: bytes) -> None: + parsed = cls(data) + self.assertEqual(cls.from_data(parsed.info).data, data) + + def test_subclasses_use_arp_schema(self) -> None: + """#1204: InARP and DRARP inherit ARP's schema and data classes.""" + from pcapkit.protocols.data.link.arp import ARP as Data_ARP + from pcapkit.protocols.schema.link.arp import ARP as Schema_ARP + + for module, name in CLASSES: + with self.subTest(cls=name): + cls = self._cls(module, name) + self.assertIs(cls.__schema__, Schema_ARP) + self.assertIs(cls.__data__, Data_ARP) + + def test_subclasses_round_trip(self) -> None: + """#1204: the issue's repro parses and rebuilds for InARP and DRARP.""" + for module, name, oper in (('pcapkit.protocols.link.arp', 'InARP', '0008'), + ('pcapkit.protocols.application.rarp', 'DRARP', '0005')): + with self.subTest(cls=name): + cls = self._cls(module, name) + data = bytes.fromhex('000108000604' + oper + '001122334455' '0a000001' + '66778899aabb' '0a000002') + parsed = cls(data) + self.assertEqual(str(parsed.info.spa), '10.0.0.1') + self.assertRoundTrips(cls, data) + + def test_ethernet_hlen_other_than_six(self) -> None: + """#1205: any Ethernet ``hlen`` that parses also rebuilds.""" + for module, name in CLASSES: + for hlen in (0, 1, 8, 20): + with self.subTest(cls=name, hlen=hlen): + cls = self._cls(module, name) + self.assertRoundTrips(cls, _header('0800', hlen, 4, '0001')) + + def test_ethernet_issue_repro_round_trips(self) -> None: + """#1205: the issue's parsed hlen-8 and hlen-0 headers rebuild.""" + from pcapkit.protocols.link.arp import ARP + + for hlen, data in ( + (8, bytes.fromhex('0001080008040001' + '01' * 8 + '0a000001' + '02' * 8 + '0a000002')), + (0, bytes.fromhex('0001080000040001' '0a000001' '0a000002')), + ): + with self.subTest(hlen=hlen): + parsed = ARP(data) + self.assertEqual(parsed.info.hlen, hlen) + self.assertRoundTrips(ARP, data) + + def test_ethernet_make_without_hlen_requires_six_octets(self) -> None: + """#1205: with ``htype`` Ethernet and no ``hlen``, a MAC is 6 octets.""" + from pcapkit.utilities.exceptions import ProtocolError + + for module, name in CLASSES: + cls = self._cls(module, name) + for addr in ('00:11:22:33:44', '', '00:11:22:33:44:55:66'): + with self.subTest(cls=name, addr=addr): + with self.assertRaises(ProtocolError): + cls(htype=1, sha=addr, tha=addr, spa='10.0.0.1', tpa='10.0.0.2') + with self.subTest(cls=name, addr='six octets'): + proto = cls(htype=1, sha='00:11:22:33:44:55', tha='66:77:88:99:aa:bb', + spa='10.0.0.1', tpa='10.0.0.2') + self.assertEqual(proto.info.hlen, 6) + + def test_make_accepts_any_octets_with_explicit_hlen_or_other_htype(self) -> None: + """#1205: an explicit ``hlen``, or a non-Ethernet ``htype``, takes any count.""" + from pcapkit.const.arp.hardware import Hardware + + for module, name in CLASSES: + cls = self._cls(module, name) + for hlen, addr in ((5, '00:11:22:33:44'), (0, ''), (8, '01:01:01:01:01:01:01:01')): + with self.subTest(cls=name, hlen=hlen): + proto = cls(htype=1, hlen=hlen, sha=addr, tha=addr, + spa='10.0.0.1', tpa='10.0.0.2') + self.assertEqual(proto.info.hlen, hlen) + self.assertEqual(proto.info.sha, addr) + with self.subTest(cls=name, htype='Fibre_Channel'): + proto = cls(htype=Hardware.Fibre_Channel, sha='aabbcc', tha='ddeeff', + spa='10.0.0.1', tpa='10.0.0.2') + self.assertEqual(proto.info.hlen, 3) + + def test_ethernet_mac_still_validated(self) -> None: + """#1205: a malformed MAC string is still rejected.""" + from pcapkit.const.arp.hardware import Hardware + from pcapkit.protocols.link.arp import ARP + from pcapkit.utilities.exceptions import ProtocolError + + arp = object.__new__(ARP) + self.assertEqual(arp._make_addr_resolve('01:02:03:04:05:06:07:08', Hardware.Ethernet), + bytes.fromhex('0102030405060708')) + self.assertEqual(arp._make_addr_resolve('', Hardware.Ethernet), b'') + for addr in ('not-a-mac', '0102', '01:2', '01:02:', ':01'): + with self.subTest(addr=addr): + with self.assertRaises(ProtocolError): + arp._make_addr_resolve(addr, Hardware.Ethernet) + + def test_protocol_address_follows_ptype_and_plen(self) -> None: + """#1206: IP decoding requires ``ptype`` and ``plen`` to agree.""" + import ipaddress + + cases = ( + ('0800', 4, ipaddress.IPv4Address), + ('86dd', 16, ipaddress.IPv6Address), + ('0800', 16, str), + ('0800', 5, str), + ('0800', 0, str), + ('86dd', 4, str), + ) + for module, name in CLASSES: + for ptype, plen, kind in cases: + with self.subTest(cls=name, ptype=ptype, plen=plen): + cls = self._cls(module, name) + data = _header(ptype, 6, plen, '0001') + parsed = cls(data) + self.assertIsInstance(parsed.info.spa, kind) + self.assertIsInstance(parsed.info.tpa, kind) + self.assertRoundTrips(cls, data) + + +if __name__ == '__main__': + unittest.main()