protocols: drop ipv6_opts' stray SMF_DPD test field, fix two length underflows

docs/source/pcapkit/protocols/internet/hip.rst

@@ -424,6 +424,7 @@ Auxiliary Functions
 
 .. autofunction:: pcapkit.protocols.schema.internet.hip.locator_value_selector
 .. autofunction:: pcapkit.protocols.schema.internet.hip.host_id_hi_selector
+.. autofunction:: pcapkit.protocols.schema.internet.hip.registration_type_list_len
 
 Data Models
 -----------

docs/source/pcapkit/protocols/internet/hopopt.rst

@@ -222,6 +222,7 @@ Auxiliary Functions
 
 .. autofunction:: pcapkit.protocols.schema.internet.hopopt.smf_dpd_data_selector
 .. autofunction:: pcapkit.protocols.schema.internet.hopopt.smf_i_dpd_tid_selector
+.. autofunction:: pcapkit.protocols.schema.internet.hopopt.smf_i_dpd_id_len
 .. autofunction:: pcapkit.protocols.schema.internet.hopopt.quick_start_data_selector
 
 Data Models

docs/source/pcapkit/protocols/internet/ipv6_opts.rst

@@ -222,6 +222,7 @@ Auxiliary Functions
 
 .. autofunction:: pcapkit.protocols.schema.internet.ipv6_opts.smf_dpd_data_selector
 .. autofunction:: pcapkit.protocols.schema.internet.ipv6_opts.smf_i_dpd_tid_selector
+.. autofunction:: pcapkit.protocols.schema.internet.ipv6_opts.smf_i_dpd_id_len
 .. autofunction:: pcapkit.protocols.schema.internet.ipv6_opts.quick_start_data_selector
 
 Data Models

pcapkit/protocols/schema/internet/hip.py

@@ -163,6 +163,33 @@ def host_id_hi_selector(pkt: 'dict[str, Any]') -> 'Field':
     return SchemaField(length=pkt['hi_len'], schema=schema)
 
 
+def registration_type_list_len(pkt: 'dict[str, Any]') -> 'int':
+    """Return registration type list length.
+
+    Used by the ``reg_request``, ``reg_response`` and ``reg_failed`` fields of
+    :class:`RegRequestParameter`, :class:`RegResponseParameter` and
+    :class:`RegFailedParameter` respectively, each of which follows a single
+    ``lifetime`` octet with a list of registration type octets sized by the
+    remainder of the parameter.
+
+    Args:
+        pkt: Parameter unpacked schema.
+
+    Returns:
+        Registration type list length.
+
+    Raises:
+        FieldValueError: If the parameter's ``Length`` on the wire is too
+            short to hold the ``lifetime`` octet already read, which would
+            otherwise underflow the list length below zero.
+
+    """
+    length = pkt['len'] - 1
+    if length < 0:
+        raise FieldValueError(f'HIP: invalid parameter length: {pkt["len"]}')
+    return length
+
+
 class Parameter(EnumSchema[Enum_Parameter]):
     """Base schema for HIP parameters."""
 
@@ -696,7 +723,7 @@ class RegRequestParameter(Parameter, code=Enum_Parameter.REG_REQUEST):
     lifetime: 'int' = UInt8Field()
     #: Registration types.
     reg_request: 'list[Enum_Registration]' = ListField(
-        length=lambda pkt: pkt['len'] - 1,
+        length=registration_type_list_len,
         item_type=EnumField(length=1, namespace=Enum_Registration),
     )
     #: Padding.
@@ -714,7 +741,7 @@ class RegResponseParameter(Parameter, code=Enum_Parameter.REG_RESPONSE):
     lifetime: 'int' = UInt8Field()
     #: Registration types.
     reg_response: 'list[Enum_Registration]' = ListField(
-        length=lambda pkt: pkt['len'] - 1,
+        length=registration_type_list_len,
         item_type=EnumField(length=1, namespace=Enum_Registration),
     )
     #: Padding.
@@ -732,7 +759,7 @@ class RegFailedParameter(Parameter, code=Enum_Parameter.REG_FAILED):
     lifetime: 'int' = UInt8Field()
     #: Registration types.
     reg_failed: 'list[Enum_RegistrationFailure]' = ListField(
-        length=lambda pkt: pkt['len'] - 1,
+        length=registration_type_list_len,
         item_type=EnumField(length=1, namespace=Enum_RegistrationFailure),
     )
     #: Padding.

