From 7b96fb51f881fe8f194c529c264355353ef3f979 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Tue, 3 Nov 2026 00:22:51 -0500 Subject: [PATCH] fix(internet): build an MH CGA parameter from its data model or a schema without public_key_test (#2725) Two defects in building the MH CGA Parameters option, both on main. - _passes_raw, which gates the re-parse of a made header that holds raw items, did not look inside a data model. A CGA parameter's data model with a raw extension, given directly or carried by from_data, left the octets in the made header, and _read_cga_extensions raised AttributeError. It looks into a data model as into a dict, and into a multi-mapping through every value of each key, so a raw item in a later option is found too. - _make_opt_cga_param set public_key_test only when hasattr() said it was missing, which it never did: a schema does not store an unset field, so hasattr() found the class's field object and packing raised KeyError. It tests the schema's keys, and sets the key test on a copy, leaving the caller's schema as given. - test_mh_unit reaches that branch without deleting the class attribute, and checks the made key test and the caller's untouched schema. The new module and the test_mh_unit case fail on 468f9c219 without the pcapkit/ change; the internet modules and the project guards pass. --- pcapkit/protocols/internet/mh.py | 23 +- pcapkit/protocols/protocol.py | 22 +- .../test_mh_cga_param_build_2725_unit.py | 198 ++++++++++++++++++ tests/protocols/internet/test_mh_unit.py | 15 +- 4 files changed, 239 insertions(+), 19 deletions(-) create mode 100644 tests/protocols/internet/test_mh_cga_param_build_2725_unit.py diff --git a/pcapkit/protocols/internet/mh.py b/pcapkit/protocols/internet/mh.py index c0ddd16c61..16685a6c1b 100644 --- a/pcapkit/protocols/internet/mh.py +++ b/pcapkit/protocols/internet/mh.py @@ -1154,6 +1154,10 @@ class MH(IPv6_Ext[Data_MH, Schema_MH], :exc:`~pcapkit.utilities.exceptions.ProtocolError`, rather than being dropped beside another item or, for most, raising an :exc:`AttributeError` alone (:issue:`2723`). + * A CGA parameter builds from a data model holding raw extensions, where + it raised an :exc:`AttributeError`, and from a schema without + ``public_key_test``, where it raised a :exc:`KeyError` + (:issue:`2725`). * Each enumeration field of :meth:`make` and of the message, option and sub-option makers, and the type of each option, sub-option, QoS attribute and CGA extension given to them, takes a member or an @@ -8705,7 +8709,9 @@ def _make_opt_cga_param(self, type: 'Enum_Option', option: 'Optional[Data_CGAPar 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. + parameter's fields, so build the data model from them. A + schema without ``public_key_test`` is built from a copy that + takes one from its key, and is left as given. **kwargs: Arbitrary keyword arguments. Returns: @@ -8730,11 +8736,16 @@ def _make_opt_cga_param(self, type: 'Enum_Option', option: 'Optional[Data_CGAPar length += len(data) param.append(data) elif isinstance(data, Schema_CGAParameter): - if not hasattr(data, 'public_key_test'): - data.public_key_test = {'len': max(len(data.public_key) - 2, 0)} - _, ext_len = self._make_cga_extensions(data.extensions) - length += 25 + len(data.public_key) + ext_len - param.append(data) + schema = data + # NOTE: Test the schema, not ``hasattr``: an unset field is not + # stored, so ``hasattr`` finds the class's field object (#2725). + if 'public_key_test' not in data: + # A copy takes the key test, leaving the caller's schema as given. + schema = Schema_CGAParameter.from_dict( + data.to_dict(), public_key_test={'len': max(len(data.public_key) - 2, 0)}) + _, ext_len = self._make_cga_extensions(schema.extensions) + length += 25 + len(schema.public_key) + ext_len + param.append(schema) elif isinstance(data, Data_CGAParameter): ext, ext_len = self._make_cga_extensions(data.extensions) schema = Schema_CGAParameter( diff --git a/pcapkit/protocols/protocol.py b/pcapkit/protocols/protocol.py index 23ae9f1cda..d981460ae7 100644 --- a/pcapkit/protocols/protocol.py +++ b/pcapkit/protocols/protocol.py @@ -44,7 +44,9 @@ from pcapkit.corekit.context import ContextRegistry from pcapkit.corekit.decorators import beholder, seekset from pcapkit.corekit.fields.misc import ConditionalField, SchemaField, SwitchField +from pcapkit.corekit.infoclass import Info from pcapkit.corekit.module import ModuleDescriptor +from pcapkit.corekit.multidict import MultiDict from pcapkit.corekit.protochain import ProtoChain from pcapkit.corekit.sentinels import ABSENT from pcapkit.protocols import data as data_module @@ -96,15 +98,20 @@ def _passes_raw(value: 'Any', nested: 'bool' = False) -> 'bool': Returns: :data:`True` if :obj:`bytes` sits anywhere inside a :obj:`list`, - :obj:`tuple` or :obj:`dict` among ``value``, or a + :obj:`tuple`, :obj:`dict` or data model among ``value``, or a :class:`~pcapkit.protocols.schema.schema.Schema` does anywhere. A - top-level :obj:`bytes` keyword (e.g. a payload) does not count. + top-level :obj:`bytes` keyword (e.g. a payload) does not count. A data + model is looked into as a :obj:`dict` is, since a maker builds from its + fields as given, and a multi-mapping through every value of each key + (:issue:`2725`). """ if isinstance(value, bytes): return nested if isinstance(value, Schema): # a ready-made item, which ``make()`` may pack into raw octets return True + if isinstance(value, (Info, MultiDict)): + return any(_passes_raw(item, nested) for key, item in value.items(multi=True) if key != '__packet__') if isinstance(value, dict): return any(_passes_raw(item, nested) for key, item in value.items() if key != '__packet__') if isinstance(value, (list, tuple)): @@ -848,8 +855,9 @@ def unpack(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> '_PT': :meth:`__post_init__` makes to construct the instance, that schema is kept, unless it still holds raw :obj:`bytes` where parsing its packed octets yields a schema -- e.g. a raw option given to - ``make()`` -- which :meth:`read` cannot consume. The header is - then replaced by the parsed one (:issue:`1312`). + ``make()``, directly or inside a data model -- which :meth:`read` + cannot consume. The header is then replaced by the parsed one + (:issue:`1312`). Raises: ProtocolError: If the made header holds raw items and its packed @@ -857,6 +865,12 @@ def unpack(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> '_PT': The message carries the original error's, which is also chained as ``__cause__``. + .. versionchanged:: 1.6.0 + + A raw item that ``make()`` takes from inside a data model, such as + the one :meth:`from_data` passes on, is replaced too, rather than + reaching :meth:`read` (:issue:`2725`). + """ packet = kwargs.get('__packet__', {}) # packet data if cast('Optional[_ST]', self.__header__) is None: diff --git a/tests/protocols/internet/test_mh_cga_param_build_2725_unit.py b/tests/protocols/internet/test_mh_cga_param_build_2725_unit.py new file mode 100644 index 0000000000..af38c5db88 --- /dev/null +++ b/tests/protocols/internet/test_mh_cga_param_build_2725_unit.py @@ -0,0 +1,198 @@ +# -*- coding: utf-8 -*- +"""An MH CGA parameter builds from a data model with a raw extension, or a schema without its key test. + +GitHub issue #2725 found two defects in building the CGA Parameters option. + +- ``make()`` keeps the header it made unless ``_passes_raw`` in + :mod:`pcapkit.protocols.protocol` finds a raw item among its keywords. + Missing one inside a data model -- a CGA parameter's, given directly or + inside the packet's own data model as + :meth:`~pcapkit.protocols.protocol.Protocol.from_data` passes it -- left the + octets in the made header, where ``_read_cga_extensions`` raised a bare + ``AttributeError``. ``_passes_raw`` looks inside a data model, and through + every value of a multi-mapping, so a raw extension in a later CGA Parameters + option is found as well. +- ``_make_opt_cga_param`` set ``public_key_test`` on a schema only when + ``hasattr`` said it was missing. A schema does not store an unset field, so + ``hasattr`` found the class's field object, and packing raised + ``KeyError: 'public_key_test'``. The key test goes on a copy, and the + caller's schema is left as given. + +Each build is held to the one a ``(type, kwargs)`` extension or a data model +gives, and to the specification: a CGA Parameters option's Option Length counts +the parameters it carries [:rfc:`4866#section-5.1`], which are the 25 octets of +modifier, subnet prefix and collision count, the public key, and the extensions +[:rfc:`3972#section-3`], each its 4-octet type and length and its data +[:rfc:`4581#section-2`]. + +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 unittest +from typing import TYPE_CHECKING + +from tests._support import reimport_once_per_class + +if TYPE_CHECKING: + from typing import Any + +#: A CGA extension of the unassigned type 1, carrying ``ab``. +EXTENSION = b'\x00\x01\x00\x02ab' +#: The same extension as a ``(type, kwargs)`` pair. +PAIR = (1, {'data': b'ab'}) +#: A public key whose DER length is the short form. +SHORT_KEY = b'\x30\x02\xaa\xbb' +#: A public key whose DER length takes the long form, ``0x81`` and one octet. +LONG_KEY = b'\x30\x81\xc8' + bytes(0xc8) +#: The octets of the fixed MH header and of a Binding Refresh Request. +HEADER = 8 +#: The octets of a CGA parameter before its public key. +FIXED = 25 + + +def _build(options: 'list[Any]') -> 'Any': + """A Binding Refresh Request carrying ``options``.""" + from pcapkit.const.mh.packet import Packet + from pcapkit.protocols.internet.mh import MH + + return MH(type=Packet.Binding_Refresh_Request, next=59, chksum=b'\x00\x00', + data={'options': options}) + + +def _fields(key: 'bytes') -> 'dict[str, Any]': + """The fields of one CGA parameter carrying ``key``, but its extensions.""" + 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': key, + } + + +def _option(parameters: 'list[Any]') -> 'tuple[Any, dict[str, Any]]': + """A CGA Parameters option carrying ``parameters``, as a ``(type, kwargs)`` pair.""" + from pcapkit.const.mh.option import Option + + return (Option.CGA_Parameters, {'parameters': parameters}) + + +def _with_raw(info: 'Any') -> 'Any': + """``info`` with each CGA parameter's extensions given as the raw :data:`EXTENSION`.""" + from pcapkit.const.mh.option import Option + from pcapkit.corekit.multidict import OrderedMultiDict + from pcapkit.protocols.data.internet.mh import CGAParameter, CGAParametersOption + + dict_ = info.to_dict() + options = OrderedMultiDict() # type: OrderedMultiDict[Any, Any] + for code, option in dict_['options'].items(multi=True): + if code == Option.CGA_Parameters and option.parameters: + option = CGAParametersOption(type=option.type, length=option.length, parameters=tuple( + CGAParameter(extensions=[EXTENSION], **_fields(param.public_key)) + for param in option.parameters)) + options.add(code, option) + dict_['options'] = options + return type(info).from_dict(dict_) + + +class _Case(unittest.TestCase): + """Shared assertions.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def assertCGAOption(self, data: 'bytes', key: 'bytes', extensions: 'bytes') -> None: + """The first option of ``data`` is a CGA Parameters option of one parameter.""" + from pcapkit.const.mh.option import Option + + self.assertEqual(data[HEADER], Option.CGA_Parameters) + self.assertEqual(data[HEADER + 1], FIXED + len(key) + len(extensions)) + start = HEADER + 2 + FIXED + self.assertEqual(data[start:start + len(key)], key) + end = start + len(key) + len(extensions) + self.assertEqual(data[start + len(key):end], extensions) + + +class TestMHCGARawExtensionInDataModel(_Case): + """A CGA parameter's data model holding a raw extension builds as its pair does.""" + + def test_data_model_builds_as_its_pair(self) -> None: + from pcapkit.protocols.data.internet.mh import CGAParameter + + for key in (SHORT_KEY, LONG_KEY): + with self.subTest(key=len(key)): + raw = _build([_option([CGAParameter(extensions=[EXTENSION], **_fields(key))])]) + pair = _build([_option([CGAParameter(extensions=[PAIR], **_fields(key))])]) + self.assertEqual(raw.data, pair.data) + self.assertCGAOption(raw.data, key, EXTENSION) + + def test_from_data_rebuilds_a_raw_extension(self) -> None: + from pcapkit.const.mh.cga_extension import CGAExtension + from pcapkit.const.mh.option import Option + from pcapkit.protocols.data.internet.mh import CGAParameter + from pcapkit.protocols.internet.mh import MH + + cases = { + 'one option': [_option([CGAParameter(extensions=[PAIR], **_fields(SHORT_KEY))])], + # NOTE: The parameters may span several options (RFC 4866, section + # 5.1); a multi-mapping's first value per key hides the second. + 'a later option': [ + _option([]), + _option([CGAParameter(extensions=[PAIR], **_fields(LONG_KEY))]), + ], + } + for label, options in cases.items(): + expected = _build(options) + info = _with_raw(expected.info) + for source, data in (('info', info), ('to_dict', info.to_dict())): + with self.subTest(case=label, source=source): + rebuilt = MH.from_data(data) + self.assertEqual(rebuilt.data, expected.data) + (*_, option) = rebuilt.info.options.getlist(Option.CGA_Parameters) + (param,) = option.parameters + (extension,) = param.extensions.getlist(CGAExtension.get(1)) + self.assertEqual(extension.data, b'ab') + self.assertEqual(extension.length, len(EXTENSION)) + + +class TestMHCGASchemaWithoutKeyTest(_Case): + """A CGA parameter schema without ``public_key_test`` builds as its data model does.""" + + def test_schema_builds_as_its_data_model(self) -> None: + from pcapkit.protocols.data.internet.mh import CGAParameter + from pcapkit.protocols.schema.internet.mh import CGAParameter as Schema_CGAParameter + + for key in (SHORT_KEY, LONG_KEY): + for extensions, pairs in (([], []), ([EXTENSION], [PAIR])): + with self.subTest(key=len(key), extensions=extensions): + schema = Schema_CGAParameter(extensions=extensions, **_fields(key)) + built = _build([_option([schema])]) + model = _build([_option([CGAParameter(extensions=pairs, **_fields(key))])]) + self.assertEqual(built.data, model.data) + self.assertCGAOption(built.data, key, b''.join(extensions)) + self.assertNotIn('public_key_test', schema) + + def test_schema_without_key_test_round_trips(self) -> None: + from pcapkit.const.mh.option import Option + from pcapkit.protocols.internet.mh import MH + from pcapkit.protocols.schema.internet.mh import CGAParameter as Schema_CGAParameter + + for key in (SHORT_KEY, LONG_KEY): + with self.subTest(key=len(key)): + built = _build([_option([Schema_CGAParameter(extensions=[EXTENSION], **_fields(key))])]) + parsed = MH(built.data) + (option,) = parsed.info.options.getlist(Option.CGA_Parameters) + (param,) = option.parameters + self.assertEqual(param.public_key, key) + self.assertEqual(MH.from_data(parsed.info).data, built.data) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/internet/test_mh_unit.py b/tests/protocols/internet/test_mh_unit.py index 915904454d..3507732605 100644 --- a/tests/protocols/internet/test_mh_unit.py +++ b/tests/protocols/internet/test_mh_unit.py @@ -750,15 +750,12 @@ def make_extension(code: CGAExtension, extension: data.UnknownExtension | None = Option.CGA_Parameters, parameters=[cga_schema], ).length, 0) - public_key_test = schema.CGAParameter.public_key_test - try: - delattr(schema.CGAParameter, 'public_key_test') - self.assertGreater(proto._make_opt_cga_param( - Option.CGA_Parameters, - parameters=[cga_schema_missing_test], - ).length, 0) - finally: - schema.CGAParameter.public_key_test = public_key_test + given = dict(cga_schema_missing_test) + made = proto._make_opt_cga_param(Option.CGA_Parameters, parameters=[cga_schema_missing_test]) + self.assertEqual(made.parameters[0].public_key_test, {'len': 1}) + self.assertEqual(made.length, 25 + 3) + self.assertNotIn('public_key_test', cga_schema_missing_test) + self.assertEqual(dict(cga_schema_missing_test), given) # NOTE: The data model's ``length`` is the whole option, the schema's is # its ``Option Length`` field. A 4-octet ``PadN`` spends 2 octets on its