diff --git a/pcapkit/protocols/internet/hip.py b/pcapkit/protocols/internet/hip.py index ccaed5966..cad41bf60 100644 --- a/pcapkit/protocols/internet/hip.py +++ b/pcapkit/protocols/internet/hip.py @@ -3450,7 +3450,9 @@ def _make_locator(locator: 'Optional[Data_Locator]' = None, *, ip=value.ip.packed, ) else: - raise ProtocolError(f'HIPv{version}: [ParamNo {code}] invalid format') + raise ProtocolError(f'HIPv{version}: [ParamNo {code}] invalid locator: ' + f'must be an ipaddress.IPv6Address or a LocatorData, ' + f'not {value!r}') traffic = locator.traffic type = locator.type diff --git a/pcapkit/protocols/internet/mh.py b/pcapkit/protocols/internet/mh.py index f40428734..b4b3fb8d3 100644 --- a/pcapkit/protocols/internet/mh.py +++ b/pcapkit/protocols/internet/mh.py @@ -1454,6 +1454,12 @@ def make(self, Returns: Constructed packet data. + Raises: + ProtocolError: If ``data`` is none of the forms this maker reads -- + raw :obj:`bytes`, a :obj:`dict` of message arguments, an + :class:`~pcapkit.protocols.data.internet.mh.MH` data model or a + :class:`~pcapkit.protocols.schema.internet.mh.Packet` schema. + .. versionchanged:: 1.5.0 Takes ``reserved`` ahead of ``chksum`` (:issue:`1329`), which shifts @@ -1483,7 +1489,8 @@ def make(self, elif isinstance(data, Schema_Packet): data_val = data else: - raise ProtocolError(f'MH: [Type {type_val}] invalid format') + raise ProtocolError(f'MH: [Type {type_val}] invalid data: must be bytes, a dict, ' + f'an MH data model or a Packet schema, not {data!r}') # NOTE: The header has to be a multiple of 8 octets, so the message data # needs padding until ``len(data) + 6`` is aligned. Rounding ``length`` up @@ -7903,6 +7910,11 @@ def _make_opt_pad(self, type: 'Enum_Option', option: 'Optional[Data_PadOption]' Returns: Constructed option schema. + Warns: + ProtocolWarning: If ``length`` is non-zero for a ``Pad1``, which + has no ``Option Length`` field to carry it; the option is made + as a ``PadN`` instead. + Note: :attr:`Data_PadOption.length ` counts the @@ -7921,7 +7933,8 @@ def _make_opt_pad(self, type: 'Enum_Option', option: 'Optional[Data_PadOption]' data = getattr(option, 'data', b'') if type == Enum_Option.Pad1 and length != 0: - warn(f'{self.alias}: [OptNo {type}] invalid format', ProtocolWarning) + warn(f'{self.alias}: [OptNo {type}] invalid length: must be 0 for a Pad1 option, ' + f'not {length!r}; making a PadN', ProtocolWarning) type = Enum_Option.PadN # type: ignore[assignment] return Schema_PadOption( @@ -8024,12 +8037,18 @@ def _make_opt_bad(self, type: 'Enum_Option', option: 'Optional[Data_Authorizatio Returns: Constructed option schema. + Raises: + ProtocolError: If ``data`` is not a multiple of eight octets, the + length :meth:`_read_opt_bad` accepts for an :rfc:`6275` + authenticator. + """ if option is not None: data = option.data if len(data) % 8 != 0: - raise ProtocolError(f'{self.alias}: [OptNo {type}] invalid format') + raise ProtocolError(f'{self.alias}: [OptNo {type}] invalid data: ' + f'must be a multiple of eight octets, not {data!r}') return Schema_AuthorizationDataOption( type=type, @@ -8349,6 +8368,11 @@ def _make_opt_auth(self, type: 'Enum_Option', option: 'Optional[Data_AuthOption] Returns: Constructed option schema. + Raises: + ProtocolError: If ``data`` is not two octets more than a multiple of + four, which is the only length leaving an ``Option Length`` of + ``5 + len(data)`` that :meth:`_read_opt_auth` accepts. + """ if option is not None: subtype_val = option.subtype @@ -8359,7 +8383,8 @@ def _make_opt_auth(self, type: 'Enum_Option', option: 'Optional[Data_AuthOption] reversed=subtype_reversed, pack=False) if (len(data) + 6) % 4 != 0: - raise ProtocolError(f'{self.alias}: [OptNo {type}] invalid format') + raise ProtocolError(f'{self.alias}: [OptNo {type}] invalid data: must be two octets ' + f'more than a multiple of four, not {data!r}') return Schema_AuthOption( type=type, @@ -8611,6 +8636,10 @@ def _make_opt_badf(self, type: 'Enum_Option', option: 'Optional[Data_BADFOption] Returns: Constructed option schema. + Raises: + ProtocolError: If ``data`` is empty, since the ``length`` field then + reads as ``0``, which :meth:`_read_opt_badf` rejects. + Note: The ``length`` field counts the authenticator only, excluding the SPI, see :meth:`_read_opt_badf`. @@ -8621,7 +8650,8 @@ def _make_opt_badf(self, type: 'Enum_Option', option: 'Optional[Data_BADFOption] data = option.data if not data: - raise ProtocolError(f'{self.alias}: [OptNo {type}] invalid format') + raise ProtocolError(f'{self.alias}: [OptNo {type}] invalid data: ' + f'must be at least one octet, not {data!r}') return Schema_BADFOption( type=type, diff --git a/pcapkit/protocols/misc/pcapng.py b/pcapkit/protocols/misc/pcapng.py index cc119ce97..07acb5b0e 100644 --- a/pcapkit/protocols/misc/pcapng.py +++ b/pcapkit/protocols/misc/pcapng.py @@ -4882,6 +4882,12 @@ def _make_block_dsb(self, block: 'Optional[Data_DecryptionSecretsBlock]' = None, Returns: Constructed block schema. + Raises: + ProtocolError: If ``secrets_data`` is none of the forms this maker + reads -- raw :obj:`bytes`, a :obj:`dict` of secrets arguments, a + :class:`~pcapkit.protocols.data.misc.pcapng.DSBSecrets` data + model or a :class:`~pcapkit.protocols.schema.schema.Schema`. + """ declared = None # type: Optional[int] if block is not None: @@ -4921,7 +4927,8 @@ def _make_block_dsb(self, block: 'Optional[Data_DecryptionSecretsBlock]' = None, elif isinstance(secrets_data, Schema): secrets_data_val = secrets_data else: - raise ProtocolError(f'PCAP-NG: [DSB] secrets {secrets_type_val} invalid format') + raise ProtocolError(f'PCAP-NG: [DSB] invalid secrets_data: must be bytes, a dict, ' + f'a DSBSecrets data model or a Schema, not {secrets_data!r}') # NOTE: Secrets Length excludes the padding to the next 32-bit boundary, # which ``padding_data`` writes and the block length has to include. if block is not None and isinstance(secrets_data_val, Schema): diff --git a/pcapkit/protocols/transport/sctp.py b/pcapkit/protocols/transport/sctp.py index 8a7ee9f53..1c7086f60 100644 --- a/pcapkit/protocols/transport/sctp.py +++ b/pcapkit/protocols/transport/sctp.py @@ -1906,7 +1906,8 @@ def _make_chunk_donone(self, code: 'Enum_Chunk', chunk: 'Optional[Data_UnknownCh value = chunk.value if len(flags) != 1: - raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid format') + raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid flags: ' + f'must be one octet, not {flags!r}') return Schema_UnknownChunk( type=code, @@ -1965,7 +1966,8 @@ def _make_chunk_data(self, code: 'Enum_Chunk', chunk: 'Optional[Data_DATAChunk]' data = chunk.data if not data: - raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid format') + raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid data: ' + f'must be at least one octet, not {data!r}') return Schema_DATAChunk( type=code, @@ -2028,7 +2030,8 @@ def _make_chunk_init(self, code: 'Enum_Chunk', chunk: 'Optional[Data_INITChunk]' parameters = chunk.parameters if len(flags) != 1: - raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid format') + raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid flags: ' + f'must be one octet, not {flags!r}') if parameters is not None: parameters_value = self._make_sctp_parameters(parameters) @@ -2089,7 +2092,8 @@ def _make_chunk_init_ack(self, code: 'Enum_Chunk', chunk: 'Optional[Data_INITACK parameters = chunk.parameters if len(flags) != 1: - raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid format') + raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid flags: ' + f'must be one octet, not {flags!r}') if parameters is not None: parameters_value = self._make_sctp_parameters(parameters) @@ -2150,7 +2154,8 @@ def _make_chunk_sack(self, code: 'Enum_Chunk', chunk: 'Optional[Data_SACKChunk]' dup_tsn = list(chunk.dup_tsn) if len(flags) != 1: - raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid format') + raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid flags: ' + f'must be one octet, not {flags!r}') blocks = [] # type: list[Schema_GapAckBlock] for block in gap_blocks or []: @@ -2200,7 +2205,8 @@ def _make_chunk_heartbeat(self, code: 'Enum_Chunk', chunk: 'Optional[Data_Heartb parameters = chunk.parameters if len(flags) != 1: - raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid format') + raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid flags: ' + f'must be one octet, not {flags!r}') if parameters is not None: parameters_value = self._make_sctp_parameters(parameters) @@ -2241,7 +2247,8 @@ def _make_chunk_heartbeat_ack(self, code: 'Enum_Chunk', chunk: 'Optional[Data_He parameters = chunk.parameters if len(flags) != 1: - raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid format') + raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid flags: ' + f'must be one octet, not {flags!r}') if parameters is not None: parameters_value = self._make_sctp_parameters(parameters) @@ -2323,7 +2330,8 @@ def _make_chunk_shutdown(self, code: 'Enum_Chunk', chunk: 'Optional[Data_Shutdow cum_tsn_ack = chunk.cum_tsn_ack if len(flags) != 1: - raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid format') + raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid flags: ' + f'must be one octet, not {flags!r}') return Schema_ShutdownChunk( type=code, @@ -2354,7 +2362,8 @@ def _make_chunk_shutdown_ack(self, code: 'Enum_Chunk', chunk: 'Optional[Data_Shu flags = chunk.flags if len(flags) != 1: - raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid format') + raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid flags: ' + f'must be one octet, not {flags!r}') return Schema_ShutdownACKChunk( type=code, @@ -2387,7 +2396,8 @@ def _make_chunk_error(self, code: 'Enum_Chunk', chunk: 'Optional[Data_ErrorChunk error = chunk.error if len(flags) != 1: - raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid format') + raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid flags: ' + f'must be one octet, not {flags!r}') if error is not None: error_value = self._make_sctp_causes(error) @@ -2429,7 +2439,8 @@ def _make_chunk_cookie_echo(self, code: 'Enum_Chunk', chunk: 'Optional[Data_Cook cookie = chunk.cookie if len(flags) != 1: - raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid format') + raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid flags: ' + f'must be one octet, not {flags!r}') return Schema_CookieEchoChunk( type=code, @@ -2460,7 +2471,8 @@ def _make_chunk_cookie_ack(self, code: 'Enum_Chunk', chunk: 'Optional[Data_Cooki flags = chunk.flags if len(flags) != 1: - raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid format') + raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid flags: ' + f'must be one octet, not {flags!r}') return Schema_CookieACKChunk( type=code, @@ -3018,7 +3030,8 @@ def _make_param_hostname(self, code: 'Enum_Parameter', param: 'Optional[Data_Hos if param is not None: name = param.name elif not name.endswith(b'\x00'): - raise ProtocolError(f'{self.alias}: [Param {code}] invalid format') + raise ProtocolError(f'{self.alias}: [Param {code}] invalid name: ' + f'must be NUL-terminated, not {name!r}') return Schema_HostNameAddressParameter( type=code, @@ -3583,7 +3596,8 @@ def _make_cause_invalid_stream(self, code: 'Enum_CauseCode', cause: 'Optional[Da reserved = cause.reserved if len(reserved) != 2: - raise ProtocolError(f'{self.alias}: [Cause {code}] invalid format') + raise ProtocolError(f'{self.alias}: [Cause {code}] invalid reserved: ' + f'must be two octets, not {reserved!r}') return Schema_InvalidStreamIdentifierCause( code=code, diff --git a/pcapkit/protocols/transport/tcp.py b/pcapkit/protocols/transport/tcp.py index 1b660fe9a..86abc1e57 100644 --- a/pcapkit/protocols/transport/tcp.py +++ b/pcapkit/protocols/transport/tcp.py @@ -2971,10 +2971,14 @@ def _make_mptcp_capable(self, subtype: 'Enum_MPTCPOption', opt: 'Optional[Data_M dl_len = opt.dl_len checksum = opt.checksum + names = ('skey', 'rkey', 'dl_len', 'checksum') fields = (skey, rkey, dl_len, checksum) present = fields.index(None) if None in fields else len(fields) - if any(field is not None for field in fields[present:]): - raise ProtocolError(f'{self.alias}: [OptNo {Enum_Option.Multipath_TCP}] invalid format') + for index in range(present, len(fields)): + if fields[index] is not None: + raise ProtocolError( + f'{self.alias}: [OptNo {Enum_Option.Multipath_TCP}] invalid {names[index]}: ' + f'must be None when {names[present]} is None, not {fields[index]!r}') return Schema_MPTCPCapable( kind=cast('Enum_Option', Enum_Option.Multipath_TCP), diff --git a/tests/protocols/test_maker_invalid_format_messages_2296_unit.py b/tests/protocols/test_maker_invalid_format_messages_2296_unit.py new file mode 100644 index 000000000..355fac522 --- /dev/null +++ b/tests/protocols/test_maker_invalid_format_messages_2296_unit.py @@ -0,0 +1,456 @@ +# -*- coding: utf-8 -*- +"""GitHub issue #2296: makers name the argument they rejected. + +Several ``_make_*`` builders refused a caller-supplied argument with the +*reader's* text -- ``: [Chunk|Param|Cause|OptNo N] invalid format``. +Nothing had been parsed at that point, so the message described wire data that +did not exist and told the caller neither which keyword was wrong nor what it +had passed. Each site now uses the house form established by #2232 and #2257, +``: : must be X, not {value!r}``. + +The parse side is deliberately untouched: ``invalid format`` is the right thing +for a reader to say, which is why the ``SCTP: [Chunk 0] invalid format`` pin in +:mod:`tests.protocols.transport.test_transport_edge_roundtrip_unit` -- a parse +case -- is left alone. :class:`ReadersKeepInvalidFormat` below pins that +division for the one maker/reader pair where the two guards check the same +thing. + +Every message is pinned whole, and every maker's valid input is exercised in the +same class, so a guard that starts refusing good input regresses here too. + +:mod:`pcapkit` is imported inside each test, after +:func:`~tests._support.reimport_once_per_class`. + +""" + +from __future__ import annotations + +import datetime +import struct +import unittest +import warnings +from typing import TYPE_CHECKING + +from tests._support import reimport_once_per_class + +if TYPE_CHECKING: + from typing import Any + + +class MakerMessageTestCase(unittest.TestCase): + """Re-import :mod:`pcapkit` once per class, as the suite's convention asks.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def protocol(self, cls: 'Any') -> 'Any': + """An uninitialised protocol instance, for calling a maker directly. + + The makers under test read nothing off ``self`` but ``alias``, so there + is no packet to construct first -- and constructing one would route the + bad argument through a schema that raises its own error instead. + + """ + return object.__new__(cls) + + +class SCTPChunkFlagsMessage(MakerMessageTestCase): + """``_make_chunk_*`` names ``flags`` and shows the octets it was handed.""" + + #: Chunk type per maker, as :rfc:`9260` assigns them. ``_make_chunk_donone`` + #: serves the unsupported types, so it gets one that has no maker of its own. + MAKERS = { + '_make_chunk_donone': 'Padding_Chunk', + '_make_chunk_init': 'Initiation', + '_make_chunk_init_ack': 'Initiation_Acknowledgement', + '_make_chunk_sack': 'Selective_Acknowledgement', + '_make_chunk_heartbeat': 'Heartbeat_Request', + '_make_chunk_heartbeat_ack': 'Heartbeat_Acknowledgement', + '_make_chunk_shutdown': 'Shutdown', + '_make_chunk_shutdown_ack': 'Shutdown_Acknowledgement', + '_make_chunk_error': 'Operation_Error', + '_make_chunk_cookie_echo': 'State_Cookie', + '_make_chunk_cookie_ack': 'Cookie_Acknowledgement', + } + + def test_wrong_width_flags_name_the_argument(self) -> None: + from pcapkit.const.sctp.chunk import Chunk + from pcapkit.protocols.transport.sctp import SCTP + from pcapkit.utilities.exceptions import ProtocolError + + sctp = self.protocol(SCTP) + for maker, name in self.MAKERS.items(): + code = Chunk[name] + for flags in (b'', b'\x00\x00', b'\x00\x00\x00\x00'): + with self.subTest(maker=maker, flags=flags): + with self.assertRaises(ProtocolError) as context: + getattr(sctp, maker)(code, flags=flags) + self.assertEqual( + str(context.exception), + f'SCTP: [Chunk {code}] invalid flags: ' + f'must be one octet, not {flags!r}') + + def test_one_octet_flags_still_build(self) -> None: + from pcapkit.const.sctp.chunk import Chunk + from pcapkit.protocols.transport.sctp import SCTP + + sctp = self.protocol(SCTP) + for maker, name in self.MAKERS.items(): + with self.subTest(maker=maker): + schema = getattr(sctp, maker)(Chunk[name], flags=b'\x01') + self.assertEqual(schema.flags, b'\x01') + + +class SCTPDataChunkMessage(MakerMessageTestCase): + """``_make_chunk_data`` names ``data`` rather than the chunk's format.""" + + def test_empty_data_names_the_argument(self) -> None: + from pcapkit.const.sctp.chunk import Chunk + from pcapkit.protocols.transport.sctp import SCTP + from pcapkit.utilities.exceptions import ProtocolError + + sctp = self.protocol(SCTP) + code = Chunk.Payload_Data + with self.assertRaises(ProtocolError) as context: + sctp._make_chunk_data(code, data=b'') + self.assertEqual(str(context.exception), + f"SCTP: [Chunk {code}] invalid data: " + f"must be at least one octet, not b''") + + def test_one_octet_of_user_data_still_builds(self) -> None: + from pcapkit.const.sctp.chunk import Chunk + from pcapkit.protocols.transport.sctp import SCTP + + sctp = self.protocol(SCTP) + schema = sctp._make_chunk_data(Chunk.Payload_Data, data=b'!') + self.assertEqual(schema.data, b'!') + self.assertEqual(schema.length, 17) + + +class SCTPHostNameMessage(MakerMessageTestCase): + """``_make_param_hostname`` names ``name`` and the terminator it wants.""" + + def test_unterminated_name_names_the_argument(self) -> None: + from pcapkit.const.sctp.parameter import Parameter + from pcapkit.protocols.transport.sctp import SCTP + from pcapkit.utilities.exceptions import ProtocolError + + sctp = self.protocol(SCTP) + code = Parameter.Host_Name_Address + for name in (b'example', b'example.com', b''): + with self.subTest(name=name): + with self.assertRaises(ProtocolError) as context: + sctp._make_param_hostname(code, name=name) + self.assertEqual( + str(context.exception), + f'SCTP: [Param {code}] invalid name: ' + f'must be NUL-terminated, not {name!r}') + + def test_terminated_name_still_builds(self) -> None: + from pcapkit.const.sctp.parameter import Parameter + from pcapkit.protocols.transport.sctp import SCTP + + sctp = self.protocol(SCTP) + schema = sctp._make_param_hostname(Parameter.Host_Name_Address, name=b'example\x00') + self.assertEqual(schema.name, b'example\x00') + + +class SCTPInvalidStreamMessage(MakerMessageTestCase): + """``_make_cause_invalid_stream`` names ``reserved`` and its width.""" + + def test_wrong_width_reserved_names_the_argument(self) -> None: + from pcapkit.const.sctp.cause_code import CauseCode + from pcapkit.protocols.transport.sctp import SCTP + from pcapkit.utilities.exceptions import ProtocolError + + sctp = self.protocol(SCTP) + code = CauseCode.Invalid_Stream_Identifier + for reserved in (b'', b'\x00', b'\x00\x00\x00'): + with self.subTest(reserved=reserved): + with self.assertRaises(ProtocolError) as context: + sctp._make_cause_invalid_stream(code, reserved=reserved) + self.assertEqual( + str(context.exception), + f'SCTP: [Cause {code}] invalid reserved: ' + f'must be two octets, not {reserved!r}') + + def test_two_octet_reserved_still_builds(self) -> None: + from pcapkit.const.sctp.cause_code import CauseCode + from pcapkit.protocols.transport.sctp import SCTP + + sctp = self.protocol(SCTP) + schema = sctp._make_cause_invalid_stream(CauseCode.Invalid_Stream_Identifier, + stream_id=7, reserved=b'\x00\x00') + self.assertEqual(schema.stream_id, 7) + self.assertEqual(schema.reserved, b'\x00\x00') + + +class MHMakeDataMessage(MakerMessageTestCase): + """:meth:`MH.make` names ``data`` and the forms it reads.""" + + def test_unreadable_data_names_the_argument(self) -> None: + from pcapkit.const.mh.packet import Packet + from pcapkit.protocols.internet.mh import MH + from pcapkit.utilities.exceptions import ProtocolError + + mh = self.protocol(MH) + code = Packet.Binding_Refresh_Request + for data in (5, None, 1.5, 'data'): + with self.subTest(data=data): + with self.assertRaises(ProtocolError) as context: + mh.make(type=code, data=data) + self.assertEqual( + str(context.exception), + f'MH: [Type {code}] invalid data: must be bytes, a dict, ' + f'an MH data model or a Packet schema, not {data!r}') + + def test_bytes_data_still_builds(self) -> None: + from pcapkit.const.mh.packet import Packet + from pcapkit.protocols.internet.mh import MH + + mh = self.protocol(MH) + schema = mh.make(type=Packet.Binding_Refresh_Request, data=b'\x00\x00') + self.assertEqual(schema.data, b'\x00\x00') + + +class MHOptionDataMessages(MakerMessageTestCase): + """The MH option makers name ``data`` and the width each one needs.""" + + def test_authorization_data_multiple_of_eight(self) -> None: + from pcapkit.const.mh.option import Option + from pcapkit.protocols.internet.mh import MH + from pcapkit.utilities.exceptions import ProtocolError + + mh = self.protocol(MH) + code = Option.Authorization_Data + for data in (b'a', b'abc', b'a' * 9): + with self.subTest(data=data): + with self.assertRaises(ProtocolError) as context: + mh._make_opt_bad(code, data=data) + self.assertEqual( + str(context.exception), + f'MH: [OptNo {code}] invalid data: ' + f'must be a multiple of eight octets, not {data!r}') + + self.assertEqual(mh._make_opt_bad(code, data=b'a' * 8).data, b'a' * 8) + + def test_badf_authenticator_not_empty(self) -> None: + from pcapkit.const.mh.option import Option + from pcapkit.protocols.internet.mh import MH + from pcapkit.utilities.exceptions import ProtocolError + + mh = self.protocol(MH) + code = Option.Binding_Authorization_Data_for_FMIPv6 + with self.assertRaises(ProtocolError) as context: + mh._make_opt_badf(code, data=b'') + self.assertEqual(str(context.exception), + f"MH: [OptNo {code}] invalid data: " + f"must be at least one octet, not b''") + + self.assertEqual(mh._make_opt_badf(code, data=b'\x01').data, b'\x01') + + def test_auth_data_keeps_the_option_aligned(self) -> None: + from pcapkit.const.mh.option import Option + from pcapkit.protocols.internet.mh import MH + from pcapkit.utilities.exceptions import ProtocolError + + mh = self.protocol(MH) + code = Option.AUTH_OPTION_TYPE + for data in (b'', b'a', b'abc', b'abcd'): + with self.subTest(data=data): + with self.assertRaises(ProtocolError) as context: + mh._make_opt_auth(code, data=data) + self.assertEqual( + str(context.exception), + f'MH: [OptNo {code}] invalid data: ' + f'must be two octets more than a multiple of four, not {data!r}') + + for data in (b'ab', b'abcdef'): + with self.subTest(data=data): + self.assertEqual(mh._make_opt_auth(code, data=data).data, data) + + +class MHPadOptionWarning(MakerMessageTestCase): + """``_make_opt_pad`` warns about ``length``, not about a format.""" + + def test_pad1_with_a_length_warns_by_name(self) -> None: + from pcapkit.const.mh.option import Option + from pcapkit.protocols.internet.mh import MH + from pcapkit.utilities.warnings import ProtocolWarning + + mh = self.protocol(MH) + code = Option.Pad1 + for length in (1, 3): + with self.subTest(length=length): + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + schema = mh._make_opt_pad(code, length=length) + messages = [str(w.message) for w in caught + if issubclass(w.category, ProtocolWarning)] + self.assertEqual(len(messages), 1, messages) + self.assertEqual( + messages[0], + f'MH: [OptNo {code}] invalid length: must be 0 for a Pad1 option, ' + f'not {length!r}; making a PadN') + self.assertEqual(schema.type, Option.PadN) + + def test_pad1_with_no_length_does_not_warn(self) -> None: + from pcapkit.const.mh.option import Option + from pcapkit.protocols.internet.mh import MH + from pcapkit.utilities.warnings import ProtocolWarning + + mh = self.protocol(MH) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + schema = mh._make_opt_pad(Option.Pad1, length=0) + self.assertEqual([str(w.message) for w in caught + if issubclass(w.category, ProtocolWarning)], []) + self.assertEqual(schema.type, Option.Pad1) + + +class TCPMPTCPCapableMessage(MakerMessageTestCase): + """``_make_mptcp_capable`` names the field given and the gap before it.""" + + #: One case per field that can be given without its predecessor: the + #: keywords to pass, then the field named and the absent one it follows. + CASES = ( + ({'skey': None, 'rkey': 1}, 'rkey', 'skey', 1), + ({'skey': None, 'rkey': None, 'dl_len': 7}, 'dl_len', 'skey', 7), + ({'rkey': None, 'dl_len': 7}, 'dl_len', 'rkey', 7), + ({'dl_len': None, 'checksum': b'ab'}, 'checksum', 'dl_len', b'ab'), + ) + + def test_a_field_without_its_predecessor_names_both(self) -> None: + from pcapkit.const.tcp.mp_tcp_option import MPTCPOption + from pcapkit.const.tcp.option import Option + from pcapkit.protocols.transport.tcp import TCP + from pcapkit.utilities.exceptions import ProtocolError + + tcp = self.protocol(TCP) + for kwargs, named, absent, value in self.CASES: + with self.subTest(**kwargs): + with self.assertRaises(ProtocolError) as context: + tcp._make_mptcp_capable(MPTCPOption.MP_CAPABLE, **kwargs) + self.assertEqual( + str(context.exception), + f'TCP: [OptNo {Option.Multipath_TCP}] invalid {named}: ' + f'must be None when {absent} is None, not {value!r}') + + def test_every_rfc_8684_length_still_builds(self) -> None: + from pcapkit.const.tcp.mp_tcp_option import MPTCPOption + from pcapkit.protocols.transport.tcp import TCP + + tcp = self.protocol(TCP) + #: :rfc:`8684` section 3.1's five MP_CAPABLE lengths, by how many of + #: ``skey``, ``rkey``, ``dl_len`` and ``checksum`` are present. + cases = ( + ({'skey': None, 'rkey': None, 'dl_len': None, 'checksum': None}, 4), + ({'skey': 1, 'rkey': None, 'dl_len': None, 'checksum': None}, 12), + ({'skey': 1, 'rkey': 2, 'dl_len': None, 'checksum': None}, 20), + ({'skey': 1, 'rkey': 2, 'dl_len': 3, 'checksum': None}, 22), + ({'skey': 1, 'rkey': 2, 'dl_len': 3, 'checksum': b'ab'}, 24), + ) + for kwargs, length in cases: + with self.subTest(length=length): + schema = tcp._make_mptcp_capable(MPTCPOption.MP_CAPABLE, **kwargs) + self.assertEqual(schema.length, length) + + +class HIPLocatorMessage(MakerMessageTestCase): + """``_make_locator`` names the locator it could not read.""" + + def locator(self, value: 'Any') -> 'Any': + from pcapkit.protocols.data.internet.hip import Locator + + return Locator(traffic=0, type=0, length=16, preferred=False, reserved=0, + lifetime=datetime.timedelta(seconds=0), locator=value) + + def parameter(self, value: 'Any') -> 'Any': + from pcapkit.const.hip.parameter import Parameter + from pcapkit.protocols.data.internet.hip import LocatorSetParameter + + return LocatorSetParameter(type=Parameter.LOCATOR_SET, critical=False, length=24, + locator_set=(self.locator(value),)) + + def test_unreadable_locator_names_the_argument(self) -> None: + from pcapkit.const.hip.parameter import Parameter + from pcapkit.protocols.internet.hip import HIP + from pcapkit.utilities.exceptions import ProtocolError + + hip = self.protocol(HIP) + code = Parameter.LOCATOR_SET + for value in ('::1', 5, None, b'\x00' * 16): + with self.subTest(value=value): + with self.assertRaises(ProtocolError) as context: + hip._make_param_locator_set(code, self.parameter(value), version=2) + self.assertEqual( + str(context.exception), + f'HIPv2: [ParamNo {code}] invalid locator: must be an ' + f'ipaddress.IPv6Address or a LocatorData, not {value!r}') + + def test_an_ipv6_address_locator_still_builds(self) -> None: + import ipaddress + + from pcapkit.const.hip.parameter import Parameter + from pcapkit.protocols.internet.hip import HIP + + hip = self.protocol(HIP) + address = ipaddress.IPv6Address('::1') + schema = hip._make_param_locator_set(Parameter.LOCATOR_SET, + self.parameter(address), version=2) + self.assertEqual(schema.locators[0]['value'], address.packed) + + +class PCAPNGDSBSecretsMessage(MakerMessageTestCase): + """``_make_block_dsb`` names ``secrets_data`` and the forms it reads.""" + + def test_unreadable_secrets_data_names_the_argument(self) -> None: + from pcapkit.protocols.misc.pcapng import PCAPNG + from pcapkit.utilities.exceptions import ProtocolError + + pcapng = self.protocol(PCAPNG) + for secrets_data in (5, None, 1.5, 'secrets'): + with self.subTest(secrets_data=secrets_data): + with self.assertRaises(ProtocolError) as context: + pcapng._make_block_dsb(secrets_data=secrets_data) + self.assertEqual( + str(context.exception), + f'PCAP-NG: [DSB] invalid secrets_data: must be bytes, a dict, ' + f'a DSBSecrets data model or a Schema, not {secrets_data!r}') + + def test_raw_secrets_data_still_builds(self) -> None: + from pcapkit.protocols.misc.pcapng import PCAPNG + + pcapng = self.protocol(PCAPNG) + schema = pcapng._make_block_dsb(secrets_data=b'CLIENT_RANDOM 00 11\n') + self.assertEqual(schema.secrets_length, 20) + + +class ReadersKeepInvalidFormat(MakerMessageTestCase): + """The parse side still says ``invalid format``, as #2296 asks. + + ``_read_chunk_data`` and ``_make_chunk_data`` guard the same thing -- a DATA + chunk with no user data -- so the pair is where a sweep over the makers is + most likely to catch a reader by accident. + + """ + + #: ``sctp/data-no-user-data`` from + #: :mod:`tests.protocols.transport.test_transport_edge_roundtrip_unit`: the + #: SCTP common header, then a DATA chunk declaring its own 16 octets and + #: carrying no user data -- TSN, stream id, stream sequence and PPID only. + NO_USER_DATA = (struct.pack('!HHII', 5000, 5001, 0xdeadbeef, 0) + + struct.pack('!BBH', 0, 0xff, 16) + + b'\x00\x00\x00\x01' + b'\x00\x01\x00\x02' + bytes(4)) + + def test_the_reader_still_reports_an_invalid_format(self) -> None: + from pcapkit.protocols.transport.sctp import SCTP + from pcapkit.utilities.exceptions import ProtocolError + + with self.assertRaises(ProtocolError) as context: + SCTP(self.NO_USER_DATA, len(self.NO_USER_DATA)) + self.assertEqual(str(context.exception), 'SCTP: [Chunk 0] invalid format') + + +if __name__ == '__main__': + unittest.main()