From e233a7390051a9dccf86834dfe72e28c508c3338 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Tue, 6 Oct 2026 17:02:31 -0400 Subject: [PATCH] fix(corekit): a length-less SchemaField passes -1 to its schema, so every list item warns 'packet length < 0' SchemaField.unpack passed its "undeclared" length of -1 on to the nested schema, seeding __length__ negative, so every sub-field of every item warned. It now measures the remainder of the buffer and passes that as a declared length. Passing None instead would let prepare() turn an exhausted buffer into its quiet StreamEOFError, ending an extraction on a truncated item; a declared zero keeps that a FieldValueError as before. Closes #1148 --- pcapkit/corekit/fields/misc.py | 16 ++- pcapkit/protocols/schema/internet/mh.py | 15 +-- ...est_fields_misc_schema_length_1148_unit.py | 118 ++++++++++++++++++ 3 files changed, 141 insertions(+), 8 deletions(-) create mode 100644 tests/corekit/test_fields_misc_schema_length_1148_unit.py diff --git a/pcapkit/corekit/fields/misc.py b/pcapkit/corekit/fields/misc.py index 07c1f35c4b..d66dcdd68e 100644 --- a/pcapkit/corekit/fields/misc.py +++ b/pcapkit/corekit/fields/misc.py @@ -742,8 +742,22 @@ def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> '_TS' else: file = buffer + # NOTE: a negative length means the field was declared without one, so + # the nested schema is sized by whatever remains in ``file`` instead. + # Passing ``-1`` on would seed its ``__length__`` negative, and every + # sub-field read would then warn ``packet length < 0``. The remainder is + # measured here rather than left to ``prepare`` (by passing + # :data:`None`), because there a remainder of zero raises the quiet + # end-of-stream signal that ends a whole extraction, whereas here it + # only means the item is truncated. + length = self.length + if length < 0: + current = file.tell() + length = file.seek(0, io.SEEK_END) - current + file.seek(current) + packet.update(self._packet) - return cast('_TS', self._schema.unpack(file, self.length, # type: ignore[call-arg,misc] + return cast('_TS', self._schema.unpack(file, length, # type: ignore[call-arg,misc] nested_packet_context(packet))) diff --git a/pcapkit/protocols/schema/internet/mh.py b/pcapkit/protocols/schema/internet/mh.py index 6545dccb05..bd7e1e8d38 100644 --- a/pcapkit/protocols/schema/internet/mh.py +++ b/pcapkit/protocols/schema/internet/mh.py @@ -1373,13 +1373,14 @@ class ContextRequestOption(Option, code=Enum_Option.Context_Request_Option): :class:`~pcapkit.corekit.fields.collections.ListField` of a nested schema. Each entry is self-describing -- a type octet, a length octet and that many octets of value [:rfc:`5949#section-6.2.1`] -- so a nested schema would need - a :class:`~pcapkit.corekit.fields.misc.SchemaField` with no fixed length, - and such a field is handed ``-1`` as its length and hands that on as the - nested schema's ``__length__``. Every field then decrements a budget that - started negative, and parsing a perfectly valid option emits a - :class:`~pcapkit.utilities.warnings.SchemaWarning` per field saying - ``packet length < 0``. Decoding the run directly costs a few lines and - keeps a valid packet quiet. + a :class:`~pcapkit.corekit.fields.misc.SchemaField` with no fixed length. + Such a field sizes the nested schema by what is left of the run, which + parses a valid option cleanly, but an entry whose length overruns the + option would then only raise a + :class:`~pcapkit.utilities.warnings.SchemaWarning` and come back with its + value padded out by zero octets. Decoding the run directly costs a few + lines and rejects that entry with + :exc:`~pcapkit.utilities.exceptions.ProtocolError` instead. """ diff --git a/tests/corekit/test_fields_misc_schema_length_1148_unit.py b/tests/corekit/test_fields_misc_schema_length_1148_unit.py new file mode 100644 index 0000000000..786a16284b --- /dev/null +++ b/tests/corekit/test_fields_misc_schema_length_1148_unit.py @@ -0,0 +1,118 @@ +# -*- coding: utf-8 -*- +"""A length-less ``SchemaField`` sizes its nested schema by what is left to read. + +GitHub issue #1148. ``SchemaField``'s default length is ``-1``, meaning "not +declared", and :meth:`SchemaField.unpack +` passed that ``-1`` straight on +as the nested schema's length. The nested schema's ``__length__`` therefore +started negative, and every sub-field it read warned ``packet length < 0`` -- +six warnings per HIP ``Locator`` and per MH ``CGAParameter``, on input that +parses correctly. + +The nested schema is now told the true remainder of the buffer. A remainder of +zero is passed as a declared length, so a truncated item is still reported as +malformed data rather than as the quiet end-of-stream signal that +:func:`~pcapkit.utilities.decorators.prepare` raises for a derived zero. +""" + +import unittest +import warnings + +from pcapkit.const.hip.parameter import Parameter as Enum_Parameter +from pcapkit.const.mh.option import Option as Enum_Option +from pcapkit.corekit.fields.collections import ListField +from pcapkit.corekit.fields.misc import SchemaField +from pcapkit.corekit.fields.numbers import UInt8Field, UInt16Field +from pcapkit.protocols.schema.internet.hip import LocatorSetParameter +from pcapkit.protocols.schema.internet.mh import CGAParametersOption +from pcapkit.protocols.schema.schema import Schema +from pcapkit.utilities.exceptions import FieldValueError +from pcapkit.utilities.warnings import SchemaWarning + +#: One plain IPv6 locator: traffic, type 0, ``Locator Length`` 4, flags, lifetime. +LOCATOR = bytes([0, 0, 4, 1]) + (16).to_bytes(4, 'big') + bytes(range(16)) + +#: One CGA Parameters structure with a 6-octet public key and no extensions. +CGA_PARAMETER = b'\x11' * 16 + b'\x22' * 8 + b'\x01' + bytes([0x30, 4]) + b'ABCD' + + +class Pair(Schema): + """Four-octet schema used as a list item.""" + + a: 'int' = UInt16Field() + b: 'int' = UInt16Field() + + +class LengthLessList(Schema): + """A list of :class:`Pair` items declared without a length.""" + + n: 'int' = UInt8Field() + items: 'list[Pair]' = ListField(length=lambda pkt: pkt['n'], item_type=SchemaField(schema=Pair)) + + +class ShortFixedList(Schema): + """A list of :class:`Pair` items declared one octet shorter than they are.""" + + n: 'int' = UInt8Field() + items: 'list[Pair]' = ListField(length=lambda pkt: pkt['n'], item_type=SchemaField(length=3, schema=Pair)) + + +def locator_set(count: 'int', truncate: 'int' = 0) -> 'bytes': + body = LOCATOR * count + raw = int(Enum_Parameter.LOCATOR_SET).to_bytes(2, 'big') + len(body).to_bytes(2, 'big') + body + raw += b'\x00' * (-len(raw) % 8) + return raw[:len(raw) - truncate] + + +def cga_option(body: 'bytes') -> 'bytes': + return bytes([int(Enum_Option.CGA_Parameters), len(CGA_PARAMETER)]) + body + + +class LengthLessSchemaFieldTests(unittest.TestCase): + + def unpack(self, schema: 'type[Schema]', data: 'bytes') -> 'tuple[Schema, list[str]]': + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + result = schema.unpack(data) + return result, [str(w.message) for w in caught if issubclass(w.category, SchemaWarning)] + + def test_hip_locators_do_not_warn(self) -> 'None': + for count in (1, 2, 3): + with self.subTest(count=count): + param, warned = self.unpack(LocatorSetParameter, locator_set(count)) + self.assertEqual(warned, []) + self.assertEqual(len(param.locators), count) + self.assertEqual(param.locators[-1].lifetime, 16) + + def test_mh_cga_parameter_does_not_warn(self) -> 'None': + option, warned = self.unpack(CGAParametersOption, cga_option(CGA_PARAMETER)) + self.assertEqual(warned, []) + self.assertEqual(len(option.parameters), 1) + self.assertEqual(option.parameters[0].public_key, b'0\x04ABCD') + + def test_length_less_items_do_not_warn(self) -> 'None': + result, warned = self.unpack(LengthLessList, b'\x08' + bytes(range(8))) + self.assertEqual(warned, []) + self.assertEqual([(item.a, item.b) for item in result.items], [(0x0001, 0x0203), (0x0405, 0x0607)]) + + def test_truncated_variable_item_still_warns_and_raises(self) -> 'None': + # A remainder of zero must not become ``prepare``'s quiet + # ``StreamEOFError``, which an extractor reads as end of capture. + for name, schema, data in ( + ('hip', LocatorSetParameter, locator_set(2, truncate=10)), + ('mh', CGAParametersOption, cga_option(CGA_PARAMETER[:-3])), + ): + with self.subTest(name=name): + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + with self.assertRaises(FieldValueError): + schema.unpack(data) + self.assertTrue(any(issubclass(w.category, SchemaWarning) for w in caught)) + + def test_fixed_length_item_running_short_still_warns(self) -> 'None': + _, warned = self.unpack(ShortFixedList, b'\x06' + bytes(range(6))) + self.assertEqual(warned, ['packet length < 0: -1'] * 2) + + +if __name__ == '__main__': + unittest.main()