From b81334ee0d3eae1ece0f391177bf23802af8578c Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Thu, 17 Sep 2026 20:04:11 -0400 Subject: [PATCH] corekit: unpack a ListField's schema items from the configured field (#433) * `ListField.unpack`'s schema branch built a per-item field via `self._item_type(packet)` -- which applies `SchemaField.__call__`'s `callback` and `length_callback` to a fresh copy -- and then unpacked from `self._item_type` itself, discarding that copy. Any per-item `callback` or `length_callback` a `SchemaField` item carried was silently dropped. * Fix: unpack from `field`, the configured copy, instead of from `self._item_type`. * Latent rather than live: all four in-tree `ListField` declarations whose item is a `SchemaField` (`mh.py`, `hip.py`, `tcp.py`, `sctp.py`) pass only `schema=` and, in two cases, a static `length=`; none passes `callback` or `length_callback`, so the discarded copy was always equivalent to the original and nothing parses differently. Regenerated sample captures are byte-identical to a pristine `origin/main` tree. * `pack` was checked for the mirror-image problem and does not have it: it never builds a per-item configured copy to discard in the first place, and `SchemaField.pack` does not consult `_length`/`_length_callback` at all. * Added `ListFieldSchemaItemTests`, constructing a `ListField` directly with a `SchemaField` item carrying a `length_callback` and one carrying a `callback`, asserting each is honoured -- the shape the four in-tree declarations hide. Full suite: 928 passed, 17 skipped, 1268 subtests passed (baseline on a pristine a4c8d62b1: 926 passed, 17 skipped, 1268 subtests -- the difference is exactly the two new tests). --- pcapkit/corekit/fields/collections.py | 2 +- tests/corekit/test_fields_collections.py | 106 +++++++++++++++++++++++ 2 files changed, 107 insertions(+), 1 deletion(-) diff --git a/pcapkit/corekit/fields/collections.py b/pcapkit/corekit/fields/collections.py index 4465eac4c2..9a7981fb86 100644 --- a/pcapkit/corekit/fields/collections.py +++ b/pcapkit/corekit/fields/collections.py @@ -166,7 +166,7 @@ def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> 'byte field = self._item_type(packet) if is_schema: - data = cast('SchemaField', self._item_type).unpack(file, packet) + data = cast('SchemaField', field).unpack(file, packet) end = file.tell() if end <= offset: diff --git a/tests/corekit/test_fields_collections.py b/tests/corekit/test_fields_collections.py index 7710b40115..8606b9b29a 100644 --- a/tests/corekit/test_fields_collections.py +++ b/tests/corekit/test_fields_collections.py @@ -314,5 +314,111 @@ def subclasses(cls: 'Any') -> 'Any': self.assertEqual(sorted(fell_back), []) +class ListFieldSchemaItemTests(unittest.TestCase): + """``ListField.unpack``'s schema branch must unpack from the *configured* + per-item field, not from ``self._item_type`` itself. + + ``field = self._item_type(packet)`` builds a per-item copy through + :meth:`SchemaField.__call__ `, + which applies both ``callback`` and ``length_callback`` to that copy -- the + schema branch then unpacked from ``self._item_type`` instead of from + ``field``, discarding whatever the copy carries. C.f. #433. + + No declaration in this package passes either argument, so nothing here + parses differently today; both cases below construct a ``ListField`` + directly rather than through one of the four in-tree declarations, since + those are exactly the shape that hides the bug. + + """ + + def setUp(self) -> None: + purge_modules(['pcapkit']) + + def test_a_length_callback_is_honoured(self) -> None: + """The per-item field's own resolved length must reach its schema. + + ``field``'s ``length_callback`` resolves to ``2``; ``self._item_type``, + never having been called, is stuck at the ``-1`` its constructor left + it with. ``SchemaField.unpack`` threads its own ``self.length`` through + to ``Item.unpack``'s ``length`` argument, which lands in + ``packet['__length__']`` -- so which one was used is directly visible + to ``Item.pre_unpack`` without ``Item`` ever needing to consume it. + + """ + from pcapkit.corekit.fields.collections import ListField + from pcapkit.corekit.fields.misc import SchemaField + from pcapkit.corekit.fields.numbers import UInt8Field + from pcapkit.protocols.schema.schema import Schema, schema_final + + recorded = [] # type: list[int] + + @schema_final + class Item(Schema): + """A single fixed-width byte: its own width never depends on + ``__length__``, so the list keeps making progress whichever length + got recorded.""" + + marker: 'int' = UInt8Field() + + @classmethod + def pre_unpack(cls, packet: 'dict[str, Any]') -> 'None': + recorded.append(packet['__length__']) + + item_field = SchemaField(schema=Item, length=lambda pkt: 2) + list_field = ListField(length=2, item_type=item_field) + + list_field.unpack(b'\x01\x02', {}) + + self.assertEqual(recorded, [2, 2]) + + def test_a_callback_is_honoured(self) -> None: + """The per-item field's ``callback`` mutation must reach ``unpack``. + + The callback is evaluated regardless: ``field = self._item_type(packet)`` + runs it for its side effect even on the unpatched code. So what this + proves is not that the callback *fires*, but that the mutation it made + on its copy is what ``unpack`` actually used. It alternates + ``field._schema`` between two otherwise-identical schemas, one per + item; unpacking from ``self._item_type`` instead would use the schema + fixed at construction time for every item. + + """ + from pcapkit.corekit.fields.collections import ListField + from pcapkit.corekit.fields.misc import SchemaField + from pcapkit.corekit.fields.numbers import UInt8Field + from pcapkit.protocols.schema.schema import Schema, schema_final + + seen = [] # type: list[str] + + @schema_final + class TypeA(Schema): + marker: 'int' = UInt8Field() + + @classmethod + def pre_unpack(cls, packet: 'dict[str, Any]') -> 'None': + seen.append('A') + + @schema_final + class TypeB(Schema): + marker: 'int' = UInt8Field() + + @classmethod + def pre_unpack(cls, packet: 'dict[str, Any]') -> 'None': + seen.append('B') + + counter = {'n': 0} + + def alternate(field: 'Any', packet: 'dict[str, Any]') -> 'None': + field._schema = TypeB if counter['n'] % 2 else TypeA # pylint: disable=protected-access + counter['n'] += 1 + + item_field = SchemaField(length=1, schema=TypeA, callback=alternate) + list_field = ListField(length=2, item_type=item_field) + + list_field.unpack(b'\x01\x02', {}) + + self.assertEqual(seen, ['A', 'B']) + + if __name__ == '__main__': unittest.main()