diff --git a/pcapkit/protocols/link/arp.py b/pcapkit/protocols/link/arp.py index 04084f7425..077177c829 100644 --- a/pcapkit/protocols/link/arp.py +++ b/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 diff --git a/tests/protocols/link/test_arp_proto_addr_length_unit.py b/tests/protocols/link/test_arp_proto_addr_length_unit.py new file mode 100644 index 0000000000..b8971a0e89 --- /dev/null +++ b/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 +` 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() diff --git a/tests/protocols/link/test_link_unit.py b/tests/protocols/link/test_link_unit.py index b8799b2dfa..96e9fee6b7 100644 --- a/tests/protocols/link/test_link_unit.py +++ b/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')