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()