From 0c22e7d271c2b6dbfb5ddefa05c8fc45173df86e Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Wed, 7 Oct 2026 19:58:35 -0400 Subject: [PATCH] fix(protocol): main is red, IPv4 make() with an unparsable raw item no longer raises the wrapped error #1344 pads the IPv4 option area once at the end, so the schema now accepts octets read() then rejects (an option whose length runs past the area), and the reader's bare ProtocolError escaped the wrap #1352 put around the schema re-parse only. unpack() now wraps the read as well when the made header holds raw items, and the wrapped message carries the original one. The same escape is fixed for TCP, HOPOPT, IPv6-Opts, MH, SCTP and HIP. Closes #1394 --- pcapkit/protocols/protocol.py | 14 ++- .../protocols/test_make_raw_item_read_unit.py | 96 +++++++++++++++++++ 2 files changed, 106 insertions(+), 4 deletions(-) create mode 100644 tests/protocols/test_make_raw_item_read_unit.py diff --git a/pcapkit/protocols/protocol.py b/pcapkit/protocols/protocol.py index fd04e0d45b..d685aea8c5 100644 --- a/pcapkit/protocols/protocol.py +++ b/pcapkit/protocols/protocol.py @@ -727,19 +727,25 @@ def unpack(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> '_PT': Raises: ProtocolError: If the made header holds raw items and its packed - octets do not parse. + octets do not parse, either as a schema or by :meth:`read`. + The message carries the original error's, which is also + chained as ``__cause__``. """ packet = kwargs.get('__packet__', {}) # packet data if cast('Optional[_ST]', self.__header__) is None: self.__header__ = cast('_ST', self.__schema__.unpack(self._file, length, packet)) # type: ignore[call-arg,misc] elif self._make_raw and _holds_raw(self.__header__): + # NOTE: A schema may accept octets that :meth:`read` then rejects, + # e.g. an option whose length runs past the option area, so the + # read is covered too (:issue:`1394`). try: parsed = cast('_ST', type(self.__header__).unpack(io.BytesIO(self._data), length, packet)) # type: ignore[call-arg,misc] + if _raw_where_parsed(self.__header__, parsed): + self.__header__ = parsed + return self.read(length, **kwargs) except (BaseError, struct.error) as error: # pylint: disable=no-member - raise ProtocolError(f'{type(self).__name__}: malformed raw item given to make()') from error - if _raw_where_parsed(self.__header__, parsed): - self.__header__ = parsed + raise ProtocolError(f'{type(self).__name__}: malformed raw item given to make(): {error}') from error return self.read(length, **kwargs) @staticmethod diff --git a/tests/protocols/test_make_raw_item_read_unit.py b/tests/protocols/test_make_raw_item_read_unit.py new file mode 100644 index 0000000000..2c1d0e230a --- /dev/null +++ b/tests/protocols/test_make_raw_item_read_unit.py @@ -0,0 +1,96 @@ +# -*- coding: utf-8 -*- +"""A raw ``make()`` item rejected by ``read`` raises the wrapped error. + +GitHub issue #1394: :meth:`~pcapkit.protocols.protocol.ProtocolBase.unpack` +wrapped only the schema re-parse of a made header holding raw items. A raw +item the schema accepts but ``read`` rejects -- an option whose length runs +past the option area, or one cut short of its own fields -- escaped as the +reader's bare :exc:`~pcapkit.utilities.exceptions.ProtocolError`. The read is +now covered too: the error says ``malformed raw item given to make()``, +carries the reader's message, and chains it as ``__cause__``. + +Every case builds its own packet in memory and reads no capture. Classes are +imported inside each test, after +:func:`~tests._support.reimport_once_per_class`, so they belong to the live +:mod:`pcapkit` import. + +""" + +import importlib +import unittest +import warnings +from typing import TYPE_CHECKING + +from tests._support import reimport_once_per_class + +if TYPE_CHECKING: + from typing import Any + + +def _attr(path: 'str') -> 'Any': + module, _, name = path.rpartition('.') + return getattr(importlib.import_module(module), name) + + +#: One raw item per protocol that the schema parses and ``read`` rejects: +#: (protocol class, construction keywords, the reader's own message). +READ_REJECTED = { + # the lone ``0x02`` takes the next item's kind octet as its length, 148, + # which runs past the 8-octet option area + 'IPv4 overrun': ('pcapkit.protocols.internet.ipv4.IPv4', + {'options': [b'\x02', b'\x94\x04\x00\x00'], 'protocol': 253}, + r'IPv4: invalid format'), + 'IPv4 short': ('pcapkit.protocols.internet.ipv4.IPv4', + {'options': [b'\x0c'], 'protocol': 253}, + r'IPv4: \[OptNo 12\] invalid format'), + 'TCP': ('pcapkit.protocols.transport.tcp.TCP', + {'options': [b'\x03']}, + r'TCP: \[OptNo 3\] invalid format'), + 'HOPOPT': ('pcapkit.protocols.internet.hopopt.HOPOPT', + {'extension': True, 'options': [b'\x23']}, + r'HOPOPT: \[OptNo 35\] invalid format'), + 'IPv6_Opts': ('pcapkit.protocols.internet.ipv6_opts.IPv6_Opts', + {'extension': True, 'options': [b'\x26']}, + r'IPv6-Opts: \[OptNo 38\] invalid format'), + 'MH': ('pcapkit.protocols.internet.mh.MH', + {'data': {'options': [b'\x2b']}}, + r'MH: \[Opt 43\] invalid format'), + 'SCTP': ('pcapkit.protocols.transport.sctp.SCTP', + {'chunks': [b'\x01']}, + r'SCTP: \[Chunk 1\] invalid format'), + # ESP_INFO needs a 12-octet body, and this one has 8 + 'HIP': ('pcapkit.protocols.internet.hip.HIP', + {'extension': True, 'parameters': [b'\x00\x41\x00\x08' + bytes(8)]}, + r'HIPv6: \[ParamNo 65\] invalid format'), +} + + +class TestRawItemRejectedByRead(unittest.TestCase): + """A raw item ``read`` rejects raises the wrapped ``ProtocolError`` (#1394).""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_read_rejection_is_wrapped(self) -> None: + exc = _attr('pcapkit.utilities.exceptions.ProtocolError') + for case, (klass, kwargs, message) in READ_REJECTED.items(): + with self.subTest(case=case): + cls = _attr(klass) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + with self.assertRaisesRegex(exc, r'malformed raw item given to make\(\)') as caught: + cls(**kwargs) + # the reader's own error is chained, and its message kept + self.assertIsInstance(caught.exception.__cause__, exc) + self.assertRegex(str(caught.exception.__cause__), message) + self.assertRegex(str(caught.exception), message) + + def test_well_formed_raw_item_still_builds(self) -> None: + # the wider ``try`` leaves a raw item that does parse alone + ipv4 = _attr('pcapkit.protocols.internet.ipv4.IPv4') + proto = ipv4(options=[b'\x94\x04\x00\x00'], protocol=253) + self.assertEqual(proto.data.hex(), '460000180000000000fd00007f0000010000000094040000') + + +if __name__ == '__main__': + unittest.main()