From bba587826d3e294fb79c8a5e4f3a7ff08b17258e Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Wed, 4 Nov 2026 09:54:10 -0500 Subject: [PATCH] fix(corekit): end a cut list of plain values with the data, not in zero-filled items (#2760) A cut ListField of plain values read on past the data, each item zero-filled, so a cut IGMPv3 group record held its counted sources as 0.0.0.0. It now ends with the data as a list of schema items has done since #2756; the item the cut lands inside reads zeros past it. - ListField._unpack_items stops a plain list at the octets present; a length no whole number of items fills is still refused. - An item read from a list's zero fill hands the fill's start on through Schema.unpack (__data_present__), so a list in it ends with the data; unread fill goes back to the ledger. - Radiotap passes __zero_tail__ for a short TLV item, whose missing octets are zeros (#2655). - IGMPv3 Query data gains num_sources and make() a num_sources keyword; _make_data keeps the count of an unedited cut source list and lets make() count edited sources. Tests failing on c211805ae: every plain-value list cut at every offset; IGMPv3, IPv6-Route. --- .../pcapkit/protocols/internet/igmpv3.rst | 2 +- pcapkit/corekit/fields/collections.py | 63 +++- pcapkit/corekit/fields/field.py | 40 ++- pcapkit/protocols/internet/igmpv3/data.py | 6 +- pcapkit/protocols/internet/igmpv3/protocol.py | 33 +- pcapkit/protocols/link/radiotap/protocol.py | 4 +- pcapkit/protocols/schema.py | 24 +- tests/protocols/internet/test_igmp_unit.py | 13 +- .../test_igmpv3_source_list_cut_2760_unit.py | 221 +++++++++++++ .../test_ipv6_route_address_cut_2760_unit.py | 105 +++++++ .../test_list_plain_value_cut_2760_unit.py | 297 ++++++++++++++++++ 11 files changed, 770 insertions(+), 38 deletions(-) create mode 100644 tests/protocols/internet/test_igmpv3_source_list_cut_2760_unit.py create mode 100644 tests/protocols/internet/test_ipv6_route_address_cut_2760_unit.py create mode 100644 tests/protocols/test_list_plain_value_cut_2760_unit.py diff --git a/docs/source/pcapkit/protocols/internet/igmpv3.rst b/docs/source/pcapkit/protocols/internet/igmpv3.rst index e9b508b16b..3bb0ae5cad 100644 --- a/docs/source/pcapkit/protocols/internet/igmpv3.rst +++ b/docs/source/pcapkit/protocols/internet/igmpv3.rst @@ -28,7 +28,7 @@ Octets Bits Name Description 8 68 ``igmpv3.s_flag`` Suppress Router-Side Processing (S) 8 69 ``igmpv3.qrv`` Querier's Robustness Variable (QRV) 9 72 ``igmpv3.qqic`` Querier's Query Interval Code - 10 80 Number of Sources (N) + 10 80 ``igmpv3.num_sources`` Number of Sources (N) 12 96 ``igmpv3.sources`` Source Address [1..N] ======= ========= ================================== =================================== diff --git a/pcapkit/corekit/fields/collections.py b/pcapkit/corekit/fields/collections.py index 8b34581fcc..9acb9d451f 100644 --- a/pcapkit/corekit/fields/collections.py +++ b/pcapkit/corekit/fields/collections.py @@ -4,7 +4,7 @@ import io from typing import TYPE_CHECKING, Generic, TypeVar, cast -from pcapkit.corekit.fields.field import FieldBase, _zero_filled_items +from pcapkit.corekit.fields.field import FieldBase, _give_back_fill, _zero_filled_items from pcapkit.corekit.fields.misc import SchemaField from pcapkit.corekit.fields.numbers import NumberField from pcapkit.corekit.multidict import OrderedMultiDict @@ -213,18 +213,29 @@ def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> 'byte consumes nothing from ``buffer``, or if schema items short of the field's length cannot be zero-filled. - When ``buffer`` falls short of the field's length, the schema item it - ends inside is read as though its missing octets were zeros, as + When ``buffer`` falls short of the field's length, the item it ends + inside is read as though its missing octets were zeros, as :meth:`FieldBase.unpack ` reads any other field, and the list ends with that item: none is read - from the zeros past it, which hold no octet of the data. + from past the data, which holds no octet of it. A ``__data_present__`` + key in ``packet`` says that ``buffer`` holds the data that many octets + in, and past them the zeros an enclosing list of schema items was + filled out with, as :meth:`Schema.unpack + ` tells the fields of the item + the data ends inside; the list then ends with the data there. + + A ``__zero_tail__`` key in ``packet``, true, says instead that the + octets past ``buffer`` are zeros of the data, as radiotap.org reads the + octets a producer leaves off a TLV item (:issue:`2655`), so the list + reads its items on through them to its length. .. versionchanged:: 1.6.0 - A list of schema items the data ends inside ends with the data, the - item it ends inside read with its missing octets as zeros, where an - item past the data was refused as consuming nothing (:issue:`2686`, - :issue:`2756`). + A list the data ends inside ends with the data, the item it ends + inside read with its missing octets as zeros, where a schema item + past the data was refused as consuming nothing and a plain value + past it was read as zeros, and so does a list in an item the data + ends inside (:issue:`2686`, :issue:`2756`, :issue:`2760`). """ length = self._length @@ -264,13 +275,23 @@ def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> 'byte # :mod:`pcapkit.corekit.fields.misc`'s entry in :data:`sys.modules` # afterwards. is_schema = isinstance(self._item_type, SchemaField) - if is_schema and length > 0: + if length > 0: start = file.tell() - present = file.seek(0, io.SEEK_END) - start + octets = file.seek(0, io.SEEK_END) - start file.seek(start, io.SEEK_SET) - if present < length: - with _zero_filled_items(self.name, length, file.read()) as filled: - return self._unpack_items(filled, packet, is_schema, present) + # NOTE: Past ``__data_present__`` octets the buffer holds an enclosing fill (#2760). + held = cast('int', packet.get('__data_present__', -1)) + present = held if 0 <= held < octets else octets + # NOTE: Octets past the data that are zeros of it are read as items too (#2655, #2760). + stop = -1 if packet.get('__zero_tail__') else present + if present < length and is_schema: + with _zero_filled_items(self.name, length, file.read(), present) as filled: + return self._unpack_items(filled, packet, is_schema, stop) + if present < length and stop >= 0: + # NOTE: A plain value the data ends inside zero-fills itself, as any field does. + items = self._unpack_items(file, packet, is_schema, present) + _give_back_fill(octets - (file.tell() - start)) + return items return self._unpack_items(file, packet, is_schema) def _unpack_items(self, file: 'IO[bytes]', packet: 'dict[str, Any]', is_schema: 'bool', @@ -281,9 +302,9 @@ def _unpack_items(self, file: 'IO[bytes]', packet: 'dict[str, Any]', is_schema: file: Stream the items are read from. packet: Packet data. is_schema: Whether the items are schemas. - present: Octets of ``file`` the data holds, past which it holds the - zeros :meth:`unpack` filled it with; negative where it holds - no zeros. + present: Octets of ``file`` the data holds, past which ``file`` + ends or holds zeros a fill made up; negative where it holds the + data to the field's length. Returns: Unpacked items. @@ -317,8 +338,11 @@ def _unpack_items(self, file: 'IO[bytes]', packet: 'dict[str, Any]', is_schema: temp = [] # type: list[_TL] while length > 0: # NOTE: An item starting past the octets present would be made of the zero-fill alone, - # one the capture never held, so the list ends with the data (#2756). + # or of nothing, one the capture never held, so the list ends with the data (#2756, #2760). if 0 <= present <= offset - start: + # NOTE: A length no whole number of plain values fills is refused, as it is whole. + if not is_schema and length % item_type(packet).length: + raise FieldValueError(f'Field {self.name} has invalid length.') break field = item_type(packet) @@ -326,6 +350,10 @@ def _unpack_items(self, file: 'IO[bytes]', packet: 'dict[str, Any]', is_schema: # 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 + # NOTE: The item the data ends inside reads the fill past it, which the lists in it + # are told of, so they end with the data too (#2760). + if present >= 0: + context = dict(context, __data_present__=present - (offset - start)) data = cast('SchemaField', field).unpack(file, context) end = file.tell() @@ -354,6 +382,7 @@ def _unpack_items(self, file: 'IO[bytes]', packet: 'dict[str, Any]', is_schema: buffer = file.read(field.length) data = field.unpack(buffer, packet) + offset = file.tell() temp.append(data) diff --git a/pcapkit/corekit/fields/field.py b/pcapkit/corekit/fields/field.py index 9a5e451fe5..038f7960f5 100644 --- a/pcapkit/corekit/fields/field.py +++ b/pcapkit/corekit/fields/field.py @@ -121,7 +121,8 @@ def _zero_pad_budget() -> 'Iterator[list[int]]': @contextlib.contextmanager -def _zero_filled_items(name: 'str', length: 'int', data: 'bytes') -> 'Iterator[IO[bytes]]': +def _zero_filled_items(name: 'str', length: 'int', data: 'bytes', + present: 'int' = -1) -> 'Iterator[IO[bytes]]': """Zero-fill the octets of a list of schema items to its declared width. :meth:`FieldBase.unpack` reads a field the data ends inside as though its @@ -134,11 +135,14 @@ def _zero_filled_items(name: 'str', length: 'int', data: 'bytes') -> 'Iterator[I Args: name: Field name. length: Octets the field declares. - data: Octets actually present, fewer than ``length``. + data: Octets of the field's buffer, fewer than ``length``, or holding + past ``present`` the zeros of an enclosing fill. + present: Octets of ``data`` the data holds; negative for all of them. Yields: A stream of ``data``, zero-filled to ``length``. The zeros left unread - past where the stream ends up are not counted as octets read. + past where the stream ends up are not counted as octets read, an + enclosing fill's included (:issue:`2760`). Raises: FieldValueError: If ``length`` is past @@ -148,11 +152,13 @@ def _zero_filled_items(name: 'str', length: 'int', data: 'bytes') -> 'Iterator[I octets it has actually read. """ + if present < 0: + present = len(data) ceiling = config.MAX_ZERO_PAD_LENGTH if ceiling != -1 and length > ceiling: raise FieldValueError( f'Field {name} declares a length of {length} octet(s), ' - f'but only {len(data)} octet(s) are available.' + f'but only {present} octet(s) are available.' ) ledger = _zero_pad_ledger.get() @@ -177,8 +183,9 @@ def _zero_filled_items(name: 'str', length: 'int', data: 'bytes') -> 'Iterator[I ledger[1] += padding # NOTE: The items credit the zeros they read as octets supplied, so they are taken off in - # advance and those left unread given back (#2756); items that fail credit an unknown share, - # so the octets read go back as they were. + # advance and those left unread given back, an enclosing fill's too, which it took off + # (#2756, #2760); items that fail credit an unknown share, so the octets read go back as + # they were. read = ledger[0] ledger[0] -= padding filled = io.BytesIO(data + bytes(padding)) @@ -187,7 +194,26 @@ def _zero_filled_items(name: 'str', length: 'int', data: 'bytes') -> 'Iterator[I except BaseException: ledger[0] = read raise - ledger[0] += length - max(filled.tell(), len(data)) + ledger[0] += length - max(filled.tell(), present) + + +def _give_back_fill(count: 'int') -> 'None': + """Give back the ``count`` zeros of a fill that a list of plain values left unread. + + Args: + count: Zeros of an enclosing fill in the list's buffer that no item + read; nothing is given back for none. + + :func:`_zero_filled_items` takes the zeros of its fill off the octets read + in advance, as the items reading them credit them, and gives back those its + stream is left short of. A list in the item the data ends inside reads none + of its buffer past the data (:issue:`2760`), so the zeros it leaves there + are given back here. + + """ + ledger = _zero_pad_ledger.get() + if ledger is not None and count > 0: + ledger[0] += count #: Pattern[str]: Matches a leading, optionally byte-order-prefixed negative diff --git a/pcapkit/protocols/internet/igmpv3/data.py b/pcapkit/protocols/internet/igmpv3/data.py index 3ae0aad2f1..4b6faaa605 100644 --- a/pcapkit/protocols/internet/igmpv3/data.py +++ b/pcapkit/protocols/internet/igmpv3/data.py @@ -86,6 +86,10 @@ class Query(IGMPv3): #: Querier's Query Interval, in seconds, decoded from :attr:`qqic` by the #: same formula as :attr:`max_resp_time` (:rfc:`3376#section-4.1.7`). qqi: 'int' + #: Number of source addresses, as received on the wire + #: (:rfc:`3376#section-4.1.8`); more than :attr:`sources` holds where the + #: data ends inside them. + num_sources: 'int' #: Source addresses being queried (:rfc:`3376#section-4.1.9`); an empty #: :obj:`tuple` for a General or Group-Specific Query. sources: 'tuple[IPv4Address, ...]' @@ -94,7 +98,7 @@ class Query(IGMPv3): def __init__(self, version: 'Literal[3]', type: 'Enum_Type', length: 'int', max_resp_code: 'int', max_resp_time: 'int', checksum: 'bytes', group: 'IPv4Address', resv: 'int', s_flag: 'bool', qrv: 'int', qqic: 'int', qqi: 'int', - sources: 'tuple[IPv4Address, ...]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,redefined-builtin,line-too-long + num_sources: 'int', sources: 'tuple[IPv4Address, ...]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,redefined-builtin,line-too-long @info_final diff --git a/pcapkit/protocols/internet/igmpv3/protocol.py b/pcapkit/protocols/internet/igmpv3/protocol.py index b2240879e7..6b75ed99b6 100644 --- a/pcapkit/protocols/internet/igmpv3/protocol.py +++ b/pcapkit/protocols/internet/igmpv3/protocol.py @@ -25,7 +25,7 @@ 8 68 ``igmpv3.s_flag`` Suppress Router-Side Processing (S) 8 69 ``igmpv3.qrv`` Querier's Robustness Variable (QRV) 9 72 ``igmpv3.qqic`` Querier's Query Interval Code - 10 80 Number of Sources (N) + 10 80 ``igmpv3.num_sources`` Number of Sources (N) 12 96 ``igmpv3.sources`` Source Address [1..N] ======= ========= ================================== =================================== @@ -50,6 +50,7 @@ .. versionadded:: 1.6.0 """ +import ipaddress from typing import TYPE_CHECKING, cast from pcapkit.const.igmp.record_type import RecordType as Enum_RecordType @@ -275,6 +276,7 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_IGMPv3' qrv=qrv, qqic=qqic, qqi=qqi, + num_sources=ext.num_sources, sources=sources, ) # type: Data_IGMPv3 elif isinstance(body, Schema_Report): @@ -318,6 +320,7 @@ def make(self, qqic: 'Optional[int]' = None, qqi: 'Optional[int]' = None, sources: 'Optional[Iterable[IPv4Address | str | bytes | int]]' = None, + num_sources: 'Optional[int]' = None, reserved_1: 'int' = 0, reserved_2: 'bytes' = b'\x00\x00', num_group_records: 'Optional[int]' = None, @@ -363,6 +366,11 @@ def make(self, is given; only meaningful for a Query. sources: Source addresses being queried (:rfc:`3376#section-4.1.9`); only meaningful for a Query. + num_sources: Number of sources, as transmitted on the wire + (:rfc:`3376#section-4.1.8`); the length of ``sources`` when not + given; only meaningful for a Query. A count above the length of + ``sources`` is written as given, as on a Query the capture cut + inside its sources. reserved_1: The reserved octet following ``type`` (:rfc:`3376#section-4.2.1`); only meaningful for a Report. reserved_2: The reserved 16 bits before the group-record count @@ -388,7 +396,8 @@ def make(self, or code outside 0 to 255, or is neither a Query nor a Report; if ``max_resp_time`` is not an :obj:`int` from 0 to 255; if ``num_group_records`` is not an :obj:`int` from the length of - ``records`` to 65535; if ``checksum`` or ``reserved_2`` is not + ``records`` to 65535, or ``num_sources`` one from the length of + ``sources``; if ``checksum`` or ``reserved_2`` is not two octets; or if a record's ``record_type`` is refused, see :meth:`_make_group_record`. @@ -425,10 +434,15 @@ def make(self, if qqic is None: qqic = _encode_resp_code(0 if qqi is None else qqi) source_list = [] if sources is None else list(sources) + if num_sources is None: + num_sources = len(source_list) + elif self._make_uint(num_sources, bits=16, name='num_sources') < len(source_list): + raise ProtocolError(f'{self.alias}: invalid num_sources: {num_sources!r} ' + f'for {len(source_list)} source(s)') ext = Schema_QueryExtension( flags={'resv': resv, 's_flag': int(bool(s_flag)), 'qrv': qrv}, qqic=qqic, - num_sources=len(source_list), + num_sources=num_sources, sources=source_list, payload=b'', ) @@ -493,6 +507,18 @@ def _make_data(cls, data: 'Data_IGMPv3') -> 'dict[str, Any]': # type: ignore[ov } query = cast('Data_Query', data) + # NOTE: The count as parsed, while the sources are those a cut begins (11 octets into the body, + # octet 12 of the message), zeros past it; the info keeps no octets captured to compare (#2760). + num_sources = None + record = query.get('__short_read__') + if record is not None and record[0] == 'body': + cut = max(record[1] - 11, 0) + try: + packed = b''.join(ipaddress.IPv4Address(source).packed for source in query.sources) + except (TypeError, ValueError): # edited, to be refused by ``make`` + packed = b'' + if len(packed) == -(-cut // 4) * 4 and not packed[cut:].strip(b'\x00'): + num_sources = query.get('num_sources') return { 'version': query.get('version'), 'type': query.type, @@ -504,6 +530,7 @@ def _make_data(cls, data: 'Data_IGMPv3') -> 'dict[str, Any]': # type: ignore[ov 'qrv': query.qrv, 'qqic': query.qqic, 'sources': query.sources, + 'num_sources': num_sources, 'payload': cls._make_payload(data), } diff --git a/pcapkit/protocols/link/radiotap/protocol.py b/pcapkit/protocols/link/radiotap/protocol.py index 617daa1e32..d7d19bb388 100644 --- a/pcapkit/protocols/link/radiotap/protocol.py +++ b/pcapkit/protocols/link/radiotap/protocol.py @@ -2150,7 +2150,9 @@ def _read_tlv_data(self, tlv_type: 'Enum_FieldType', data: 'bytes') -> 'Data_Fie # what it is given; the refusal is handled here, so it is quiet. with CaughtByCaller(FieldValueError): try: - schema = cast('Schema', field_schema.unpack(octets, length, None)) # type: ignore[call-arg,misc] + # NOTE: The octets left off are zeros of the item, so its lists read on (#2655, #2760). + schema = cast('Schema', field_schema.unpack(octets, length, # type: ignore[call-arg,misc] + {'__zero_tail__': True})) except FieldValueError: return data if layout is None: diff --git a/pcapkit/protocols/schema.py b/pcapkit/protocols/schema.py index bf3f738dc3..f7a21ef159 100644 --- a/pcapkit/protocols/schema.py +++ b/pcapkit/protocols/schema.py @@ -767,12 +767,23 @@ def unpack(cls, data: 'bytes | IO[bytes]', records, and warn nothing (:issue:`2749`). The key is cleared as unpacking starts too. + A ``__data_present__`` key in ``packet`` says that ``data`` holds + the data that many octets in, and past them the zeros a + :class:`~pcapkit.corekit.fields.collections.ListField` of schema + items filled it out with, as the list says of the item the data + ends inside. The field a read crosses them in is given how many of + its octets the data holds, under the same key, so that a list there + ends with the data, as it does where the data itself ends + (:issue:`2760`). 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`), and a schema whose + ``__short_read__`` in ``packet`` (:issue:`2727`), a schema whose ``length`` ends with the data reads its fields past it as cut, not as - a negative length (:issue:`2749`). + a negative length (:issue:`2749`), and the field a read crosses the + end of the data in, inside zeros filled out past it, is told where it + ends (:issue:`2760`). """ # force cast arg type since decorator changed their signatures @@ -787,6 +798,10 @@ def unpack(cls, data: 'bytes | IO[bytes]', # 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) + # NOTE: Where ``data`` holds the data ends, past it the zero fill of a list of schema + # items the data ends inside, which the field a read crosses it in is told of (#2760). + present = cast('int', packet.pop('__data_present__', -1)) + fill = data.tell() + present if present >= 0 else -1 self = cls.__new__(cls) for field in self.__fields__.values(): @@ -850,7 +865,10 @@ def unpack(cls, data: 'bytes | IO[bytes]', if not isinstance(field, (ForwardMatchField, OptionField)): self._note_short_read(field.name, length, byte, packet) - value = field.unpack(byte, packet.copy()) + context = packet.copy() + if 0 <= fill < data.tell(): + context['__data_present__'] = max(fill - data.tell() + len(byte), 0) + value = field.unpack(byte, context) if isinstance(field, ListField) and field.shortfall > 0: # NOTE: The data ended before the items the list counts, each at least an # octet, so it declares that many past those read (#2711). diff --git a/tests/protocols/internet/test_igmp_unit.py b/tests/protocols/internet/test_igmp_unit.py index cd54f4ddcd..aa02ff3c9b 100644 --- a/tests/protocols/internet/test_igmp_unit.py +++ b/tests/protocols/internet/test_igmp_unit.py @@ -1267,7 +1267,9 @@ def test_cut_additional_data_goes_to_the_next_layer(self) -> None: def test_record_list_reads_bytes_and_a_live_stream_alike(self) -> None: """The record list reads the octets present zero-filled to its length, - from a buffer or a stream part-way through, and keeps that length (#2686).""" + from a buffer or a stream part-way through, and keeps that length (#2686); + the record the data ends inside reads none of the sources it counts past + the data from the zeros (#2760).""" from pcapkit.protocols.internet.igmpv3.schema import Report records = IGMPV3_TWO_RECORD_REPORT[8:28] @@ -1283,7 +1285,7 @@ def test_record_list_reads_bytes_and_a_live_stream_alike(self) -> None: self.assertEqual([str(record.multicast_address) for record in parsed], ['239.1.1.1', '0.0.0.0']) self.assertEqual(parsed[1].num_sources, 3) - self.assertEqual([str(source) for source in parsed[1].sources], ['0.0.0.0'] * 3) + self.assertEqual(parsed[1].sources, []) self.assertEqual(field.length, 36) @@ -1340,7 +1342,8 @@ def test_whole_messages_are_not_marked(self) -> None: def test_a_query_cut_at_every_offset(self) -> None: """Fewer than 12 octets is no IGMPv3 Query (:rfc:`3376#section-7.1`), so - those cuts are refused; every longer one ends inside the source list.""" + those cuts are refused; every longer one ends inside the source list, which + holds the sources the cut begins under the Number of Sources declared (#2760).""" from pcapkit.utilities.exceptions import ProtocolError packet = IGMPV3_TWO_SOURCE_QUERY @@ -1352,8 +1355,8 @@ def test_a_query_cut_at_every_offset(self) -> None: self.parse(raw) continue igmp = self.parse(raw) - self.assertEqual((str(igmp.info.group), igmp.info.qqic, len(igmp.info.sources)), - ('239.1.2.3', 125, 2)) + self.assertEqual((str(igmp.info.group), igmp.info.qqic, igmp.info.num_sources, + len(igmp.info.sources)), ('239.1.2.3', 125, 2, -(-(keep - 12) // 4))) self.assert_rebuilds(raw, igmp, ('body', keep - 1)) def test_a_report_cut_at_every_offset(self) -> None: diff --git a/tests/protocols/internet/test_igmpv3_source_list_cut_2760_unit.py b/tests/protocols/internet/test_igmpv3_source_list_cut_2760_unit.py new file mode 100644 index 0000000000..1506b758ff --- /dev/null +++ b/tests/protocols/internet/test_igmpv3_source_list_cut_2760_unit.py @@ -0,0 +1,221 @@ +# -*- coding: utf-8 -*- +"""An IGMPv3 group record cut inside its sources reads no source past the cut. + +GitHub issue #2760: a Report the capture cut inside a group record's sources held, after the +source the cut lands inside, every source its Number of Sources counts, read as ``0.0.0.0`` +from no octet of the capture. The source list ends with the data instead, as a list of +schema items does (#2756): the source the cut lands inside is read as far as it was captured, +with zeros after, and no source is made up past it. Behind IPv4, the records the data ends +inside are read from the zeros their list is filled out with (#2686), and their sources end +with the data too, so a framed cut reads as the same octets parsed on their own. + +A Query cut inside its sources reads them the same way. The record and the Query keep the +Number of Sources they declare, which their rebuild writes as captured, so every cut +rebuilds as its octets, from its info and from the info's dict, the datagram and the IGMPv3 +message alike; a cut Query whose sources are edited counts them again, as a whole one does. + +The messages are built here; this module is unit-tier and reads no capture file. + +""" +from __future__ import annotations + +import ipaddress +import unittest +import warnings +from typing import TYPE_CHECKING + +from tests._support import reimport_once_per_class +from tests.foundation import _roundtrip as wire + +if TYPE_CHECKING: + from typing import Any + +#: Test tier (#1958): hand-picked parse -> rebuild round trips. +__tier__ = 'ALL' + + +def record(kind: int, sources: 'tuple[str, ...]', aux: int = 0) -> bytes: + """A record of type ``kind`` for 239.1.1.1 with ``sources`` and ``aux`` words of + auxiliary data (RFC 3376, section 4.2.4).""" + return (bytes([kind, aux]) + len(sources).to_bytes(2, 'big') + bytes([239, 1, 1, 1]) + + b''.join(ipaddress.IPv4Address(source).packed for source in sources) + b'AUX!' * aux) + + +def checksummed(head: bytes, rest: bytes) -> bytes: + """``head`` and ``rest`` around a checksum of both.""" + return head + wire.checksum(head + bytes(2) + rest).to_bytes(2, 'big') + rest + + +#: MODE_IS_INCLUDE with three sources, then ALLOW_NEW_SOURCES with two and a word of +#: auxiliary data. +RECORDS = (record(1, ('10.0.0.1', '10.0.0.2', '10.0.0.3')), record(5, ('10.0.0.4', '10.0.0.5'), aux=1)) +#: The Report. +REPORT = checksummed(b'\x22\x00', b'\x00\x00' + len(RECORDS).to_bytes(2, 'big') + b''.join(RECORDS)) + + +def query(count: int) -> bytes: + """A Group-and-Source-Specific Query for 239.1.2.3 with ``count`` sources, from 10.0.0.1.""" + return checksummed(b'\x11\x64', bytes([239, 1, 2, 3, 0x0A, 125]) + count.to_bytes(2, 'big') + + b''.join(ipaddress.IPv4Address(f'10.0.0.{index}').packed for index in range(1, count + 1))) + + +#: The Query with three sources; one with one is beside it in :data:`QUERIES`. +QUERY = query(3) +#: Queries cut at every offset from the end of their fixed fields, which IGMPv3 needs. +QUERIES = (query(1), QUERY) + + +def query_sources(message: bytes, keep: int) -> 'tuple[str, ...]': + """The sources the first ``keep`` octets of the Query ``message`` begin, the last read as its + octets with zeros after them.""" + cut = message[:keep] + return tuple(str(ipaddress.IPv4Address(cut[at:at + 4].ljust(4, b'\x00'))) for at in range(12, keep, 4)) + + +def expected(keep: int) -> 'list[tuple[int, int, tuple[str, ...], bytes]]': + """The records the first ``keep`` octets of :data:`REPORT` begin, each as its type, Number + of Sources, the sources those octets begin, the last read as its octets with zeros after + them, and its auxiliary data, zero-filled to its length.""" + cut, out, start = REPORT[:keep], [], 8 + for item in RECORDS: + if start >= keep: + break + head = cut[start:start + 8].ljust(8, b'\x00') + count, aux = int.from_bytes(head[2:4], 'big'), head[1] * 4 + sources = tuple(str(ipaddress.IPv4Address(cut[at:at + 4].ljust(4, b'\x00'))) + for at in range(start + 8, min(start + 8 + 4 * count, keep), 4)) + tail = start + 8 + 4 * count + out.append((head[0], count, sources, cut[tail:tail + aux].ljust(aux, b'\x00'))) + start += len(item) + return out + + +class SourceListCutTests(unittest.TestCase): + """A record cut inside its sources holds the sources its octets begin, and rebuilds.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + from pcapkit.protocols.internet.igmpv3.protocol import IGMPv3 + from pcapkit.protocols.internet.ipv4.protocol import IPv4 + + self.IGMPv3, self.IPv4 = IGMPv3, IPv4 + catcher = warnings.catch_warnings() + catcher.__enter__() + self.addCleanup(catcher.__exit__, None, None, None) + warnings.simplefilter('ignore') + + @staticmethod + def records(layer: 'Any') -> 'list[tuple[int, int, tuple[str, ...], bytes]]': + """``layer``'s records, each as its type, Number of Sources, sources and auxiliary data.""" + return [(int(item.record_type), item.num_sources, tuple(map(str, item.sources)), item.aux_data) + for item in layer.info.records] + + def parses(self, message: bytes, keep: int) -> 'dict[str, tuple[Any, Any, bytes]]': + """``message`` cut ``keep`` octets in, parsed on its own unsized and sized and behind + IPv4 declaring all of it: each as the protocol, its IGMPv3 layer and the octets.""" + packet = wire.ipv4(message, proto=2) + raw = packet[:len(packet) - len(message) + keep] + alone, sized, framed = (self.IGMPv3(message[:keep]), self.IGMPv3(message[:keep], keep), + self.IPv4(raw)) + return {'alone': (alone, alone, message[:keep]), 'sized': (sized, sized, message[:keep]), + 'framed': (framed, framed.payload, raw)} + + def test_the_issue_repro(self) -> None: + rec = bytes([1, 0, 0, 3, 224, 0, 0, 1, 10, 0, 0, 1, 10, 0, 0, 2, 10, 0, 0, 3]) + rep = bytes([0x22, 0, 0, 0, 0, 0, 0, 1]) + rec + self.assertEqual([str(source) for source in self.IGMPv3(rep[:20]).info.records[0].sources], + ['10.0.0.1']) + + def test_the_cuts_reach_every_source(self) -> None: + self.assertEqual(len(REPORT), 8 + 20 + 20) + self.assertEqual(self.records(self.IGMPv3(REPORT)), expected(len(REPORT))) + + def test_a_cut_record_holds_the_sources_it_begins(self) -> None: + for keep in range(8, len(REPORT)): + for shape, (_, igmp, _) in self.parses(REPORT, keep).items(): + with self.subTest(captured=keep, shape=shape): + self.assertIsInstance(igmp, self.IGMPv3) + self.assertEqual(self.records(igmp), expected(keep)) + + def test_a_cut_inside_the_second_source_reads_as_captured(self) -> None: + # Two octets into the second source: it reads ``10.0.0.0``, and none follows it. + for shape, (_, igmp, _) in self.parses(REPORT, 8 + 8 + 6).items(): + with self.subTest(shape=shape): + self.assertEqual(self.records(igmp), [(1, 3, ('10.0.0.1', '10.0.0.0'), b'')]) + + def test_a_cut_query_holds_the_sources_it_begins(self) -> None: + held = set() + for message in QUERIES: + for keep in range(12, len(message)): + for shape, (_, igmp, _) in self.parses(message, keep).items(): + with self.subTest(sources=message[11], captured=keep, shape=shape): + self.assertEqual(igmp.info.num_sources, message[11]) + self.assertEqual(tuple(map(str, igmp.info.sources)), query_sources(message, keep)) + held.add(len(igmp.info.sources)) + # A cut leaves no source, one, and more than one. + self.assertEqual(held, {0, 1, 2, 3}) + + def test_an_edited_cut_query_counts_its_sources(self) -> None: + """Sources no longer those a cut Query began with, zeros past the cut, are counted as a + whole Query's are, not given the count the cut Query declares; its record still cuts it + back to the octets it captured.""" + edits = ( + ('one source fewer', lambda sources: sources[:-1]), + ('no sources', lambda sources: []), + ('one source more', lambda sources: sources + ['10.9.9.9']), + ('a value past the cut', lambda sources: sources[:-1] + ['10.1.1.1']), + ) + whole = self.IGMPv3(QUERY).info + for keep in range(13, len(QUERY)): + for label, edit in edits: + if label == 'a value past the cut' and (keep - 12) % 4 == 0: + continue # the last source is whole, so no value of it is past the cut + for shape, (_, igmp, _) in self.parses(QUERY, keep).items(): + with self.subTest(captured=keep, edit=label, shape=shape): + sources = edit([str(source) for source in igmp.info.sources]) + built = {} + for name, info in (('cut', igmp.info), ('whole', whole)): + data = info.to_dict() + data['sources'] = tuple(map(ipaddress.IPv4Address, sources)) + built[name] = self.IGMPv3.from_data(data) + self.assertEqual(built['whole'].info.num_sources, len(sources)) + self.assertEqual(built['cut'].info.num_sources, len(sources)) + self.assertEqual(built['cut'].data, built['whole'].data[:keep]) + + def test_make_writes_num_sources_as_given(self) -> None: + # Counted from the sources when not given; a count above them is written as given, and one + # below them, or that does not fit 16 bits, is refused. + from pcapkit.const.igmp.type import Type + from pcapkit.utilities.exceptions import ProtocolError + + sources = ['10.0.0.1', '10.0.0.2'] + self.assertEqual(self.IGMPv3(type=Type.IGMP_Membership_Query, sources=sources).info.num_sources, 2) + made = self.IGMPv3(type=Type.IGMP_Membership_Query, sources=sources, num_sources=5) + self.assertEqual(made.data[10:12], (5).to_bytes(2, 'big')) + for count in (1, 0x10000, True, '2'): + with self.subTest(num_sources=count), self.assertRaisesRegex(ProtocolError, 'num_sources'): + self.IGMPv3(type=Type.IGMP_Membership_Query, sources=sources, num_sources=count) + + def test_a_cut_rebuilds_as_captured(self) -> None: + for message in (REPORT, *QUERIES): + for keep in range(12, len(message)): + for shape, (proto, igmp, raw) in self.parses(message, keep).items(): + for path, layer, octets in (('outer', proto, raw), ('igmpv3', igmp, message[:keep])): + with self.subTest(message=message[:1].hex(), captured=keep, shape=shape, layer=path): + cls = type(layer) + self.assertEqual(cls.from_data(layer.info).data, octets) + self.assertEqual(cls.from_data(layer.info.to_dict()).data, octets) + + def test_a_whole_report_is_the_control(self) -> None: + for shape, (proto, igmp, raw) in self.parses(REPORT, len(REPORT)).items(): + with self.subTest(shape=shape): + self.assertIsNone(igmp.info.get('__short_read__')) + self.assertEqual(self.records(igmp), [ + (1, 3, ('10.0.0.1', '10.0.0.2', '10.0.0.3'), b''), + (5, 2, ('10.0.0.4', '10.0.0.5'), b'AUX!'), + ]) + self.assertEqual(type(proto).from_data(proto.info).data, raw) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/internet/test_ipv6_route_address_cut_2760_unit.py b/tests/protocols/internet/test_ipv6_route_address_cut_2760_unit.py new file mode 100644 index 0000000000..90aa4b6d8c --- /dev/null +++ b/tests/protocols/internet/test_ipv6_route_address_cut_2760_unit.py @@ -0,0 +1,105 @@ +# -*- coding: utf-8 -*- +"""A Type 0 Routing header cut inside its addresses reads no address past the cut. + +GitHub issue #2760: a Type 0 Routing header the capture cut inside its addresses held every +address its Hdr Ext Len leaves room for, those past the cut read as ``::`` from no octet of +the capture. The address list ends with the data instead, as a list of schema items does +(#2756): the address the cut lands inside is read as far as it was captured, with zeros +after, and no address is made up past it. + +The header keeps the length it declares, which its rebuild writes as captured, so every +cut rebuilds as its octets, from its info and from the info's dict, on its own and behind +IPv6 alike. + +The headers are built here; this module is unit-tier and reads no capture file. + +""" +from __future__ import annotations + +import ipaddress +import unittest +import warnings +from typing import TYPE_CHECKING + +from tests._support import reimport_once_per_class +from tests.foundation import _roundtrip as wire + +if TYPE_CHECKING: + from typing import Any + +#: Test tier (#1958): hand-picked parse -> rebuild round trips. +__tier__ = 'ALL' + +#: The addresses of the header. +ADDRESSES = ('2001:db8::1', '2001:db8::2', '2001:db8::3') +#: A Type 0 Routing header naming no next header, its Hdr Ext Len sized for +#: :data:`ADDRESSES`, three segments left (RFC 2460, section 4.4). +ROUTE = (bytes([59, 2 * len(ADDRESSES), 0, len(ADDRESSES)]) + bytes(4) + + b''.join(ipaddress.IPv6Address(address).packed for address in ADDRESSES)) + + +def begun(keep: int) -> 'tuple[str, ...]': + """The addresses the first ``keep`` octets of :data:`ROUTE` begin, the last read as its + octets with zeros after them.""" + cut = ROUTE[:keep] + return tuple(str(ipaddress.IPv6Address(cut[at:at + 16].ljust(16, b'\x00'))) for at in range(8, keep, 16)) + + +class AddressListCutTests(unittest.TestCase): + """A header cut inside its addresses holds the addresses its octets begin, and rebuilds.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + from pcapkit.protocols.internet.ipv6.protocol import IPv6 + from pcapkit.protocols.internet.ipv6_route.protocol import IPv6_Route + + self.IPv6, self.Route = IPv6, IPv6_Route + catcher = warnings.catch_warnings() + catcher.__enter__() + self.addCleanup(catcher.__exit__, None, None, None) + warnings.simplefilter('ignore') + + def parses(self, keep: int) -> 'dict[str, tuple[Any, Any, bytes]]': + """:data:`ROUTE` cut ``keep`` octets in, parsed on its own and behind IPv6 declaring + all of it: each as the protocol, its Routing header and the octets.""" + packet = wire.ipv6(ROUTE, nxt=43) + raw = packet[:40 + keep] + alone, framed = self.Route(ROUTE[:keep], keep), self.IPv6(raw, len(raw)) + (route,) = [ext for ext in framed.extension_headers.values() if isinstance(ext, self.Route)] + return {'alone': (alone, alone, ROUTE[:keep]), 'framed': (framed, route, raw)} + + def test_the_cuts_reach_every_address(self) -> None: + self.assertEqual(len(ROUTE), 8 + 16 * len(ADDRESSES)) + self.assertEqual(tuple(map(str, self.Route(ROUTE, len(ROUTE)).info.ip)), ADDRESSES) + + def test_a_cut_header_holds_the_addresses_it_begins(self) -> None: + for keep in range(8, len(ROUTE)): + for shape, (_, route, _) in self.parses(keep).items(): + with self.subTest(captured=keep, shape=shape): + self.assertEqual(tuple(map(str, route.info.ip)), begun(keep)) + self.assertEqual(route.info.length, len(ROUTE)) + + def test_a_cut_inside_the_second_address_reads_as_captured(self) -> None: + for shape, (_, route, _) in self.parses(8 + 16 + 6).items(): + with self.subTest(shape=shape): + self.assertEqual(tuple(map(str, route.info.ip)), ('2001:db8::1', '2001:db8::')) + + def test_a_cut_rebuilds_as_captured(self) -> None: + for keep in range(8, len(ROUTE)): + for shape, (proto, route, raw) in self.parses(keep).items(): + for path, layer, octets in (('outer', proto, raw), ('route', route, ROUTE[:keep])): + with self.subTest(captured=keep, shape=shape, layer=path): + cls = type(layer) + self.assertEqual(cls.from_data(layer.info).data, octets) + self.assertEqual(cls.from_data(layer.info.to_dict()).data, octets) + + def test_a_whole_header_is_the_control(self) -> None: + for shape, (proto, route, raw) in self.parses(len(ROUTE)).items(): + with self.subTest(shape=shape): + self.assertIsNone(route.info.get('__short_read__')) + self.assertEqual(tuple(map(str, route.info.ip)), ADDRESSES) + self.assertEqual(type(proto).from_data(proto.info).data, raw) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/test_list_plain_value_cut_2760_unit.py b/tests/protocols/test_list_plain_value_cut_2760_unit.py new file mode 100644 index 0000000000..226260fd61 --- /dev/null +++ b/tests/protocols/test_list_plain_value_cut_2760_unit.py @@ -0,0 +1,297 @@ +# -*- coding: utf-8 -*- +"""A list of plain values its length runs past the data reads no item past the data. + +GitHub issue #2760: an IGMPv3 group record cut inside its sources held the sources its +Number of Sources counts past the cut, as ``0.0.0.0``. #2756 ended a +:class:`~pcapkit.corekit.fields.collections.ListField` of schema items with the data, the +item the data ends inside read with its missing octets as zeros, but a list of plain values +read on past the data, each item zero-filled as a field the data ends inside is. Such a list +ends with the data too, and so does a list in the item of a list of schema items the data +ends inside, which reads the zeros that list was filled out with (#2686). + +The sweep finds every list of plain values under :mod:`pcapkit.protocols` from its schema +modules, as :file:`tests/protocols/test_list_item_cut_2749_unit.py` finds the lists of schema +items, and reads three items of each behind a header field that sizes them, at every cut: +on its own, and as the one item of a list of schema items whose length runs past the data. +Each cut holds the items its octets begin and no more, the one the data ends inside read as +its captured octets with zeros after them, and packs back to the octets read, from the +schema and from its dict. The octets read are those present: the zeros of the fill the +items leave unread are taken back. + +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 unittest import mock + +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 plain values. +__tier__ = 'LATEST' + +#: Every list of plain values the sweep finds, as ``holder.field``. +LISTS = frozenset({ + 'internet.hip:ACKParameter.update_id', + 'internet.hip:AckDataParameter.ack', + 'internet.hip:DHGroupListParameter.groups', + 'internet.hip:ESPTransformParameter.suites', + 'internet.hip:HIPCipherParameter.ciphers', + 'internet.hip:HIPTransformParameter.suites', + 'internet.hip:HIPTransportModeParameter.mode', + 'internet.hip:HITSuiteListParameter.suites', + 'internet.hip:NATTraversalModeParameter.modes', + 'internet.hip:RegFailedParameter.reg_failed', + 'internet.hip:RegInfoParameter.reg_info', + 'internet.hip:RegRequestParameter.reg_request', + 'internet.hip:RegResponseParameter.reg_response', + 'internet.hip:RouteDstParameter.hit', + 'internet.hip:RouteViaParameter.hit', + 'internet.hip:TransportFormatListParameter.formats', + 'internet.hip:ViaRVSParameter.address', + 'internet.igmpv3:GroupRecord.sources', + 'internet.igmpv3:QueryExtension.sources', + 'internet.ipv4:LSROption.route', + 'internet.ipv4:RROption.route', + 'internet.ipv4:SSROption.route', + 'internet.ipv4:TSOption.ts_data', + 'internet.ipv6_route:SourceRoute.ip', + 'internet.mh:BIDReferenceSuboption.bid', + 'internet.mh:FlowSummaryOption.fid', + 'internet.mh:HomeAgentSwitchMessage.addresses', + 'internet.mh:MultiPrefixExtension.prefixes', + 'link.radiotap:EHT.data', + 'link.radiotap:EHT.user_info', + 'link.radiotap:HEMU.ru_channel1', + 'link.radiotap:HEMU.ru_channel2', + 'link.radiotap:VHT.mcs_nss', + 'transport.sctp:MissingMandatoryParameterCause.types', + 'transport.sctp:SACKChunk.dup_tsn', + 'transport.sctp:SupportedAddressTypesParameter.types', + 'transport.tcp:MPTCPRemoveAddress.addr_id', +}) + +#: Items each list is read with. +ITEMS = 3 + + +def plain_lists() -> 'Iterator[tuple[str, type, Any]]': + """``holder.field``, the holder schema and the item field of every list of plain values + under :mod:`pcapkit.protocols`.""" + from pcapkit.corekit.fields.collections import ListField, OptionField + from pcapkit.corekit.fields.misc import ConditionalField, 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, ConditionalField): + field = field.field + item = getattr(field, '_item_type', None) + if isinstance(field, ListField) and not isinstance(field, OptionField) \ + and item is not None and not isinstance(item, SchemaField): + yield f'{label}.{name}', cls, item + + +def octets(holder: type, item: 'Any') -> 'tuple[int, bytes]': + """The width of ``item`` and the octets of :data:`ITEMS` items of it, each distinct + where the item is a number or an address, a member of its enumeration otherwise.""" + from pcapkit.corekit.fields.numbers import EnumField + from tests.corekit.test_schema_roundtrip_unit import minimal + + width = item({}).length + if isinstance(item, EnumField): + return width, item.pack(minimal(item, holder, 0, {}), {}) * ITEMS + return width, bytes((index % 255) + 1 for index in range(width * ITEMS)) + + +def holders(item: 'Any') -> 'tuple[type, type]': + """A list of ``item`` sized by a header field with the rest of the data after it, and a + list of schema items, each such a list without the rest, sized by a header field too.""" + from pcapkit.corekit.fields.collections import ListField + from pcapkit.corekit.fields.misc import PayloadField, SchemaField + from pcapkit.corekit.fields.numbers import UInt16Field + from pcapkit.protocols.schema import Schema, schema_final + + @schema_final + class Sized(Schema): + size = UInt16Field() + items = ListField(length=lambda pkt: pkt['size'], item_type=item) + rest = PayloadField() + + @schema_final + class Record(Schema): + size = UInt16Field() + items = ListField(length=lambda pkt: pkt['size'], item_type=item) + + @schema_final + class Records(Schema): + size = UInt16Field() + records = ListField(length=lambda pkt: pkt['size'], item_type=SchemaField(schema=Record)) + + return Sized, Records + + +def begun(item: 'Any', data: bytes, width: int) -> 'list[Any]': + """The items ``data`` begins, the last read as its octets with zeros after them.""" + return [item.unpack(data[start:start + width].ljust(width, b'\x00'), {}) + for start in range(0, len(data), width)] + + +class PlainListCutTests(unittest.TestCase): + """A list of plain values the data ends inside ends with the data.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def assert_packs_back(self, parsed: 'Any', data: bytes) -> None: + """``parsed`` packs back to ``data``, and so does the schema rebuilt from its dict.""" + self.assertEqual(parsed.pack(), data) + self.assertEqual(type(parsed).from_dict(parsed.to_dict()).pack(), data) + + def test_the_sweep_finds_every_list_of_plain_values(self) -> None: + self.assertEqual({label for label, _, _ in plain_lists()}, LISTS) + + def test_a_cut_list_holds_the_items_it_begins(self) -> None: + for label, holder, item in plain_lists(): + width, whole = octets(holder, item) + sized, _ = holders(item) + for keep in range(len(whole)): + data = len(whole).to_bytes(2, 'big') + whole[:keep] + with self.subTest(list=label, captured=keep), warnings.catch_warnings(): + warnings.simplefilter('ignore') + parsed = sized.unpack(data, 2 + len(whole), {}) + self.assertEqual(parsed.items, begun(item, whole[:keep], width)) + self.assertEqual(parsed.__dict__['__short_read__'], ('items', keep)) + self.assert_packs_back(parsed, data) + + def test_a_list_in_a_cut_item_holds_the_items_it_begins(self) -> None: + # The record list is zero-filled to its length, and the record the data ends inside + # reads its list from the fill (#2686, #2756). + for label, holder, item in plain_lists(): + width, whole = octets(holder, item) + _, records = holders(item) + record = len(whole).to_bytes(2, 'big') + whole + for keep in range(2, len(record)): + data = len(record).to_bytes(2, 'big') + record[:keep] + with self.subTest(list=label, captured=keep), warnings.catch_warnings(): + warnings.simplefilter('ignore') + parsed = records.unpack(data, 2 + len(record), {}) + self.assertEqual([entry.items for entry in parsed.records], + [begun(item, whole[:keep - 2], width)]) + self.assert_packs_back(parsed, data) + + def test_a_whole_list_is_the_control(self) -> None: + for label, holder, item in plain_lists(): + width, whole = octets(holder, item) + sized, records = holders(item) + data = len(whole).to_bytes(2, 'big') + whole + nested = len(data).to_bytes(2, 'big') + data + with self.subTest(list=label), warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + parsed = sized.unpack(data, len(data), {}) + self.assertIsNone(parsed.__dict__.get('__short_read__')) + self.assertEqual(parsed.items, begun(item, whole, width)) + self.assert_packs_back(parsed, data) + within = records.unpack(nested, len(nested), {}) + self.assertEqual([entry.items for entry in within.records], [parsed.items]) + self.assert_packs_back(within, nested) + self.assertEqual(caught, []) + + def test_a_zero_tail_is_read_as_items(self) -> None: + # Where the octets past the data are zeros of it, as radiotap.org reads the octets a + # producer leaves off a TLV item (#2655), a list reads its items on through them. + from pcapkit.corekit.fields.collections import ListField + from pcapkit.corekit.fields.misc import SchemaField + from pcapkit.corekit.fields.numbers import UInt16Field, UInt32Field + from pcapkit.protocols.schema import Schema, schema_final + + @schema_final + class Pair(Schema): + id = UInt16Field() + value = UInt32Field() + + plain = ListField(length=12, item_type=UInt32Field()) + pairs = ListField(length=18, item_type=SchemaField(length=6, schema=Pair)) + for keep in range(12): + with self.subTest(items='plain', captured=keep): + data = bytes(range(1, 13))[:keep] + self.assertEqual(plain.unpack(data, {'__zero_tail__': True}), + begun(UInt32Field(), data.ljust(12, b'\x00'), 4)) + self.assertEqual(len(plain.unpack(data, {})), -(-keep // 4)) + for keep in range(18): + with self.subTest(items='schema', captured=keep): + data = bytes(range(1, 19))[:keep] + self.assertEqual([item.pack() for item in pairs.unpack(data, {'__zero_tail__': True})], + [chunk for chunk in (data.ljust(18, b'\x00')[at:at + 6] for at in (0, 6, 12))]) + self.assertEqual(len(pairs.unpack(data, {})), -(-keep // 6)) + + def test_a_length_no_whole_number_of_items_fills_is_refused(self) -> None: + # The items run past such a length read whole, so a cut of it is refused as well, rather + # than read as far as the data goes. + from pcapkit.corekit.fields.collections import ListField + from pcapkit.corekit.fields.numbers import UInt32Field + from pcapkit.utilities.exceptions import FieldValueError + + field = ListField(length=10, item_type=UInt32Field()) + whole = bytes(range(1, 11)) + for keep in range(len(whole) + 1): + with self.subTest(captured=keep), self.assertRaisesRegex(FieldValueError, 'invalid length'): + field.unpack(whole[:keep], {}) + + def test_the_octets_read_are_those_present(self) -> None: + # Behind the fill, the zeros the list leaves unread past the item the data ends + # inside are given back, as the fill takes them off the octets read in advance. + from pcapkit.corekit.fields.field import _zero_pad_budget + from pcapkit.corekit.fields.numbers import UInt32Field + from pcapkit.utilities import config + + item = UInt32Field() + sized, records = holders(item) + whole = bytes(range(1, 13)) + record = len(whole).to_bytes(2, 'big') + whole + for keep in range(len(record)): + with self.subTest(captured=keep), mock.patch.object(config, 'ZERO_PAD_BUDGET_RATIO', -1): + with _zero_pad_budget() as ledger: + sized.unpack(record[:keep], len(record), {}) + self.assertEqual(ledger, [keep, 0]) + with _zero_pad_budget() as ledger: + data = len(record).to_bytes(2, 'big') + record[:keep] + records.unpack(data, 2 + len(record), {}) + self.assertEqual(ledger, [len(data), len(record) - keep]) + + def test_a_list_of_schema_items_in_a_cut_item_ends_with_the_data(self) -> None: + # A list of schema items in the record the data ends inside reads its buffer from the + # fill, so it ends with the data and gives back the zeros of the fill it leaves unread. + from pcapkit.corekit.fields.field import _zero_pad_budget + from pcapkit.corekit.fields.misc import SchemaField + from pcapkit.corekit.fields.numbers import UInt16Field, UInt32Field + from pcapkit.protocols.schema import Schema, schema_final + from pcapkit.utilities import config + + @schema_final + class Pair(Schema): + id = UInt16Field() + value = UInt32Field() + + _, records = holders(SchemaField(length=6, schema=Pair)) + pairs = bytes(range(1, 19)) + record = len(pairs).to_bytes(2, 'big') + pairs + for keep in range(2, len(record)): + data = len(record).to_bytes(2, 'big') + record[:keep] + with self.subTest(captured=keep), mock.patch.object(config, 'ZERO_PAD_BUDGET_RATIO', -1), \ + _zero_pad_budget() as ledger: + parsed = records.unpack(data, 2 + len(record), {}) + self.assertEqual([len(entry.items) for entry in parsed.records], [-(-(keep - 2) // 6)]) + self.assertEqual(parsed.pack(), data) + self.assertEqual(ledger, [len(data), len(record) - keep]) + + +if __name__ == '__main__': + unittest.main()