From faa02c75e01c23572f9771e14113603c4bc6e31b Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Mon, 2 Nov 2026 17:00:00 -0500 Subject: [PATCH] fix(internet)!: refuse or rebuild exactly an MH mobility option changed in one octet (#2684) - Option fields were read at their own widths whatever the length said, so a length they overran or fell short of read octets that were not the option's and rebuilt another. A mobility option, FID, ANI or LCMP sub-option, QoS attribute or CGA extension whose fields do not take its declared length is refused, and so is a target care-of address neither IPv4 nor IPv6, as BID's is. - A CGA public key takes the long form of a DER length, and a CGA parameter whose fixed fields or key run past its option is refused in the schema. - The IPv4 Address Acknowledgement, IPv4 Home Address Request and Reply, and Delegated MNP makers check a prefix length against its field width alone, so one past the address that the reader keeps no longer refuses to rebuild. - The Service Selection and MN-ID NAI makers count the octets the identifier packs as, rather than its text re-encoded or its characters. - The QoS option keeps the two reserved traffic-class bits as tc_reserved, a new required field of its data model, which a direct construction now has to pass: the breaking change. - Adds a one-octet mutation sweep over every registered mobility option. New tests fail on be8306140 without the pcapkit/ change; MH and guard modules pass. --- docs/source/pcapkit/protocols/internet/mh.rst | 1 + pcapkit/protocols/data/internet/mh.py | 13 +- pcapkit/protocols/internet/mh.py | 190 +++++++++-- pcapkit/protocols/schema/internet/mh.py | 93 +++++- .../test_internet_maker_ranges_2614_unit.py | 13 +- .../test_mh_option_mutation_2684_unit.py | 311 ++++++++++++++++++ tests/protocols/internet/test_mh_unit.py | 51 ++- 7 files changed, 603 insertions(+), 69 deletions(-) create mode 100644 tests/protocols/internet/test_mh_option_mutation_2684_unit.py diff --git a/docs/source/pcapkit/protocols/internet/mh.rst b/docs/source/pcapkit/protocols/internet/mh.rst index 4f26ec952c..f14da8075e 100644 --- a/docs/source/pcapkit/protocols/internet/mh.rst +++ b/docs/source/pcapkit/protocols/internet/mh.rst @@ -987,6 +987,7 @@ Auxiliary Functions .. autofunction:: pcapkit.protocols.schema.internet.mh.pad_opt_data_len .. autofunction:: pcapkit.protocols.schema.internet.mh.bid_address_selector .. autofunction:: pcapkit.protocols.schema.internet.mh.br_code_selector +.. autofunction:: pcapkit.protocols.schema.internet.mh.cga_public_key_length .. autofunction:: pcapkit.protocols.schema.internet.mh.dmnp_prefix_selector .. autofunction:: pcapkit.protocols.schema.internet.mh.fb_code_selector .. autofunction:: pcapkit.protocols.schema.internet.mh.lma_address_selector diff --git a/pcapkit/protocols/data/internet/mh.py b/pcapkit/protocols/data/internet/mh.py index 4e88a683e6..82f10e2512 100644 --- a/pcapkit/protocols/data/internet/mh.py +++ b/pcapkit/protocols/data/internet/mh.py @@ -2168,13 +2168,22 @@ def __init__(self, type: 'Enum_QoSAttribute', length: 'int', reserved: 'bytes', @info_final class QualityOfServiceOption(Option): - """Data model for MH Quality-of-Service option.""" + """Data model for MH Quality-of-Service option. + + .. versionchanged:: 1.6.0 + + New required field ``tc_reserved`` (:issue:`2684`). + + """ #: Service request identifier. sr_id: 'int' #: Differentiated services code point, the significant 6 bits of the traffic #: class octet. dscp: 'int' + #: The 2 reserved bits of the traffic class octet, after the code point, as + #: captured [:rfc:`7222#section-4.1`]. + tc_reserved: 'int' #: Operational code -- ``0`` response, ``1`` allocate, ``2`` de-allocate, #: ``3`` modify, ``4`` query, ``5`` negotiate. oc: 'int' @@ -2185,7 +2194,7 @@ class QualityOfServiceOption(Option): if TYPE_CHECKING: def __init__(self, type: 'Enum_Option', length: 'int', sr_id: 'int', dscp: 'int', - oc: 'int', + tc_reserved: 'int', oc: 'int', reserved: 'bytes', attributes: 'OrderedMultiDict[Enum_QoSAttribute, QoSAttribute]') -> 'None': ... diff --git a/pcapkit/protocols/internet/mh.py b/pcapkit/protocols/internet/mh.py index 11f20a116d..fe7a5f7f0a 100644 --- a/pcapkit/protocols/internet/mh.py +++ b/pcapkit/protocols/internet/mh.py @@ -3044,6 +3044,17 @@ def _read_mh_options(self, options_schema: 'list[Schema_Option]') -> 'Option': Returns: Parsed MH options data. + Raises: + ProtocolError: If an option read by a ``_read_opt_*`` method has fields + that do not take exactly the octets its ``Option Length`` declares + [:rfc:`6275#section-6.2.1`]. A parser registered as a callable + reports its own length, and is not checked. + + .. versionchanged:: 1.6.0 + + Such an option is refused, rather than read from octets outside it and + rebuilt with the length its fields give (:issue:`2684`). + """ options = OrderedMultiDict() # type: Option @@ -3059,6 +3070,11 @@ def _read_mh_options(self, options_schema: 'list[Schema_Option]') -> 'Option': meth = name[0] data = meth(schema, options=options) + # NOTE: Fields are read at their own widths, so a length they do not fill + # exactly reads octets not the option's, and rebuilds as another (#2684). + if isinstance(name, str) and len(schema) != data.length: + raise ProtocolError(f'{self.alias}: [OptNo {type}] invalid option length: {schema.length}') + # record option data options.add(type, data) @@ -4962,6 +4978,19 @@ def _read_fid_suboptions( Returns: Parsed sub-option data. + Raises: + ProtocolError: If a sub-option's fields do not take exactly the octets + its ``Sub-Opt Length`` declares [:rfc:`6089#section-4.2.1`], or a + target care-of address is neither IPv4 nor IPv6 + [:rfc:`7109#section-6.2.2`], as the binding identifier option's + is refused. + + .. versionchanged:: 1.6.0 + + Such a sub-option is refused, as a mobility option is by + :meth:`_read_mh_options`, rather than read from octets outside it or + left unable to rebuild (:issue:`2684`). + """ suboptions = OrderedMultiDict() # type: FlowIDSuboption @@ -5002,6 +5031,8 @@ def _read_fid_suboptions( ) elif code == Enum_FlowIDSuboption.Target_Care_of_Address: target = cast('Schema_TargetCareofAddressSuboption', schema) + if target.length not in (6, 18): + raise ProtocolError(f'{self.alias}: [FID SubOpt {code}] invalid sub-option length: {target.length}') data = Data_TargetCareofAddressSuboption( type=code, length=self._mh_option_length(target.length), @@ -5016,6 +5047,8 @@ def _read_fid_suboptions( data=unknown.data, ) + if len(schema) != data.length: + raise ProtocolError(f'{self.alias}: [FID SubOpt {code}] invalid sub-option length: {schema.length}') suboptions.add(code, data) return suboptions @@ -5354,6 +5387,15 @@ def _read_ani_suboptions( Returns: Parsed sub-option data. + Raises: + ProtocolError: If a sub-option's fields do not take exactly the octets + its ``ANI Length`` declares [:rfc:`6757#section-3.1`]. + + .. versionchanged:: 1.6.0 + + Such a sub-option is refused, as a mobility option is by + :meth:`_read_mh_options` (:issue:`2684`). + """ suboptions = OrderedMultiDict() # type: ANISuboption @@ -5421,6 +5463,8 @@ def _read_ani_suboptions( data=unknown.data, ) + if len(schema) != data.length: + raise ProtocolError(f'{self.alias}: [ANI SubOpt {code}] invalid sub-option length: {schema.length}') suboptions.add(code, data) return suboptions @@ -5749,6 +5793,15 @@ def _read_qos_attributes( Returns: Parsed attribute data. + Raises: + ProtocolError: If an attribute's fields do not take exactly the octets + its ``Length`` declares [:rfc:`7222#section-4.2`]. + + .. versionchanged:: 1.6.0 + + Such an attribute is refused, as a mobility option is by + :meth:`_read_mh_options` (:issue:`2684`). + """ attributes = OrderedMultiDict() # type: QoSAttribute @@ -5816,6 +5869,8 @@ def _read_qos_attributes( data=unknown.data, ) + if len(schema) != data.length: + raise ProtocolError(f'{self.alias}: [QoS Attr {code}] invalid attribute length: {schema.length}') attributes.add(code, data) return attributes @@ -5840,8 +5895,9 @@ def _read_opt_qos(self, schema: 'Schema_QualityOfServiceOption', *, Note: The traffic class octet is a 6-bit differentiated services code point - with two reserved bits below it, so only the code point is carried into - the data model. + with two reserved bits below it, which the receiver ignores + [:rfc:`7222#section-4.1`]; both are carried into the data model, so + that the octet rebuilds as captured. Args: schema: Parsed option schema. @@ -5850,6 +5906,11 @@ def _read_opt_qos(self, schema: 'Schema_QualityOfServiceOption', *, Returns: Constructed option data. + .. versionchanged:: 1.6.0 + + The two reserved bits of the traffic class are kept, as + ``tc_reserved``, rather than dropped (:issue:`2684`). + """ if schema.length < 6: raise ProtocolError(f'{self.alias}: [Opt {schema.type}] invalid format') @@ -5859,6 +5920,7 @@ def _read_opt_qos(self, schema: 'Schema_QualityOfServiceOption', *, length=self._mh_option_length(schema.length), sr_id=schema.sr_id, dscp=schema.tc >> 2, + tc_reserved=schema.tc & 0b11, oc=schema.oc, reserved=schema.reserved, attributes=self._read_qos_attributes(schema.attributes), @@ -6034,6 +6096,15 @@ def _read_lcmp_suboptions( Returns: Parsed sub-option data. + Raises: + ProtocolError: If a sub-option's fields do not take exactly the octets + its ``LCMP Length`` declares [:rfc:`8127#section-3.1`]. + + .. versionchanged:: 1.6.0 + + Such a sub-option is refused, as a mobility option is by + :meth:`_read_mh_options` (:issue:`2684`). + """ suboptions = OrderedMultiDict() # type: LMAControlledMAGSuboption @@ -6068,6 +6139,8 @@ def _read_lcmp_suboptions( data=unknown.data, ) + if len(schema) != data.length: + raise ProtocolError(f'{self.alias}: [LCMP SubOpt {code}] invalid sub-option length: {schema.length}') suboptions.add(code, data) return suboptions @@ -6535,6 +6608,17 @@ def _read_cga_extensions(self, extensions_schema: 'list[Schema_CGAExtension]') - Returns: Parsed CGA extensions data. + Raises: + ProtocolError: If an extension read by a ``_read_ext_*`` method has + fields that do not take exactly the octets its ``Extension Data + Length`` declares [:rfc:`4581#section-2`]; a parser registered as + a callable is not checked. + + .. versionchanged:: 1.6.0 + + Such an extension is refused, as a mobility option is by + :meth:`_read_mh_options` (:issue:`2684`). + """ extensions = OrderedMultiDict() # type: Extension @@ -6549,6 +6633,8 @@ def _read_cga_extensions(self, extensions_schema: 'list[Schema_CGAExtension]') - else: meth = name[0] data = meth(schema, extensions=extensions) + if isinstance(name, str) and len(schema) != data.length: + raise ProtocolError(f'{self.alias}: [CGA Ext {type}] invalid extension length: {schema.length}') # record extension data extensions.add(type, data) @@ -8291,6 +8377,11 @@ def _make_opt_mn_id(self, type: 'Enum_Option', option: 'Optional[Data_MNIDOption Also if ``subtype`` does not fit in one octet, or the option is too long for its one-octet ``Option Length``. + .. versionchanged:: 1.6.0 + + An ``NAI`` identifier's length counts the octets it packs as, rather + than its characters (:issue:`2684`). + """ if option is not None: subtype_val = option.subtype @@ -8417,7 +8508,9 @@ def _make_opt_mn_id(self, type: 'Enum_Option', option: 'Optional[Data_MNIDOption raise ProtocolError( f'{self.alias}: [OptNo {type}] MN-ID subtype NAI ' f'identifier must be str, not {identifier!r}') - id_len = len(identifier) + # NOTE: The field packs UTF-8 octets, or a DecodedString's ``raw`` + # ones, so the length counts those rather than characters (#2684). + id_len = len(getattr(identifier, 'raw', None) or identifier.encode()) else: # NOTE: every other subtype's field is a # BytesField(length=pkt['length'] - 1) (see mn_id_selector), @@ -8952,11 +9045,19 @@ def _make_opt_service(self, type: 'Enum_Option', option: 'Optional[Data_ServiceS ProtocolError: If the option is too long for its one-octet ``Option Length``. + .. versionchanged:: 1.6.0 + + The length counts the octets a + :class:`~pcapkit.corekit.fields.strings.DecodedString` identifier + packs back as, rather than its text re-encoded (:issue:`2684`). + """ if option is not None: identifier = option.identifier - encoded = identifier.encode() + # NOTE: An identifier that is not UTF-8 parses as a DecodedString, whose + # ``raw`` octets are what packs back, so the length counts those (#2684). + encoded = getattr(identifier, 'raw', None) or identifier.encode() if not encoded: raise ProtocolError(f'{self.alias}: [OptNo {type}] service identifier must not be empty') @@ -9294,11 +9395,20 @@ def _make_opt_ipv4_ack(self, type: 'Enum_Option', Constructed option schema. Raises: - ProtocolError: If ``prefix_length`` is negative, a :obj:`bool`, - or longer than the address, or ``reserved`` does not fit - its field. + ProtocolError: If ``prefix_length`` or ``reserved`` is negative, a + :obj:`bool`, or does not fit its field. Also if ``status`` does not fit in one octet. + Note: + The prefix length is checked against its 6-bit field alone, as + :rfc:`5555#section-3.2.1` does not have the receiver discard one longer + than the address, so whatever the reader keeps is written back. + + .. versionchanged:: 1.6.0 + + A prefix length longer than the address is written, rather than + refused, so that every one the reader keeps rebuilds (:issue:`2684`). + """ if option is not None: status_val = option.status # type: Enum_DSMIPv6HomeAddress @@ -9310,8 +9420,6 @@ def _make_opt_ipv4_ack(self, type: 'Enum_Option', reversed=status_reversed, pack=False) prefix_length = self._make_uint(prefix_length, bits=6, name='prefix length') - if prefix_length > 32: - raise ProtocolError(f'{self.alias}: [OptNo {type}] invalid prefix length: {prefix_length}') self._make_uint(status_val, bits=8, name='status') return Schema_IPv4AddressAcknowledgementOption( @@ -9524,9 +9632,18 @@ def _make_opt_ipv4_hoa_req(self, type: 'Enum_Option', Constructed option schema. Raises: - ProtocolError: If ``prefix_length`` is negative, a :obj:`bool`, - or longer than the address, or ``reserved`` does not fit - its field. + ProtocolError: If ``prefix_length`` or ``reserved`` is negative, a + :obj:`bool`, or does not fit its field. + + Note: + The prefix length is checked against its 6-bit field alone, as + :rfc:`5844#section-3.3.1` does not have the receiver discard one longer + than the address, so whatever the reader keeps is written back. + + .. versionchanged:: 1.6.0 + + A prefix length longer than the address is written, rather than + refused, so that every one the reader keeps rebuilds (:issue:`2684`). """ if option is not None: @@ -9535,8 +9652,6 @@ def _make_opt_ipv4_hoa_req(self, type: 'Enum_Option', address = option.address prefix_length = self._make_uint(prefix_length, bits=6, name='prefix length') - if prefix_length > 32: - raise ProtocolError(f'{self.alias}: [OptNo {type}] invalid prefix length: {prefix_length}') return Schema_IPv4HomeAddressRequestOption( type=type, @@ -9573,11 +9688,20 @@ def _make_opt_ipv4_hoa_rep(self, type: 'Enum_Option', option: 'Optional[Data_IPv Constructed option schema. Raises: - ProtocolError: If ``prefix_length`` is negative, a :obj:`bool`, - or longer than the address, or ``reserved`` does not fit - its field. + ProtocolError: If ``prefix_length`` or ``reserved`` is negative, a + :obj:`bool`, or does not fit its field. Also if ``status`` does not fit in one octet. + Note: + The prefix length is checked against its 6-bit field alone, as + :rfc:`5844#section-3.3.2` does not have the receiver discard one longer + than the address, so whatever the reader keeps is written back. + + .. versionchanged:: 1.6.0 + + A prefix length longer than the address is written, rather than + refused, so that every one the reader keeps rebuilds (:issue:`2684`). + """ if option is not None: status_val = option.status # type: Enum_HomeAddressReply @@ -9589,8 +9713,6 @@ def _make_opt_ipv4_hoa_rep(self, type: 'Enum_Option', option: 'Optional[Data_IPv reversed=status_reversed, pack=False) prefix_length = self._make_uint(prefix_length, bits=6, name='prefix length') - if prefix_length > 32: - raise ProtocolError(f'{self.alias}: [OptNo {type}] invalid prefix length: {prefix_length}') self._make_uint(status_val, bits=8, name='status') return Schema_IPv4HomeAddressReplyOption( @@ -10625,12 +10747,18 @@ def _make_opt_dmnp(self, type: 'Enum_Option', option: 'Optional[Data_DelegatedMN Note: The ``V`` flag is derived from the prefix's own family rather than taken as an argument, so that the flag and the emitted width cannot - disagree. + disagree. The prefix length is checked against its 8-bit field alone, + as :rfc:`7148#section-4.1` does not have the receiver discard one + longer than the address, so whatever the reader keeps is written back. Raises: - ProtocolError: If ``prefix_length`` is negative, a :obj:`bool`, - or longer than the address, or ``reserved`` does not fit - its field. + ProtocolError: If ``prefix_length`` or ``reserved`` is negative, a + :obj:`bool`, or does not fit its field. + + .. versionchanged:: 1.6.0 + + A prefix length longer than the address is written, rather than + refused, so that every one the reader keeps rebuilds (:issue:`2684`). """ if option is not None: @@ -10649,8 +10777,6 @@ def _make_opt_dmnp(self, type: 'Enum_Option', option: 'Optional[Data_DelegatedMN ipv4 = addr.version == 4 prefix_length = self._make_uint(prefix_length, bits=8, name='prefix length') - if prefix_length > (32 if ipv4 else 128): - raise ProtocolError(f'{self.alias}: [OptNo {type}] invalid prefix length: {prefix_length}') return Schema_DelegatedMNPOption( type=type, @@ -10881,6 +11007,7 @@ def _make_opt_qos(self, type: 'Enum_Option', option: 'Optional[Data_QualityOfSer oc: 'int' = 0, attributes: 'Optional[QoSAttribute | list[Schema_QoSAttribute | tuple[Enum_QoSAttribute, dict[str, Any]] | bytes]]' = None, reserved: 'bytes' = b'\x00\x00\x00', + tc_reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_QualityOfServiceOption': """Make MH quality-of-service option. @@ -10893,26 +11020,35 @@ def _make_opt_qos(self, type: 'Enum_Option', option: 'Optional[Data_QualityOfSer attributes: Quality-of-service attributes. reserved: The three reserved octets as captured, truncated or zero- filled to 3 octets. + tc_reserved: The 2 reserved bits of the traffic class, after ``dscp``. **kwargs: Arbitrary keyword arguments. Returns: Constructed option schema. Raises: - ProtocolError: If ``sr_id``, ``dscp`` or ``oc`` does not fit its field. + ProtocolError: If ``sr_id``, ``dscp``, ``tc_reserved`` or ``oc`` does + not fit its field. Also if the option is too long for its one-octet ``Option Length``. + .. versionchanged:: 1.6.0 + + Takes ``tc_reserved``, which a data model passed as ``option`` carries + as well (:issue:`2684`). + """ if option is not None: reserved = getattr(option, 'reserved', b'\x00\x00\x00') sr_id = option.sr_id dscp = option.dscp + tc_reserved = getattr(option, 'tc_reserved', 0) oc = option.oc attributes = option.attributes else: attributes = attributes or [] dscp = self._make_uint(dscp, bits=6, name='DSCP') + tc_reserved = self._make_uint(tc_reserved, bits=2, name='traffic class reserved bits') entries = self._make_qos_attributes(attributes) length = 6 + sum(len(entry) if isinstance(entry, bytes) else len(entry.pack()) @@ -10922,7 +11058,7 @@ def _make_opt_qos(self, type: 'Enum_Option', option: 'Optional[Data_QualityOfSer type=type, length=self._make_uint(length, bits=8, name='Option Length'), sr_id=self._make_uint(sr_id, bits=8, name='service request identifier'), - tc=dscp << 2, + tc=dscp << 2 | tc_reserved, oc=self._make_uint(oc, bits=8, name='operational code'), reserved=reserved, attributes=entries, diff --git a/pcapkit/protocols/schema/internet/mh.py b/pcapkit/protocols/schema/internet/mh.py index 726556ccbe..ad4fac4bb3 100644 --- a/pcapkit/protocols/schema/internet/mh.py +++ b/pcapkit/protocols/schema/internet/mh.py @@ -62,6 +62,7 @@ from pcapkit.corekit.fields.strings import BitField, BytesField, PaddingField, StringField from pcapkit.protocols.schema.schema import EnumSchema, Schema, schema_final from pcapkit.utilities.datetime import fromtimestamp +from pcapkit.utilities.exceptions import ProtocolError from pcapkit.utilities.logging import SPHINX_TYPE_CHECKING __all__ = [ @@ -137,13 +138,24 @@ from pcapkit.protocols.protocol import ProtocolBase if SPHINX_TYPE_CHECKING: # pragma: no cover - from typing_extensions import TypedDict + from typing_extensions import NotRequired, TypedDict class ANSIKeyLengthTest(TypedDict): """Length test for ASN.1 encoded data, see - :attr:`CGAParameter.public_key_test`.""" + :attr:`CGAParameter.public_key_test`. + .. versionchanged:: 1.6.0 + + New key ``long`` (:issue:`2684`). + + """ + + #: First length octet: the length itself, or ``0x80`` plus the number of + #: length octets that follow. len: int + #: The two octets after it, which hold a long-form length; absent from a + #: test the maker sets. + long: 'NotRequired[int]' class MultiPrefixExtensionFlags(TypedDict): """Flags for :attr:`MultiPrefixExtension.flags`.""" @@ -592,6 +604,50 @@ def fb_code_selector(pkt: 'dict[str, Any]') -> 'Field': return EnumField(length=1, namespace=Enum_FlowBindingIndicationTrigger) +def cga_public_key_length(pkt: 'dict[str, Any]') -> 'int': + """Return the length of :attr:`CGAParameter.public_key`. + + Args: + pkt: Packet data. + + Returns: + The octets of the key's ASN.1 type, length and value. The key is a DER + ``SubjectPublicKeyInfo`` [:rfc:`3972#section-3`], whose length octet is + either the length or, from ``0x80``, the count of length octets after it + [ITU-T X.690 section 8.1.3]; a key of more than 127 octets takes the + latter. + + Raises: + ProtocolError: If the key runs past the CGA Parameters option it is read + from, or its length is indefinite or wider than an option can hold. + :rfc:`4866#section-5.1` lets the parameters span several options, + which are not joined here, so such a key is refused rather than read + from the octets after the option (:issue:`2684`). + + Note: + The check is skipped when ``__length__`` is absent or negative, which is + what :meth:`Schema.pack ` + leaves it as. A key test set by + :meth:`MH._make_opt_cga_param ` + carries no ``long``, and its ``len`` is the key's length less two. + + """ + test = pkt['public_key_test'] + size = test['len'] + if 'long' not in test or size < 0x80: + length = size + 2 + elif size in (0x81, 0x82): + length = 2 + (size & 0x7F) + (test['long'] >> 8 if size == 0x81 else test['long']) + else: + raise ProtocolError(f'MH: [OptNo {Enum_Option.CGA_Parameters}] invalid CGA public key ' + f'length octet: {size:#04x}') + remaining = pkt.get('__length__') + if isinstance(remaining, int) and 0 <= remaining < length: + raise ProtocolError(f'MH: [OptNo {Enum_Option.CGA_Parameters}] CGA public key of ' + f'{length} octet(s) runs past the option, with {remaining} left') + return length + + @schema_final class MH(Schema): """Header schema for MH packets. @@ -938,7 +994,15 @@ def __init__(self, type: 'Enum_CGAExtension', length: 'int', data: 'bytes') -> ' @schema_final class CGAParameter(Schema): - """Header schema for CGA Parameters.""" + """Header schema for CGA Parameters. + + .. versionchanged:: 1.6.0 + + The public key takes the long form of a DER length, and parameters whose + fixed fields or public key run past the CGA Parameters option are refused, + rather than read from the octets after it (:issue:`2684`). + + """ #: Modifier. modifier: 'Enum_CGAType' = EnumField(length=16, namespace=Enum_CGAType) @@ -947,11 +1011,12 @@ class CGAParameter(Schema): #: Collision count. collision_count: 'int' = UInt8Field() #: Public key length test. - public_key_test: 'ANSIKeyLengthTest' = ForwardMatchField(BitField(length=2, namespace={ + public_key_test: 'ANSIKeyLengthTest' = ForwardMatchField(BitField(length=4, namespace={ 'len': (8, 8), + 'long': (16, 16), })) #: Public key (ASN.1 encoded). - public_key: 'bytes' = BytesField(length=lambda pkt: pkt['public_key_test']['len'] + 2) # 2 bytes for type & length + public_key: 'bytes' = BytesField(length=cga_public_key_length) #: Extension fields. extensions: 'list[CGAExtension]' = OptionField( length=lambda pkt: pkt['length'] - 25 - len(pkt['public_key']), @@ -961,6 +1026,24 @@ class CGAParameter(Schema): eool=None, ) + @classmethod + def pre_unpack(cls, packet: 'dict[str, Any]') -> 'None': + """Refuse parameters whose fixed fields do not fit in what is left. + + Args: + packet: Packet data. + + Raises: + ProtocolError: If fewer octets are left than the 25 of the modifier, + subnet prefix and collision count [:rfc:`3972#section-3`], which + :func:`cga_public_key_length` gives its reason for refusing. + + """ + remaining = packet.get('__length__') + if isinstance(remaining, int) and 0 <= remaining < 25: + raise ProtocolError(f'MH: [OptNo {Enum_Option.CGA_Parameters}] CGA parameters of ' + f'{remaining} octet(s) cannot hold their fixed fields') + if TYPE_CHECKING: def __init__(self, modifier: 'Enum_CGAType', prefix: 'int', collision_count: 'int', public_key: 'bytes', extensions: 'list[CGAExtension | bytes]') -> 'None': ... diff --git a/tests/protocols/internet/test_internet_maker_ranges_2614_unit.py b/tests/protocols/internet/test_internet_maker_ranges_2614_unit.py index 15af14fa5b..ed3e135f7f 100644 --- a/tests/protocols/internet/test_internet_maker_ranges_2614_unit.py +++ b/tests/protocols/internet/test_internet_maker_ranges_2614_unit.py @@ -440,8 +440,8 @@ class MHMakerTests(_Case): } #: Options by maker suffix: ``{suffix: (option, base arguments, {argument: (maximum, name)})}``. - #: A prefix length's maximum is the address width, which the maker checks - #: past the field width. + #: A prefix length's maximum is the address width where the maker checks it + #: past the field width, and the field width where it does not (#2684). OPTIONS = { 'pad': ('PadN', {}, {'length': (0xFF, 'Option Length')}), 'bra': ('Binding_Refresh_Advice', {}, {'interval': (0xFFFF, 'refresh interval')}), @@ -461,16 +461,16 @@ class MHMakerTests(_Case): 'restart': ('Restart_Counter', {}, {'counter': (0xFFFFFFFF, 'restart counter')}), 'ipv4_hoa': ('IPv4_Home_Address', {}, {'prefix_length': (32, 'prefix length'), 'reserved': (0x1FF, 'reserved bits')}), - 'ipv4_ack': ('IPv4_Address_Acknowledgement', {}, {'prefix_length': (32, 'prefix length'), + 'ipv4_ack': ('IPv4_Address_Acknowledgement', {}, {'prefix_length': (0x3F, 'prefix length'), 'reserved': (0x3, 'reserved bits')}), 'nat': ('NAT_Detection', {}, {'refresh': (0xFFFFFFFF, 'refresh interval'), 'reserved': (0x7FFF, 'reserved bits')}), 'gre': ('GRE_Key_Option', {}, {'key': (0xFFFFFFFF, 'GRE key')}), 'bid': ('Binding_Identifier', {}, {'bid': (0xFFFF, 'binding identifier'), 'bid_pri': (0x7F, 'binding priority')}), - 'ipv4_hoa_req': ('IPv4_Home_Address_Request', {}, {'prefix_length': (32, 'prefix length'), + 'ipv4_hoa_req': ('IPv4_Home_Address_Request', {}, {'prefix_length': (0x3F, 'prefix length'), 'reserved': (0x3FF, 'reserved bits')}), - 'ipv4_hoa_rep': ('IPv4_Home_Address_Reply', {}, {'prefix_length': (32, 'prefix length'), + 'ipv4_hoa_rep': ('IPv4_Home_Address_Reply', {}, {'prefix_length': (0x3F, 'prefix length'), 'reserved': (0x3, 'reserved bits')}), 'ipv4_dhcp': ('IPv4_DHCP_Support_Mode', {}, {'reserved': (0x7FFF, 'reserved bits')}), 'transient': ('Transient_Binding', {}, {'lifetime': (0xFF, 'lifetime'), @@ -491,12 +491,13 @@ class MHMakerTests(_Case): 'mcast_sel': ('Dynamic_IP_Multicast_Selector', {}, {'protocol': (0xFF, 'protocol'), 'records': (0xFFFF, 'number of records'), 'reserved': (0x7F, 'reserved bits')}), - 'dmnp': ('Delegated_Mobile_Network_Prefix', {}, {'prefix_length': (128, 'prefix length'), + 'dmnp': ('Delegated_Mobile_Network_Prefix', {}, {'prefix_length': (0xFF, 'prefix length'), 'reserved': (0x7F, 'reserved bits')}), 'ams4': ('Active_Multicast_Subscription_IPv4', {}, {'igmp_type': (0xFF, 'IGMP type')}), 'ams6': ('Active_Multicast_Subscription_IPv6', {}, {'mld_type': (0xFF, 'MLD type')}), 'qos': ('Quality_of_Service', {}, {'sr_id': (0xFF, 'service request identifier'), 'dscp': (0x3F, 'DSCP'), + 'tc_reserved': (0x3, 'traffic class reserved bits'), 'oc': (0xFF, 'operational code')}), 'mcast': ('Multicast_Mobility_Option', {}, {'code': (0xFF, 'code')}), 'mcast_ack': ('Multicast_Acknowledgement_Option', {}, {'code': (0xFF, 'code'), diff --git a/tests/protocols/internet/test_mh_option_mutation_2684_unit.py b/tests/protocols/internet/test_mh_option_mutation_2684_unit.py new file mode 100644 index 0000000000..340ba7f7d5 --- /dev/null +++ b/tests/protocols/internet/test_mh_option_mutation_2684_unit.py @@ -0,0 +1,311 @@ +# -*- coding: utf-8 -*- +"""A mobility option changed in one octet is refused or rebuilds byte for byte. + +GitHub issue #2684. Option fields were read at their own widths whatever the +length said, so a length they overran or fell short of rebuilt as another; a +prefix length longer than its address parsed and then refused to rebuild; a text +identifier rebuilt the length of its text re-encoded; and the Quality-of-Service +traffic class lost its reserved bits. + +Every registered option is built from a valid example inside a Binding Refresh +Request, and each octet of it changed in turn. The cases the issue reports are +pinned by name, against what their RFCs say the octets hold. + +Every case builds its own octets in memory and reads no capture. + +""" + +import struct +import unittest +import warnings + +from tests._support import reimport_once_per_class +from tests.utilities._harness import capture + +#: Test tier (#1958): a sweep over every registered mobility option. +__tier__ = 'LATEST' + +V6 = '2001:db8::1' +V4 = '198.51.100.7' +#: A DER SEQUENCE of 159 octets, whose length takes the long form. +LONG_KEY = b'\x30\x81\x9f' + bytes(range(159)) + +#: Options a one-octet change makes the reader raise an error that is not +#: pcapkit's, by the reader: its data model has no form for the octets, a prefix +#: with host bits set or a time past datetime.MAXYEAR. Inside IPv6 the header is +#: then Raw, so the round trip holds. Found by this sweep, not fixed here. +KNOWN_CRASHES = { + 'Mobile_Network_Prefix_Option': '_read_opt_mnp', + 'Timestamp_Option': '_read_opt_timestamp', +} + + +def _ipv6(body: bytes) -> bytes: + return (bytes([0x60, 0, 0, 0]) + struct.pack('!HBB', len(body), 135, 64) + + bytes(15) + b'\x01' + bytes(15) + b'\x02' + body) + + +def _examples() -> 'dict': + """Valid arguments for the options whose defaults are not, or are too thin.""" + from pcapkit.const.mh.ani_suboption import ANISuboption as ANI + from pcapkit.const.mh.cga_extension import CGAExtension + from pcapkit.const.mh.cga_type import CGAType + from pcapkit.const.mh.flow_id_suboption import FlowIDSuboption as FID + from pcapkit.const.mh.lma_mag_suboption import LMAControlledMAGSuboption as LCMP + from pcapkit.const.mh.mn_id_subtype import MNIDSubtype + from pcapkit.const.mh.option import Option + from pcapkit.const.mh.qos_attribute import QoSAttribute as QoS + from pcapkit.protocols.data.internet.mh import CGAParameter + + fields = {'modifier': CGAType.Tag_086F_CA5E_10B2_00C9_9C8C_E001_6427_7C08, + 'prefix': 0x20010db800000000, 'collision_count': 0} # type: dict + parameter = CGAParameter(**fields, public_key=b'\x30\x02\xaa\xbb', extensions=[ + (CGAExtension.Multi_Prefix, {'prefixes': [0x20010db800000001]})]) + # a DER key past 127 octets, as a 1024-bit RSA SubjectPublicKeyInfo is + long_key = CGAParameter(**fields, public_key=LONG_KEY, extensions=[]) + return { + Option.PadN: [{'length': 4}], + Option.Authorization_Data: [{'data': bytes(range(8))}], + Option.Mobility_Header_Link_Layer_Address_option: [{'address': b'\x00\x11\x22\x33\x44\x55'}], + Option.MN_ID_OPTION_TYPE: [{}, {'subtype': MNIDSubtype.NAI, 'identifier': 'usér@example.com'}, + {'subtype': MNIDSubtype.IMSI, 'identifier': b'\x12\x34\x56\x78\x9f'}], + Option.AUTH_OPTION_TYPE: [{'spi': 0xdeadbeef, 'data': bytes(range(10))}], + Option.MESG_ID_OPTION_TYPE: [{'seconds': 0x83aa7e80, 'fraction': 0x40000000}], + Option.CGA_Parameters: [{'parameters': [parameter]}, {'parameters': [long_key]}], + Option.Signature: [{'signature': b'\xaa' * 4}], + Option.Permanent_Home_Keygen_Token: [{'token': bytes(range(8))}], + Option.Experimental_Mobility_Option: [{'data': b'\x01\x02\x03\x04'}], + Option.Binding_Authorization_Data_for_FMIPv6: [{'spi': 1, 'data': bytes(range(12))}], + Option.Service_Selection_Mobility_Option: [{'identifier': 'ims'}], + Option.Binding_Identifier: [{'bid': 5, 'address': V6}, {'bid': 5, 'address': V4}], + Option.Local_Mobility_Anchor_Address_Option: [{'address': V6}, {'address': V4}], + Option.Flow_Identification_Mobility_Option: [{'fid': 7, 'suboptions': [ + (FID.BID_Reference, {'bid': [1, 2]}), + (FID.Traffic_Selector, {'ts_format': 2, 'selector': b'\x00\x01\x02\x03'}), + (FID.Flow_Binding_Action, {'action': 11}), + (FID.Target_Care_of_Address, {'address': V6}), + (FID.Target_Care_of_Address, {'address': V4}), + (FID.PadN, {'length': 2}), (FID.Pad, {}), + ]}], + Option.Redirect_Mobility_Option: [{'ipv6': V6}, {'ipv4': V4}], + Option.Access_Network_Identifier: [{'suboptions': [ + (ANI.Network_Identifier, {'net_name': b'wifi', 'ap_name': b'\x00\x11\x22\x33\x44\x55'}), + (ANI.Geo_Location, {'raw_latitude': -1234567, 'raw_longitude': 987654}), + (ANI.Operator_Identifier, {'op_id_type': 2, 'identifier': b'example.com'}), + (ANI.Civic_Location, {'format': 0, 'location': b'GB\x01\x02'}), + (ANI.MAG_Group_Identifier, {'group_id': 9}), + (ANI.ANI_Update_Timer, {'timer': 15}), + ]}], + Option.Delegated_Mobile_Network_Prefix: [{}, {'prefix_length': 24, 'prefix': '192.0.2.0'}], + Option.Quality_of_Service: [{'sr_id': 3, 'dscp': 46, 'oc': 1, 'attributes': [ + (QoS.Per_MN_Agg_Max_DL_Bit_Rate, {'rate': 1_000_000}), + (QoS.Per_Session_Agg_Max_UL_Bit_Rate, {'service': True, 'rate': 500_000}), + (QoS.Allocation_Retention_Priority, {'priority_level': 5}), + (QoS.QoS_Traffic_Selector, {'ts_format': 2, 'selector': b'\x01\x02'}), + (QoS.QoS_Vendor_Specific_Attribute, {'vendor': 32473, 'subtype': 1, 'data': b'\xff'}), + ]}], + Option.LMA_User_Plane_Address: [{}, {'address': V6}, {'address': V4}], + Option.LMA_Controlled_MAG_Parameters: [{'suboptions': [ + (LCMP.Binding_Re_registration_Control, {'start_time': 10, 'initial_retransmission': 2, + 'max_retransmission': 30}), + (LCMP.Heartbeat_Control, {'interval': 60, 'retransmission_delay': 3, + 'max_retransmissions': 5}), + ]}], + } + + +class TestMHOptionMutation(unittest.TestCase): + """Pin byte-exact rebuilds of mobility options changed in one octet.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + catcher = warnings.catch_warnings() + catcher.__enter__() + self.addCleanup(catcher.__exit__, None, None, None) + warnings.simplefilter('ignore') + + @staticmethod + def _brr(code: 'object', kwargs: 'dict') -> bytes: + 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': [(code, kwargs)]}).data + + def _assert_identity_or_refused(self, raw: bytes, crashed: 'set[str]', name: str) -> 'object': + """Parse ``raw`` alone and inside IPv6; return the parsed MH, or None if refused.""" + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.internet.mh import MH + from pcapkit.utilities.exceptions import BaseError + + try: + mh = MH(raw, len(raw)) + except BaseError: + mh = None + except Exception: # pylint: disable=broad-except + crashed.add(name) + mh = None + else: + self.assertEqual(MH.from_data(mh.info).data, raw) + self.assertEqual(MH.from_data(mh.info.to_dict()).data, raw) + + packet = _ipv6(raw) + ipv6 = IPv6(packet, len(packet)) + self.assertEqual(IPv6.from_data(ipv6.info).data, packet) + return mh + + def test_reported_cases(self) -> None: + from pcapkit.const.mh.option import Option + from pcapkit.corekit.fields.strings import DecodedString + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.internet.mh import MH + from pcapkit.protocols.misc.raw import Raw + from pcapkit.utilities import exceptions + from pcapkit.utilities.exceptions import ProtocolError + from pcapkit.utilities.logging import logger + + examples = _examples() + # the issue's lone sub-options, so the octet changed is the first one's length + examples[Option.LMA_Controlled_MAG_Parameters] = [{'suboptions': [ + examples[Option.LMA_Controlled_MAG_Parameters][0]['suboptions'][1]]}] + examples[Option.Access_Network_Identifier] = [{'suboptions': [ + examples[Option.Access_Network_Identifier][0]['suboptions'][0]]}] + # (option, octet of it, new value, refusal or the option the RFC reads) + cases = ( + # (a) a length of 0 cannot hold the fields RFC 4283 section 3, RFC 8127 + # section 3.1.2 and RFC 6757 section 3.1.1 give + (Option.MN_ID_OPTION_TYPE, 1, 0x00, 'invalid option length'), + (Option.LMA_Controlled_MAG_Parameters, 3, 0x00, 'invalid sub-option length'), + (Option.Access_Network_Identifier, 3, 0x00, 'invalid sub-option length'), + # (b) a prefix length past the address is kept, as no RFC discards it + (Option.IPv4_Address_Acknowledgement, 3, 0xff, lambda opt: opt.prefix_length == 63), + (Option.IPv4_Home_Address_Request, 2, 0xff, lambda opt: opt.prefix_length == 63), + (Option.IPv4_Home_Address_Reply, 3, 0xff, lambda opt: opt.prefix_length == 63), + (Option.Delegated_Mobile_Network_Prefix, 3, 0xc0, lambda opt: opt.prefix_length == 192), + # (c) octets that are not UTF-8 keep their own length, RFC 5149 section 3 + (Option.Service_Selection_Mobility_Option, 2, 0xe9, + lambda opt: isinstance(opt.identifier, DecodedString) and opt.identifier.raw == b'\xe9ms'), + # (d) the traffic class is the DSCP then two reserved bits, RFC 7222 section 4.1 + (Option.Quality_of_Service, 3, 0xb9, lambda opt: (opt.dscp, opt.tc_reserved) == (46, 1)), + # (e) a public key past the option, and an extension shorter than its fields + (Option.CGA_Parameters, 28, 0x82, 'runs past the option'), + (Option.CGA_Parameters, 34, 0x00, 'invalid extension length'), + # (f) a Signature retyped as CGA Parameters, and an option retyped as Pad1 + (Option.Signature, 0, Option.CGA_Parameters, 'cannot hold their fixed fields'), + (Option.Mobile_Node_Link_local_Address_Interface_Identifier_Option, 0, Option.Pad1, + 'invalid option length'), + ) + for code, offset, value, expected in cases: + whole = self._brr(code, examples.get(code, [{}])[0]) + raw = whole[:8 + offset] + bytes([value]) + whole[8 + offset + 1:] + with self.subTest(option=code.name, offset=offset, raw=raw.hex()): + packet = _ipv6(raw) + with capture(logger) as package, capture(exceptions.logger) as module: + ipv6 = IPv6(packet, len(packet)) + self.assertEqual(package.messages + module.messages, []) + self.assertEqual(IPv6.from_data(ipv6.info).data, packet) + + if isinstance(expected, str): + with self.assertRaisesRegex(ProtocolError, expected): + MH(raw, len(raw)) + self.assertIsInstance(ipv6.payload, Raw) + self.assertRegex(str(ipv6.payload.info.error), expected) + continue + + mh = MH(raw, len(raw)) + self.assertTrue(expected(mh.info.options[code])) + self.assertEqual(MH.from_data(mh.info).data, raw) + self.assertEqual(MH.from_data(mh.info.to_dict()).data, raw) + + def test_text_identifiers_count_their_octets(self) -> None: + from pcapkit.const.mh.mn_id_subtype import MNIDSubtype + from pcapkit.const.mh.option import Option + from pcapkit.protocols.internet.mh import MH + + # RFC 4283 section 3 and RFC 5149 section 3 count octets, and NAI and the + # service identifier are UTF-8, so a non-ASCII character takes two + for code, kwargs, octets in ( + (Option.MN_ID_OPTION_TYPE, {'subtype': MNIDSubtype.NAI, 'identifier': 'usér@x'}, 8), + (Option.Service_Selection_Mobility_Option, {'identifier': 'voip.é'}, 7), + ): + with self.subTest(option=code.name): + raw = self._brr(code, kwargs) + self.assertEqual(raw[9], octets) + mh = MH(raw, len(raw)) + self.assertEqual(MH.from_data(mh.info).data, raw) + + def test_prefix_length_is_checked_against_its_field(self) -> None: + from pcapkit.const.mh.option import Option + from pcapkit.const.mh.packet import Packet + from pcapkit.corekit.multidict import OrderedMultiDict + from pcapkit.protocols.internet.mh import MH + from pcapkit.utilities.exceptions import ProtocolError + + # RFC 5555 section 3.2.1, RFC 5844 sections 3.3.1 and 3.3.2 and RFC 7148 + # section 4.1 discard no prefix length, so a caller, a hand-built model and + # a parse are all held to the field's width alone, and agree + for code, widest in ((Option.IPv4_Address_Acknowledgement, 0x3F), + (Option.IPv4_Home_Address_Request, 0x3F), + (Option.IPv4_Home_Address_Reply, 0x3F), + (Option.Delegated_Mobile_Network_Prefix, 0xFF)): + whole = self._brr(code, {}) + parsed = MH(whole, len(whole)).info.options[code] + for value in sorted({33, min(129, widest), widest, widest + 1}): + model = type(parsed)(**{**parsed.to_dict(), 'prefix_length': value}) + options = OrderedMultiDict([(code, model)]) + with self.subTest(option=code.name, prefix_length=value): + if value > widest: + with self.assertRaisesRegex(ProtocolError, 'invalid prefix length'): + self._brr(code, {'prefix_length': value}) + with self.assertRaisesRegex(ProtocolError, 'invalid prefix length'): + MH(type=Packet.Binding_Refresh_Request, next=59, chksum=b'\x00\x00', + data={'options': options}) + continue + raw = self._brr(code, {'prefix_length': value}) + self.assertEqual(MH(type=Packet.Binding_Refresh_Request, next=59, chksum=b'\x00\x00', + data={'options': options}).data, raw) + self.assertEqual(MH(raw, len(raw)).info.options[code].prefix_length, value) + + def test_cga_public_key_takes_its_der_length(self) -> None: + from pcapkit.const.mh.cga_type import CGAType + from pcapkit.const.mh.option import Option + from pcapkit.protocols.data.internet.mh import CGAParameter + from pcapkit.protocols.internet.mh import MH + + # RFC 3972 section 3 has the key a DER SubjectPublicKeyInfo, whose length + # past 127 octets is 0x80 plus the count of the octets holding it (X.690) + for key in (LONG_KEY, b'\x30\x82\x00\xa2' + bytes(162), b'\x30\x02\xaa\xbb'): + with self.subTest(key=key[:4].hex()): + raw = self._brr(Option.CGA_Parameters, {'parameters': [CGAParameter( + modifier=CGAType.Tag_086F_CA5E_10B2_00C9_9C8C_E001_6427_7C08, prefix=1, + collision_count=0, public_key=key, extensions=[])]}) + mh = MH(raw, len(raw)) + parameter, = mh.info.options[Option.CGA_Parameters].parameters + self.assertEqual(parameter.public_key, key) + self.assertEqual(len(parameter.extensions), 0) + self.assertEqual(MH.from_data(mh.info).data, raw) + + def test_every_option_changed_in_one_octet(self) -> None: + from pcapkit.protocols.internet.mh import MH + + examples = _examples() + crashed = set() # type: set[str] + for code in sorted(MH.__option__, key=int): + for kwargs in examples.get(code, [{}]): + whole = self._brr(code, kwargs) + self.assertIsNotNone(self._assert_identity_or_refused(whole, crashed, code.name), code.name) + end = 9 if whole[8] == 0 else 10 + whole[9] + for pos in range(8, end): + octet = whole[pos] + values = {0x00, 0xff, octet ^ 0x01, octet ^ 0x80, (octet + 1) & 0xff, (octet - 1) & 0xff} + if pos == 9: + values.update(range(octet + 9)) # the Option Length + for value in sorted(values - {octet}): + raw = whole[:pos] + bytes([value]) + whole[pos + 1:] + with self.subTest(option=code.name, pos=pos - 8, raw=raw.hex()): + self._assert_identity_or_refused(raw, crashed, code.name) + self.assertEqual(crashed, set(KNOWN_CRASHES)) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/internet/test_mh_unit.py b/tests/protocols/internet/test_mh_unit.py index 29802f80fe..9c9a2b5e47 100644 --- a/tests/protocols/internet/test_mh_unit.py +++ b/tests/protocols/internet/test_mh_unit.py @@ -2935,19 +2935,13 @@ def test_mh_bool_is_still_a_valid_value_for_a_code_field(self) -> None: code=True, seq=1).pack(), bytes.fromhex('010100010000')) - def test_mh_dmnp_prefix_length_range_check_survives_the_bool_guard(self) -> None: - """The prefix-length range check needs a test that does not depend on #508. - - Without #508's fix, the only thing reaching this check with an IPv4 - address is ``prefix=True`` at the default ``prefix_length=64``: the bool - is laundered into ``0.0.0.1``, and ``64 > 32`` then raises -- which is - exactly why #508's own sweep read this site as already guarded. It was - not; the ``ProtocolError`` was about the *prefix length*, not the prefix. - - With the bool refused earlier, that accident does not exercise the range - check at all, so this test reaches it with a real address instead. - Without this test, the branch would be covered only by the defect #508 - removes. + def test_mh_dmnp_prefix_length_is_checked_against_its_field(self) -> None: + """The prefix length is held to its 8-bit field, whatever the family. + + :rfc:`7148#section-4.1` has no prefix length past the address discarded, + and the reader keeps one as captured, so the maker writes it too rather + than refusing what parses (#2684). A :obj:`bool` prefix is still refused + before it can pick a family (#508). """ from pcapkit.const.mh.option import Option from pcapkit.protocols.internet.mh import MH @@ -2955,26 +2949,25 @@ def test_mh_dmnp_prefix_length_range_check_survives_the_bool_guard(self) -> None proto = object.__new__(MH) - # an IPv4 prefix caps at /32 and an IPv6 one at /128 - for prefix, prefix_length in (('198.51.100.0', 33), ('198.51.100.0', 64), - ('2001:db8:3::', 129)): - with self.subTest(prefix=prefix, prefix_length=prefix_length): - with self.assertRaises(ProtocolError) as context: - proto._make_opt_dmnp( # type: ignore[arg-type] - Option.Delegated_Mobile_Network_Prefix, - prefix_length=prefix_length, prefix=prefix) - self.assertIsInstance(context.exception, BaseError) - self.assertIn(f'invalid prefix length: {prefix_length}', - str(context.exception)) - - # and the boundary values either side are accepted - for prefix, prefix_length, length in (('198.51.100.0', 32, 6), - ('2001:db8:3::', 128, 18)): + # past the address but inside the field: written, the length by family + for prefix, prefix_length, length in (('198.51.100.0', 33, 6), ('198.51.100.0', 255, 6), + ('2001:db8:3::', 129, 18)): with self.subTest(prefix=prefix, prefix_length=prefix_length): schema = proto._make_opt_dmnp( # type: ignore[arg-type] Option.Delegated_Mobile_Network_Prefix, prefix_length=prefix_length, prefix=prefix) - self.assertEqual(schema.length, length) + self.assertEqual((schema.length, schema.prefix_length), (length, prefix_length)) + + # past the field: refused, naming the prefix length + for prefix in ('198.51.100.0', '2001:db8:3::'): + with self.subTest(prefix=prefix, prefix_length=256): + with self.assertRaisesRegex(ProtocolError, 'invalid prefix length: 256'): + proto._make_opt_dmnp( # type: ignore[arg-type] + Option.Delegated_Mobile_Network_Prefix, prefix_length=256, prefix=prefix) + + with self.assertRaisesRegex(BaseError, 'invalid mobile network prefix'): + proto._make_opt_dmnp( # type: ignore[arg-type] + Option.Delegated_Mobile_Network_Prefix, prefix_length=24, prefix=True) def test_mh_bid_rejects_an_out_of_range_binding_priority(self) -> None: """``BID-PRI`` is a 7-bit field, so ``0x80`` cannot be packed.