pcapkit/protocols/schema/internet/hopopt.py

@@ -141,6 +141,27 @@ def mpl_opt_seed_id_len(pkt: 'dict[str, Any]') -> 'int':
     raise FieldValueError(f'HOPOPT: invalid MPL Seed-ID type: {s_type}')
 
 
+def smf_i_dpd_id_len(pkt: 'dict[str, Any]') -> 'int':
+    """Return SMF I-DPD identifier length.
+
+    Args:
+        pkt: SMF identification-based DPD option unpacked schema.
+
+    Returns:
+        SMF I-DPD identifier length.
+
+    Raises:
+        FieldValueError: If ``Opt Data Len`` on the wire is too short to hold
+            the TaggerID it declares, which would otherwise underflow the
+            identifier length below zero.
+
+    """
+    length = pkt['len'] - (1 if pkt['info']['type'] == 0 else (pkt['info']['len'] + 2))
+    if length < 0:
+        raise FieldValueError(f'HOPOPT: invalid SMF I-DPD option length: {pkt["len"]}')
+    return length
+
+
 def smf_dpd_data_selector(pkt: 'dict[str, Any]') -> 'Field':
     """Selector function for :attr:`_SMFDPDOption.data` field.
 
@@ -443,9 +464,7 @@ class SMFIdentificationBasedDPDOption(SMFDPDOption, code=Enum_SMFDPDMode.I_DPD):
         lambda pkt: pkt['info']['type'] != 0,
     )
     #: Identifier.
-    id: 'bytes' = BytesField(length=lambda pkt: pkt['len'] - (
-        1 if pkt['info']['type'] == 0 else (pkt['info']['len'] + 2)
-    ))
+    id: 'bytes' = BytesField(length=smf_i_dpd_id_len)
 
     def post_process(self, packet: 'dict[str, Any]') -> 'SMFIdentificationBasedDPDOption':
         """Revise ``schema`` data after unpacking process.

pcapkit/protocols/schema/internet/ipv6_opts.py

@@ -141,6 +141,27 @@ def mpl_opt_seed_id_len(pkt: 'dict[str, Any]') -> 'int':
     raise FieldValueError(f'IPv6-Opts: invalid MPL Seed-ID type: {s_type}')
 
 
+def smf_i_dpd_id_len(pkt: 'dict[str, Any]') -> 'int':
+    """Return SMF I-DPD identifier length.
+
+    Args:
+        pkt: SMF identification-based DPD option unpacked schema.
+
+    Returns:
+        SMF I-DPD identifier length.
+
+    Raises:
+        FieldValueError: If ``Opt Data Len`` on the wire is too short to hold
+            the TaggerID it declares, which would otherwise underflow the
+            identifier length below zero.
+
+    """
+    length = pkt['len'] - (1 if pkt['info']['type'] == 0 else (pkt['info']['len'] + 2))
+    if length < 0:
+        raise FieldValueError(f'IPv6-Opts: invalid SMF I-DPD option length: {pkt["len"]}')
+    return length
+
+
 def smf_dpd_data_selector(pkt: 'dict[str, Any]') -> 'Field':
     """Selector function for :attr:`_SMFDPDOption.data` field.
 
@@ -431,9 +452,6 @@ class SMFDPDOption(Option, EnumSchema[Enum_SMFDPDMode]):
 class SMFIdentificationBasedDPDOption(SMFDPDOption, code=Enum_SMFDPDMode.I_DPD):
     """Header schema for IPv6-Opts SMF identification-based DPD options."""
 
-    test: 'SMFDPDTestFlag' = ForwardMatchField(BitField(length=1, namespace={
-        'mode': (0, 1),
-    }))
     #: TaggerID information.
     info: 'TaggerIDInfo' = BitField(length=1, namespace={
         'mode': (0, 1),
@@ -446,9 +464,7 @@ class SMFIdentificationBasedDPDOption(SMFDPDOption, code=Enum_SMFDPDMode.I_DPD):
         lambda pkt: pkt['info']['type'] != 0,
     )
     #: Identifier.
-    id: 'bytes' = BytesField(length=lambda pkt: pkt['len'] - (
-        1 if pkt['info']['type'] == 0 else (pkt['info']['len'] + 2)
-    ))
+    id: 'bytes' = BytesField(length=smf_i_dpd_id_len)
 
     def post_process(self, packet: 'dict[str, Any]') -> 'SMFIdentificationBasedDPDOption':
         """Revise ``schema`` data after unpacking process.

