diff --git a/pcapkit/protocols/internet/hip.py b/pcapkit/protocols/internet/hip.py index a7ef32c44c..e7c8a44083 100644 --- a/pcapkit/protocols/internet/hip.py +++ b/pcapkit/protocols/internet/hip.py @@ -431,6 +431,17 @@ class HIP(IPv6_Ext[Data_HIP, Schema_HIP], Internet[Data_HIP, Schema_HIP], :exc:`~pcapkit.utilities.exceptions.ProtocolError`, rather than resolving against an enumeration inferred from the caller's frame or writing a different value (:issue:`2594`). + * A header constructed without ``version`` is HIP version 2, as + :meth:`make` writes, rather than 6, the default IP version + :class:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext` hands on under + that name; and the ``ENCRYPTED`` maker, given an IV with a cipher + other than ``AES-128-CBC`` and ``AES-256-CBC``, raises + :exc:`~pcapkit.utilities.exceptions.ProtocolError`, rather than + counting an IV in the Length that the parameter does not carry. A + parsed ``ENCRYPTED`` parameter takes its cipher from the + ``HIP_CIPHER`` parameters before it, so an AES cipher's IV is read + as the IV, rather than warning of a missing ``HIP_CIPHER`` and + reading the IV as encrypted data (:issue:`2708`). .. versionremoved:: 1.6.0 @@ -794,30 +805,44 @@ def id(cls) -> 'tuple[Literal["HIP"]]': # type: ignore[override] @overload def __post_init__(self, file: 'IO[bytes] | bytes', length: 'Optional[int]' = ..., *, # pylint: disable=arguments-differ - extension: 'bool' = ..., **kwargs: 'Any') -> 'None': ... + version: 'int' = ..., extension: 'bool' = ..., **kwargs: 'Any') -> 'None': ... @overload def __post_init__(self, **kwargs: 'Any') -> 'None': ... # pylint: disable=arguments-differ def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None, length: 'Optional[int]' = None, *, # pylint: disable=arguments-differ - extension: 'bool' = False, **kwargs: 'Any') -> 'None': + version: 'int' = 2, extension: 'bool' = False, **kwargs: 'Any') -> 'None': """Post initialisation hook. Args: file: Source packet stream. length: Length of packet data. + version: When constructing, the HIP version :meth:`make` writes, + with :meth:`make`'s default, the :rfc:`7401` version; when + parsing, the version of the enclosing IP layer, which is not + read. extension: If the protocol is used as an IPv6 extension header. **kwargs: Arbitrary keyword arguments. See Also: For construction arguments, please refer to :meth:`self.make `. + Note: + ``version`` is declared here because :meth:`IPv6_Ext.__post_init__ + ` + defaults it to ``6``, the IP version, and hands it on to + :meth:`make`, which writes it as the HIP version. Left to that + default, a header constructed without one would be HIP version 6, + which no RFC defines, rather than the version 2 :meth:`make` + called directly writes (:issue:`2708`). + """ #: bool: If the protocol is used as an IPv6 extension header. self._extf = extension # call super __post_init__ - super().__post_init__(file, length, extension=extension, **kwargs) # type: ignore[arg-type] + super().__post_init__(file, length, version=version, # type: ignore[call-overload] + extension=extension, **kwargs) def __length_hint__(self) -> 'Literal[40]': """Return an estimated length for the object.""" @@ -4187,7 +4212,8 @@ def _make_param_encrypted(self, code: 'Enum_Parameter', param: 'Optional[Data_En param: parameter data version: HIP protocol version cipher: cipher ID, of the width ``HIP_CIPHER`` carries it in - iv: initialization vector (optional depending on cipher ID) + iv: initialization vector, given for the AES ciphers and only + for them data: encrypted data reserved: the four reserved octets, truncated or zero-filled to 4 octets @@ -4198,8 +4224,9 @@ def _make_param_encrypted(self, code: 'Enum_Parameter', param: 'Optional[Data_En Raises: ProtocolError: If ``cipher`` is a name or does not fit its field, - or the cipher is ``AES-128-CBC`` or ``AES-256-CBC`` and ``iv`` - is missing or **NOT** 16 octets. + the cipher is ``AES-128-CBC`` or ``AES-256-CBC`` and ``iv`` + is missing or **NOT** 16 octets, or ``iv`` is given with any + other cipher, which takes none. """ if param is not None: @@ -4216,6 +4243,10 @@ def _make_param_encrypted(self, code: 'Enum_Parameter', param: 'Optional[Data_En raise ProtocolError(f'HIPv{version}: [ParamNo {code}] IV is required for AES cipher') if len(iv) != 16: raise ProtocolError(f'HIPv{version}: [ParamNo {code}] IV length must be 16 bytes for AES cipher') + elif iv is not None: + # NOTE: The schema writes an IV for the AES ciphers alone, so one + # given here would count in Length and be dropped (#2708). + raise ProtocolError(f'HIPv{version}: [ParamNo {code}] IV is not used by cipher {cipher_id!r}') schema = Schema_EncryptedParameter( reserved=reserved, @@ -4229,7 +4260,7 @@ def _make_param_encrypted(self, code: 'Enum_Parameter', param: 'Optional[Data_En # parameter does. Passed as a constructor keyword it draws an # ``UnknownFieldWarning`` and is dropped, leaving ``pre_unpack`` to fall # back to its sibling lookup, which a standalone ``make`` call gives no - # ``options`` to search; the parameter would then be treated as + # ``HIP_CIPHER`` parameter to search; the parameter would then be treated as # cipher-less and silently packed without its IV. Setting the # already-resolved ``cipher_id`` as a plain attribute instead reaches # ``pack()``'s packet context via ``packet.update(self.__dict__)``, diff --git a/pcapkit/protocols/internet/mh.py b/pcapkit/protocols/internet/mh.py index 7b7760df08..bfa7920c11 100644 --- a/pcapkit/protocols/internet/mh.py +++ b/pcapkit/protocols/internet/mh.py @@ -8636,14 +8636,17 @@ def _make_opt_cga_pr(self, type: 'Enum_Option', option: 'Optional[Data_CGAParame ) def _make_opt_cga_param(self, type: 'Enum_Option', option: 'Optional[Data_CGAParametersOption]' = None, *, - parameters: 'Optional[list[Schema_CGAParameter | Data_CGAParameter | dict[str, Any] | bytes]]' = None, + parameters: 'Optional[list[Schema_CGAParameter | Data_CGAParameter | bytes]]' = None, **kwargs: 'Any') -> 'Schema_CGAParametersOption': """Make MH CGA parameters option. Args: type: Option type. option: Option data model. - parameters: CGA parameters. + parameters: CGA parameters, each as octets, a schema or a data + model. A :obj:`dict` is refused: a mapping stands for a + maker's keyword arguments, and no maker takes a CGA + parameter's fields, so build the data model from them. **kwargs: Arbitrary keyword arguments. Returns: @@ -8651,7 +8654,8 @@ def _make_opt_cga_param(self, type: 'Enum_Option', option: 'Optional[Data_CGAPar Raises: ProtocolError: If the option is too long for its one-octet - ``Option Length``. + ``Option Length``, or an item of ``parameters`` is not octets, + a schema or a data model, such as a :obj:`dict`. """ if option is not None: @@ -8686,7 +8690,8 @@ def _make_opt_cga_param(self, type: 'Enum_Option', option: 'Optional[Data_CGAPar length += 25 + len(data.public_key) + ext_len param.append(schema) else: - raise ProtocolError(f'{self.alias}: [OptNo {type}] unknown CGA parameter format: {data}') + raise ProtocolError(f'{self.alias}: [OptNo {type}] unknown CGA parameter format: must be bytes, ' + f'a CGA parameter schema or data model, not {data!r}') return Schema_CGAParametersOption( type=type, diff --git a/pcapkit/protocols/schema/internet/hip.py b/pcapkit/protocols/schema/internet/hip.py index 9038472775..282ee1feb2 100644 --- a/pcapkit/protocols/schema/internet/hip.py +++ b/pcapkit/protocols/schema/internet/hip.py @@ -2,7 +2,7 @@ # mypy: disable-error-code=assignment """header schema for Host Identity Protocol""" -from typing import TYPE_CHECKING, cast +from typing import TYPE_CHECKING from pcapkit.const.hip.certificate import Certificate as Enum_Certificate from pcapkit.const.hip.cipher import Cipher as Enum_Cipher @@ -64,8 +64,6 @@ from typing import Any, Optional from pcapkit.corekit.fields.field import FieldBase as Field - from pcapkit.protocols.data.internet.hip import EncryptedParameter as Data_EncryptedParameter - from pcapkit.protocols.data.internet.hip import HIPCipherParameter as Data_HIPCipherParameter from pcapkit.protocols.protocol import ProtocolBase if SPHINX_TYPE_CHECKING: # pragma: no cover @@ -845,6 +843,28 @@ class HIPCipherParameter(Parameter, code=Enum_Parameter.HIP_CIPHER): #: Padding. padding: 'bytes' = PaddingField(length=parameter_padding_len) + def post_process(self, packet: 'dict[str, Any]') -> 'Schema': + """Revise ``schema`` data after unpacking process. + + Args: + packet: Unpacked data. + + Returns: + Revised schema. + + Note: + The cipher IDs are added to the ``__hip_ciphers__`` key of + ``packet``, the one context every parameter of the area is + unpacked with, for an ``ENCRYPTED`` parameter after this one to + read its cipher from (see :meth:`EncryptedParameter.pre_unpack`). + The parameters already parsed reach neither that context nor the + reader's ``options`` before the whole area is read + (:issue:`2708`). + + """ + packet['__hip_ciphers__'] = (*packet.get('__hip_ciphers__', ()), *self.ciphers) + return self + if TYPE_CHECKING: def __init__(self, type: 'Enum_Parameter', len: 'int', ciphers: 'list[Enum_Cipher]') -> 'None': ... @@ -910,40 +930,34 @@ def pre_unpack(cls, packet: 'dict[str, Any]') -> 'None': `, which sets it as a plain attribute so ``pack()``'s own ``packet.update(self.__dict__)`` carries it in here -- that value - is trusted over the ``HIP_CIPHER`` sibling lookup below, which a - parameter packed on its own has no ``options`` list for. + is trusted over the ``HIP_CIPHER`` lookup below. + + Otherwise the cipher comes from the cipher IDs that the + ``HIP_CIPHER`` parameters before this one in the area carry, in + order, under ``__hip_ciphers__`` (see + :meth:`HIPCipherParameter.post_process`): the first ``ENCRYPTED`` + parameter takes the first cipher ID, the next one the next, as + ``__hip_encrypted__`` counts them (:issue:`2708`). """ + index = packet.get('__hip_encrypted__', 0) + packet['__hip_encrypted__'] = index + 1 + if 'cipher' in packet: packet['__cipher__'] = packet.pop('cipher') return - if 'options' in packet: - cipher_list = cast('list[Data_HIPCipherParameter]', - packet['options'].getlist(Enum_Parameter.HIP_CIPHER)) - if not cipher_list: - warn(f'HIP: [ParamNo {Enum_Parameter.ENCRYPTED}] ' - 'missing HIP_CIPHER parameter', ProtocolWarning) - cipher_id = Enum_Cipher(0xffff) - else: - cipher_ids = [] # type: list[Enum_Cipher] - for cipher in cipher_list: - cipher_ids.extend(cipher.cipher_id) - - encrypted_list = cast('list[Data_EncryptedParameter]', - packet['options'].getlist(Enum_Parameter.ENCRYPTED)) - encrypted_index = len(encrypted_list) - - if encrypted_index >= len(cipher_ids): - warn(f'HIP: [ParamNo {Enum_Parameter.ENCRYPTED}] ' - 'too many ENCRYPTED parameters', ProtocolWarning) - cipher_id = Enum_Cipher(0xfffe) - else: - cipher_id = cipher_ids[encrypted_index] - else: + cipher_ids = packet.get('__hip_ciphers__', ()) # type: tuple[Enum_Cipher, ...] + if not cipher_ids: warn(f'HIP: [ParamNo {Enum_Parameter.ENCRYPTED}] ' 'missing HIP_CIPHER parameter', ProtocolWarning) cipher_id = Enum_Cipher(0xffff) + elif index >= len(cipher_ids): + warn(f'HIP: [ParamNo {Enum_Parameter.ENCRYPTED}] ' + 'too many ENCRYPTED parameters', ProtocolWarning) + cipher_id = Enum_Cipher(0xfffe) + else: + cipher_id = cipher_ids[index] packet['__cipher__'] = cipher_id diff --git a/pcapkit/protocols/transport/tcp.py b/pcapkit/protocols/transport/tcp.py index ce2fa22c20..cdb8ab8cb9 100644 --- a/pcapkit/protocols/transport/tcp.py +++ b/pcapkit/protocols/transport/tcp.py @@ -360,6 +360,14 @@ class TCP(Transport[Data_TCP, Schema_TCP], resolving against an enumeration inferred from the caller's frame, writing a different value or raising another exception (:issue:`2594`). + * An option given to :meth:`make` as raw octets, of a kind no member of + :class:`~pcapkit.const.tcp.option.Option` holds, such as 255, which + parses as an unassigned option, builds as its ``(code, arguments)`` + pair does, rather than raising + :exc:`~pcapkit.utilities.exceptions.EnumValueError`; and empty raw + octets, which hold no kind, raise + :exc:`~pcapkit.utilities.exceptions.ProtocolError` rather than an + :exc:`IndexError` (:issue:`2708`). .. versionremoved:: 1.6.0 @@ -952,7 +960,8 @@ def _make_tcp_options(self, Raises: ProtocolError: If an option's kind is a name, a :obj:`bool`, or a - code the 8-bit kind octet cannot carry (:issue:`2594`), or if + code the 8-bit kind octet cannot carry (:issue:`2594`), if an + option given as raw octets is empty (:issue:`2708`), or if an option maker rejects its option, e.g. a user timeout too large for 15 bits of minutes, an MPTCP ``subtype`` or an alternate checksum ``algorithm`` given by name, or an MPTCP @@ -976,7 +985,12 @@ def _make_tcp_options(self, if isinstance(options, list): for schema in options: if isinstance(schema, bytes): - code = Enum_Option.get(schema[0]) + if not schema: + raise ProtocolError(f'{self.alias}: invalid option: {schema!r} holds no kind octet') + # NOTE: Resolved as the kind of a ``(code, arguments)`` pair + # is, so a kind no ``Option`` member holds, such as 255, + # comes back as its code rather than raising (#2708). + code = self._make_index(schema[0], Enum_Option, bits=8, name='option kind') data = schema # type: Schema_Option | bytes data_len = len(data) elif isinstance(schema, Schema): diff --git a/tests/protocols/internet/test_hip_make_codes_2594_unit.py b/tests/protocols/internet/test_hip_make_codes_2594_unit.py index 91a2261060..5f85b90f22 100644 --- a/tests/protocols/internet/test_hip_make_codes_2594_unit.py +++ b/tests/protocols/internet/test_hip_make_codes_2594_unit.py @@ -150,6 +150,12 @@ '_make_param_encrypted.cipher': {'iv': None}, } +#: Other arguments a row needs to build without its own: the default cipher, +#: ``NULL_ENCRYPT``, refuses an IV (#2708). +BARE_EXTRA = { + '_make_param_encrypted.cipher': {'iv': None}, +} + #: ``maker.argument`` to the attribute its parameter's data model keeps the #: parsed values in, for the rows whose enumeration is narrower than the field. PARSED = { @@ -277,13 +283,19 @@ def test_a_member_name_is_refused(self) -> None: def test_the_member_and_its_code_build_the_same_octets(self) -> None: """#2594: the member and ``int(member)`` resolve with ``get`` to one field value.""" + rows = {f'{maker}.{argument}': (argument, code, extra, listed) + for maker, argument, _, _, _, _, code, extra, listed in ROWS} for label, member, _, _, _, build in self._rows(): with self.subTest(row=label): expected = build(member) self.assertEqual(build(int(member)).hex(), expected.hex()) # the member is not what the maker builds without it, so a # value that never reached the packet cannot pass - self.assertNotEqual(build().hex(), expected.hex()) + bare = build + if label in BARE_EXTRA: + argument, code, extra, listed = rows[label] + _, bare = self._builder(argument, code, {**extra, **BARE_EXTRA[label]}, listed) + self.assertNotEqual(bare().hex(), expected.hex()) def test_the_widest_code_builds_parses_back_and_rebuilds(self) -> None: """#2594: the field width decides what is refused, so ``(1 << bits) - 1`` builds.""" diff --git a/tests/protocols/internet/test_hip_unit.py b/tests/protocols/internet/test_hip_unit.py index 3c47685b9c..591a9f02e2 100644 --- a/tests/protocols/internet/test_hip_unit.py +++ b/tests/protocols/internet/test_hip_unit.py @@ -1534,7 +1534,7 @@ def test_hip_parameter_constructors_cover_data_model_and_default_paths(self) -> hip_data.EncryptedParameter(type=Parameter.ENCRYPTED, critical=False, reserved=bytes(4), length=8, cipher=Cipher.NULL_ENCRYPT, - iv=b'iv', data=b'data'), + iv=None, data=b'data'), version=2, ).data, b'data') @@ -2427,7 +2427,6 @@ def test_hip_schema_selectors_and_encrypted_parameter_branches(self) -> None: from pcapkit.const.hip.cipher import Cipher from pcapkit.const.hip.hi_algorithm import HIAlgorithm from pcapkit.const.hip.parameter import Parameter - from pcapkit.corekit.multidict import OrderedMultiDict from pcapkit.protocols.schema.internet import hip as hip_schema from pcapkit.utilities.exceptions import FieldValueError @@ -2478,33 +2477,28 @@ def test_hip_schema_selectors_and_encrypted_parameter_branches(self) -> None: self.assertEqual(missing_packet['__cipher__'], Cipher.get(0xffff)) warn.assert_called_once() - empty_options = OrderedMultiDict() - no_cipher_packet = {'options': empty_options} + # #2708: the cipher comes from the ``HIP_CIPHER`` parameters unpacked + # before it into the same context, as the parameters of one area are. + no_cipher_packet: dict[str, object] = {} + hip_schema.HIPCipherParameter.unpack(b'\x02\x43\x00\x00' + bytes(4), packet=no_cipher_packet) with mock.patch('pcapkit.protocols.schema.internet.hip.warn') as warn: hip_schema.EncryptedParameter.pre_unpack(no_cipher_packet) self.assertEqual(no_cipher_packet['__cipher__'], Cipher.get(0xffff)) warn.assert_called_once() - cipher_options = OrderedMultiDict() - cipher_options.add(Parameter.HIP_CIPHER, types.SimpleNamespace( - cipher_id=(Cipher.AES_128_CBC, Cipher.NULL_ENCRYPT), - )) - cipher_packet = {'options': cipher_options} + cipher_packet: dict[str, object] = {} + hip_schema.HIPCipherParameter.unpack(b'\x02\x43\x00\x04\x00\x02\x00\x01', packet=cipher_packet) with mock.patch('pcapkit.protocols.schema.internet.hip.warn') as warn: hip_schema.EncryptedParameter.pre_unpack(cipher_packet) self.assertIs(cipher_packet['__cipher__'], Cipher.AES_128_CBC) warn.assert_not_called() - cipher_options.add(Parameter.ENCRYPTED, object()) - next_cipher_packet = {'options': cipher_options} - hip_schema.EncryptedParameter.pre_unpack(next_cipher_packet) - self.assertIs(next_cipher_packet['__cipher__'], Cipher.NULL_ENCRYPT) + hip_schema.EncryptedParameter.pre_unpack(cipher_packet) + self.assertIs(cipher_packet['__cipher__'], Cipher.NULL_ENCRYPT) - cipher_options.add(Parameter.ENCRYPTED, object()) - too_many_packet = {'options': cipher_options} with mock.patch('pcapkit.protocols.schema.internet.hip.warn') as warn: - hip_schema.EncryptedParameter.pre_unpack(too_many_packet) - self.assertEqual(too_many_packet['__cipher__'], Cipher.get(0xfffe)) + hip_schema.EncryptedParameter.pre_unpack(cipher_packet) + self.assertEqual(cipher_packet['__cipher__'], Cipher.get(0xfffe)) warn.assert_called_once() encrypted = hip_schema.EncryptedParameter( diff --git a/tests/protocols/internet/test_hip_version_encrypted_iv_2708_unit.py b/tests/protocols/internet/test_hip_version_encrypted_iv_2708_unit.py new file mode 100644 index 0000000000..aca1d2b17a --- /dev/null +++ b/tests/protocols/internet/test_hip_version_encrypted_iv_2708_unit.py @@ -0,0 +1,249 @@ +# -*- coding: utf-8 -*- +"""HIP builds version 2 by default, and ``ENCRYPTED`` carries an IV with an AES cipher alone. + +GitHub issue #2708, three defects of :class:`~pcapkit.protocols.internet.hip.HIP`: + +* **Version.** :class:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext`'s + ``__post_init__`` defaults ``version`` to 6, the IP version, and hands it on + to :meth:`HIP.make `, which writes it + as the HIP version. HIP's own ``__post_init__`` takes ``version`` with + ``make``'s default, so ``HIP(next=59, parameters=[])`` writes version 2, as + ``make`` called directly does. Version 6 is no HIP version: + :rfc:`7401#section-5.1` gives 2, and :rfc:`5201` 1. While parsing, the keyword + is the enclosing IP layer's version and the octets decide. +* **IV.** ``ENCRYPTED`` writes an IV for ``AES-128-CBC`` and ``AES-256-CBC`` + alone (:rfc:`7401#section-5.2.18`). An IV given with any other cipher would + count in Length while the schema drops it, and the packet would not parse + back, so the maker refuses it, from a data model as from keywords. +* **Cipher on parse.** ``ENCRYPTED``'s schema takes its cipher from the + ``HIP_CIPHER`` parameters before it in the area, the Nth ``ENCRYPTED`` the + Nth cipher ID. Each parameter of the area is unpacked with one shared + context, and ``HIP_CIPHER`` records its cipher IDs there. The parsed + parameters themselves never reach that context, so a lookup through them + would find none, warn of a missing ``HIP_CIPHER`` and read an AES IV as + encrypted data. A parameter with no ``HIP_CIPHER`` before it still warns. + +The ciphers are read off :class:`~pcapkit.const.hip.cipher.Cipher`, with one +code it does not assign. + +Every case builds its own octets in memory and reads no capture. :class:`HIP` +is imported inside each test, after +:func:`~tests._support.reimport_once_per_class`. + +""" +# Deferring these is the design: each build runs after the per-class re-import. +# pylint: disable=import-outside-toplevel +from __future__ import annotations + +import inspect +import unittest +import warnings + +from tests._support import reimport_once_per_class + +#: A cipher code :class:`~pcapkit.const.hip.cipher.Cipher` assigns no member. +UNASSIGNED_CIPHER = 0x1234 + + +class TestHIPDefaultVersion(unittest.TestCase): + """A HIP header constructed without ``version`` is the one ``make`` writes.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_constructor_writes_make_default(self) -> None: + from pcapkit.protocols.internet.hip import HIP + + default = inspect.signature(HIP.make).parameters['version'].default + self.assertEqual(default, 2) + + for kwargs in ({'next': 59}, {'next': 59, 'parameters': []}): + with self.subTest(**kwargs): + proto = HIP(**kwargs) + self.assertEqual(proto.info.version, default) + # Version 2, three reserved zero bits, then the fixed bit 1. + self.assertEqual(proto.data[3], 0x21) + self.assertEqual(proto.data, HIP(version=default, **kwargs).data) + self.assertEqual(proto.data, proto.make(**kwargs).pack()) + + def test_explicit_version_is_written(self) -> None: + from pcapkit.protocols.internet.hip import HIP + + for version in (1, 2): + with self.subTest(version=version): + proto = HIP(next=59, version=version) + self.assertEqual(proto.info.version, version) + self.assertEqual(proto.data[3], version << 4 | 0x01) + + def test_parse_reads_version_from_octets(self) -> None: + from pcapkit.protocols.internet.hip import HIP + + for version in (1, 2): + data = HIP(next=59, version=version).data + for ip_version in (4, 6): + with self.subTest(version=version, ip_version=ip_version): + parsed = HIP(data, version=ip_version) + self.assertEqual(parsed.info.version, version) + self.assertEqual(HIP.from_data(parsed.info).data, data) + + +class TestHIPEncryptedIV(unittest.TestCase): + """``ENCRYPTED`` takes an IV with the AES ciphers and with no other.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + @staticmethod + def _build(cipher: 'object', iv: 'bytes | None') -> 'bytes': + from pcapkit.const.hip.parameter import Parameter + from pcapkit.protocols.internet.hip import HIP + + return HIP(next=59, parameters=[ + (Parameter.HIP_CIPHER, {'ciphers': [cipher]}), + (Parameter.ENCRYPTED, {'cipher': cipher, 'iv': iv, 'data': b'abcd'}), + ]).data + + @staticmethod + def _ciphers() -> 'tuple[list[object], list[object]]': + from pcapkit.const.hip.cipher import Cipher + + aes = [Cipher.AES_128_CBC, Cipher.AES_256_CBC] + others = [member for member in Cipher if member not in aes] # type: list[object] + others.append(UNASSIGNED_CIPHER) + return list(aes), others + + def test_other_cipher_refuses_an_iv(self) -> None: + from pcapkit.utilities.exceptions import ProtocolError + + _, others = self._ciphers() + self.assertTrue(others) + for cipher in others: + for iv in (b'', bytes(16)): + with self.subTest(cipher=cipher, iv=iv): + with self.assertRaisesRegex(ProtocolError, 'IV is not used by cipher'): + self._build(cipher, iv) + + def test_data_model_with_other_cipher_refuses_an_iv(self) -> None: + from pcapkit.const.hip.parameter import Parameter + from pcapkit.protocols.data.internet.hip import EncryptedParameter + from pcapkit.protocols.internet.hip import HIP + from pcapkit.utilities.exceptions import ProtocolError + + _, others = self._ciphers() + for cipher in others: + for iv in (None, bytes(16)): + with self.subTest(cipher=cipher, iv=iv): + param = EncryptedParameter(type=Parameter.ENCRYPTED, critical=True, length=8, + reserved=bytes(4), cipher=cipher, iv=iv, data=b'abcd') + if iv is None: + schema = HIP()._make_param_encrypted(Parameter.ENCRYPTED, param, version=2) + self.assertEqual(bytes(schema), b'\x02\x81\x00\x08' + bytes(4) + b'abcd' + bytes(4)) + else: + with self.assertRaisesRegex(ProtocolError, 'IV is not used by cipher'): + HIP()._make_param_encrypted(Parameter.ENCRYPTED, param, version=2) + + def test_other_cipher_without_iv_round_trips(self) -> None: + from pcapkit.protocols.internet.hip import HIP + + _, others = self._ciphers() + for cipher in others: + with self.subTest(cipher=cipher): + data = self._build(cipher, None) + # ENCRYPTED, after the HIP_CIPHER parameter: Length 8 is the + # four reserved octets and the data, with no IV. + self.assertEqual(data[48:52], b'\x02\x81\x00\x08') + self.assertEqual(data[56:60], b'abcd') + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + parsed = HIP(data) + self.assertEqual(HIP.from_data(parsed.info).data, data) + + def test_aes_cipher_writes_its_iv(self) -> None: + aes, _ = self._ciphers() + for cipher in aes: + with self.subTest(cipher=cipher): + data = self._build(cipher, bytes(range(16))) + # Length 24: the reserved octets, the 16-octet IV and the data. + self.assertEqual(data[48:52], b'\x02\x81\x00\x18') + self.assertEqual(data[56:72], bytes(range(16))) + self.assertEqual(data[72:76], b'abcd') + + +class TestHIPEncryptedCipherOnParse(unittest.TestCase): + """A parsed ``ENCRYPTED`` takes its cipher from the ``HIP_CIPHER`` before it.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _parse(self, data: 'bytes') -> 'tuple[object, list[str]]': + """Parse ``data`` as HIP, returning the layer and the warnings it drew.""" + from pcapkit.protocols.internet.hip import HIP + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + parsed = HIP(data) + return parsed, [str(warning.message) for warning in caught] + + def test_each_cipher_reads_its_own_layout(self) -> None: + from pcapkit.const.hip.cipher import Cipher + from pcapkit.const.hip.parameter import Parameter + from pcapkit.protocols.internet.hip import HIP + + aes, others = TestHIPEncryptedIV._ciphers() # pylint: disable=protected-access + for cipher in [*aes, *others]: + iv = bytes(range(16)) if cipher in aes else None + with self.subTest(cipher=cipher): + data = TestHIPEncryptedIV._build(cipher, iv) # pylint: disable=protected-access + parsed, caught = self._parse(data) + self.assertEqual(caught, []) + (encrypted,) = parsed.info.parameters.getlist(Parameter.ENCRYPTED) + self.assertEqual(int(encrypted.cipher), int(cipher)) + self.assertIsInstance(encrypted.cipher, Cipher) + self.assertEqual(encrypted.iv, iv) + self.assertEqual(encrypted.data, b'abcd') + self.assertEqual(HIP.from_data(parsed.info).data, data) + + def test_each_encrypted_takes_the_next_cipher(self) -> None: + from pcapkit.const.hip.cipher import Cipher + from pcapkit.const.hip.parameter import Parameter + from pcapkit.protocols.internet.hip import HIP + + ciphers = [Cipher.AES_256_CBC, Cipher.NULL_ENCRYPT, Cipher.AES_128_CBC] + ivs = [bytes(16), None, bytes(range(16))] + data = HIP(next=59, parameters=[ + (Parameter.HIP_CIPHER, {'ciphers': ciphers[:1]}), + (Parameter.HIP_CIPHER, {'ciphers': ciphers[1:]}), + *[(Parameter.ENCRYPTED, {'cipher': cipher, 'iv': iv, 'data': b'abcd'}) + for cipher, iv in zip(ciphers, ivs)], + ]).data + parsed, caught = self._parse(data) + self.assertEqual(caught, []) + encrypted = parsed.info.parameters.getlist(Parameter.ENCRYPTED) + self.assertEqual([param.cipher for param in encrypted], ciphers) + self.assertEqual([param.iv for param in encrypted], ivs) + self.assertEqual(HIP.from_data(parsed.info).data, data) + + def test_without_a_cipher_for_it_still_warns(self) -> None: + from pcapkit.const.hip.cipher import Cipher + from pcapkit.const.hip.parameter import Parameter + from pcapkit.protocols.internet.hip import HIP + + lone = HIP(next=59, parameters=[ + (Parameter.ENCRYPTED, {'cipher': Cipher.NULL_ENCRYPT, 'data': b'abcd'}), + ]).data + extra = HIP(next=59, parameters=[ + (Parameter.HIP_CIPHER, {'ciphers': [Cipher.NULL_ENCRYPT]}), + (Parameter.ENCRYPTED, {'cipher': Cipher.NULL_ENCRYPT, 'data': b'abcd'}), + (Parameter.ENCRYPTED, {'cipher': Cipher.NULL_ENCRYPT, 'data': b'efgh'}), + ]).data + for data, message in ((lone, 'missing HIP_CIPHER parameter'), + (extra, 'too many ENCRYPTED parameters')): + with self.subTest(message=message): + parsed, caught = self._parse(data) + self.assertEqual(len(caught), 1) + self.assertIn(message, caught[0]) + self.assertEqual(HIP.from_data(parsed.info).data, data) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/internet/test_mh_cga_param_dict_2708_unit.py b/tests/protocols/internet/test_mh_cga_param_dict_2708_unit.py new file mode 100644 index 0000000000..61bcb6d18d --- /dev/null +++ b/tests/protocols/internet/test_mh_cga_param_dict_2708_unit.py @@ -0,0 +1,83 @@ +# -*- coding: utf-8 -*- +"""The MH CGA Parameters option maker neither takes nor advertises a mapping. + +GitHub issue #2708: the annotation of ``parameters`` on +:meth:`MH._make_opt_cga_param ` +names the forms the maker builds, octets, a schema and a data model, and no +mapping, which raises ``ProtocolError``. A mapping given to a maker stands for +some maker's keyword arguments, and no maker takes a CGA parameter's fields. A +parse gives each parameter as a data model, which +:meth:`Info.to_dict ` leaves intact +inside the option's tuple, so whatever parses still builds. + +Every case builds its own octets in memory and reads no capture. :class:`MH` +is imported inside each test, after +:func:`~tests._support.reimport_once_per_class`. + +""" +# Deferring these is the design: each build runs after the per-class re-import. +# pylint: disable=import-outside-toplevel +from __future__ import annotations + +import inspect +import unittest + +from tests._support import reimport_once_per_class + + +def _fields() -> 'dict[str, object]': + """Return the fields of one CGA parameter, as its data model takes them.""" + from pcapkit.const.mh.cga_type import CGAType + + return { + 'modifier': CGAType.Tag_086F_CA5E_10B2_00C9_9C8C_E001_6427_7C08, + 'prefix': 0x20010db800000000, + 'collision_count': 0, + 'public_key': b'\x30\x02\xaa\xbb', + 'extensions': [], + } + + +class TestMHCGAParamDict(unittest.TestCase): + """A CGA parameter is octets, a schema or a data model, and never a mapping.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_annotation_names_no_mapping(self) -> None: + from pcapkit.protocols.internet.mh import MH + + annotation = inspect.signature(MH._make_opt_cga_param).parameters['parameters'].annotation + self.assertIsInstance(annotation, str) + self.assertNotIn('dict', annotation) + for name in ('Schema_CGAParameter', 'Data_CGAParameter', 'bytes'): + with self.subTest(name=name): + self.assertIn(name, annotation) + + def test_mapping_is_refused(self) -> None: + from pcapkit.const.mh.option import Option + from pcapkit.protocols.internet.mh import MH + from pcapkit.utilities.exceptions import ProtocolError + + with self.assertRaisesRegex(ProtocolError, 'unknown CGA parameter format'): + MH()._make_opt_cga_param(Option.CGA_Parameters, parameters=[_fields()]) + + def test_data_model_of_the_same_fields_round_trips(self) -> None: + from pcapkit.const.mh.option import Option + from pcapkit.const.mh.packet import Packet + from pcapkit.protocols.data.internet.mh import CGAParameter + from pcapkit.protocols.internet.mh import MH + + built = MH(type=Packet.Binding_Refresh_Request, next=59, chksum=b'\x00\x00', data={ + 'options': [(Option.CGA_Parameters, {'parameters': [CGAParameter(**_fields())]})], + }) + parsed = MH(built.data) + (option,) = parsed.info.options.getlist(Option.CGA_Parameters) + # 25 octets of modifier, prefix and collision count, then the key. + self.assertEqual(option.length, 2 + 25 + 4) + self.assertIsInstance(option.to_dict()['parameters'][0], CGAParameter) + self.assertEqual(MH.from_data(parsed.info).data, built.data) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/test_protocol_make_unit.py b/tests/protocols/test_protocol_make_unit.py index a30b8d449f..a9eda33966 100644 --- a/tests/protocols/test_protocol_make_unit.py +++ b/tests/protocols/test_protocol_make_unit.py @@ -41,7 +41,7 @@ def _attr(path: 'str') -> 'Any': RAW_ITEMS = { 'HIP parameters': ('pcapkit.protocols.internet.hip.HIP', {'extension': True, 'parameters': [b'\x00\x41\x00\x0c' + bytes(12)]}, - '1106206100000000' + '00' * 32 + '0041000c' + '00' * 12), + '1106202100000000' + '00' * 32 + '0041000c' + '00' * 12), 'IPv4 options': ('pcapkit.protocols.internet.ipv4.IPv4', {'options': [b'\x94\x04\x00\x00'], 'protocol': 253}, '460000180000000000fda5e47f0000010000000094040000'), diff --git a/tests/protocols/transport/test_tcp_raw_option_kind_2708_unit.py b/tests/protocols/transport/test_tcp_raw_option_kind_2708_unit.py new file mode 100644 index 0000000000..cc80dac5af --- /dev/null +++ b/tests/protocols/transport/test_tcp_raw_option_kind_2708_unit.py @@ -0,0 +1,93 @@ +# -*- coding: utf-8 -*- +"""A TCP option given as raw octets builds whatever kind its first octet holds. + +GitHub issue #2708: ``_make_tcp_options`` resolves the kind of a raw-octets +option through ``_make_index``, as it does the kind of a ``(code, arguments)`` +pair, so a kind no member of :class:`~pcapkit.const.tcp.option.Option` holds, +255 among them, comes back as its code. ``Option.get`` itself raises +``EnumValueError`` for such a kind, which a parse gives back as an unassigned +option, and whose octets have to build through this branch as well. Empty +octets hold no kind at all, and raise ``ProtocolError`` as the other malformed +items do. + +The kinds are the ones ``Option.get`` refuses, read off the registry rather +than listed, and each case asserts the octets against :rfc:`9293#section-3.1`: +the option is written as given, after the 20-octet header, and Data Offset +counts it. + +Every case builds its own octets in memory and reads no capture. :class:`TCP` +is imported inside each test, after +:func:`~tests._support.reimport_once_per_class`. + +""" +# Deferring these is the design: each build runs after the per-class re-import. +# pylint: disable=import-outside-toplevel +from __future__ import annotations + +import unittest + +from tests._support import reimport_once_per_class + + +def _refused_kinds() -> 'list[int]': + """Return the kinds that ``Option.get`` refuses.""" + from pcapkit.const.tcp.option import Option + + kinds = [] + for kind in range(256): + try: + Option.get(kind) + except ValueError: + kinds.append(kind) + return kinds + + +class TestTCPRawOptionKind(unittest.TestCase): + """A raw-octets TCP option of a kind no member holds builds and rebuilds.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_registry_refuses_a_kind(self) -> None: + """The cases below are not vacuous: 255 is a kind ``Option.get`` refuses.""" + self.assertIn(255, _refused_kinds()) + + def test_raw_octets_build_as_the_pair_does(self) -> None: + from pcapkit.protocols.transport.tcp import TCP + + for kind in _refused_kinds(): + with self.subTest(kind=kind): + raw = bytes([kind, 4]) + b'ab' + built = TCP(options=[raw]).data + self.assertEqual(built, TCP(options=[(kind, {'data': b'ab'})]).data) + # Data Offset is 6 words: the 20-octet header and the option. + self.assertEqual(built[12] >> 4, 6) + self.assertEqual(built[20:], raw) + + def test_parsed_option_rebuilds_from_its_octets(self) -> None: + from pcapkit.protocols.transport.tcp import TCP + + for kind in _refused_kinds(): + with self.subTest(kind=kind): + packet = TCP(options=[(kind, {'data': b'ab'})]).data + parsed = TCP(packet) + (option,) = parsed.info.options.values() + self.assertEqual(int(option.kind), kind) + self.assertEqual(option.data, b'ab') + + octets = packet[20:20 + option.length] + self.assertEqual(TCP(options=[octets]).data, packet) + + def test_empty_octets_are_refused(self) -> None: + """Raw octets with no kind octet raise ``ProtocolError``, not ``IndexError``.""" + from pcapkit.protocols.transport.tcp import TCP + from pcapkit.utilities.exceptions import ProtocolError + + for options in ([b''], [b'\x01', b'']): + with self.subTest(options=options): + with self.assertRaisesRegex(ProtocolError, 'holds no kind octet'): + TCP(options=options) + + +if __name__ == '__main__': + unittest.main()