diff --git a/docs/source/contributing/conventions/utilities-or-corekit.rst b/docs/source/contributing/conventions/utilities-or-corekit.rst index 341ed2c144..ebe6d27e79 100644 --- a/docs/source/contributing/conventions/utilities-or-corekit.rst +++ b/docs/source/contributing/conventions/utilities-or-corekit.rst @@ -84,6 +84,14 @@ through them, never through an ``unpack`` override of its own or a length callab clamped to the octets present, which hides the cut from the record. Ruled in review of :pr:`2682` (:issue:`2680`), and applied to option areas on :issue:`2694`. +A field callback tells a value the cut left as zero from one read through the same +record: :meth:`Schema.unpack ` gives +it to the fields after the one the data ends inside, as ``__short_read__`` in their +packet. A callback that would refuse such a zero leaves it to the short read, and +refuses a value read with a :exc:`~pcapkit.utilities.exceptions.ProtocolError`, as +the ``Opt Data Len`` helpers of +:mod:`pcapkit.protocols.schema.internet.ipv6_option` do (:issue:`2727`). + **A helper in one protocol's own schema module moves there too once code outside that protocol uses it and it is generic enough to reuse**, so that another implementor can find it; a helper only its own protocol calls stays beside the schema. Ruled on diff --git a/pcapkit/protocols/schema/internet/ipv6_option.py b/pcapkit/protocols/schema/internet/ipv6_option.py index 3023c9557a..ac851a5bb9 100644 --- a/pcapkit/protocols/schema/internet/ipv6_option.py +++ b/pcapkit/protocols/schema/internet/ipv6_option.py @@ -23,7 +23,7 @@ from pcapkit.corekit.fields.ipaddress import IPv4AddressField, IPv6AddressField from pcapkit.corekit.fields.misc import NoValueField, SchemaField from pcapkit.corekit.fields.strings import BytesField -from pcapkit.utilities.exceptions import FieldValueError +from pcapkit.utilities.exceptions import FieldValueError, ProtocolError __all__ = [ 'rpl_opt_sub_tlv_len', 'mpl_opt_seed_id_len', 'smf_i_dpd_id_len', @@ -38,6 +38,23 @@ from pcapkit.protocols.schema.schema import EnumMeta, Schema +def _opt_data_len_cut(pkt: 'dict[str, Any]') -> 'bool': + """Whether the data ends before an option's ``Opt Data Len``. + + Args: + pkt: Option unpacked schema. + + Returns: + Whether the field :meth:`Schema.unpack + ` gives as the one the + data ends inside is ``len``, so that its value is the zero a capture cut + leaves rather than a length read (:issue:`2727`). + + """ + record = pkt.get('__short_read__') + return record is not None and record[0] == 'len' + + def rpl_opt_sub_tlv_len(pkt: 'dict[str, Any]') -> 'int': """Return RPL option sub-TLV length. @@ -87,17 +104,27 @@ def smf_i_dpd_id_len(pkt: 'dict[str, Any]', *, prefix: 'str') -> 'int': prefix: Name of the extension header, leading the error message. Returns: - SMF I-DPD identifier length. + SMF I-DPD identifier length, or zero if the data ends before + ``Opt Data Len``, which leaves the option to the short read. Raises: - FieldValueError: If ``Opt Data Len`` on the wire is too short to hold + ProtocolError: 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. + .. versionchanged:: 1.6.0 + + A length the data ends before is left to the short read, and one too + short raises :exc:`~pcapkit.utilities.exceptions.ProtocolError` rather + than :exc:`~pcapkit.utilities.exceptions.FieldValueError` + (:issue:`2727`). + """ length = pkt['len'] - (1 if pkt['info']['type'] == 0 else (pkt['info']['len'] + 2)) if length < 0: - raise FieldValueError(f'{prefix}: invalid SMF I-DPD option length: {pkt["len"]}') + if _opt_data_len_cut(pkt): + return 0 + raise ProtocolError(f'{prefix}: invalid SMF I-DPD option length: {pkt["len"]}') return length @@ -120,8 +147,8 @@ def smf_dpd_data_selector(pkt: 'dict[str, Any]', *, prefix: 'str', wrapped ``SMFHashBasedDPDOption`` instance. Raises: - FieldValueError: If ``Opt Data Len`` is ``0``, or if no schema is - registered for the DPD mode. + ProtocolError: If ``Opt Data Len`` is ``0`` on the wire. + FieldValueError: If no schema is registered for the DPD mode. Note: The field is sized ``Opt Data Len + 2`` rather than ``Opt Data Len``. @@ -133,11 +160,26 @@ def smf_dpd_data_selector(pkt: 'dict[str, Any]', *, prefix: 'str', :class:`~pcapkit.corekit.fields.collections.OptionField` would then mis-count against the option area. + Where the data ends before ``Opt Data Len``, the forward match reads + it as zero, and the option is the short read of the schema chosen by + the mode bit, likewise zero. + + .. versionchanged:: 1.6.0 + + A length the data ends before is left to the short read, and a zero + on the wire raises :exc:`~pcapkit.utilities.exceptions.ProtocolError` + rather than :exc:`~pcapkit.utilities.exceptions.FieldValueError` + (:issue:`2727`). + """ # NOTE: With ``Opt Data Len`` 0 the option has no data octet, so the mode - # bit forward-matched above belongs to whatever follows the option. - if pkt['test']['len'] == 0: - raise FieldValueError(f'{prefix}: invalid SMF DPD option length: 0') + # bit forward-matched above belongs to whatever follows the option. Unless + # the data ends inside the forward match before ``Opt Data Len``, the + # option's second octet, which then reads as zero (#2727). + record = pkt.get('__short_read__') + cut = record is not None and record[0] == 'test' and record[1] < 2 + if pkt['test']['len'] == 0 and not cut: + raise ProtocolError(f'{prefix}: invalid SMF DPD option length: 0') mode = Enum_SMFDPDMode.get(pkt['test']['mode']) schema: 'Optional[Type[Schema]]' = base.registry[mode] @@ -268,18 +310,28 @@ def calipso_pad_len(pkt: 'dict[str, Any]', *, prefix: 'str') -> 'int': prefix: Name of the extension header, leading the error message. Returns: - CALIPSO option padding length. + CALIPSO option padding length, or zero if the data ends before + ``Opt Data Len``, which leaves the option to the short read. Raises: - FieldValueError: If ``Opt Data Len`` on the wire is too short to hold + ProtocolError: If ``Opt Data Len`` on the wire is too short to hold the fixed header and the compartment bitmap declared by ``cmpt_len``, which would otherwise underflow the padding length below zero. + .. versionchanged:: 1.6.0 + + A length the data ends before is left to the short read, and one too + short raises :exc:`~pcapkit.utilities.exceptions.ProtocolError` rather + than :exc:`~pcapkit.utilities.exceptions.FieldValueError` + (:issue:`2727`). + """ length = pkt['len'] - 8 - pkt['cmpt_len'] * 4 if length < 0: - raise FieldValueError(f'{prefix}: invalid CALIPSO option length: {pkt["len"]}') + if _opt_data_len_cut(pkt): + return 0 + raise ProtocolError(f'{prefix}: invalid CALIPSO option length: {pkt["len"]}') return length @@ -291,16 +343,26 @@ def mpl_opt_pad_len(pkt: 'dict[str, Any]', *, prefix: 'str') -> 'int': prefix: Name of the extension header, leading the error message. Returns: - MPL option padding length. + MPL option padding length, or zero if the data ends before + ``Opt Data Len``, which leaves the option to the short read. Raises: - FieldValueError: If ``Opt Data Len`` on the wire is too short to hold + ProtocolError: If ``Opt Data Len`` on the wire is too short to hold the fixed header and the Seed-ID declared by ``flags.type``, which would otherwise underflow the padding length below zero. + .. versionchanged:: 1.6.0 + + A length the data ends before is left to the short read, and one too + short raises :exc:`~pcapkit.utilities.exceptions.ProtocolError` rather + than :exc:`~pcapkit.utilities.exceptions.FieldValueError` + (:issue:`2727`). + """ length = pkt['len'] - 2 - (0 if pkt['flags']['type'] == 0 else mpl_opt_seed_id_len(pkt, prefix=prefix)) if length < 0: - raise FieldValueError(f'{prefix}: invalid MPL option length: {pkt["len"]}') + if _opt_data_len_cut(pkt): + return 0 + raise ProtocolError(f'{prefix}: invalid MPL option length: {pkt["len"]}') return length diff --git a/pcapkit/protocols/schema/schema.py b/pcapkit/protocols/schema/schema.py index 9a5ea5a7da..2d21a96a1f 100644 --- a/pcapkit/protocols/schema/schema.py +++ b/pcapkit/protocols/schema/schema.py @@ -738,6 +738,21 @@ def unpack(cls, data: 'bytes | IO[bytes]', remainder itself rather than the same number of octets from beyond it, and record a cut inside it (:issue:`2685`). + The fields after the one the data ends inside find it in ``packet`` + as ``__short_read__``: the ``(name, octets read)`` pair recorded as + the short read, or that of a + :class:`~pcapkit.corekit.fields.misc.ForwardMatchField` the data + ends inside first, which the record leaves out as it consumes + nothing. So a field callback can tell a value the end of the data + left as zero from one read, and leave such a value to the short + read rather than refuse it. The key is cleared as unpacking starts, + so a nested schema finds only its own. + + .. versionchanged:: 1.6.0 + + The field the data ends inside is given to the fields after it, as + ``__short_read__`` in ``packet`` (:issue:`2727`). + """ # force cast arg type since decorator changed their signatures if TYPE_CHECKING: @@ -745,6 +760,10 @@ def unpack(cls, data: 'bytes | IO[bytes]', length = cast('int', length) packet = cast('dict[str, Any]', packet) + # NOTE: ``packet`` may be an enclosing schema's, copied into a nested + # context, so its short read is not this schema's (#2727). + packet.pop('__short_read__', None) + self = cls.__new__(cls) for field in self.__fields__.values(): field = field(packet) @@ -779,7 +798,7 @@ def unpack(cls, data: 'bytes | IO[bytes]', byte = data.read(length) self.__buffer__[field.name] = byte - self._note_short_read(field.name, length, byte) + self._note_short_read(field.name, length, byte, packet) packet[field.name] = byte packet['__length__'] -= length @@ -801,7 +820,7 @@ def unpack(cls, data: 'bytes | IO[bytes]', byte = data.read(min(length, sys.maxsize)) self.__buffer__[field.name] = byte if not isinstance(field, (ForwardMatchField, OptionField)): - self._note_short_read(field.name, length, byte) + self._note_short_read(field.name, length, byte, packet) value = field.unpack(byte, packet.copy()) if isinstance(field, SchemaField) and isinstance(value, Schema): @@ -843,6 +862,12 @@ def unpack(cls, data: 'bytes | IO[bytes]', # preceding octet as their first. data.seek(-len(byte), io.SEEK_CUR) self.__buffer__[field.name] = b'' + + # NOTE: Left out of the record, but the values it matched past the + # end of the data are zeros all the same, so the fields after it + # are told (#2727). + if len(byte) < length: + packet.setdefault('__short_read__', (field.name, len(byte))) elif isinstance(field, OptionField) and field.option_padding > 0: # NOTE: The option list ended before the declared area did, and # ``option_padding`` is the rest of that area; give back as much @@ -856,10 +881,10 @@ def unpack(cls, data: 'bytes | IO[bytes]', self.__buffer__[field.name] = byte[:kept] packet['__length__'] -= consumed if kept == len(byte): - self._note_short_read(field.name, length, byte) + self._note_short_read(field.name, length, byte, packet) else: if isinstance(field, OptionField): - self._note_short_read(field.name, length, byte) + self._note_short_read(field.name, length, byte, packet) packet['__length__'] -= length if packet['__length__'] < 0: @@ -1187,17 +1212,24 @@ def __delattr__(self, name: 'str') -> 'None': return return super().__delattr__(name) - def _note_short_read(self, name: 'str', length: 'int', byte: 'bytes') -> 'None': + def _note_short_read(self, name: 'str', length: 'int', byte: 'bytes', + packet: 'Optional[dict[str, Any]]' = None) -> 'None': """Record the first field the data ends inside, for :meth:`_cut_short_read`. Args: name: Field name. length: Octets the field declares. byte: Octets actually read for it. + packet: Unpacked data, where the fields after it find the record, + unless a forward match the data ended inside is there first + (see :meth:`unpack`). Without it, the record is kept on the + schema alone. """ if len(byte) < length and '__short_read__' not in self.__dict__: self.__dict__['__short_read__'] = (name, len(byte)) + if packet is not None: + packet.setdefault('__short_read__', self.__dict__['__short_read__']) def _cut_short_read(self) -> 'None': """Cut the packed buffers back to the octets a short read consumed. diff --git a/tests/protocols/internet/test_ipv6_extension_unit.py b/tests/protocols/internet/test_ipv6_extension_unit.py index 0e30ed84b2..f0325ed6a7 100644 --- a/tests/protocols/internet/test_ipv6_extension_unit.py +++ b/tests/protocols/internet/test_ipv6_extension_unit.py @@ -1422,7 +1422,7 @@ def _assert_option_schema_helpers_and_post_process_branches(self, schema: types. from pcapkit.corekit.fields.ipaddress import IPv4AddressField, IPv6AddressField from pcapkit.corekit.fields.misc import NoValueField, SchemaField from pcapkit.corekit.fields.strings import BytesField - from pcapkit.utilities.exceptions import FieldValueError + from pcapkit.utilities.exceptions import FieldValueError, ProtocolError self.assertEqual(schema.mpl_opt_seed_id_len({'flags': {'type': 0}}), 0) self.assertEqual(schema.mpl_opt_seed_id_len({'flags': {'type': 1}}), 2) @@ -1439,10 +1439,11 @@ def _assert_option_schema_helpers_and_post_process_branches(self, schema: types. 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'): + # unguarded subtraction reached :func:`struct.calcsize` as ``'-1s'``. A + # ``ProtocolError``, so that a header refused for it is refused quietly (#2727). + with self.assertRaisesRegex(ProtocolError, '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'): + with self.assertRaisesRegex(ProtocolError, '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): @@ -2110,14 +2111,15 @@ def _assert_identification_based_dpd_option_rejects_underflowing_length(self, pr 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`. + through :mod:`pcapkit.utilities.exceptions`. It is a ``ProtocolError``, so + that a header refused for it is refused quietly (#2727). """ - from pcapkit.utilities.exceptions import FieldValueError + from pcapkit.utilities.exceptions import ProtocolError raw = bytes.fromhex('3b00080000000000') - with self.assertRaisesRegex(FieldValueError, 'invalid SMF (I-)?DPD option length'): + with self.assertRaisesRegex(ProtocolError, 'invalid SMF (I-)?DPD option length'): with time_limit(): protocol_cls(raw, extension=True) @@ -2143,17 +2145,18 @@ def _assert_calipso_option_rejects_underflowing_length(self, protocol_cls: type) length to ``-8``. That reached :func:`struct.calcsize` as the template ``'-8s'`` 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`. + through :mod:`pcapkit.utilities.exceptions`. It is a ``ProtocolError``, so + that a header refused for it is refused quietly (#2727). """ - from pcapkit.utilities.exceptions import FieldValueError + from pcapkit.utilities.exceptions import ProtocolError # next(1) hdr_ext_len(1)=1 -> 16-octet extension header; option # type(1)=CALIPSO(0x07) len(1)=0, then domain(4) cmpt_len(1) level(1) # checksum(2) all zero, with four trailing zero octets. raw = bytes.fromhex('3b0007000000000000000000000000') - with self.assertRaisesRegex(FieldValueError, 'invalid CALIPSO option length'): + with self.assertRaisesRegex(ProtocolError, 'invalid CALIPSO option length'): with time_limit(): protocol_cls(raw, extension=True) @@ -2178,17 +2181,18 @@ def _assert_mpl_option_rejects_underflowing_length(self, protocol_cls: type) -> :func:`struct.calcsize` as the template ``'-2s'`` 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`. + :mod:`pcapkit.utilities.exceptions`. It is a ``ProtocolError``, so that a + header refused for it is refused quietly (#2727). """ - from pcapkit.utilities.exceptions import FieldValueError + from pcapkit.utilities.exceptions import ProtocolError # next(1) hdr_ext_len(1)=0 -> 8-octet extension header; option # type(1)=MPL_Option(0x6d) len(1)=0, flags(1)=0 (Seed-ID type 0), # seq(1)=0, with two trailing zero octets. raw = bytes.fromhex('3b006d0000000000') - with self.assertRaisesRegex(FieldValueError, 'invalid MPL option length'): + with self.assertRaisesRegex(ProtocolError, 'invalid MPL option length'): with time_limit(): protocol_cls(raw, extension=True) diff --git a/tests/protocols/internet/test_ipv6_option_helpers_1519_unit.py b/tests/protocols/internet/test_ipv6_option_helpers_1519_unit.py index e89ba62d43..d0662d7105 100644 --- a/tests/protocols/internet/test_ipv6_option_helpers_1519_unit.py +++ b/tests/protocols/internet/test_ipv6_option_helpers_1519_unit.py @@ -35,22 +35,26 @@ ('pcapkit.protocols.schema.internet.ipv6_opts', 'IPv6-Opts'), ) -#: ``(helper, packet, message after the prefix)`` for every ``raise`` in the -#: helpers whose packet alone triggers it. +#: ``(helper, packet, message after the prefix, error)`` for every ``raise`` in +#: the helpers whose packet alone triggers it. An ``Opt Data Len`` too short is a +#: ``ProtocolError``, so that a header refused for it is refused quietly (#2727). ERRORS = ( - ('mpl_opt_seed_id_len', {'flags': {'type': 4}}, 'invalid MPL Seed-ID type: 4'), + ('mpl_opt_seed_id_len', {'flags': {'type': 4}}, 'invalid MPL Seed-ID type: 4', 'FieldValueError'), ('smf_i_dpd_id_len', {'len': 0, 'info': {'type': 0, 'len': 0}}, - 'invalid SMF I-DPD option length: 0'), + 'invalid SMF I-DPD option length: 0', 'ProtocolError'), ('smf_i_dpd_id_len', {'len': 3, 'info': {'type': 2, 'len': 3}}, - 'invalid SMF I-DPD option length: 3'), + 'invalid SMF I-DPD option length: 3', 'ProtocolError'), ('smf_dpd_data_selector', {'test': {'len': 0, 'mode': 0}}, - 'invalid SMF DPD option length: 0'), - ('smf_i_dpd_tid_selector', {'info': {'type': 0, 'len': 1}}, 'invalid TaggerID length: 1'), - ('smf_i_dpd_tid_selector', {'info': {'type': 2, 'len': 4}}, 'invalid TaggerID length: 4'), - ('smf_i_dpd_tid_selector', {'info': {'type': 3, 'len': 16}}, 'invalid TaggerID length: 16'), - ('calipso_pad_len', {'len': 11, 'cmpt_len': 1}, 'invalid CALIPSO option length: 11'), - ('mpl_opt_pad_len', {'len': 3, 'flags': {'type': 1}}, 'invalid MPL option length: 3'), - ('mpl_opt_pad_len', {'len': 1, 'flags': {'type': 0}}, 'invalid MPL option length: 1'), + 'invalid SMF DPD option length: 0', 'ProtocolError'), + ('smf_i_dpd_tid_selector', {'info': {'type': 0, 'len': 1}}, 'invalid TaggerID length: 1', + 'FieldValueError'), + ('smf_i_dpd_tid_selector', {'info': {'type': 2, 'len': 4}}, 'invalid TaggerID length: 4', + 'FieldValueError'), + ('smf_i_dpd_tid_selector', {'info': {'type': 3, 'len': 16}}, 'invalid TaggerID length: 16', + 'FieldValueError'), + ('calipso_pad_len', {'len': 11, 'cmpt_len': 1}, 'invalid CALIPSO option length: 11', 'ProtocolError'), + ('mpl_opt_pad_len', {'len': 3, 'flags': {'type': 1}}, 'invalid MPL option length: 3', 'ProtocolError'), + ('mpl_opt_pad_len', {'len': 1, 'flags': {'type': 0}}, 'invalid MPL option length: 1', 'ProtocolError'), ) #: The helpers each header binds, and the keywords it binds them with beyond @@ -85,15 +89,15 @@ def test_shared_module_holds_every_helper(self) -> None: self.assertEqual(sorted(shared.__all__), sorted([*BOUND, 'quick_start_data_selector', *UNBOUND])) def test_shared_helpers_report_the_prefix_they_are_given(self) -> None: - from pcapkit.utilities.exceptions import FieldValueError + from pcapkit.utilities import exceptions shared = self._shared() hopopt = importlib.import_module(HEADERS[0][0]) for prefix in ('HOPOPT', 'IPv6-Opts', 'any-header'): - for name, packet, message in ERRORS: + for name, packet, message, error in ERRORS: extra = {'base': hopopt.SMFDPDOption} if name == 'smf_dpd_data_selector' else {} with self.subTest(prefix=prefix, helper=name, packet=packet): - with self.assertRaises(FieldValueError) as caught: + with self.assertRaises(getattr(exceptions, error)) as caught: getattr(shared, name)(copy.deepcopy(packet), prefix=prefix, **extra) self.assertEqual(str(caught.exception), f'{prefix}: {message}') @@ -165,12 +169,12 @@ def test_shared_selectors_return_the_schemas_they_are_given(self) -> None: self.assertEqual(field.length, length) def test_headers_keep_their_exact_error_text(self) -> None: - from pcapkit.utilities.exceptions import FieldValueError + from pcapkit.utilities import exceptions for module, prefix in self._headers(): - for name, packet, message in ERRORS: + for name, packet, message, error in ERRORS: with self.subTest(module=module.__name__, helper=name, packet=packet): - with self.assertRaises(FieldValueError) as caught: + with self.assertRaises(getattr(exceptions, error)) as caught: getattr(module, name)(copy.deepcopy(packet)) self.assertEqual(str(caught.exception), f'{prefix}: {message}') diff --git a/tests/protocols/internet/test_ipv6_options_roundtrip_unit.py b/tests/protocols/internet/test_ipv6_options_roundtrip_unit.py index d3213a3306..34d125af32 100644 --- a/tests/protocols/internet/test_ipv6_options_roundtrip_unit.py +++ b/tests/protocols/internet/test_ipv6_options_roundtrip_unit.py @@ -111,14 +111,15 @@ def test_unassigned_quick_start_function_is_kept_as_raw_data(self) -> None: self.assertEqual(opt.data, bytes.fromhex('534012345678')) def test_zero_length_smf_dpd_is_rejected_in_both_modes(self) -> None: - from pcapkit.utilities.exceptions import FieldValueError + # a ``ProtocolError``, so that a header refused for it is refused quietly (#2727) + from pcapkit.utilities.exceptions import ProtocolError for cls in self._protocols(): # the octet after ``Opt Data Len`` 0 is the next option's type, so its # top bit must not be taken as the H-DPD (``80``) or I-DPD mode bit for mode, hexstr in (('H-DPD', '3b00080080000000'), ('I-DPD', '3b00080000000000')): with self.subTest(protocol=cls.__name__, mode=mode): - with self.assertRaisesRegex(FieldValueError, 'invalid SMF DPD option length: 0'): + with self.assertRaisesRegex(ProtocolError, 'invalid SMF DPD option length: 0'): self._parse(cls, bytes.fromhex(hexstr)) def test_ip_dff_data_length_is_three(self) -> None: diff --git a/tests/protocols/internet/test_option_length_cut_2727_unit.py b/tests/protocols/internet/test_option_length_cut_2727_unit.py new file mode 100644 index 0000000000..9a284202d4 --- /dev/null +++ b/tests/protocols/internet/test_option_length_cut_2727_unit.py @@ -0,0 +1,393 @@ +# -*- coding: utf-8 -*- +"""A cut after an IPv6 option's type octet reads as a short read, quietly (#2727). + +In HOPOPT and IPv6-Opts, a capture cut that leaves only the type octet of a +CALIPSO, SMF_DPD or MPL option logged ``FieldValueError: ... invalid option +length: 0``, at CRITICAL and again from ``beholder`` inside IPv6. The callbacks +sizing those options from ``Opt Data Len`` refused the zero the cut left in place +of a length, before the short read #2694 extended to option areas could record +the cut. A cut one octet later was clean. + +``Schema.unpack`` gives the fields after the one the data ends inside the short +read, as ``__short_read__`` in their packet, a forward match included. A callback +whose ``Opt Data Len`` the data ends before leaves the option to the short read, +and the option is then kept or refused by #2694's rule: kept if its parser takes +the zeros and its maker writes the octets captured back, else refused quietly. +CALIPSO and SMF_DPD are kept; MPL's parser refuses a length below two, as Router +Alert's does. A length on the wire too short is refused with a ``ProtocolError``, +which a handled refusal keeps quiet (#2588), where a ``FieldValueError`` logged. + +Every case builds its own octets and reads no capture. Classes are imported inside +each test, after :func:`~tests._support.reimport_once_per_class`. + +""" + +import contextlib +import struct +import unittest +import warnings +from typing import TYPE_CHECKING + +from tests._support import reimport_once_per_class, time_limit +from tests.utilities._harness import capture + +if TYPE_CHECKING: + from typing import Any, Callable, Iterator, Optional + +#: Test tier (#1958): a sweep that cuts every IPv6 option at every offset. +__tier__ = 'LATEST' + +#: The octets of a HOPOPT or IPv6-Opts header IPv6 keeps as its own layer: its +#: ``__length_hint__``, under which it is kept as raw octets (#1170). +OPTIONS_HINT = 8 + +#: The SchemaWarning a layer kept with its header cut short gives, which +#: ``Protocol._parse_next_layer`` documents and #2526 pins: the header's fields run +#: past the length it was handed. It is not a refusal. +CUT_HEADER_WARNING = 'packet length < 0' + + +def ipv6(code: int, body: bytes, length: 'Optional[int]' = None) -> bytes: + """An IPv6 packet carrying ``body`` as next header ``code``, its Payload Length + ``length``, or that of ``body`` if not given.""" + declared = len(body) if length is None else length + return struct.pack('!IHBB', 6 << 28, declared, code, 64) + bytes(32) + body + + +@contextlib.contextmanager +def reported() -> 'Iterator[list[str]]': + """What the block reports, on the ``pcapkit`` logger and on :mod:`warnings`.""" + from pcapkit.utilities.logging import logger + + messages = [] # type: list[str] + with capture(logger) as logs, warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + yield messages + messages.extend(f'{level}: {text}' for level, text in logs.messages) + messages.extend(f'warned: {item.message}' for item in caught) + + +def refusals(messages: 'list[str]') -> 'list[str]': + """``messages`` but the warnings a header cut short gives.""" + return [text for text in messages if CUT_HEADER_WARNING not in text] + + +def variants() -> 'list[tuple[Any, dict[str, Any]]]': + """The options the issue names, with their defaults and with the fields their + lengths count filled in.""" + import ipaddress + + from pcapkit.const.ipv6.option import Option + from pcapkit.const.ipv6.smf_dpd_mode import SMFDPDMode + from pcapkit.const.ipv6.tagger_id import TaggerID + + return [ + (Option.CALIPSO, {}), + (Option.CALIPSO, {'domain': 0x12345678, 'level': 5, 'checksum': b'\xab\xcd', + 'bitmap': bytes([1, 2, 3, 4]), 'pad': b'\x01\x02'}), + (Option.SMF_DPD, {}), + (Option.SMF_DPD, {'tid_type': TaggerID.IPv4, 'tid': ipaddress.ip_address('192.0.2.1'), + 'id': b'\x01\x02'}), + (Option.SMF_DPD, {'tid_type': TaggerID.IPv6, 'tid': ipaddress.ip_address('2001:db8::1'), + 'id': b'\x01'}), + (Option.SMF_DPD, {'tid_type': TaggerID.DEFAULT, 'tid': b'\xaa\xbb', 'id': b'\x01'}), + (Option.SMF_DPD, {'mode': SMFDPDMode.H_DPD, 'hav': b'\x81\x02\x03'}), + (Option.MPL_Option, {}), + (Option.MPL_Option, {'seed_type': 1, 'seed': 0x1234, 'seq': 7}), + (Option.MPL_Option, {'seed_type': 2, 'seed': 0x0102030405060708}), + (Option.MPL_Option, {'seed_type': 3, 'seed': 1 << 100}), + ] + + +class OptionLengthCutTests(unittest.TestCase): + """Pin a cut before an option's ``Opt Data Len`` as a quiet short read.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def headers(self) -> 'list[tuple[Any, int, Any]]': + """HOPOPT and IPv6-Opts, with the next header code IPv6 gives each and the + key IPv6 keeps each under.""" + from pcapkit.const.ipv6.extension_header import ExtensionHeader + from pcapkit.protocols.internet.hopopt import HOPOPT + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + + return [(HOPOPT, 0, ExtensionHeader.HOPOPT), (IPv6_Opts, 60, ExtensionHeader.IPv6_Opts)] + + def build(self, cls: 'Any', code: 'Any', kwargs: 'dict[str, Any]', lead: bool) -> 'tuple[bytes, int]': + """A whole ``cls`` header carrying option ``code``, after a PadN of six octets + if ``lead``, and the offset the option starts at.""" + from pcapkit.const.ipv6.option import Option + + options = [(Option.PadN, {'length': 4})] * lead + [(code, kwargs)] + whole = bytes(cls(next=59, options=options, payload=b'')) + start = 2 + for key, option in cls(whole).info.options.items(multi=True): + if key == code: + break + start += option.length + return whole, start + + def assertRebuildsQuietly(self, cls: 'Any', layer: 'Any', raw: bytes, # pylint: disable=invalid-name + reparsed: bool = False) -> None: + """``cls`` rebuilt from ``layer``'s info and its dict gives ``raw``, reporting nothing. + + A ``reparsed`` rebuild parses a header kept cut short again, as an IPv6 + packet does its extension headers, which then warns as its parse did. + + """ + for path, data in (('info', layer.info), ('dict', layer.info.to_dict())): + with self.subTest(path=path): + with reported() as messages: + rebuilt = cls.from_data(data) + if reparsed: + messages = refusals(messages) + self.assertEqual((rebuilt.data.hex(), messages), (raw.hex(), [])) + + def parse(self, build: 'Callable[[], Any]') -> 'tuple[Any, list[str]]': + """What ``build`` returns, and what it reported.""" + with time_limit(), reported() as messages: + layer = build() + return layer, messages + + def test_every_option_at_every_cut_reads_short_or_is_refused_quietly(self) -> None: + # every registered option, first in its header and after a PadN, cut at + # every offset: on its own, a short read that rebuilds or a ProtocolError; + # inside IPv6, nothing reported, and the packet rebuilds + from pcapkit.const.ipv6.option import Option + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.utilities.exceptions import ProtocolError + + overrides = {Option.ILNP_Nonce: {'nonce': 0xFFFFFF}, + Option.Line_Identification_Option: {'id': b'line-1'}} + for cls, next_header, _ in self.headers(): + cases = [(code, overrides.get(code, {})) for code in cls.__option__] + variants() + for code, kwargs in cases: + for lead in (False, True): + whole, _ = self.build(cls, code, kwargs, lead) + for cut in range(1, len(whole)): + raw = whole[:cut] + with self.subTest(header=cls.__name__, option=whole.hex(), cut=cut): + try: + header, messages = self.parse(lambda: cls(raw)) # pylint: disable=cell-var-from-loop + except ProtocolError: + pass + else: + self.assertEqual((header.data, refusals(messages)), (raw, [])) + self.assertIsNotNone(header.info.get('__short_read__')) + self.assertRebuildsQuietly(cls, header, raw) + + packet = ipv6(next_header, raw, len(whole)) + ip6, messages = self.parse(lambda: IPv6(packet)) # pylint: disable=cell-var-from-loop + self.assertEqual((ip6.data, refusals(messages)), (packet, [])) + self.assertRebuildsQuietly(IPv6, ip6, packet, reparsed=True) + + def test_a_cut_after_the_type_reads_short(self) -> None: + # a CALIPSO or SMF_DPD option cut after its type is a short read, as one cut + # an octet later is, on its own and inside IPv6 + from pcapkit.const.ipv6.option import Option + from pcapkit.protocols.internet.ipv6 import IPv6 + + for cls, next_header, key in self.headers(): + for code, kwargs in variants(): + if code == Option.MPL_Option: + continue + for lead in (False, True): + whole, start = self.build(cls, code, kwargs, lead) + self.assertEqual(whole[start], code) + for cut in (start + 1, start + 2): + raw = whole[:cut] + with self.subTest(header=cls.__name__, option=whole.hex(), cut=cut): + header, messages = self.parse(lambda: cls(raw)) # pylint: disable=cell-var-from-loop + # NOTE: An SMF_DPD option the data ends in before its + # ``Opt Data Len`` sizes its identifier schema by the zero + # read there, at two octets, which its fields run past. + if code != Option.SMF_DPD or cut > start + 1: + self.assertEqual(messages, []) + self.assertEqual((header.data, refusals(messages)), (raw, [])) + self.assertEqual(header.info.get('__short_read__'), ('options', cut - 2)) + self.assertEqual(list(header.info.options)[-1], code) + self.assertRebuildsQuietly(cls, header, raw) + + packet = ipv6(next_header, raw, len(whole)) + ip6, messages = self.parse(lambda: IPv6(packet)) # pylint: disable=cell-var-from-loop + self.assertEqual((ip6.data, refusals(messages)), (packet, [])) + if cut >= OPTIONS_HINT: + kept = ip6.extension_headers.get(key) + self.assertIsNotNone(kept) + self.assertEqual(kept.info.get('__short_read__'), ('options', cut - 2)) + self.assertRebuildsQuietly(IPv6, ip6, packet, reparsed=True) + + def test_an_mpl_option_cut_after_the_type_is_refused_quietly(self) -> None: + # its parser refuses the zero length the cut leaves, as Router Alert's does + # (#2694), so the header is refused, and kept as Raw inside IPv6, quietly + from pcapkit.const.ipv6.option import Option + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.misc.raw import Raw + from pcapkit.utilities.exceptions import ProtocolError + + error = r'\[OptNo 109\] option cut short: .*\[OptNo 109\] invalid format' + for cls, next_header, _ in self.headers(): + for code, kwargs in variants(): + if code != Option.MPL_Option: + continue + for lead in (False, True): + whole, start = self.build(cls, code, kwargs, lead) + self.assertEqual(whole[start], code) + raw = whole[:start + 1] + with self.subTest(header=cls.__name__, option=whole.hex()): + with reported(), self.assertRaisesRegex(ProtocolError, error): + cls(raw) + + packet = ipv6(next_header, raw, len(whole)) + ip6, messages = self.parse(lambda: IPv6(packet)) # pylint: disable=cell-var-from-loop + self.assertEqual((ip6.data, messages), (packet, [])) + self.assertIs(type(ip6.payload), Raw) + self.assertRebuildsQuietly(IPv6, ip6, packet) + + def test_a_length_on_the_wire_too_short_is_refused_quietly(self) -> None: + # whole headers whose option declares an ``Opt Data Len`` too short for its + # fields: refused with a ProtocolError, and kept as Raw inside IPv6 quietly + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.misc.raw import Raw + from pcapkit.utilities.exceptions import ProtocolError + + cases = ( + # CALIPSO of Opt Data Len 0, and of 8 with a compartment word declared + ('3b0007000000000000000000000000', 'invalid CALIPSO option length: 0'), + ('3b01070800000000010000000000000000', 'invalid CALIPSO option length: 8'), + # MPL of Opt Data Len 0, and of 3 with a 16-bit Seed-ID + ('3b006d0000000000', 'invalid MPL option length: 0'), + ('3b016d03400000000000000000000000', 'invalid MPL option length: 3'), + # SMF_DPD of Opt Data Len 0 before an octet with the mode bit clear and + # one with it set, and in the last two octets of the header (#1195) + ('3b00080000000000', 'invalid SMF DPD option length: 0'), + ('3b00080080000000', 'invalid SMF DPD option length: 0'), + ('3b00010200000800', 'invalid SMF DPD option length: 0'), + # SMF I-DPD of Opt Data Len 3 with an IPv4 TaggerID + ('3b010803230000000000000000000000', 'invalid SMF I-DPD option length: 3'), + ) + for cls, next_header, _ in self.headers(): + for whole, error in cases: + raw = bytes.fromhex(whole) + with self.subTest(header=cls.__name__, raw=whole): + with reported(), self.assertRaisesRegex(ProtocolError, error): + cls(raw) + + packet = ipv6(next_header, raw) + ip6, messages = self.parse(lambda: IPv6(packet)) # pylint: disable=cell-var-from-loop + self.assertEqual((ip6.data, messages), (packet, [])) + self.assertIs(type(ip6.payload), Raw) + self.assertRebuildsQuietly(IPv6, ip6, packet) + + def test_the_helpers_leave_a_cut_length_to_the_short_read(self) -> None: + # a packet whose short read names ``len`` was cut before it; one naming a + # later field, or none, holds a length read off the wire + from pcapkit.corekit.fields.misc import SchemaField + from pcapkit.protocols.schema.internet import hopopt, ipv6_option + from pcapkit.utilities.exceptions import ProtocolError + + helpers = ( + (ipv6_option.calipso_pad_len, {'len': 0, 'cmpt_len': 0}, 'domain'), + (ipv6_option.mpl_opt_pad_len, {'len': 0, 'flags': {'type': 0}}, 'flags'), + (ipv6_option.smf_i_dpd_id_len, {'len': 0, 'info': {'type': 0, 'len': 0}}, 'info'), + ) + for helper, packet, later in helpers: + with self.subTest(helper=helper.__name__): + self.assertEqual(helper({**packet, '__short_read__': ('len', 0)}, prefix='X'), 0) + for record in ((later, 0), None): + with self.assertRaisesRegex(ProtocolError, r'^X: invalid .* option length: 0$'): + helper({**packet, '__short_read__': record}, prefix='X') + + # the forward match of an SMF_DPD option, ended before its second octet, and + # after it + select = ipv6_option.smf_dpd_data_selector + field = select({'test': {'len': 0, 'mode': 0}, '__short_read__': ('test', 1)}, + prefix='X', base=hopopt.SMFDPDOption) + self.assertIsInstance(field, SchemaField) + self.assertEqual((field.length, field.schema), (2, hopopt.SMFIdentificationBasedDPDOption)) + for record in (('test', 2), None): + with self.assertRaisesRegex(ProtocolError, '^X: invalid SMF DPD option length: 0$'): + select({'test': {'len': 0, 'mode': 0}, '__short_read__': record}, + prefix='X', base=hopopt.SMFDPDOption) + + +class SchemaShortReadContextTests(unittest.TestCase): + """Pin the short read ``Schema.unpack`` gives the fields after it.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_fields_after_the_end_of_the_data_are_told(self) -> None: + from pcapkit.corekit.fields.misc import ForwardMatchField, SchemaField + from pcapkit.corekit.fields.numbers import UInt8Field + from pcapkit.corekit.fields.strings import BytesField + from pcapkit.protocols.schema.schema import Schema, schema_final + + seen = [] # type: list[tuple[str, Any]] + + def probe(name: str) -> 'Callable[[dict[str, Any]], int]': + def length(pkt: 'dict[str, Any]') -> int: + seen.append((name, pkt.get('__short_read__'))) + return 1 + return length + + @schema_final + class Inner(Schema): + a: int = UInt8Field() + b: bytes = BytesField(length=probe('inner')) + + @schema_final + class Outer(Schema): + peek: bytes = ForwardMatchField(BytesField(length=3)) + x: bytes = BytesField(length=probe('x')) + inner: Inner = SchemaField(length=2, schema=Inner) + y: bytes = BytesField(length=probe('y')) + z: bytes = BytesField(length=probe('z')) + + # whole, a stale record in the packet handed dropped; cut after the nested + # schema; and cut inside the forward match, where the nested schema is told + # only of its own cut + for raw, packet, short, expected in ( + (b'\x01\x02\x03\x04\x05', {'__short_read__': ('stale', 0)}, None, + [('x', None), ('inner', None), ('y', None), ('z', None)]), + (b'\x01\x02\x03', {}, ('y', 0), + [('x', None), ('inner', None), ('y', None), ('z', ('y', 0))]), + (b'\x01', {}, ('inner', 0), + [('x', ('peek', 1)), ('inner', ('a', 0)), ('y', ('peek', 1)), ('z', ('peek', 1))]), + ): + with self.subTest(raw=raw.hex()): + seen.clear() + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + schema = Outer.unpack(raw, len(raw), packet) + self.assertEqual(seen, expected) + self.assertEqual(schema.__dict__.get('__short_read__'), short) + self.assertEqual(bytes(schema), raw) + + def test_a_short_read_noted_without_a_packet_is_kept_on_the_schema(self) -> None: + # ``packet`` is optional, for a caller with no packet to hand the record to, + # and the first record still wins + from pcapkit.corekit.fields.numbers import UInt8Field + from pcapkit.protocols.schema.schema import Schema, schema_final + + @schema_final + class Tiny(Schema): + a: int = UInt8Field() + b: int = UInt8Field() + + # pylint: disable=protected-access + schema = Tiny.unpack(b'\x01\x02', 2, {}) + schema._note_short_read('b', 1, b'') + self.assertEqual(schema.__dict__['__short_read__'], ('b', 0)) + schema._note_short_read('a', 1, b'') + self.assertEqual(schema.__dict__['__short_read__'], ('b', 0)) + + packet = {} # type: dict[str, Any] + schema = Tiny.unpack(b'\x01\x02', 2, {}) + schema._note_short_read('a', 1, b'', packet) + self.assertEqual((schema.__dict__['__short_read__'], packet), (('a', 0), {'__short_read__': ('a', 0)})) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/internet/test_smf_dpd_truncated_len_unit.py b/tests/protocols/internet/test_smf_dpd_truncated_len_unit.py index 7d67bd52d0..e457af10bc 100644 --- a/tests/protocols/internet/test_smf_dpd_truncated_len_unit.py +++ b/tests/protocols/internet/test_smf_dpd_truncated_len_unit.py @@ -47,14 +47,16 @@ def setUp(self) -> None: def _protocols(self) -> 'list[type]': return [getattr(importlib.import_module(mod), name) for mod, name in PROTOCOLS] - def test_smf_dpd_raises_field_value_error(self) -> None: - from pcapkit.utilities.exceptions import FieldValueError + def test_smf_dpd_raises_protocol_error(self) -> None: + # the header is whole, so its ``Opt Data Len`` of zero is on the wire, and + # a ``ProtocolError`` refuses it quietly in a header that is handled (#2727) + from pcapkit.utilities.exceptions import ProtocolError for cls in self._protocols(): with self.subTest(protocol=cls.__name__): with warnings.catch_warnings(): warnings.simplefilter('ignore') - with self.assertRaisesRegex(FieldValueError, 'invalid SMF DPD option length: 0'): + with self.assertRaisesRegex(ProtocolError, 'invalid SMF DPD option length: 0'): cls(io.BytesIO(PREFIX + bytes.fromhex('0800'))) def test_no_option_type_raises_a_non_library_error(self) -> None: diff --git a/tests/protocols/test_dead_branches_2318_unit.py b/tests/protocols/test_dead_branches_2318_unit.py index 48b7ad136b..7925646d0a 100644 --- a/tests/protocols/test_dead_branches_2318_unit.py +++ b/tests/protocols/test_dead_branches_2318_unit.py @@ -375,7 +375,9 @@ class CALIPSOLengthTests(unittest.TestCase): ``calipso_pad_len`` sizes ``pad`` on both routes that reach the reader -- a parse, where the option is unpacked off the wire, and a construction, - where the caller's own option schema is packed -- so each is pinned. + where the caller's own option schema is packed -- so each is pinned. It + rejects one with a ``ProtocolError``, so that a header refused for it is + refused quietly (#2727). """ @@ -385,14 +387,14 @@ def setUp(self) -> None: def test_short_option_raises_from_the_pad_length(self) -> None: from pcapkit.protocols.internet.hopopt import HOPOPT from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts - from pcapkit.utilities.exceptions import FieldValueError + from pcapkit.utilities.exceptions import ProtocolError for length, cmpt_len in ((7, 0), (5, 0), (8, 1), (11, 1)): option = (bytes([7, length]) + bytes(4) + bytes([cmpt_len, 0]) + b'\xab\xcd' + bytes(max(0, length - 8))) for protocol in (HOPOPT, IPv6_Opts): with self.subTest(protocol=protocol.__name__, length=length): - with self.assertRaises(FieldValueError) as caught: + with self.assertRaises(ProtocolError) as caught: protocol(hopopt_bytes(option)) self.assertIn(f'invalid CALIPSO option length: {length}', str(caught.exception)) @@ -412,7 +414,7 @@ def test_a_constructed_option_raises_from_the_pad_length(self) -> None: from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts from pcapkit.protocols.schema.internet import hopopt as schema_hopopt from pcapkit.protocols.schema.internet import ipv6_opts as schema_ipv6_opts - from pcapkit.utilities.exceptions import FieldValueError + from pcapkit.utilities.exceptions import ProtocolError for length, cmpt_len in ((7, 0), (5, 0), (8, 1), (11, 1)): # ``bitmap`` is conditional on a non-zero ``cmpt_len``, and is the @@ -424,10 +426,10 @@ def test_a_constructed_option_raises_from_the_pad_length(self) -> None: type=Option.CALIPSO, len=length, domain=0x12345678, cmpt_len=cmpt_len, level=5, checksum=b'\xab\xcd', bitmap=bitmap, pad=b'') - with self.assertRaises(FieldValueError) as caught: + with self.assertRaises(ProtocolError) as caught: option.pack() self.assertIn(f'invalid CALIPSO option length: {length}', str(caught.exception)) - with self.assertRaises(FieldValueError) as caught: + with self.assertRaises(ProtocolError) as caught: protocol(options=[option], next=59, payload=b'') self.assertIn(f'invalid CALIPSO option length: {length}', str(caught.exception))