diff --git a/pcapkit/protocols/internet/hip.py b/pcapkit/protocols/internet/hip.py index 2224ed4ebb..227d5cc440 100644 --- a/pcapkit/protocols/internet/hip.py +++ b/pcapkit/protocols/internet/hip.py @@ -442,6 +442,9 @@ class HIP(IPv6_Ext[Data_HIP, Schema_HIP], Internet[Data_HIP, Schema_HIP], ``HIP_CIPHER`` parameters before it, so an AES cipher's IV is read as the IV, rather than warning of a missing ``HIP_CIPHER`` and reading the IV as encrypted data (:issue:`2708`). + * A parameter given to :meth:`make` as empty raw octets, which hold no + type, raises :exc:`~pcapkit.utilities.exceptions.ProtocolError`, + rather than being dropped from the header (:issue:`2723`). .. versionremoved:: 1.6.0 @@ -965,7 +968,8 @@ def _make_hip_param(self, parameters: 'list[Schema_Parameter | tuple[Enum_Parame Raises: ProtocolError: If the code of a ``(code, value)`` pair, or the key of a parameter given as parsed data, is a name, a :obj:`bool` or - a code its field cannot carry. + a code its field cannot carry, or if a parameter given as raw + octets is empty (:issue:`2723`). """ total_length = 0 @@ -973,6 +977,8 @@ def _make_hip_param(self, parameters: 'list[Schema_Parameter | tuple[Enum_Parame parameters_list = [] # type: list[Schema_Parameter | bytes] for schema in parameters: if isinstance(schema, bytes): + if not schema: + raise ProtocolError(f'HIPv{version}: invalid parameter: {schema!r} holds no type field') # :rfc:`7401#section-5.2.1` pads every parameter to a # multiple of 8 octets, and Header Length counts in those # units; a raw parameter is taken to omit its padding. diff --git a/pcapkit/protocols/internet/hopopt.py b/pcapkit/protocols/internet/hopopt.py index 9af668fbeb..0302a40a37 100644 --- a/pcapkit/protocols/internet/hopopt.py +++ b/pcapkit/protocols/internet/hopopt.py @@ -234,6 +234,9 @@ class HOPOPT(IPv6_Ext[Data_HOPOPT, Schema_HOPOPT], Quick-Start report, given the wrong number of octets, raise :exc:`~pcapkit.utilities.exceptions.ProtocolError`, rather than being padded or cut to their field (:issue:`2676`). + * An option given to :meth:`make` as empty raw octets, which hold no + type octet, raises :exc:`~pcapkit.utilities.exceptions.ProtocolError` + rather than an :exc:`IndexError` (:issue:`2723`). .. versionremoved:: 1.6.0 @@ -645,6 +648,12 @@ def _make_hopopt_options(self, options: 'list[Schema_Option | tuple[Enum_Option Returns: Tuple of options and total length of options. + Raises: + ProtocolError: If the type of an option, given in a ``(type, + kwargs)`` pair or as the key of a container, is a name, a + :obj:`bool` or a code its field cannot carry, or if an option + given as raw octets is empty (:issue:`2723`). + Note: Options given as a :obj:`list` are fresh input: padding options in it are dropped, and padding is inserted after each option to align @@ -666,6 +675,8 @@ def _make_hopopt_options(self, options: 'list[Schema_Option | tuple[Enum_Option options_list = [] # type: list[Schema_Option | bytes] for schema in options: if isinstance(schema, bytes): + if not schema: + raise ProtocolError(f'{self.alias}: invalid option: {schema!r} holds no type octet') code = Enum_Option.get(schema[0]) if code in (Enum_Option.Pad1, Enum_Option.PadN): # ignore padding options by default continue diff --git a/pcapkit/protocols/internet/ipv4.py b/pcapkit/protocols/internet/ipv4.py index 9d6d1233aa..a6c455548a 100644 --- a/pcapkit/protocols/internet/ipv4.py +++ b/pcapkit/protocols/internet/ipv4.py @@ -243,6 +243,9 @@ class IPv4(IP[Data_IPv4, Schema_IPv4], the option has no room for, rather than cutting it to nine or four, and :meth:`make` for a checksum that is not two octets, rather than padding or cutting it (:issue:`2676`). + * An option given to :meth:`make` as empty raw octets, which hold no + type octet, raises :exc:`~pcapkit.utilities.exceptions.ProtocolError` + rather than an :exc:`IndexError` (:issue:`2723`). .. versionremoved:: 1.6.0 @@ -882,7 +885,8 @@ def _make_ipv4_options(self, options: 'list[Schema_Option | tuple[Enum_OptionNum Raises: ProtocolError: If the type of an option, given in a ``(type, kwargs)`` pair or as the key of a container, is a name, a - :obj:`bool` or a code its field cannot carry. + :obj:`bool` or a code its field cannot carry, or if an option + given as raw octets is empty (:issue:`2723`). """ total_length = 0 @@ -891,6 +895,8 @@ def _make_ipv4_options(self, options: 'list[Schema_Option | tuple[Enum_OptionNum if isinstance(options, list): for schema in options: if isinstance(schema, bytes): + if not schema: + raise ProtocolError(f'{self.alias}: invalid option: {schema!r} holds no type octet') code = Enum_OptionNumber.get(schema[0]) if code in (Enum_OptionNumber.NOP, Enum_OptionNumber.EOOL): # ignore padding options by default continue diff --git a/pcapkit/protocols/internet/ipv6_opts.py b/pcapkit/protocols/internet/ipv6_opts.py index f3cfa7d0ec..fd3d7f3d9f 100644 --- a/pcapkit/protocols/internet/ipv6_opts.py +++ b/pcapkit/protocols/internet/ipv6_opts.py @@ -238,6 +238,9 @@ class IPv6_Opts(IPv6_Ext[Data_IPv6_Opts, Schema_IPv6_Opts], Quick-Start report, given the wrong number of octets, raise :exc:`~pcapkit.utilities.exceptions.ProtocolError`, rather than being padded or cut to their field (:issue:`2676`). + * An option given to :meth:`make` as empty raw octets, which hold no + type octet, raises :exc:`~pcapkit.utilities.exceptions.ProtocolError` + rather than an :exc:`IndexError` (:issue:`2723`). .. versionremoved:: 1.6.0 @@ -633,6 +636,12 @@ def _make_ipv6_opts(self, options: 'list[Schema_Option | tuple[Enum_Option | int Returns: Tuple of options and total length of options. + Raises: + ProtocolError: If the type of an option, given in a ``(type, + kwargs)`` pair or as the key of a container, is a name, a + :obj:`bool` or a code its field cannot carry, or if an option + given as raw octets is empty (:issue:`2723`). + Note: Options given as a :obj:`list` are fresh input: padding options in it are dropped, and padding is inserted after each option to align @@ -654,6 +663,8 @@ def _make_ipv6_opts(self, options: 'list[Schema_Option | tuple[Enum_Option | int options_list = [] # type: list[Schema_Option | bytes] for schema in options: if isinstance(schema, bytes): + if not schema: + raise ProtocolError(f'{self.alias}: invalid option: {schema!r} holds no type octet') code = Enum_Option.get(schema[0]) if code in (Enum_Option.Pad1, Enum_Option.PadN): # ignore padding options by default continue diff --git a/pcapkit/protocols/internet/mh.py b/pcapkit/protocols/internet/mh.py index 87c1fa4e10..8ef1a369a1 100644 --- a/pcapkit/protocols/internet/mh.py +++ b/pcapkit/protocols/internet/mh.py @@ -1148,6 +1148,11 @@ class MH(IPv6_Ext[Data_MH, Schema_MH], degree that does not fit 24 bits as two's complement, where it masked it. The option makers check their registry codes and their one-octet ``Option Length`` too (:issue:`2614`). + * An option, sub-option, QoS attribute or CGA extension given to a + maker as empty raw octets, which hold no type, raises + :exc:`~pcapkit.utilities.exceptions.ProtocolError`, rather than being + dropped beside another item or, for most, raising an + :exc:`AttributeError` alone (:issue:`2723`). * 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 @@ -8011,13 +8016,16 @@ def _make_mh_options(self, options: 'Option | list[Schema_Option | tuple[Enum_Op Raises: ProtocolError: If the type of an option, given in a ``(type, kwargs)`` pair or as the key of parsed data, is a name, a - :obj:`bool` or a code its field cannot carry. + :obj:`bool` or a code its field cannot carry, or if an option + given as raw octets is empty (:issue:`2723`). """ if isinstance(options, list): options_list = [] # type: list[Schema_Option | bytes] for schema in options: if isinstance(schema, bytes): + if not schema: + raise ProtocolError(f'{self.alias}: invalid option: {schema!r} holds no type octet') code = Enum_Option.get(int.from_bytes(schema[0:1], 'big', signed=False)) data = schema # type: Schema_Option | bytes @@ -10020,12 +10028,15 @@ def _make_fid_suboptions( Raises: ProtocolError: If the type of a sub-option, given in a ``(type, kwargs)`` pair or as the key of parsed data, is a name, a - :obj:`bool` or a code its field cannot carry. + :obj:`bool` or a code its field cannot carry, or if a + sub-option given as raw octets is empty (:issue:`2723`). """ if isinstance(suboptions, list): entries = [] # type: list[Schema_FlowIdentificationSuboption | bytes] for item in suboptions: + if isinstance(item, bytes) and not item: + raise ProtocolError(f'{self.alias}: invalid sub-option: {item!r} holds no type octet') if isinstance(item, (bytes, Schema)): entries.append(cast('Schema_FlowIdentificationSuboption | bytes', item)) else: @@ -10441,12 +10452,15 @@ def _make_ani_suboptions( Raises: ProtocolError: If the type of a sub-option, given in a ``(type, kwargs)`` pair or as the key of parsed data, is a name, a - :obj:`bool` or a code its field cannot carry. + :obj:`bool` or a code its field cannot carry, or if a + sub-option given as raw octets is empty (:issue:`2723`). """ if isinstance(suboptions, list): entries = [] # type: list[Schema_ANISuboption | bytes] for item in suboptions: + if isinstance(item, bytes) and not item: + raise ProtocolError(f'{self.alias}: invalid sub-option: {item!r} holds no type octet') if isinstance(item, (bytes, Schema)): entries.append(cast('Schema_ANISuboption | bytes', item)) else: @@ -10846,12 +10860,15 @@ def _make_qos_attributes( Raises: ProtocolError: If the type of an attribute, given in a ``(type, kwargs)`` pair or as the key of parsed data, is a name, a - :obj:`bool` or a code its field cannot carry. + :obj:`bool` or a code its field cannot carry, or if an + attribute given as raw octets is empty (:issue:`2723`). """ if isinstance(attributes, list): entries = [] # type: list[Schema_QoSAttribute | bytes] for item in attributes: + if isinstance(item, bytes) and not item: + raise ProtocolError(f'{self.alias}: invalid attribute: {item!r} holds no type octet') if isinstance(item, (bytes, Schema)): entries.append(cast('Schema_QoSAttribute | bytes', item)) else: @@ -11200,12 +11217,15 @@ def _make_lcmp_suboptions( Raises: ProtocolError: If the type of a sub-option, given in a ``(type, kwargs)`` pair or as the key of parsed data, is a name, a - :obj:`bool` or a code its field cannot carry. + :obj:`bool` or a code its field cannot carry, or if a + sub-option given as raw octets is empty (:issue:`2723`). """ if isinstance(suboptions, list): entries = [] # type: list[Schema_LMAControlledMAGSuboption | bytes] for item in suboptions: + if isinstance(item, bytes) and not item: + raise ProtocolError(f'{self.alias}: invalid sub-option: {item!r} holds no type octet') if isinstance(item, (bytes, Schema)): entries.append(cast('Schema_LMAControlledMAGSuboption | bytes', item)) else: @@ -11658,7 +11678,8 @@ def _make_cga_extensions(self, extensions: 'Extension | list[Schema_CGAExtension Raises: ProtocolError: If the type of an extension, given in a ``(type, kwargs)`` pair or as the key of parsed data, is a name, a - :obj:`bool` or a code its field cannot carry. + :obj:`bool` or a code its field cannot carry, or if an + extension given as raw octets is empty (:issue:`2723`). """ total_length = 0 @@ -11666,6 +11687,8 @@ def _make_cga_extensions(self, extensions: 'Extension | list[Schema_CGAExtension extensions_list = [] # type: list[Schema_CGAExtension | bytes] for schema in extensions: if isinstance(schema, bytes): + if not schema: + raise ProtocolError(f'{self.alias}: invalid CGA extension: {schema!r} holds no type field') code = Enum_CGAExtension.get(int.from_bytes(schema[0:2], 'big', signed=False)) data = schema # type: Schema_CGAExtension | bytes diff --git a/pcapkit/protocols/misc/pcapng.py b/pcapkit/protocols/misc/pcapng.py index 19cbf7a5a2..a640bb7d2a 100644 --- a/pcapkit/protocols/misc/pcapng.py +++ b/pcapkit/protocols/misc/pcapng.py @@ -723,16 +723,21 @@ class PCAPNG(ProtocolBase[Data_PCAPNG, Schema_PCAPNG], .. versionchanged:: 1.6.0 - Each enumeration field of :meth:`make` and of the block and option - makers, and the code of each option and the type of each name - resolution record given to a block maker, takes a member or an - :class:`int` code, resolved with the enumeration's ``get``, an option - code in the block's namespace; a name, a :obj:`bool` for a field wider - than one bit, or a code the field cannot carry raises - :exc:`~pcapkit.utilities.exceptions.ProtocolError`, rather than - resolving against an enumeration inferred from the caller's frame, - writing a different value or raising another exception - (:issue:`2594`). + * Each enumeration field of :meth:`make` and of the block and option + makers, and the code of each option and the type of each name + resolution record given to a block maker, takes a member or an + :class:`int` code, resolved with the enumeration's ``get``, an option + code in the block's namespace; a name, a :obj:`bool` for a field + wider than one bit, or a code the field cannot carry raises + :exc:`~pcapkit.utilities.exceptions.ProtocolError`, rather than + resolving against an enumeration inferred from the caller's frame, + writing a different value or raising another exception + (:issue:`2594`). + * An option or a name resolution record given to a block maker as raw + octets too short to hold its 16-bit code raises + :exc:`~pcapkit.utilities.exceptions.ProtocolError`, rather than + reading empty octets as code 0 and writing an end-of-options or + end-of-records marker for them (:issue:`2723`). .. versionremoved:: 1.6.0 @@ -5150,7 +5155,8 @@ def _make_pcapng_options(self, options: 'Option | list[Schema_Option | tuple[Enu Raises: ProtocolError: If an option's code is a name, a :obj:`bool`, or a code the 16-bit option code field cannot carry - (:issue:`2594`). + (:issue:`2594`), or if an option given as raw octets is + shorter than that field (:issue:`2723`). """ # an option area kept as captured is written back verbatim @@ -5163,6 +5169,10 @@ def _make_pcapng_options(self, options: 'Option | list[Schema_Option | tuple[Enu options_list = [] # type: list[Schema_Option | bytes] for schema in options: if isinstance(schema, bytes): + # NOTE: Octets short of the code field would read as code 0 + # and be taken for an opt_endofopt (#2723). + if len(schema) < 2: + raise ProtocolError(f'PCAP-NG: invalid option: {schema!r} holds no whole option code') code = Enum_OptionType.get(int.from_bytes(schema[0:2], self._byte, signed=False), namespace=namespace) if code == Enum_OptionType.opt_endofopt: # moved to the end, written once @@ -6630,7 +6640,8 @@ def _make_nrb_records(self, records: 'Record | list[Schema_NameResolutionRecord Raises: ProtocolError: If a record's type is a name, a :obj:`bool`, or a code the 16-bit record type field cannot carry - (:issue:`2594`). + (:issue:`2594`), or if a record given as raw octets is shorter + than that field (:issue:`2723`). """ # a record area kept as captured is written back verbatim @@ -6643,6 +6654,10 @@ def _make_nrb_records(self, records: 'Record | list[Schema_NameResolutionRecord records_list = [] # type: list[Schema_NameResolutionRecord | bytes] for schema in records: if isinstance(schema, bytes): + # NOTE: Octets short of the type field would read as type 0 + # and be taken for an nrb_record_end (#2723). + if len(schema) < 2: + raise ProtocolError(f'PCAP-NG: invalid record: {schema!r} holds no whole record type') code = Enum_RecordType.get(int.from_bytes(schema[0:2], self._byte, signed=False)) if code == Enum_RecordType.nrb_record_end: # moved to the end, written once has_record_end = True diff --git a/pcapkit/protocols/transport/dccp.py b/pcapkit/protocols/transport/dccp.py index cdd4557a34..88a9114ab2 100644 --- a/pcapkit/protocols/transport/dccp.py +++ b/pcapkit/protocols/transport/dccp.py @@ -585,6 +585,12 @@ def _make_dccp_options(self, options: 'list[Schema_Option | tuple[Enum_Option | Returns: Tuple of options and total length of options. + Raises: + ProtocolError: If the type of an option, given in a ``(type, + kwargs)`` pair or as the key of a container, is a name, a + :obj:`bool` or a code its field cannot carry, or if an option + given as raw octets is empty (:issue:`2723`). + Note: The options are emitted in the given order, Padding included, so that a parsed option list re-emits the same octets. Padding @@ -595,6 +601,8 @@ def _make_dccp_options(self, options: 'list[Schema_Option | tuple[Enum_Option | options_list = [] # type: list[Schema_Option | bytes] if isinstance(options, list): for item in options: + if isinstance(item, bytes) and not item: + raise ProtocolError(f'{self.alias}: invalid option: {item!r} holds no type octet') if isinstance(item, (bytes, Schema)): options_list.append(item) else: diff --git a/pcapkit/protocols/transport/sctp.py b/pcapkit/protocols/transport/sctp.py index 748c8ee1d7..31fbaa66fd 100644 --- a/pcapkit/protocols/transport/sctp.py +++ b/pcapkit/protocols/transport/sctp.py @@ -465,6 +465,10 @@ class SCTP(Transport[Data_SCTP, Schema_SCTP], code no member holds is written as given; a name, a :obj:`bool` for a field wider than one bit, or a code the field cannot carry raises :exc:`~pcapkit.utilities.exceptions.ProtocolError` (:issue:`2594`). + * A parameter or an error cause given to a chunk maker as empty raw + octets, which hold no type or code, raises + :exc:`~pcapkit.utilities.exceptions.ProtocolError` rather than an + :exc:`AttributeError` (:issue:`2723`). """ @@ -1169,11 +1173,19 @@ def _make_sctp_parameters(self, parameters: 'list[Schema_Parameter | tuple[Enum_ Returns: Constructed parameter schemas. + Raises: + ProtocolError: If the type of a parameter, given in a ``(type, + kwargs)`` pair or as the key of a container, is a name, a + :obj:`bool` or a code its field cannot carry, or if a parameter + given as raw octets is empty (:issue:`2723`). + """ parameters_list = [] # type: list[Schema_Parameter | bytes] if isinstance(parameters, list): for schema in parameters: + if isinstance(schema, bytes) and not schema: + raise ProtocolError(f'{self.alias}: invalid parameter: {schema!r} holds no type field') if isinstance(schema, (bytes, Schema)): parameters_list.append(schema) else: @@ -1243,11 +1255,19 @@ def _make_sctp_causes(self, causes: 'list[Schema_ErrorCause | tuple[Enum_CauseCo Returns: Constructed error cause schemas. + Raises: + ProtocolError: If the code of an error cause, given in a ``(code, + kwargs)`` pair or as the key of a container, is a name, a + :obj:`bool` or a code its field cannot carry, or if an error + cause given as raw octets is empty (:issue:`2723`). + """ causes_list = [] # type: list[Schema_ErrorCause | bytes] if isinstance(causes, list): for schema in causes: + if isinstance(schema, bytes) and not schema: + raise ProtocolError(f'{self.alias}: invalid error cause: {schema!r} holds no code field') if isinstance(schema, (bytes, Schema)): causes_list.append(schema) else: diff --git a/tests/protocols/internet/test_empty_raw_option_2723_unit.py b/tests/protocols/internet/test_empty_raw_option_2723_unit.py new file mode 100644 index 0000000000..2082dd12dc --- /dev/null +++ b/tests/protocols/internet/test_empty_raw_option_2723_unit.py @@ -0,0 +1,83 @@ +# -*- coding: utf-8 -*- +"""An IPv4, HOPOPT or IPv6-Opts option given as empty raw octets is refused. + +GitHub issue #2723: each of ``_make_ipv4_options``, ``_make_hopopt_options`` +and ``_make_ipv6_opts`` reads the type octet of a raw-octets option before it +checks that the octets hold one, so ``[b'']`` raised a bare ``IndexError``. +Empty octets raise ``ProtocolError``, as the other malformed items in those +branches do, wherever in the list they sit. + +A raw option that holds its type octet builds as given, which the cases +below pin against the specifications: the IPv4 Router Alert option of +:rfc:`2113#section-2.1` after the 20-octet header, with IHL counting it; the +IPv6 Router Alert option of :rfc:`2711#section-2.1` and the Tunnel +Encapsulation Limit option of :rfc:`2473#section-5.1` after the two fixed +octets, padded with a PadN option of :rfc:`8200#section-4.2` to 8 octets. + +Every case builds its own octets in memory and reads no capture. The protocol +classes are 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 importlib +import unittest + +from tests._support import reimport_once_per_class + +#: Module and class name of each protocol whose option maker is covered. +PROTOCOLS = ( + ('pcapkit.protocols.internet.ipv4', 'IPv4'), + ('pcapkit.protocols.internet.hopopt', 'HOPOPT'), + ('pcapkit.protocols.internet.ipv6_opts', 'IPv6_Opts'), +) + +#: A whole raw option for each protocol: Router Alert for IPv4 and HOPOPT, +#: Tunnel Encapsulation Limit for IPv6-Opts. +RAW = { + 'IPv4': b'\x94\x04\x00\x00', + 'HOPOPT': b'\x05\x02\x00\x00', + 'IPv6_Opts': b'\x04\x01\x05', +} + + +class TestEmptyRawOption(unittest.TestCase): + """Empty raw octets raise ``ProtocolError``; a whole raw option builds.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_empty_octets_are_refused(self) -> None: + """Raw octets with no type octet raise ``ProtocolError``, not ``IndexError``.""" + from pcapkit.utilities.exceptions import ProtocolError + + for module, name in PROTOCOLS: + protocol = getattr(importlib.import_module(module), name) + for options in ([b''], [RAW[name], b'']): + with self.subTest(protocol=name, options=options): + with self.assertRaisesRegex(ProtocolError, r"invalid option: b'' holds no type octet"): + protocol(options=options) + + def test_ipv4_raw_option_builds_as_given(self) -> None: + from pcapkit.protocols.internet.ipv4 import IPv4 + + built = IPv4(options=[RAW['IPv4']]).data + self.assertEqual(built[0] & 0x0F, 6) # IHL: the 20-octet header and the option + self.assertEqual(built[20:24], RAW['IPv4']) + + def test_ipv6_extension_raw_option_builds_as_given(self) -> None: + from pcapkit.protocols.internet.hopopt import HOPOPT + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + + built = HOPOPT(options=[RAW['HOPOPT']]).data + self.assertEqual(built[1:], b'\x00' + RAW['HOPOPT'] + b'\x01\x00') + + built = IPv6_Opts(options=[RAW['IPv6_Opts']]).data + self.assertEqual(built[1:], b'\x00' + RAW['IPv6_Opts'] + b'\x01\x01\x00') + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/internet/test_hip_empty_raw_param_2723_unit.py b/tests/protocols/internet/test_hip_empty_raw_param_2723_unit.py new file mode 100644 index 0000000000..c4de13df84 --- /dev/null +++ b/tests/protocols/internet/test_hip_empty_raw_param_2723_unit.py @@ -0,0 +1,59 @@ +# -*- coding: utf-8 -*- +"""A HIP parameter given as empty raw octets is refused. + +GitHub issue #2723: ``_make_hip_param`` pads a raw-octets parameter to a +multiple of 8 octets and reads no type off it, so ``[b'']`` padded to nothing +and built the header an empty parameter list builds: the item was dropped. +Empty octets raise ``ProtocolError``, as the other malformed items in that +branch do, wherever in the list they sit. + +A raw parameter that holds its type builds as given, which the cases below pin +against :rfc:`7401`: an ``ECHO_REQUEST_SIGNED`` parameter of +:rfc:`7401#section-5.2.20` follows the 40-octet fixed header, padded to 8 +octets as :rfc:`7401#section-5.2.1` requires, and Header Length counts it in +8-octet units past the first 8 (:rfc:`7401#section-5.1`). + +Every case builds its own octets in memory and reads no capture. :class:`HIP` +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 tests._support import reimport_once_per_class + +#: An ECHO_REQUEST_SIGNED parameter, type 897, carrying ``ab`` and no padding. +ECHO_REQUEST_SIGNED = b'\x03\x81\x00\x02ab' + + +class TestHIPEmptyRawParam(unittest.TestCase): + """Empty raw octets raise ``ProtocolError``; a whole raw parameter builds.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_empty_octets_are_refused(self) -> None: + from pcapkit.protocols.internet.hip import HIP + from pcapkit.utilities.exceptions import ProtocolError + + for parameters in ([b''], [ECHO_REQUEST_SIGNED, b''], [b'', ECHO_REQUEST_SIGNED]): + with self.subTest(parameters=parameters): + with self.assertRaisesRegex(ProtocolError, r"invalid parameter: b'' holds no type field"): + HIP(parameters=parameters) + + def test_raw_parameter_builds_as_given(self) -> None: + from pcapkit.const.hip.parameter import Parameter + from pcapkit.protocols.internet.hip import HIP + + built = HIP(parameters=[ECHO_REQUEST_SIGNED]).data + self.assertEqual(built[1], 5) # Header Length: (40 + 8 - 8) / 8 + self.assertEqual(built[40:], ECHO_REQUEST_SIGNED + b'\x00\x00') + self.assertIn(Parameter.ECHO_REQUEST_SIGNED, HIP(built).info.parameters) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/internet/test_mh_empty_raw_item_2723_unit.py b/tests/protocols/internet/test_mh_empty_raw_item_2723_unit.py new file mode 100644 index 0000000000..32235d4a35 --- /dev/null +++ b/tests/protocols/internet/test_mh_empty_raw_item_2723_unit.py @@ -0,0 +1,153 @@ +# -*- coding: utf-8 -*- +"""An MH option, sub-option, QoS attribute or CGA extension given as empty raw octets is refused. + +GitHub issue #2723: the MH makers that take raw octets in a list -- +``_make_mh_options``, ``_make_fid_suboptions``, ``_make_ani_suboptions``, +``_make_qos_attributes``, ``_make_lcmp_suboptions`` and ``_make_cga_extensions`` +-- kept an empty item as it was. Packed, it adds no octets, so the header parsed +back held nothing in its place: alone, it reached ``read()`` as a schema and +raised a bare ``AttributeError``, or, for the access network identifier and +LMA-controlled MAG sub-options, failed as an option of length 0; beside a whole +item it was dropped. Empty octets raise ``ProtocolError``, as the +other malformed items in those branches do, wherever in the list they sit. + +A whole raw item builds as given. Each case asserts it against +:rfc:`6275#section-6.2.1`: an option's Option Length counts the octets after +its own two, so a sub-option or attribute ends its option and lengthens it by +its own size. A raw CGA extension builds the octets its ``(type, kwargs)`` pair +does. + +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 PadN option carrying no padding octets. +PADN = b'\x01\x00' +#: A sub-option or attribute of the unassigned type 200, carrying ``ab``. +ITEM = b'\xc8\x02ab' +#: A CGA extension of the unassigned type 1, carrying ``ab``. +EXTENSION = b'\x00\x01\x00\x02ab' +#: The octets of the fixed MH header and of a Binding Refresh Request. +HEADER = 8 + + +def _build(options: 'list[Any]') -> 'bytes': + """The octets of 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}).data + + +def _lists() -> 'list[tuple[str, Any, str, str]]': + """Each sub-list as its label, its option type, its keyword and its item noun.""" + from pcapkit.const.mh.option import Option + + return [ + ('FID', Option.Flow_Identification_Mobility_Option, 'suboptions', 'sub-option'), + ('ANI', Option.Access_Network_Identifier, 'suboptions', 'sub-option'), + ('QoS', Option.Quality_of_Service, 'attributes', 'attribute'), + ('LCMP', Option.LMA_Controlled_MAG_Parameters, 'suboptions', 'sub-option'), + ] + + +def _cga_fields() -> 'dict[str, Any]': + """The fields of one CGA parameter, 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': b'\x30\x02\xaa\xbb', + } + + +def _cga_schema(extensions: 'list[Any]') -> 'Any': + """A CGA parameter schema carrying ``extensions``.""" + from pcapkit.protocols.schema.internet.mh import CGAParameter + + return CGAParameter(extensions=extensions, public_key_test={'len': 2}, **_cga_fields()) + + +class TestMHEmptyRawItem(unittest.TestCase): + """Empty raw octets raise ``ProtocolError``; a whole raw item builds.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_empty_option_is_refused(self) -> None: + from pcapkit.protocols.internet.mh import MH + from pcapkit.utilities.exceptions import ProtocolError + + for options in ([b''], [PADN, b''], [b'', PADN]): + with self.subTest(options=options): + with self.assertRaisesRegex(ProtocolError, r"invalid option: b'' holds no type octet"): + MH(data={'options': options}) + + def test_empty_sub_item_is_refused(self) -> None: + from pcapkit.utilities.exceptions import ProtocolError + + for label, option, keyword, noun in _lists(): + for items in ([b''], [ITEM, b''], [b'', ITEM]): + with self.subTest(option=label, items=items): + with self.assertRaisesRegex(ProtocolError, rf"invalid {noun}: b'' holds no type octet"): + _build([(option, {keyword: items})]) + + def test_empty_cga_extension_is_refused(self) -> None: + from pcapkit.const.mh.option import Option + from pcapkit.protocols.data.internet.mh import CGAParameter + from pcapkit.utilities.exceptions import ProtocolError + + parameters = { + 'schema': _cga_schema([b'']), + 'schema beside a whole one': _cga_schema([EXTENSION, b'']), + 'data model': CGAParameter(extensions=[b''], **_cga_fields()), + } + for label, parameter in parameters.items(): + with self.subTest(parameter=label): + with self.assertRaisesRegex(ProtocolError, r"invalid CGA extension: b'' holds no type field"): + _build([(Option.CGA_Parameters, {'parameters': [parameter]})]) + + def test_whole_option_builds_as_given(self) -> None: + built = _build([PADN]) + self.assertEqual(built[HEADER:HEADER + len(PADN)], PADN) + + def test_whole_sub_item_builds_as_given(self) -> None: + for label, option, keyword, _ in _lists(): + with self.subTest(option=label): + # ANI and LCMP options need a sub-option, so one item is the base. + one = _build([(option, {keyword: [ITEM]})]) + built = _build([(option, {keyword: [ITEM, ITEM]})]) + self.assertEqual(built[HEADER], option) + self.assertEqual(built[HEADER + 1], one[HEADER + 1] + len(ITEM)) + end = HEADER + 2 + built[HEADER + 1] + self.assertEqual(built[end - 2 * len(ITEM):end], ITEM * 2) + + def test_whole_cga_extension_builds_as_its_pair(self) -> None: + from pcapkit.const.mh.option import Option + from pcapkit.protocols.data.internet.mh import CGAParameter + + raw = _build([(Option.CGA_Parameters, {'parameters': [_cga_schema([EXTENSION])]})]) + pair = _build([(Option.CGA_Parameters, {'parameters': [ + CGAParameter(extensions=[(1, {'data': b'ab'})], **_cga_fields())]})]) + self.assertEqual(raw, pair) + self.assertIn(EXTENSION, raw) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/misc/test_pcapng_short_raw_item_2723_unit.py b/tests/protocols/misc/test_pcapng_short_raw_item_2723_unit.py new file mode 100644 index 0000000000..29b3f56aaa --- /dev/null +++ b/tests/protocols/misc/test_pcapng_short_raw_item_2723_unit.py @@ -0,0 +1,116 @@ +# -*- coding: utf-8 -*- +"""A PCAP-NG option or record given as raw octets shorter than its code is refused. + +GitHub issue #2723: ``_make_pcapng_options`` and ``_make_nrb_records`` read the +16-bit Option Code or Record Type of a raw item before they check that the +octets hold one. An empty item read as code 0, so ``[b'']`` was taken for an +``opt_endofopt`` or an ``nrb_record_end`` and written as one, rather than +refused; a single octet read as a code of its own. Octets shorter than the +field raise ``ProtocolError``, as the other malformed items in those +branches do, in a little-endian section and a big-endian one. + +A raw item that holds its code builds as given: a whole ``if_name`` +option sits at the start of an Interface Description Block's option area, +after the 8 octets of the block's Type and Total Length and the 8 of its +LinkType, Reserved and SnapLen fields, and a whole ``nrb_record_ipv4`` record +at the start of a Name Resolution Block's record area, after the 8 octets of +the block's Type and Total Length. + +Every case builds its own octets in memory and reads no capture. :mod:`pcapkit` +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 re +import unittest +from typing import TYPE_CHECKING + +from tests._support import reimport_once_per_class + +if TYPE_CHECKING: + from typing import Any, Literal + +#: The section byte orders covered. +ORDERS = ('little', 'big') # type: tuple[Literal['little', 'big'], ...] +#: Raw octets too short to hold a 16-bit code. +SHORT = (b'', b'\x00', b'\x01') + + +def _raw(code: 'int', value: 'bytes', order: 'Literal["little", "big"]') -> 'bytes': + """A whole option or record of ``code`` carrying ``value``, padded to 32 bits.""" + return (int(code).to_bytes(2, order) + len(value).to_bytes(2, order) + + value + bytes(-len(value) % 4)) + + +class TestPCAPNGShortRawItem(unittest.TestCase): + """Raw octets short of the code raise ``ProtocolError``; a whole item builds.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _block(self, order: 'Literal["little", "big"]', name: 'str', block: 'dict[str, Any]') -> 'bytes': + """The octets of the block named ``name``, built from ``block`` in a section of ``order``.""" + from pcapkit.const.pcapng.block_type import BlockType + from pcapkit.const.reg.linktype import LinkType + from pcapkit.foundation.engines.pcapng import Context + from pcapkit.protocols.misc.pcapng import PCAPNG + + section = PCAPNG(num=0, sct=1, ctx=None, type=BlockType.Section_Header_Block, + block={'byteorder': order}) + context = Context(section.info) # type: Any + interface = PCAPNG(num=1, sct=1, ctx=context, type=BlockType.Interface_Description_Block, + block={'linktype': LinkType.ETHERNET, 'snaplen': 0x40000}) + context.interfaces.append(interface.info) + return PCAPNG(num=2, sct=1, ctx=context, type=BlockType[name], block=block).data + + def test_short_option_is_refused(self) -> None: + from pcapkit.const.pcapng.option_type import OptionType + from pcapkit.utilities.exceptions import ProtocolError + + for order in ORDERS: + whole = _raw(OptionType.if_name, b'eth0', order) + for short in SHORT: + message = re.escape(f'invalid option: {short!r} holds no whole option code') + for options in ([short], [whole, short]): + with self.subTest(order=order, options=options): + with self.assertRaisesRegex(ProtocolError, message): + self._block(order, 'Interface_Description_Block', + {'linktype': 1, 'snaplen': 0xFFFF, 'options': options}) + + def test_short_record_is_refused(self) -> None: + from pcapkit.const.pcapng.record_type import RecordType + from pcapkit.utilities.exceptions import ProtocolError + + for order in ORDERS: + whole = _raw(RecordType.nrb_record_ipv4, bytes([192, 0, 2, 1]) + b'host\x00', order) + for short in SHORT: + message = re.escape(f'invalid record: {short!r} holds no whole record type') + for records in ([short], [whole, short]): + with self.subTest(order=order, records=records): + with self.assertRaisesRegex(ProtocolError, message): + self._block(order, 'Name_Resolution_Block', {'records': records}) + + def test_whole_items_build_as_given(self) -> None: + from pcapkit.const.pcapng.option_type import OptionType + from pcapkit.const.pcapng.record_type import RecordType + + for order in ORDERS: + with self.subTest(order=order): + option = _raw(OptionType.if_name, b'eth0', order) + built = self._block(order, 'Interface_Description_Block', + {'linktype': 1, 'snaplen': 0xFFFF, 'options': [option]}) + self.assertEqual(int.from_bytes(built[4:8], order), len(built)) + self.assertEqual(built[16:16 + len(option)], option) + + record = _raw(RecordType.nrb_record_ipv4, bytes([192, 0, 2, 1]) + b'host\x00', order) + built = self._block(order, 'Name_Resolution_Block', {'records': [record]}) + self.assertEqual(int.from_bytes(built[4:8], order), len(built)) + self.assertEqual(built[8:8 + len(record)], record) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/transport/test_dccp_empty_raw_option_2723_unit.py b/tests/protocols/transport/test_dccp_empty_raw_option_2723_unit.py new file mode 100644 index 0000000000..a3b389595e --- /dev/null +++ b/tests/protocols/transport/test_dccp_empty_raw_option_2723_unit.py @@ -0,0 +1,57 @@ +# -*- coding: utf-8 -*- +"""A DCCP option given as empty raw octets is refused. + +GitHub issue #2723: ``_make_dccp_options`` appends a raw-octets option as it +is, and reads no type octet off it, so ``[b'']`` added no octets and built the +header an empty option list builds: the item was dropped. Empty octets raise +``ProtocolError``, as the other malformed items in that branch do, wherever in +the list they sit. + +A raw option that holds its type octet builds as given, which the cases below +pin against :rfc:`4340`: the one-octet Slow Receiver option of +:rfc:`4340#section-11.6` follows the 20 octets of a DCCP-Request with 48-bit +sequence numbers, padded with Padding options to a 32-bit boundary, and Data +Offset counts it (:rfc:`4340#section-5.1`). + +Every case builds its own octets in memory and reads no capture. :class:`DCCP` +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 tests._support import reimport_once_per_class + +#: A Slow Receiver option. +SLOW_RECEIVER = b'\x02' + + +class TestDCCPEmptyRawOption(unittest.TestCase): + """Empty raw octets raise ``ProtocolError``; a whole raw option builds.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_empty_octets_are_refused(self) -> None: + from pcapkit.protocols.transport.dccp import DCCP + from pcapkit.utilities.exceptions import ProtocolError + + for options in ([b''], [SLOW_RECEIVER, b''], [b'', SLOW_RECEIVER]): + with self.subTest(options=options): + with self.assertRaisesRegex(ProtocolError, r"invalid option: b'' holds no type octet"): + DCCP(options=options) + + def test_raw_option_builds_as_given(self) -> None: + from pcapkit.protocols.transport.dccp import DCCP + + built = DCCP(options=[SLOW_RECEIVER]).data + self.assertEqual(built[4], 6) # Data Offset: the 20-octet header and one word of options + self.assertEqual(built[20:], SLOW_RECEIVER + b'\x00\x00\x00') + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/transport/test_sctp_empty_raw_item_2723_unit.py b/tests/protocols/transport/test_sctp_empty_raw_item_2723_unit.py new file mode 100644 index 0000000000..06dab722b6 --- /dev/null +++ b/tests/protocols/transport/test_sctp_empty_raw_item_2723_unit.py @@ -0,0 +1,85 @@ +# -*- coding: utf-8 -*- +"""An SCTP parameter or error cause given as empty raw octets is refused. + +GitHub issue #2723: ``_make_sctp_parameters`` and ``_make_sctp_causes`` kept an +empty raw item in the list they return. Its packed octets parse back with no +item in its place, so the made header was kept, and reading it then took the +empty octets for a schema: ``[b'']`` raised a bare ``AttributeError``. Empty +octets raise ``ProtocolError``, as the other malformed items in those +branches do. + +A raw item that holds its type or code builds as given, which the cases +below pin against :rfc:`9260`: the IPv4 Address parameter of +:rfc:`9260#section-3.3.2.1` after the 20 octets of an INIT chunk's fixed +fields, and the Invalid Stream Identifier cause of +:rfc:`9260#section-3.3.10.1` after the 4 octets of an ABORT chunk's header, +each chunk's Chunk Length counting it. + +Every case builds its own octets in memory and reads no capture. :class:`SCTP` +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 tests._support import reimport_once_per_class + +#: An IPv4 Address parameter, of 192.0.2.1. +PARAMETER = b'\x00\x05\x00\x08\xc0\x00\x02\x01' +#: An Invalid Stream Identifier cause, of stream 1. +CAUSE = b'\x00\x01\x00\x08\x00\x01\x00\x00' + + +class TestSCTPEmptyRawItem(unittest.TestCase): + """Empty raw octets raise ``ProtocolError``; a whole raw item builds.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_empty_parameter_is_refused(self) -> None: + from pcapkit.const.sctp.chunk import Chunk + from pcapkit.protocols.transport.sctp import SCTP + from pcapkit.utilities.exceptions import ProtocolError + + for chunk in (Chunk.Initiation, Chunk.Initiation_Acknowledgement): + for parameters in ([b''], [PARAMETER, b'']): + with self.subTest(chunk=chunk, parameters=parameters): + with self.assertRaisesRegex(ProtocolError, r"invalid parameter: b'' holds no type field"): + SCTP(chunks=[(chunk, {'parameters': parameters})]) + + def test_empty_cause_is_refused(self) -> None: + from pcapkit.const.sctp.chunk import Chunk + from pcapkit.protocols.transport.sctp import SCTP + from pcapkit.utilities.exceptions import ProtocolError + + for chunk in (Chunk.Abort, Chunk.Operation_Error): + for error in ([b''], [CAUSE, b'']): + with self.subTest(chunk=chunk, error=error): + with self.assertRaisesRegex(ProtocolError, r"invalid error cause: b'' holds no code field"): + SCTP(chunks=[(chunk, {'error': error})]) + + def test_raw_parameter_builds_as_given(self) -> None: + from pcapkit.const.sctp.chunk import Chunk + from pcapkit.protocols.transport.sctp import SCTP + + built = SCTP(chunks=[(Chunk.Initiation, {'parameters': [PARAMETER]})]).data + self.assertEqual(built[12], Chunk.Initiation) + self.assertEqual(int.from_bytes(built[14:16], 'big'), 28) # 20 fixed octets and the parameter + self.assertEqual(built[32:], PARAMETER) + + def test_raw_cause_builds_as_given(self) -> None: + from pcapkit.const.sctp.chunk import Chunk + from pcapkit.protocols.transport.sctp import SCTP + + built = SCTP(chunks=[(Chunk.Abort, {'error': [CAUSE]})]).data + self.assertEqual(built[12], Chunk.Abort) + self.assertEqual(int.from_bytes(built[14:16], 'big'), 12) # the chunk header and the cause + self.assertEqual(built[16:], CAUSE) + + +if __name__ == '__main__': + unittest.main()