fix(arp): non-IP protocol addresses pack as ASCII text, and hlen/plen are never checked

pcapkit/protocols/link/arp.py

@@ -215,8 +215,8 @@ def make(self, *,
              ptype_default: 'Optional[int]' = None,
              ptype_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None,  # pylint: disable=line-too-long
              ptype_reversed: 'bool' = False,
-             hlen: 'int' = 6,
-             plen: 'int' = 4,
+             hlen: 'Optional[int]' = None,
+             plen: 'Optional[int]' = None,
              oper: 'Enum_Operation | StdlibEnum | AenumEnum | str | int' = Enum_Operation.REQUEST,
              oper_default: 'Optional[int]' = None,
              oper_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None,  # pylint: disable=line-too-long
@@ -238,8 +238,10 @@ 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.
-            plen: Protocol address length.
+            hlen: Hardware address length; derived from the packed ``sha`` if
+                not given.
+            plen: Protocol address length; derived from the packed ``spa`` if
+                not given.
             oper: Operation.
             oper_default: Default value of operation.
             oper_namespace: Namespace of operation.
@@ -254,6 +256,10 @@ def make(self, *,
         Returns:
             Constructed packet data.
 
+        Raises:
+            ProtocolError: If a packed hardware address is not ``hlen`` octets
+                long, or a packed protocol address is not ``plen`` octets long.
+
         """
         _htype = self._make_index(htype, htype_default, namespace=htype_namespace,
                                   reversed=htype_reversed, pack=False)
@@ -262,16 +268,29 @@ def make(self, *,
         _oper = self._make_index(oper, oper_default, namespace=oper_namespace,
                                  reversed=oper_reversed, pack=False)
 
+        _sha = self._make_addr_resolve(sha, _htype)
+        _spa = self._make_proto_resolve(spa, _ptype)
+        _tha = self._make_addr_resolve(tha, _htype)
+        _tpa = self._make_proto_resolve(tpa, _ptype)
+
+        _hlen = len(_sha) if hlen is None else 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,
+            hlen=_hlen,
+            plen=_plen,
             oper=_oper,
-            sha=self._make_addr_resolve(sha, _htype),
-            spa=self._make_proto_resolve(spa, _ptype),
-            tha=self._make_addr_resolve(tha, _htype),
-            tpa=self._make_proto_resolve(tpa, _ptype),
+            sha=_sha,
+            spa=_spa,
+            tha=_tha,
+            tpa=_tpa,
             payload=payload,
         )
 
@@ -401,17 +420,21 @@ def _make_proto_resolve(self, addr: 'IPv4Address | IPv6Address | str | bytes', p
 
         Arguments:
             addr: Protocol address.
+            ptype: Protocol type.
 
         Returns:
             Packed protocol address. If ``ptype`` is ``0x0800`` (IPv4) or
             ``0x86dd`` (IPv6), the address is validated and packed to 4 or 16
-            octets; otherwise, a :data:`str` is encoded, an
+            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
             packed, and anything else is returned as given.
 
         Raises:
             FieldValueError: If ``addr`` is a :obj:`bool` (c.f.
                 :func:`~pcapkit.corekit.fields.ipaddress.parse_ip_address`).
+            ProtocolError: If ``ptype`` is neither IPv4 nor IPv6 and ``addr``
+                is a :data:`str` that is not *hex* encoded.
 
         Notes:
             Through :func:`parse_ip_address` rather than
@@ -440,7 +463,11 @@ def _make_proto_resolve(self, addr: 'IPv4Address | IPv6Address | str | bytes', p
                 addr, f'{self.__class__.__name__}: invalid protocol address', version=6)).packed
 
         if isinstance(addr, str):
-            return addr.encode()
+            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
         return addr

tests/protocols/link/test_arp_proto_addr_length_unit.py

@@ -0,0 +1,85 @@
+# -*- coding: utf-8 -*-
+"""ARP packs non-IP protocol addresses as octets and checks ``hlen``/``plen``.
+
+GitHub issue #1139: :meth:`ARP._make_proto_resolve
+<pcapkit.protocols.link.arp.ARP._make_proto_resolve>` encoded a :data:`str`
+protocol address as its ASCII text when ``ptype`` was neither IPv4 nor IPv6,
+while :meth:`~pcapkit.protocols.link.arp.ARP._read_proto_resolve` returns hex;
+and :meth:`~pcapkit.protocols.link.arp.ARP.make` let the schema silently
+truncate an address that disagreed with ``hlen``/``plen``.
+
+Every case builds its own octets in memory and reads no capture.
+
+"""
+
+import unittest
+
+from pcapkit.const.arp.hardware import Hardware
+from pcapkit.const.reg.ethertype import EtherType
+from pcapkit.protocols.link.arp import ARP
+from pcapkit.utilities.exceptions import ProtocolError
+
+#: A protocol type that is neither IPv4 nor IPv6.
+NON_IP = EtherType.get(0x88B5)
+
+#: ARP request: htype, ptype, hlen, plen, oper, then sha/spa/tha/tpa.
+ARP_IPV4 = bytes.fromhex('0001' '0800' '06' '04' '0001'
+                         '0123456789ab' 'c0000201' 'fedcba987654' 'c0000202')
+ARP_IPV6 = bytes.fromhex('0001' '86dd' '06' '10' '0001'
+                         '0123456789ab' '20010db8000000000000000000000001'
+                         'fedcba987654' '20010db8000000000000000000000002')
+ARP_NON_IP = bytes.fromhex('0001' '88b5' '06' '03' '0001'
+                           '0123456789ab' 'a1b2c3' 'fedcba987654' 'd4e5f6')
+
+
+class TestARPProtocolAddress(unittest.TestCase):
+    """Pin the wire form of non-IP protocol addresses."""
+
+    def test_non_ip_str_packs_as_octets(self) -> None:
+        proto = ARP(ptype=NON_IP, spa='a1b2c3', tpa='d4e5f6',
+                    sha='01:23:45:67:89:ab', tha='fe:dc:ba:98:76:54')
+        self.assertEqual(proto.data, ARP_NON_IP)
+        self.assertEqual(proto.info.spa, 'a1b2c3')
+
+    def test_non_ip_str_not_hex_raises(self) -> None:
+        with self.assertRaises(ProtocolError):
+            ARP(ptype=NON_IP, spa='raw', tpa='raw')
+
+    def test_from_data_round_trips(self) -> None:
+        for name, data in (('IPv4', ARP_IPV4), ('IPv6', ARP_IPV6), ('non-IP', ARP_NON_IP)):
+            with self.subTest(ptype=name):
+                parsed = ARP(data)
+                self.assertEqual(ARP.from_data(parsed.info).data, data)
+
+
+class TestARPLengthFields(unittest.TestCase):
+    """``hlen``/``plen`` are derived when omitted and checked when given."""
+
+    def test_lengths_derived_from_addresses(self) -> None:
+        proto = ARP(ptype=EtherType.Internet_Protocol_version_6,
+                    sha='01:23:45:67:89:ab', spa='2001:db8::1',
+                    tha='fe:dc:ba:98:76:54', tpa='2001:db8::2')
+        self.assertEqual(proto.data, ARP_IPV6)
+
+    def test_mismatched_lengths_raise(self) -> None:
+        cases = {
+            'hlen': dict(hlen=4),
+            'plen': dict(plen=2),
+            'ipv6 plen': dict(ptype=EtherType.Internet_Protocol_version_6, plen=4,
+                              spa='::1', tpa='::2'),
+            'tha': dict(htype=Hardware.Fibre_Channel, sha='aabbcc', tha='aabb'),
+            'tpa': dict(ptype=NON_IP, spa='a1b2c3', tpa='d4e5'),
+        }
+        for name, kwargs in cases.items():
+            with self.subTest(case=name):
+                with self.assertRaises(ProtocolError):
+                    ARP(**kwargs)
+
+    def test_matching_explicit_lengths_pass(self) -> None:
+        proto = ARP(hlen=6, plen=4, sha='01:23:45:67:89:ab', spa='192.0.2.1',
+                    tha='fe:dc:ba:98:76:54', tpa='192.0.2.2')
+        self.assertEqual(proto.data, ARP_IPV4)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/protocols/link/test_link_unit.py

@@ -593,7 +593,7 @@ def read_operation(oper: int) -> tuple[str, str]:
         self.assertEqual(arp._make_proto_resolve('2001:db8::1',
                                                  EtherType.Internet_Protocol_version_6),
                          ip_address('2001:db8::1').packed)
-        self.assertEqual(arp._make_proto_resolve('raw', EtherType.get(0x88B5)), b'raw')
+        self.assertEqual(arp._make_proto_resolve('c0000209', EtherType.get(0x88B5)), b'\xc0\x00\x02\x09')
         self.assertEqual(arp._make_proto_resolve(ip_address('192.0.2.9'), EtherType.get(0x88B5)),
                          ip_address('192.0.2.9').packed)
         self.assertEqual(arp._make_proto_resolve(b'bytes', EtherType.get(0x88B5)), b'bytes')