diff --git a/pcapkit/protocols/internet/hip.py b/pcapkit/protocols/internet/hip.py index ccaed5966e..2914db63f0 100644 --- a/pcapkit/protocols/internet/hip.py +++ b/pcapkit/protocols/internet/hip.py @@ -3450,7 +3450,8 @@ 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 IPv6Address or LocatorData, not {value!r}') traffic = locator.traffic type = locator.type diff --git a/pcapkit/protocols/internet/mh.py b/pcapkit/protocols/internet/mh.py index f404287345..d5d670b9db 100644 --- a/pcapkit/protocols/internet/mh.py +++ b/pcapkit/protocols/internet/mh.py @@ -1483,7 +1483,7 @@ 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: unsupported type {data.__class__.__name__}') # 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 @@ -7921,7 +7921,7 @@ 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 for Pad1: {length}; falling back to PadN', ProtocolWarning) type = Enum_Option.PadN # type: ignore[assignment] return Schema_PadOption( @@ -8029,7 +8029,7 @@ def _make_opt_bad(self, type: 'Enum_Option', option: 'Optional[Data_Authorizatio 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: length must be a multiple of 8, not {len(data)}') return Schema_AuthorizationDataOption( type=type, @@ -8359,7 +8359,7 @@ 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: length must make the option a multiple of 4, not {len(data)}') return Schema_AuthOption( type=type, @@ -8621,7 +8621,7 @@ 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: must not be empty, not {data!r}') return Schema_BADFOption( type=type, diff --git a/pcapkit/protocols/misc/pcapng.py b/pcapkit/protocols/misc/pcapng.py index cc119ce97d..fae4601c60 100644 --- a/pcapkit/protocols/misc/pcapng.py +++ b/pcapkit/protocols/misc/pcapng.py @@ -4921,7 +4921,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: ' + f'unsupported type {type(secrets_data).__name__}') # 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 8a7ee9f53c..bd3576760b 100644 --- a/pcapkit/protocols/transport/sctp.py +++ b/pcapkit/protocols/transport/sctp.py @@ -1906,7 +1906,7 @@ 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: must be one octet, not {flags!r}') return Schema_UnknownChunk( type=code, @@ -1965,7 +1965,7 @@ 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: must not be empty') return Schema_DATAChunk( type=code, @@ -2028,7 +2028,7 @@ 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: must be one octet, not {flags!r}') if parameters is not None: parameters_value = self._make_sctp_parameters(parameters) @@ -2089,7 +2089,7 @@ 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: must be one octet, not {flags!r}') if parameters is not None: parameters_value = self._make_sctp_parameters(parameters) @@ -2150,7 +2150,7 @@ 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: must be one octet, not {flags!r}') blocks = [] # type: list[Schema_GapAckBlock] for block in gap_blocks or []: @@ -2200,7 +2200,7 @@ 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: must be one octet, not {flags!r}') if parameters is not None: parameters_value = self._make_sctp_parameters(parameters) @@ -2241,7 +2241,7 @@ 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: must be one octet, not {flags!r}') if parameters is not None: parameters_value = self._make_sctp_parameters(parameters) @@ -2323,7 +2323,7 @@ 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: must be one octet, not {flags!r}') return Schema_ShutdownChunk( type=code, @@ -2354,7 +2354,7 @@ 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: must be one octet, not {flags!r}') return Schema_ShutdownACKChunk( type=code, @@ -2387,7 +2387,7 @@ 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: must be one octet, not {flags!r}') if error is not None: error_value = self._make_sctp_causes(error) @@ -2429,7 +2429,7 @@ 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: must be one octet, not {flags!r}') return Schema_CookieEchoChunk( type=code, @@ -2460,7 +2460,7 @@ 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: must be one octet, not {flags!r}') return Schema_CookieACKChunk( type=code, @@ -3018,7 +3018,7 @@ 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: must be NUL-terminated, not {name!r}') return Schema_HostNameAddressParameter( type=code, @@ -3583,7 +3583,7 @@ 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: 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 1b660fe9a8..ff8075ab4b 100644 --- a/pcapkit/protocols/transport/tcp.py +++ b/pcapkit/protocols/transport/tcp.py @@ -2974,7 +2974,10 @@ def _make_mptcp_capable(self, subtype: 'Enum_MPTCPOption', opt: 'Optional[Data_M 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') + field_names = ('skey', 'rkey', 'dl_len', 'checksum') + index = next(i for i, field in enumerate(fields) if i >= present and field is not None) + raise ProtocolError(f'{self.alias}: [OptNo {Enum_Option.Multipath_TCP}] ' + f'invalid fields: {field_names[index]!r} given without {field_names[index - 1]!r}') return Schema_MPTCPCapable( kind=cast('Enum_Option', Enum_Option.Multipath_TCP), diff --git a/tests/protocols/misc/test_maker_invalid_argument_message_2296_unit.py b/tests/protocols/misc/test_maker_invalid_argument_message_2296_unit.py new file mode 100644 index 0000000000..f473a365cc --- /dev/null +++ b/tests/protocols/misc/test_maker_invalid_argument_message_2296_unit.py @@ -0,0 +1,235 @@ +"""#2296 -- makers name the rejected argument in their error messages. + +Guards against :exc:`~pcapkit.utilities.exceptions.ProtocolError` (and +:class:`~pcapkit.protocols.protocol.ProtocolWarning`) messages in which a +``_make_*`` builder rejects a caller-supplied argument with the parse-error +text ``invalid format``. That text says the wire data is malformed, but here +nothing was parsed: the caller passed a bad value, and the message gave no +hint of which keyword was wrong or what value was received. + +The house form these tests pin is ``invalid : , not +`` -- the maker names the argument, states the requirement, and shows +the offending value -- already used elsewhere in the codebase, e.g. the +mobile network prefix check in ``MH._make_opt_mnp``. The reader side keeps +its ``invalid format`` text, so the ``SCTP: [Chunk 0] invalid format`` pin in +``test_transport_edge_roundtrip_unit`` (a parse case) is untouched. + +Every message here is asserted **in full** rather than by substring, so that +a regression back to the generic text cannot slip through a partial match. + +""" +from __future__ import annotations + +import unittest +import warnings +from datetime import timedelta +from types import SimpleNamespace + +from tests._support import reimport_once_per_class + + +class MakerInvalidArgumentMessageUnitTests(unittest.TestCase): + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_sctp_flags_guards(self) -> None: + """The ``flags`` guards name the argument and show the value. + + ``_make_chunk_init`` stands in for the eleven identical guards; the + guard rejects a two-octet ``flags`` with the requirement spelled out. + + """ + from pcapkit.protocols.transport.sctp import SCTP + from pcapkit.utilities.exceptions import ProtocolError + + proto = SCTP.__new__(SCTP) + with self.assertRaises(ProtocolError) as caught: + proto._make_chunk_init(0, flags=b'\x00\x00') + + self.assertEqual( + str(caught.exception), + "SCTP: [Chunk 0] invalid flags: must be one octet, not b'\\x00\\x00'", + ) + + def test_sctp_data_guard(self) -> None: + """The ``DATA`` chunk's empty-``data`` guard names the argument.""" + from pcapkit.protocols.transport.sctp import SCTP + from pcapkit.utilities.exceptions import ProtocolError + + proto = SCTP.__new__(SCTP) + with self.assertRaises(ProtocolError) as caught: + proto._make_chunk_data(0) + + self.assertEqual(str(caught.exception), 'SCTP: [Chunk 0] invalid data: must not be empty') + + def test_sctp_hostname_guard(self) -> None: + """The host name parameter guard quotes the rejected name.""" + from pcapkit.protocols.transport.sctp import SCTP + from pcapkit.utilities.exceptions import ProtocolError + + proto = SCTP.__new__(SCTP) + with self.assertRaises(ProtocolError) as caught: + proto._make_param_hostname(11, name=b'example') + + self.assertEqual( + str(caught.exception), + "SCTP: [Param 11] invalid name: must be NUL-terminated, not b'example'", + ) + + def test_sctp_cause_reserved_guard(self) -> None: + """The invalid stream identifier cause quotes the ``reserved`` value.""" + from pcapkit.protocols.transport.sctp import SCTP + from pcapkit.utilities.exceptions import ProtocolError + + proto = SCTP.__new__(SCTP) + with self.assertRaises(ProtocolError) as caught: + proto._make_cause_invalid_stream(3, reserved=b'\x00') + + self.assertEqual( + str(caught.exception), + "SCTP: [Cause 3] invalid reserved: must be two octets, not b'\\x00'", + ) + + def test_mh_make_data_guard(self) -> None: + """``MH.make`` names the ``data`` argument and its runtime type. + + The value is deliberately described by type rather than ``repr``: + this branch is reached with an arbitrary object, whose repr may be + unbounded, while the type name is what tells the caller how to fix + the call. + + """ + from pcapkit.protocols.internet.mh import MH + from pcapkit.utilities.exceptions import ProtocolError + + proto = MH.__new__(MH) + with self.assertRaises(ProtocolError) as caught: + proto.make(type=1, data=object()) + + self.assertEqual( + str(caught.exception), + 'MH: [Type 1] invalid data: unsupported type object', + ) + + def test_mh_pad1_length_warning(self) -> None: + """The ``Pad1`` fallback warning states the length and the fallback.""" + from pcapkit.const.mh.option import Option as Enum_Option + from pcapkit.protocols.internet.mh import MH + + proto = MH.__new__(MH) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + proto._make_opt_pad(Enum_Option.Pad1, length=3) + + self.assertEqual(len(caught), 1) + self.assertEqual( + str(caught[0].message), + 'MH: [OptNo 0] invalid length for Pad1: 3; falling back to PadN', + ) + + def test_mh_auth_data_length_guard(self) -> None: + """The binding authorization data guard quotes the data length.""" + from pcapkit.const.mh.option import Option as Enum_Option + from pcapkit.protocols.internet.mh import MH + from pcapkit.utilities.exceptions import ProtocolError + + proto = MH.__new__(MH) + with self.assertRaises(ProtocolError) as caught: + proto._make_opt_bad(Enum_Option.Authorization_Data, data=b'123') + + self.assertEqual( + str(caught.exception), + f'MH: [OptNo {Enum_Option.Authorization_Data}] ' + 'invalid data: length must be a multiple of 8, not 3', + ) + + def test_mh_auth_option_length_guard(self) -> None: + """The auth option guard quotes the data length it cannot align.""" + from pcapkit.const.mh.option import Option as Enum_Option + from pcapkit.protocols.internet.mh import MH + from pcapkit.utilities.exceptions import ProtocolError + + proto = MH.__new__(MH) + with self.assertRaises(ProtocolError) as caught: + proto._make_opt_auth(Enum_Option.Authorization_Data, data=b'123') + + self.assertEqual( + str(caught.exception), + f'MH: [OptNo {Enum_Option.Authorization_Data}] ' + 'invalid data: length must make the option a multiple of 4, not 3', + ) + + def test_mh_badf_empty_data_guard(self) -> None: + """The FMIPv6 binding authorization guard shows the empty value.""" + from pcapkit.const.mh.option import Option as Enum_Option + from pcapkit.protocols.internet.mh import MH + from pcapkit.utilities.exceptions import ProtocolError + + proto = MH.__new__(MH) + with self.assertRaises(ProtocolError) as caught: + proto._make_opt_badf(Enum_Option.Binding_Authorization_Data_for_FMIPv6) + + self.assertEqual( + str(caught.exception), + f'MH: [OptNo {Enum_Option.Binding_Authorization_Data_for_FMIPv6}] ' + "invalid data: must not be empty, not b''", + ) + + def test_tcp_mptcp_field_order_guard(self) -> None: + """The MP_CAPABLE guard names the field and its missing predecessor. + + The check is a prefix rule -- each field requires every field before + it -- so the message pins both ends of the violated pair. + + """ + from pcapkit.const.tcp.option import Option as Enum_Option + from pcapkit.protocols.transport.tcp import TCP + from pcapkit.utilities.exceptions import ProtocolError + + proto = TCP.__new__(TCP) + with self.assertRaises(ProtocolError) as caught: + proto._make_mptcp_capable(Enum_Option.Multipath_TCP, skey=None, rkey=2) + + self.assertEqual( + str(caught.exception), + f"TCP: [OptNo {Enum_Option.Multipath_TCP}] " + "invalid fields: 'rkey' given without 'skey'", + ) + + def test_hip_locator_value_guard(self) -> None: + """The locator guard quotes the value that is neither address nor data.""" + from pcapkit.protocols.data.internet.hip import Locator as Data_Locator + from pcapkit.protocols.internet.hip import HIP + from pcapkit.utilities.exceptions import ProtocolError + + proto = HIP.__new__(HIP) + bad = Data_Locator(locator='not-an-ip', traffic=0, type=0, length=16, + preferred=False, reserved=0, lifetime=timedelta(0)) + param = SimpleNamespace(locator_set=[bad]) + with self.assertRaises(ProtocolError) as caught: + proto._make_param_locator_set(193, version=2, param=param) # type: ignore[arg-type] + + self.assertEqual( + str(caught.exception), + "HIPv2: [ParamNo 193] invalid locator: " + "must be an IPv6Address or LocatorData, not 'not-an-ip'", + ) + + def test_pcapng_dsb_secrets_data_guard(self) -> None: + """The DSB guard names the ``secrets_data`` argument and its type. + + As with ``MH.make``, the unsupported branch receives an arbitrary + object, so the message reports the type name rather than a repr. + + """ + from pcapkit.protocols.misc.pcapng import PCAPNG + from pcapkit.utilities.exceptions import ProtocolError + + proto = PCAPNG.__new__(PCAPNG) + with self.assertRaises(ProtocolError) as caught: + proto._make_block_dsb(secrets_data=object()) + + self.assertEqual( + str(caught.exception), + 'PCAP-NG: [DSB] invalid secrets_data: unsupported type object', + )