fix(protocols): refuse or build what the TCP, HIP and MH makers mishandled (#2708)

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 <HIP.make>`.
 
+        Note:
+            ``version`` is declared here because :meth:`IPv6_Ext.__post_init__
+            <pcapkit.protocols.internet.ipv6_ext.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__)``,

pcapkit/protocols/internet/mh.py

@@ -8636,22 +8636,26 @@ 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:
             Constructed option schema.
 
         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,

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':
             <pcapkit.protocols.internet.hip.HIP._make_param_encrypted>`, 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
 

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):

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."""