tests/protocols/internet/test_hip_unit.py

@@ -1763,6 +1763,44 @@ def test_hip_parameter_constructors_cover_data_model_and_default_paths(self) ->
             version=2,
         ).hmac, b'relh')
 
+    def test_hip_registration_parameters_reject_underflowing_length(self) -> None:
+        """#438: a registration parameter's ``Length`` too small for its own
+        ``lifetime`` octet must raise, not silently drop the registration list.
+
+        ``reg_request``, ``reg_response`` and ``reg_failed`` each size their
+        list of registration-type octets as ``Length - 1``. Nothing floored
+        that at zero, so a peer declaring ``Length = 0`` drove the list length
+        to ``-1``. Unlike a :class:`~pcapkit.corekit.fields.strings.BytesField`,
+        :class:`~pcapkit.corekit.fields.collections.ListField` never reaches
+        :func:`struct.calcsize` for a negative length -- its own ``while length
+        > 0`` loop just returns an empty list instead -- so this parsed to an
+        empty ``reg_type`` with no exception and no diagnostic, rather than
+        rejecting the malformed ``Length``.
+
+        """
+        from pcapkit.protocols.internet.hip import HIP
+        from pcapkit.utilities.exceptions import FieldValueError
+
+        # next(1) len(1)=5 pkt(1) ver(1)=0x01 (the reserved bit that must be 1)
+        # checksum(2) control(2) shit(16) rhit(16) -- the fixed 40-octet header,
+        # declaring one 8-octet parameter to follow: (5 - 4) * 8 == 8.
+        fixed = bytes([0x3b, 0x05, 0x00, 0x01]) + bytes(2) + bytes(2) + bytes(16) + bytes(16)
+        self.assertEqual(len(fixed), 40)
+
+        for name, code in (
+            ('REG_REQUEST', 932),
+            ('REG_RESPONSE', 934),
+            ('REG_FAILED', 936),
+        ):
+            with self.subTest(parameter=name):
+                # type(2) len(2)=0 lifetime(1), then 3 octets padding out the
+                # 8-octet parameter area the outer header declared.
+                param = code.to_bytes(2, 'big') + (0).to_bytes(2, 'big') + bytes(4)
+                raw = fixed + param
+
+                with self.assertRaisesRegex(FieldValueError, 'invalid parameter length'):
+                    HIP(raw, len(raw), extension=True)
+
     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
@@ -1791,6 +1829,16 @@ def test_hip_schema_selectors_and_encrypted_parameter_branches(self) -> None:
         self.assertEqual(type(unknown_host_field).__name__, 'BytesField')
         self.assertEqual(unknown_host_field.length, 6)
 
+        # #438: ``reg_request``/``reg_response``/``reg_failed`` size their
+        # registration-type list as ``Length - 1``, the ``- 1`` accounting for
+        # the ``lifetime`` octet already read unconditionally.
+        self.assertEqual(hip_schema.registration_type_list_len({'len': 1}), 0)
+        self.assertEqual(hip_schema.registration_type_list_len({'len': 4}), 3)
+        # a ``Length`` too small to hold that ``lifetime`` octet must raise
+        # rather than drive the list length negative.
+        with self.assertRaisesRegex(FieldValueError, 'invalid parameter length'):
+            hip_schema.registration_type_list_len({'len': 0})
+
         missing_packet: dict[str, object] = {}
         with mock.patch('pcapkit.protocols.schema.internet.hip.warn') as warn:
             hip_schema.EncryptedParameter.pre_unpack(missing_packet)

tests/protocols/internet/test_ipv6_extension_unit.py

