From f81b25462a302c1c5d020e4f5fe39f81614492e1 Mon Sep 17 00:00:00 2001 From: lux-liang <249971141+lux-liang@users.noreply.github.com> Date: Mon, 21 Sep 2026 13:09:31 +0800 Subject: [PATCH] corekit: reject short dynamic field buffers Stop callable field lengths from padding beyond the bytes that were read. Preserve the existing fixed-width padding behavior used by higher-level parser diagnostics. --- pcapkit/corekit/fields/field.py | 17 +++++++++- tests/corekit/test_fields_strings.py | 46 ++++++++++++++++++++++++++++ 2 files changed, 62 insertions(+), 1 deletion(-) diff --git a/pcapkit/corekit/fields/field.py b/pcapkit/corekit/fields/field.py index e6f191c6a2..e99bcb3493 100644 --- a/pcapkit/corekit/fields/field.py +++ b/pcapkit/corekit/fields/field.py @@ -7,7 +7,7 @@ from typing import TYPE_CHECKING, Generic, TypeVar, cast from pcapkit.utilities.compat import final -from pcapkit.utilities.exceptions import NoDefaultValue +from pcapkit.utilities.exceptions import FieldValueError, NoDefaultValue __all__ = ['Field'] @@ -233,6 +233,10 @@ def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> '_T': Returns: Unpacked field value. + Raises: + FieldValueError: If ``buffer`` contains fewer octets than a + dynamically sized field declares. + """ # NOTE: ``length`` recomputes struct.calcsize() on every read, so the # three reads this method used to make were three calcsize() calls for @@ -241,6 +245,17 @@ def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> '_T': if not isinstance(buffer, bytes): buffer = buffer.read(length) + buffer_length = len(buffer) + # Fixed-width fields have historically left-padded a short read, and + # callers rely on that while producing their own field-relative errors. + # A callable length is different: it can be derived from the packet being + # parsed, so padding to it must not allocate bytes that never arrived. + if buffer_length < length and getattr(self, '_length_callback', None) is not None: + raise FieldValueError( + f'Field {self.name} requires {length} octets, but only ' + f'{buffer_length} are available.' + ) + value = struct.unpack(self.template, buffer[:length].rjust(length, b'\x00'))[0] return self.post_process(value, packet) diff --git a/tests/corekit/test_fields_strings.py b/tests/corekit/test_fields_strings.py index a5e4b07eee..5b06d557ef 100644 --- a/tests/corekit/test_fields_strings.py +++ b/tests/corekit/test_fields_strings.py @@ -5,6 +5,52 @@ from tests._support import purge_modules +class BytesFieldTests(unittest.TestCase): + """Packing and parsing of :class:`~pcapkit.corekit.fields.strings.BytesField`.""" + + def setUp(self) -> None: + purge_modules(['pcapkit']) + + from pcapkit.corekit.fields.strings import BytesField + + self.BytesField = BytesField + + def test_unpack_rejects_a_short_dynamic_buffer(self) -> None: + """A packet-derived length must not cause a short value to be zero-padded.""" + from pcapkit.utilities.exceptions import FieldValueError + + packet = {'length': 64} + field = self.BytesField(length=lambda pkt: pkt['length'])(packet) + + with self.assertRaisesRegex(FieldValueError, r'requires 64 octets.*only 1'): + field.unpack(b'X', packet) + + def test_unpack_rejects_a_short_dynamic_stream(self) -> None: + """The same length check applies when a field reads from a stream.""" + import io + + from pcapkit.utilities.exceptions import FieldValueError + + packet = {'length': 64} + field = self.BytesField(length=lambda pkt: pkt['length'])(packet) + + with self.assertRaisesRegex(FieldValueError, r'requires 64 octets.*only 1'): + field.unpack(io.BytesIO(b'X'), packet) + + def test_unpack_preserves_complete_and_trailing_data(self) -> None: + """Complete values still parse, and bytes after the field remain ignored.""" + field = self.BytesField(length=4) + + self.assertEqual(field.unpack(b'data', {}), b'data') + self.assertEqual(field.unpack(b'dataextra', {}), b'data') + + def test_unpack_preserves_fixed_length_short_input_compatibility(self) -> None: + """Fixed-width fields retain their historical short-input padding.""" + field = self.BytesField(length=4) + + self.assertEqual(field.unpack(b'X', {}), b'\x00\x00\x00X') + + class BitFieldTests(unittest.TestCase): """Packing and parsing of :class:`~pcapkit.corekit.fields.strings.BitField`."""