From b9f4f21db1171f4c273da73c5ffe7e1a7293c318 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Tue, 3 Nov 2026 22:54:44 -0500 Subject: [PATCH] fix(corekit): read a counted list's cut item as a short read, not a negative length (#2749) - A standalone IGMPv3 Report cut inside a group record read TRUNCATED with its short-read record, and also warned `SchemaWarning: packet length < 0` for each field of the record past the cut. Report.records counts its records and runs to the end of the data, and each record is sized by the octets left to it, so Schema.unpack read the cut record's fields as overrunning a given length. - ListField now reads a schema item it counts with `__data_end__` in its packet; Schema.unpack clears the key as it starts and warns no negative length for a schema whose length ends with the data. Uncounted lists, items past the count and cut headers warn as before; inside a frame the count is unset, so frames parsed through IPv4 are unchanged. - Tests: the Report cut inside each record's fixed fields, source list and aux data, with a malformed record behind IPv4 still warning; a sweep over every list of schema items in the schema registry, with malformed-length controls. Closes #2749 --- pcapkit/corekit/fields/collections.py | 13 +- pcapkit/protocols/schema.py | 17 +- .../test_igmpv3_record_cut_quiet_2749_unit.py | 181 +++++++++++++++ .../protocols/test_list_item_cut_2749_unit.py | 215 ++++++++++++++++++ 4 files changed, 421 insertions(+), 5 deletions(-) create mode 100644 tests/protocols/internet/test_igmpv3_record_cut_quiet_2749_unit.py create mode 100644 tests/protocols/test_list_item_cut_2749_unit.py diff --git a/pcapkit/corekit/fields/collections.py b/pcapkit/corekit/fields/collections.py index f1de762efc..17ba1dd80c 100644 --- a/pcapkit/corekit/fields/collections.py +++ b/pcapkit/corekit/fields/collections.py @@ -87,11 +87,15 @@ class ListField(FieldBase[List[_TL]], Generic[_TL]): A list whose octets hold fewer items than ``count`` ended with the data before the rest, and reports them as :attr:`shortfall`, so that :meth:`Schema.unpack ` - records the cut; the items absent are not made up. + records the cut; the items absent are not made up. A schema item it counts + is read with ``__data_end__`` in its packet, since its octets end where the + data does, so an item the data ends inside records the cut as its short + read rather than warn that its fields run past the length it was given. .. versionchanged:: 1.6.0 - Takes ``count`` (:issue:`2711`). + Takes ``count`` (:issue:`2711`), and reads a schema item it counts as + ending with the data (:issue:`2749`). """ @@ -307,7 +311,10 @@ def _unpack_items(self, file: 'IO[bytes]', packet: 'dict[str, Any]', is_schema: field = item_type(packet) if is_schema: - data = cast('SchemaField', field).unpack(file, packet) + # NOTE: A counted list runs to the end of the data, so an item it counts + # running past the octets left is cut, not overrunning a length (#2749). + context = dict(packet, __data_end__=True) if len(temp) < self._count else packet + data = cast('SchemaField', field).unpack(file, context) end = file.tell() if end <= offset: diff --git a/pcapkit/protocols/schema.py b/pcapkit/protocols/schema.py index 2ece409a3b..bf3f738dc3 100644 --- a/pcapkit/protocols/schema.py +++ b/pcapkit/protocols/schema.py @@ -759,10 +759,20 @@ def unpack(cls, data: 'bytes | IO[bytes]', read rather than refuse it. The key is cleared as unpacking starts, so a nested schema finds only its own. + Fields running past ``length`` warn that ``__length__`` fell below + zero, unless a ``__data_end__`` key in ``packet`` says that + ``length`` ends where the data does, as a + :class:`~pcapkit.corekit.fields.collections.ListField` says of an + item it counts. The fields past it are then the cut the short read + records, and warn nothing (:issue:`2749`). The key is cleared as + unpacking starts too. + .. versionchanged:: 1.6.0 The field the data ends inside is given to the fields after it, as - ``__short_read__`` in ``packet`` (:issue:`2727`). + ``__short_read__`` in ``packet`` (:issue:`2727`), and a schema whose + ``length`` ends with the data reads its fields past it as cut, not as + a negative length (:issue:`2749`). """ # force cast arg type since decorator changed their signatures @@ -774,6 +784,9 @@ def unpack(cls, data: 'bytes | IO[bytes]', # NOTE: ``packet`` may be an enclosing schema's, copied into a nested # context, so its short read is not this schema's (#2727). packet.pop('__short_read__', None) + # NOTE: Fields running past a length that ends with the data are the cut its + # short read records, not an overrun (#2749). + data_end = packet.pop('__data_end__', False) self = cls.__new__(cls) for field in self.__fields__.values(): @@ -906,7 +919,7 @@ def unpack(cls, data: 'bytes | IO[bytes]', self._note_short_read(field.name, length, byte, packet) packet['__length__'] -= length - if packet['__length__'] < 0: + if packet['__length__'] < 0 and not data_end: warn(f'packet length < 0: {packet["__length__"]}', SchemaWarning, stacklevel=stacklevel()) diff --git a/tests/protocols/internet/test_igmpv3_record_cut_quiet_2749_unit.py b/tests/protocols/internet/test_igmpv3_record_cut_quiet_2749_unit.py new file mode 100644 index 0000000000..0ffe4528b1 --- /dev/null +++ b/tests/protocols/internet/test_igmpv3_record_cut_quiet_2749_unit.py @@ -0,0 +1,181 @@ +# -*- coding: utf-8 -*- +"""An IGMPv3 Report parsed on its own and cut inside a group record reads quietly. + +GitHub issue #2749: built directly on octets that end inside a group record, an IGMPv3 +Report recorded the cut and read ``TRUNCATED``, and also warned ``SchemaWarning: packet +length < 0`` once for each field of the record past the cut, between one and five times. +Its record list counts the records the Number of Group Records gives and runs to the end +of the data (#2711), and each record is sized by the octets left to it, so +:meth:`Schema.unpack ` read the fields of the +record the data ends inside as running past a length it had been given. +:class:`~pcapkit.corekit.fields.collections.ListField` now reads a schema item it counts +as ending with the data, and ``Schema.unpack`` reads such an item's fields past its length +as the cut its short read records. + +The Report below is cut inside its first, a middle and its last record, inside a record's +fixed fields, its source list and its auxiliary data, and built directly with and without +a length. It must read ``TRUNCATED`` with its short-read record, report nothing on the +warning channel or the ``pcapkit`` logger, and rebuild from its data model and its dict as +captured. Behind IPv4 the count sizes nothing, so the change does not reach it: a whole +Report there whose record counts more sources than it holds is no cut, and still warns. + +The messages are built here; this module is unit-tier and reads no capture file. + +""" +from __future__ import annotations + +import contextlib +import logging +import unittest +import warnings +from typing import TYPE_CHECKING +from unittest import mock + +from tests._support import reimport_once_per_class +from tests.foundation import _roundtrip as wire + +if TYPE_CHECKING: + from typing import Any, Iterator + +#: Test tier (#1958): hand-picked parse -> rebuild round trips. +__tier__ = 'ALL' + + +def record(sources: int, aux: int = 0, *, declared: 'int | None' = None) -> bytes: + """A MODE_IS_INCLUDE record for 239.1.1.1 with ``sources`` sources and ``aux`` words of + auxiliary data, its Number of Sources ``declared``, or ``sources`` if not given.""" + count = sources if declared is None else declared + return (bytes([1, aux]) + count.to_bytes(2, 'big') + bytes([239, 1, 1, 1]) + + bytes([10, 0, 0, 1]) * sources + b'AUX!' * aux) + + +def report(*records: bytes) -> bytes: + """An IGMPv3 Report of ``records``, checksummed.""" + rest = len(records).to_bytes(2, 'big') + b''.join(records) + head = b'\x22\x00' + return head + wire.checksum(head + bytes(4) + rest).to_bytes(2, 'big') + b'\x00\x00' + rest + + +def wrong(msg: bytes) -> bytes: + """``msg`` with its checksum flipped in both octets.""" + return msg[:2] + (int.from_bytes(msg[2:4], 'big') ^ 0x0101).to_bytes(2, 'big') + msg[4:] + + +#: The Report's records: two sources; one source and a word of auxiliary data; two +#: sources and two words of it. +RECORDS = (record(2), record(1, aux=1), record(2, aux=2)) +#: The Report itself. +MESSAGE = report(*RECORDS) +#: Each record's ordinal name, in order. +ORDINALS = ('first', 'middle', 'last') + + +def cuts() -> 'Iterator[tuple[str, str, int]]': + """Every cut strictly inside a record: the record, the part its first missing octet is + in, and the octets kept.""" + start = 8 + for ordinal, item in zip(ORDINALS, RECORDS): + sources = int.from_bytes(item[2:4], 'big') + for offset in range(1, len(item)): + part = 'fixed fields' if offset < 8 else 'source list' if offset < 8 + 4 * sources else 'aux data' + yield ordinal, part, start + offset + start += len(item) + + +class _Logged(logging.Handler): + """Collect what the ``pcapkit`` logger reports.""" + + def __init__(self) -> None: + super().__init__(logging.DEBUG) + self.messages = [] # type: list[str] + + def emit(self, record: 'logging.LogRecord') -> None: # pylint: disable=redefined-outer-name + self.messages.append(record.getMessage()) + + +@contextlib.contextmanager +def _reported() -> 'Iterator[tuple[list[str], list[warnings.WarningMessage]]]': + """Collect what the block reports, on the logger and the :mod:`warnings` channel.""" + handler = _Logged() + logger = logging.getLogger('pcapkit') + logger.addHandler(handler) + try: + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + yield handler.messages, caught + finally: + logger.removeHandler(handler) + + +class ReportRecordCutTests(unittest.TestCase): + """A Report parsed on its own and cut inside a record records the cut and reads quietly.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + from pcapkit.corekit.checksum.internet import ChecksumStatus + from pcapkit.protocols.internet.igmpv3.protocol import IGMPv3 + from pcapkit.protocols.internet.ipv4.protocol import IPv4 + from pcapkit.utilities import config + + self.IGMPv3 = IGMPv3 + self.IPv4 = IPv4 + self.Status = ChecksumStatus + patch = mock.patch.object(config, 'CHECKSUM_VERIFY', True) + patch.start() + self.addCleanup(patch.stop) + + def direct(self, data: bytes, sized: bool) -> 'tuple[Any, list[str]]': + """``data`` built directly as IGMPv3, given its length if ``sized``, and what it reported.""" + with _reported() as (logged, caught): + layer = self.IGMPv3(data, len(data)) if sized else self.IGMPv3(data) + return layer, logged + [f'{item.category.__name__}: {item.message}' for item in caught] + + def test_the_cuts_reach_every_part_of_every_record(self) -> None: + parts = {(ordinal, part) for ordinal, part, _ in cuts()} + self.assertEqual(parts, {(ordinal, part) for ordinal in ORDINALS for part in ('fixed fields', 'source list')} + | {('middle', 'aux data'), ('last', 'aux data')}) + + def test_a_cut_inside_a_record_reads_truncated_and_quietly(self) -> None: + for ordinal, part, keep in cuts(): + for bad in (False, True): + for sized in (False, True): + with self.subTest(record=ordinal, part=part, captured=keep, bad=bad, sized=sized): + layer, reported = self.direct((wrong(MESSAGE) if bad else MESSAGE)[:keep], sized) + self.assertEqual(reported, []) + self.assertEqual(layer.info.get('__short_read__'), ('body', keep - 1)) + self.assertIs(layer.info.checksum_status, self.Status.TRUNCATED) + self.assertEqual(len(layer.info.records), ORDINALS.index(ordinal) + 1) + + def test_a_cut_inside_a_record_rebuilds_as_captured_and_quietly(self) -> None: + for ordinal, part, keep in cuts(): + layer, _ = self.direct(MESSAGE[:keep], True) + for path, data in (('model', layer.info), ('dict', layer.info.to_dict())): + with self.subTest(record=ordinal, part=part, captured=keep, path=path): + with _reported() as (logged, caught): + rebuilt = self.IGMPv3.from_data(data) + self.assertEqual((rebuilt.data, rebuilt.info.checksum_status), + (MESSAGE[:keep], self.Status.TRUNCATED)) + self.assertEqual((logged, [str(item.message) for item in caught]), ([], [])) + + def test_a_whole_report_reads_valid_and_quietly(self) -> None: + for sized in (False, True): + with self.subTest(sized=sized): + layer, reported = self.direct(MESSAGE, sized) + self.assertEqual(reported, []) + self.assertIsNone(layer.info.get('__short_read__')) + self.assertIs(layer.info.checksum_status, self.Status.VALID) + self.assertEqual(self.IGMPv3.from_data(layer.info).data, MESSAGE) + + def test_a_record_overstating_its_sources_behind_ipv4_still_warns(self) -> None: + # IPv4 says where the message ends, so the record's source list and auxiliary data + # run past a length the frame declares: a malformed record, not a cut. + msg = report(record(1, declared=5)) + with _reported() as (_, caught): + layer = self.IPv4(wire.ipv4(msg, proto=2)).payload + self.assertIsInstance(layer, self.IGMPv3) + self.assertEqual([str(item.message) for item in caught if item.category.__name__ == 'SchemaWarning'], + ['packet length < 0: -16', 'packet length < 0: -16']) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/test_list_item_cut_2749_unit.py b/tests/protocols/test_list_item_cut_2749_unit.py new file mode 100644 index 0000000000..d1f5684481 --- /dev/null +++ b/tests/protocols/test_list_item_cut_2749_unit.py @@ -0,0 +1,215 @@ +# -*- coding: utf-8 -*- +"""Every list of schema items reads an item the data ends inside as a cut, quietly. + +GitHub issue #2749: an IGMPv3 Report parsed on its own and cut inside a group record +warned ``SchemaWarning: packet length < 0`` for each field of the record past the cut. Its +record list counts its records and runs to the end of the data, and each record is sized +by the octets left to it, so the record the data ends inside read its fields past them as +running past a length it had been given. A +:class:`~pcapkit.corekit.fields.collections.ListField` now reads a schema item it counts +as ending with the data, and :meth:`Schema.unpack +` reads its fields past that length as the cut +its short read records. + +The sweep finds every list of schema items under :mod:`pcapkit.protocols` from its schema +modules, as :file:`tests/corekit/test_schema_roundtrip_unit.py` finds the schemas, and +reads a minimal item of each, built as that module builds one, in the two shapes a list +ends in: counted, running to the end of the data, as the Report's records do, and sized +by a header field behind a length that runs past the data, as the others are. Each cut +inside the item is refused with an in-library error, or reads with no ``packet length < +0`` warning, records the cut, and packs back to the octets read. Option lists read their +items through their registries and record a cut in their area themselves (#2694), so the +sweep leaves them out. + +Beside it, a list sized by a header field shorter than its item, with the data whole, is +a malformed length rather than a cut, and so is an item past the items a list counts; +both still warn. + +Each test imports :mod:`pcapkit` itself, after +:func:`~tests._support.reimport_once_per_class`, and builds its own octets. + +""" +from __future__ import annotations + +import unittest +import warnings +from typing import TYPE_CHECKING + +from tests._support import reimport_once_per_class + +if TYPE_CHECKING: + from typing import Any, Iterator + +#: Test tier (#1958): a sweep over every list of schema items. +__tier__ = 'LATEST' + +#: Every list of schema items the sweep finds, as ``holder.field``, and its item schema. +LISTS = { + 'application.httpv2:SettingsFrame.settings': 'application.httpv2:SettingPair', + 'internet.hip:LocatorSetParameter.locators': 'internet.hip:Locator', + 'internet.igmpv3:Report.records': 'internet.igmpv3:GroupRecord', + 'internet.mh:CGAParametersOption.parameters': 'internet.mh:CGAParameter', + 'transport.sctp:SACKChunk.gap_blocks': 'transport.sctp:GapAckBlock', + 'transport.tcp:SACK.sack': 'transport.tcp:SACKBlock', +} + +#: Item schemas a counted list refuses at every cut, and why. +REFUSED = { + 'internet.mh:CGAParameter': 'its pre_unpack refuses octets short of its fixed fields and ' + 'public key, rather than read on past the option (#2684)', +} + +#: The message every negative length warns with. +NEGATIVE = 'packet length < 0' + + +def item_lists() -> 'Iterator[tuple[str, Any]]': + """``holder.field`` and the item field of every list of schema items under + :mod:`pcapkit.protocols`, option lists aside.""" + from pcapkit.corekit.fields.collections import ListField, OptionField + from pcapkit.corekit.fields.misc import SchemaField + from tests.corekit.test_schema_roundtrip_unit import schema_classes + + for label, cls in sorted(schema_classes().items()): + for name, field in cls.__fields__.items(): + if isinstance(field, ListField) and not isinstance(field, OptionField) \ + and isinstance(field._item_type, SchemaField): # pylint: disable=protected-access + yield f'{label}.{name}', field._item_type # pylint: disable=protected-access + + +def holders(item: 'Any') -> 'tuple[type, type]': + """A list of ``item`` counted by a header field and running to the end of the data, and + one sized by a header field with the rest of the data after it.""" + from pcapkit.corekit.fields.collections import ListField + from pcapkit.corekit.fields.misc import PayloadField + from pcapkit.corekit.fields.numbers import UInt8Field, UInt16Field + from pcapkit.protocols.schema import Schema, schema_final + + @schema_final + class Counted(Schema): + count = UInt8Field() + items = ListField(length=lambda pkt: pkt['__length__'], item_type=item, count=lambda pkt: pkt['count']) + + @schema_final + class Sized(Schema): + size = UInt16Field() + items = ListField(length=lambda pkt: pkt['size'], item_type=item) + rest = PayloadField() + + return Counted, Sized + + +def read(cls: type, data: bytes, length: int, ctx: 'dict[str, Any]') -> 'tuple[Any, list[str]]': + """``data`` unpacked as ``cls`` under ``length``, or :data:`None` if refused, and its + schema warnings.""" + from pcapkit.utilities.exceptions import BaseError + from pcapkit.utilities.warnings import SchemaWarning + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + try: + parsed = cls.unpack(data, length, dict(ctx)) + except BaseError: + parsed = None + return parsed, [str(entry.message) for entry in caught if issubclass(entry.category, SchemaWarning)] + + +class ListItemCutTests(unittest.TestCase): + """Every list of schema items reads an item the data ends inside as a quiet cut.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def items(self) -> 'Iterator[tuple[str, Any, bytes, dict[str, Any]]]': + """Each list, its item field, a minimal item's octets, and the context it reads in.""" + from tests.corekit.test_schema_roundtrip_unit import build, label_of + + for label, item in item_lists(): + _, raw, ctx = build(item.schema) + yield label_of(item.schema), item, raw, ctx + + def assert_cut(self, parsed: 'Any', caught: 'list[str]', data: bytes, ctx: 'dict[str, Any]') -> None: + """A cut ``parsed`` from ``data`` warned no negative length, recorded, and packs back.""" + self.assertEqual([text for text in caught if NEGATIVE in text], []) + self.assertTrue(parsed.__dict__.get('__short_read__') is not None + or any('__short_read__' in entry.__dict__ for entry in parsed.items)) + self.assertEqual(parsed.pack(dict(ctx)), data) + + def test_the_sweep_finds_every_list_of_schema_items(self) -> None: + from tests.corekit.test_schema_roundtrip_unit import label_of + + self.assertEqual({label: label_of(item.schema) for label, item in item_lists()}, LISTS) + + def test_an_item_a_counted_list_ends_inside_reads_as_cut(self) -> None: + read_any = set() + for name, item, raw, ctx in self.items(): + counted, _ = holders(item) + for keep in range(1, len(raw)): + data = b'\x01' + raw[:keep] + with self.subTest(item=name, captured=keep): + parsed, caught = read(counted, data, len(data), ctx) + if parsed is not None: + read_any.add(name) + self.assert_cut(parsed, caught, data, ctx) + self.assertEqual(read_any, set(LISTS.values()) - set(REFUSED)) + + def test_an_item_a_sized_list_ends_inside_reads_as_cut(self) -> None: + # The enclosing length runs past the data, as an IP layer's does over a cut capture. + for name, item, raw, ctx in self.items(): + _, sized = holders(item) + for keep in range(1, len(raw)): + data = len(raw).to_bytes(2, 'big') + raw[:keep] + with self.subTest(item=name, captured=keep): + parsed, caught = read(sized, data, 2 + len(raw), ctx) + if parsed is not None: + self.assert_cut(parsed, caught, data, ctx) + + def test_a_whole_item_reads_whole(self) -> None: + for name, item, raw, ctx in self.items(): + counted, _ = holders(item) + data = b'\x01' + raw + with self.subTest(item=name): + parsed, caught = read(counted, data, len(data), ctx) + self.assertEqual(caught, []) + self.assertIsNone(parsed.__dict__.get('__short_read__')) + self.assertEqual((len(parsed.items), parsed.pack(dict(ctx))), (1, data)) + + +class MalformedLengthTests(unittest.TestCase): + """A length the items run past with the data still there is no cut, and still warns.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + from pcapkit.corekit.fields.misc import SchemaField + from pcapkit.corekit.fields.numbers import UInt8Field, UInt16Field + from pcapkit.protocols.schema import Schema, schema_final + + @schema_final + class Item(Schema): + kind = UInt8Field() + value = UInt16Field() + + self.Counted, self.Sized = holders(SchemaField(schema=Item)) + + def test_a_list_sized_short_of_its_item_warns(self) -> None: + data = b'\x00\x02' + b'\x01\x00\x02' + b'\xee\xee' + parsed, caught = read(self.Sized, data, len(data), {}) + self.assertEqual(caught, [f'{NEGATIVE}: -1']) + self.assertEqual(parsed.rest, b'\x02\xee\xee') + + def test_an_item_past_the_count_warns(self) -> None: + # The list counts no item, so the octets after the count are no counted item cut short. + data = b'\x00' + b'\x01\x00' + _, caught = read(self.Counted, data, len(data), {}) + self.assertEqual(caught, [f'{NEGATIVE}: -1']) + + def test_an_item_the_count_covers_reads_as_cut(self) -> None: + data = b'\x01' + b'\x01\x00' + parsed, caught = read(self.Counted, data, len(data), {}) + self.assertEqual(caught, []) + self.assertEqual(parsed.items[0].__dict__.get('__short_read__'), ('value', 1)) + self.assertEqual(parsed.pack(), data) + + +if __name__ == '__main__': + unittest.main()