@@ -1040,7 +1040,6 @@ def assert_bad(reader, option_schema) -> None:
         ident = schema.SMFIdentificationBasedDPDOption(
             type=Option.SMF_DPD,
             len=7,
-            test={'mode': SMFDPDMode.I_DPD},
             info={'mode': 0, 'type': TaggerID.IPv4, 'len': 3},
             tid=ip_address('192.0.2.1'),
             id=b'id',
@@ -1264,6 +1263,20 @@ def _assert_option_schema_helpers_and_post_process_branches(self, schema: types.
         with self.assertRaises(FieldValueError):
             schema.mpl_opt_seed_id_len({'flags': {'type': 4}})
 
+        # #438: a null TaggerID consumes nothing, so the identifier is the whole
+        # of ``Opt Data Len`` less the one octet ``info`` itself already took.
+        self.assertEqual(schema.smf_i_dpd_id_len({'len': 5, 'info': {'type': 0, 'len': 0}}), 4)
+        # a non-null TaggerID additionally takes ``TidLen + 1`` octets (the ``+ 2``
+        # below also counts the octet ``info`` itself took).
+        self.assertEqual(schema.smf_i_dpd_id_len({'len': 7, 'info': {'type': 2, 'len': 3}}), 2)
+        # ``Opt Data Len`` too short to hold the TaggerID it declares must raise
+        # rather than drive the identifier length negative -- c.f. #438, where the
+        # unguarded subtraction reached :func:`struct.calcsize` as ``'-1s'``.
+        with self.assertRaisesRegex(FieldValueError, 'invalid SMF I-DPD option length'):
+            schema.smf_i_dpd_id_len({'len': 0, 'info': {'type': 0, 'len': 0}})
+        with self.assertRaisesRegex(FieldValueError, 'invalid SMF I-DPD option length'):
+            schema.smf_i_dpd_id_len({'len': 0, 'info': {'type': 2, 'len': 3}})
+
         for mode in (SMFDPDMode.I_DPD, SMFDPDMode.H_DPD):
             field = schema.smf_dpd_data_selector({'test': {'mode': mode, 'len': 4}})
             self.assertIsInstance(field, SchemaField)
@@ -1319,8 +1332,6 @@ def _assert_option_schema_helpers_and_post_process_branches(self, schema: types.
             'info': {'mode': 0, 'type': TaggerID.NULL, 'len': 0},
             'id': b'id',
         }
-        if schema.__name__.endswith('ipv6_opts'):
-            ident_kwargs['test'] = {'mode': SMFDPDMode.I_DPD}
         ident = schema.SMFIdentificationBasedDPDOption(**ident_kwargs)
         self.assertIs(ident.post_process({}), ident)
         self.assertEqual(ident.mode, SMFDPDMode.I_DPD)
@@ -1536,14 +1547,16 @@ def _assert_identification_based_dpd_options_parse_from_the_wire(self, protocol_
         identifier rather than the whole option. Both cases below hang the
         pristine tree exactly as the hash-based ones do.
 
-        Only the option itself is asserted, not the padding after it: HOPOPT and
-        IPv6-Opts disagree on how much of the option area an
-        ``SMFIdentificationBasedDPDOption`` leaves over, because the IPv6-Opts
-        schema carries an extra forward-matched octet which
-        :attr:`Schema.__buffer__ <pcapkit.protocols.schema.schema.Schema.__buffer__>`
-        records although the stream never consumes it. That is its own
-        ``len(data)``-is-not-consumed defect, distinct from the one #431 is about,
-        and pinning either count here would bless one of the two.
+        Only the option itself is asserted here, not the padding after it -- that
+        parity is what :meth:`_assert_identification_based_dpd_null_tid_option_area_matches`
+        below pins instead. Until #441, HOPOPT and IPv6-Opts disagreed on how much
+        of the option area an ``SMFIdentificationBasedDPDOption`` left over,
+        because the IPv6-Opts schema carried an extra forward-matched ``test``
+        field its HOPOPT twin did not: the field consumed no bytes but still
+        occupied a :attr:`Schema.__buffer__ <pcapkit.protocols.schema.schema.Schema.__buffer__>`
+        slot, so ``len(schema)`` over-reported the option by one octet under
+        IPv6-Opts. That was its own ``len(data)``-is-not-consumed defect, distinct
+        from the one #431 is about.
 
         """
         from pcapkit.const.ipv6.option import Option
@@ -1586,6 +1599,88 @@ def test_ipv6_opts_identification_based_dpd_options_parse_from_the_wire(self) ->
 
         self._assert_identification_based_dpd_options_parse_from_the_wire(IPv6_Opts)
 
+    def _assert_identification_based_dpd_null_tid_option_area_matches(self, protocol_cls: type) -> None:
+        """#441: a null-TaggerID I-DPD option must consume exactly what HOPOPT does.
+
+        This is the issue's own reproduction: a null-TaggerID I-DPD option with a
+        zero-octet identifier, followed by one octet of ``PadN``. ``ipv6_opts.py``'s
+        ``SMFIdentificationBasedDPDOption`` carried a stray ``test``
+        :class:`~pcapkit.corekit.fields.misc.ForwardMatchField` that its HOPOPT
+        twin did not -- see
+        :meth:`_assert_identification_based_dpd_options_parse_from_the_wire` above
+        for the mechanism. On the pristine tree this option parses under HOPOPT
+        but raises ``ProtocolError: IPv6-Opts: invalid format`` under IPv6-Opts,
+        for the identical octets.
+
+        """
+        from pcapkit.const.ipv6.option import Option
+        from pcapkit.const.ipv6.smf_dpd_mode import SMFDPDMode
+        from pcapkit.const.ipv6.tagger_id import TaggerID
+
+        raw = bytes.fromhex('1100080100010100')
+
+        with time_limit(5):
+            proto = protocol_cls(raw, extension=True)
+
+        options = list(proto.info.options.items(multi=True))
+        self.assertEqual([code for code, _ in options], [Option.SMF_DPD, Option.PadN])
+
+        smf_dpd = options[0][1]
+        self.assertEqual(smf_dpd.length, 3)
+        self.assertEqual(smf_dpd.dpd_type, SMFDPDMode.I_DPD)
+        self.assertEqual(smf_dpd.tid_type, TaggerID.NULL)
+        self.assertEqual(smf_dpd.tid_len, 0)
+        self.assertIsNone(smf_dpd.tid)
+        self.assertEqual(smf_dpd.id, b'')
+
+        padn = options[1][1]
+        self.assertEqual(padn.length, 3)
+
+        # the reader and the writer must agree: repacking the parsed schema has
+        # to give back the very bytes it was read from
+        self.assertEqual(bytes(proto.__header__), raw)
+
+    def test_hopopt_identification_based_dpd_null_tid_option_area_matches(self) -> None:
+        from pcapkit.protocols.internet.hopopt import HOPOPT
+
+        self._assert_identification_based_dpd_null_tid_option_area_matches(HOPOPT)
+
+    def test_ipv6_opts_identification_based_dpd_null_tid_option_area_matches(self) -> None:
+        from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts
+
+        self._assert_identification_based_dpd_null_tid_option_area_matches(IPv6_Opts)
+
+    def _assert_identification_based_dpd_option_rejects_underflowing_length(self, protocol_cls: type) -> None:
+        """#438: ``Opt Data Len`` too small for a null TaggerID must raise, not crash.
+
+        This is the issue's own reproduction: an I-DPD option (null TaggerID)
+        declaring ``Opt Data Len = 0``, which leaves no room for the one octet
+        ``info`` itself already consumes. ``id``'s length is ``Opt Data Len - 1``
+        for a null TaggerID, and nothing floored that at zero, so it drove the
+        identifier length to ``-1``. That reached :func:`struct.calcsize` as the
+        template ``'-1s'`` and raised a bare ``struct.error: bad char in struct
+        format`` -- not one of pcapkit's own exception types, and uncatchable
+        through :mod:`pcapkit.utilities.exceptions`.
+
+        """
+        from pcapkit.utilities.exceptions import FieldValueError
+
+        raw = bytes.fromhex('3b00080000000000')
+
+        with self.assertRaisesRegex(FieldValueError, 'invalid SMF I-DPD option length'):
+            with time_limit(5):
+                protocol_cls(raw, extension=True)
+
+    def test_hopopt_identification_based_dpd_option_rejects_underflowing_length(self) -> None:
+        from pcapkit.protocols.internet.hopopt import HOPOPT
+
+        self._assert_identification_based_dpd_option_rejects_underflowing_length(HOPOPT)
+
+    def test_ipv6_opts_identification_based_dpd_option_rejects_underflowing_length(self) -> None:
+        from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts
+
+        self._assert_identification_based_dpd_option_rejects_underflowing_length(IPv6_Opts)
+
     def _assert_a_truncated_option_area_is_diagnosed(self, protocol_cls: type) -> None:
         """An option area with nothing behind it is an error, not a hang.