From c3fc0d3d50600b4f53846a911edaaabd0df48cd4 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Mon, 2 Nov 2026 16:07:06 -0500 Subject: [PATCH] feat(link): decode radiotap TLV items a producer cut short (#2655) radiotap.org lets a producer leave off the octets it does not fill at the end of a TLV item, and a parser reads them as zeros. Such an item stayed raw. - _read_tlv_data unpacks data short of its schema, or of the vendor header, at the schema's full length, rounded up to a whole list item, so Schema.unpack zero-fills it and records the cut; the decoded data keeps it as __short_read__. Data past a fixed layout stays raw. - _make_tlv_data packs through _pack_tlv_data, which hands the record back to the schema, so Schema.pack writes the octets as captured. A malformed record, or vendor data after a cut header, raises ProtocolError. - Both unpacks share the CaughtByCaller(FieldValueError) guard, so a cut the zero-pad limits refuse stays raw, quietly. - The module docstring, radiotap.rst, the TLV data model and make()'s Raises describe it. - Tests: U-SIG, S1G, EHT and vendor items cut at a field boundary and inside a field, each against its zero-padded control; from_data and make() round trips; quietness; the padding-limit refusal; a registered layout. The #2576 pins on a short S1G and a ragged EHT now expect them decoded. --- .../pcapkit/protocols/link/radiotap.rst | 11 +- pcapkit/protocols/data/link/radiotap.py | 6 +- pcapkit/protocols/link/radiotap.py | 122 +++++-- .../link/test_radiotap_fields_2576_unit.py | 15 +- .../link/test_radiotap_short_tlv_2655_unit.py | 301 ++++++++++++++++++ 5 files changed, 416 insertions(+), 39 deletions(-) create mode 100644 tests/protocols/link/test_radiotap_short_tlv_2655_unit.py diff --git a/docs/source/pcapkit/protocols/link/radiotap.rst b/docs/source/pcapkit/protocols/link/radiotap.rst index bfeac758e3..a98e9ac18d 100644 --- a/docs/source/pcapkit/protocols/link/radiotap.rst +++ b/docs/source/pcapkit/protocols/link/radiotap.rst @@ -84,9 +84,14 @@ decoded the same way. octets after it, kept whole. Without either, the next word continues the namespace, each bit numbered 32 past the same bit of the word before. * **TLVs.** Bit 28 makes the rest of the region a list of type-length-value - items. An item's data is decoded where its type has a schema whose size it - matches, and kept raw otherwise; a vendor item is split into its header and - raw data. + items. An item's data is decoded where its type has a schema it does not run + past, and kept raw otherwise; a vendor item is split into its header and raw + data. radiotap.org lets a producer leave out the octets that end an item, which + are then zeros, so an item short of its schema, or of the vendor header, is + decoded as though they were there. Its decoded data records the cut as + ``__short_read__``, as :meth:`Schema.unpack ` + does, so :meth:`Radiotap.make ` + writes the octets that were captured. * **What is not decoded.** A radiotap-namespace bit no field is defined for -- 18, FreeBSD's XChannel, which radiotap.org lists as unofficial, and every one numbered 32 or more, the TLV-only types included -- or a field that runs diff --git a/pcapkit/protocols/data/link/radiotap.py b/pcapkit/protocols/data/link/radiotap.py index 6221e071bd..c5d471684d 100644 --- a/pcapkit/protocols/data/link/radiotap.py +++ b/pcapkit/protocols/data/link/radiotap.py @@ -448,8 +448,10 @@ class TLV(Data): type: 'FieldType' #: Length of :attr:`data`, in octets, less the padding after it. length: 'int' - #: The data, decoded where the type defines a layout and :attr:`length` - #: matches it; the raw octets otherwise. + #: The data, decoded where the type defines a layout :attr:`length` does + #: not run past; the raw octets otherwise. Data short of the layout is + #: decoded with the octets it lacks as zeros, and records the cut as + #: ``__short_read__``. data: 'RadiotapField | VendorTLV | bytes' if TYPE_CHECKING: diff --git a/pcapkit/protocols/link/radiotap.py b/pcapkit/protocols/link/radiotap.py index e36f022930..cd5def914f 100644 --- a/pcapkit/protocols/link/radiotap.py +++ b/pcapkit/protocols/link/radiotap.py @@ -89,9 +89,14 @@ octets after it, kept whole. Without either, the next word continues the namespace, each bit numbered 32 past the same bit of the word before. * **TLVs.** Bit 28 makes the rest of the region a list of type-length-value - items. An item's data is decoded where its type has a schema whose size it - matches, and kept raw otherwise; a vendor item is split into its header and - raw data. + items. An item's data is decoded where its type has a schema it does not run + past, and kept raw otherwise; a vendor item is split into its header and raw + data. radiotap.org lets a producer leave out the octets that end an item, which + are then zeros, so an item short of its schema, or of the vendor header, is + decoded as though they were there. Its decoded data records the cut as + ``__short_read__``, as :meth:`Schema.unpack ` + does, so :meth:`Radiotap.make ` + writes the octets that were captured. * **What is not decoded.** A radiotap-namespace bit no field is defined for -- 18, FreeBSD's XChannel, which radiotap.org lists as unofficial, and every one numbered 32 or more, the TLV-only types included -- or a field that runs @@ -497,8 +502,10 @@ def make(self, fit in 32 bits; if ``fields`` is neither :class:`bytes` nor fields; if the fields do not match the bits of ``present``, a word but the last of which ends the chain, or a value does - not fit its subfield; or if ``it_len`` is given and does not - fit in 16 bits. + not fit its subfield; if a TLV item's ``__short_read__`` names + no field and an octet count, or cuts short a vendor header that + data follows; or if ``it_len`` is given and does not fit in 16 + bits. """ for name, value in (('it_version', it_version), ('it_pad', it_pad)): @@ -2053,33 +2060,55 @@ def _read_tlv_data(self, tlv_type: 'Enum_FieldType', data: 'bytes') -> 'Data_Fie Returns: A vendor TLV for the Vendor Namespace type, the field for a type - whose schema ``data`` fills exactly, and the raw octets otherwise. + whose schema ``data`` does not run past, and the raw octets + otherwise. Data short of its schema, or of the vendor header, is + read with the octets it lacks as zeros, and the result records the + cut as ``__short_read__``, which :meth:`_make_tlv_data` replays; + it stays raw if the zero-padding limits of + :mod:`pcapkit.utilities.config` refuse the zeros. """ + # NOTE: radiotap.org reads the octets a producer leaves off an item as zeros, so short data is unpacked + # at its schema's full length, to the end of a list item, and ``Schema.unpack`` records the cut (#2655). + layout = None # type: Optional[tuple[FieldParser, FieldConstructor, Type[Schema_Field]]] if tlv_type == Enum_FieldType.Vendor_Namespace: - size, _ = self._field_size(Schema_VendorTLV) - if len(data) < size: + field_schema = Schema_VendorTLV # type: Type[Schema] + length, _ = self._field_size(field_schema) + octets = data[:length] + else: + layout = self._lookup_field(tlv_type) + if layout is None: return data - vendor = cast('Schema_VendorTLV', - Schema_VendorTLV.unpack(data[:size], size, None)) # type: ignore[call-arg,misc] - return Data_VendorTLV(oui=vendor.oui, subtype=vendor.subtype, presence_type=vendor.presence_type, - reserved=vendor.reserved, data=data[size:]) - - layout = self._lookup_field(tlv_type) - if layout is None: - return data - parser, _, field_schema = layout - size, rest = self._field_size(field_schema) - if len(data) < size or (not rest and len(data) != size): - return data - # NOTE: a list taking the rest refuses octets that are not whole items - # of it; the refusal is handled here, so it is quiet. + field_schema = layout[2] + size, rest = self._field_size(field_schema) + if not rest and len(data) > size: + return data + length = max(len(data), size) + if rest and len(data) > size: + tail = next(field for field in field_schema.__fields__.values() if field.length < 0) + item = tail._item_type if isinstance(tail, ListField) else None # pylint: disable=protected-access + if item is not None and item.length > 0: + length += -(len(data) - size) % item.length + octets = data + + # NOTE: a limit on zero-padding, or a list taking the rest, may refuse + # what it is given; the refusal is handled here, so it is quiet. with CaughtByCaller(FieldValueError): try: - schema = cast('Schema_Field', field_schema.unpack(data, len(data), None)) # type: ignore[call-arg,misc] + schema = cast('Schema', field_schema.unpack(octets, length, None)) # type: ignore[call-arg,misc] except FieldValueError: return data - return parser(tlv_type, schema) + if layout is None: + vendor = cast('Schema_VendorTLV', schema) + entry = Data_VendorTLV(oui=vendor.oui, subtype=vendor.subtype, presence_type=vendor.presence_type, + reserved=vendor.reserved, data=data[length:]) # type: Data_Field | Data_VendorTLV + else: + entry = layout[0](tlv_type, cast('Schema_Field', schema)) + + record = schema.__dict__.get('__short_read__') + if record is not None: + entry.__update__({'__short_read__': record}) + return entry def _make_tlv_data(self, tlv_type: 'Enum_FieldType', data: 'Data_Field | Data_VendorTLV | Mapping[str, Any] | bytes') -> 'bytes': @@ -2087,14 +2116,17 @@ def _make_tlv_data(self, tlv_type: 'Enum_FieldType', Args: tlv_type: Type of the item. - data: The item's data: raw octets, written as they are, or decoded. + data: The item's data: raw octets, written as they are, or decoded, + cut back as its ``__short_read__`` says when it has one. Returns: The item's data. Raises: ProtocolError: If decoded data is given for a type with no layout, - or does not encode. + or does not encode; if its ``__short_read__`` is not a field + name and an octet count; or if a vendor TLV whose header is cut + short has data. """ if isinstance(data, (bytes, bytearray)): @@ -2102,17 +2134,49 @@ def _make_tlv_data(self, tlv_type: 'Enum_FieldType', if tlv_type == Enum_FieldType.Vendor_Namespace: oui, vendor_data = self._make_vendor_data(tlv_type, Schema_VendorTLV, data) values = {'reserved': 0, **data} - return Schema_VendorTLV( + header = self._pack_tlv_data(tlv_type, Schema_VendorTLV( oui=oui, subtype=self._make_subfield(tlv_type, Schema_VendorTLV, values, 'subtype'), presence_type=self._make_subfield(tlv_type, Schema_VendorTLV, values, 'presence_type'), reserved=self._make_subfield(tlv_type, Schema_VendorTLV, values, 'reserved'), - ).pack() + vendor_data + ), data) + if vendor_data and len(header) < self._field_size(Schema_VendorTLV)[0]: + raise ProtocolError(f'{self.alias}: invalid VendorTLV.data: {vendor_data!r}, ' + 'the vendor header before it is cut short') + return header + vendor_data layout = self._lookup_field(tlv_type) if layout is None: raise ProtocolError(f'{self.alias}: invalid TLV.data: {tlv_type!r} defines no layout, ' 'so its data must be bytes') - return layout[1](tlv_type, data).pack() + return self._pack_tlv_data(tlv_type, layout[1](tlv_type, data), data) + + def _pack_tlv_data(self, tlv_type: 'Enum_FieldType', schema: 'Schema', + data: 'Data_Field | Data_VendorTLV | Mapping[str, Any]') -> 'bytes': + """Pack a TLV item's schema, cut back as :meth:`_read_tlv_data` recorded. + + Args: + tlv_type: Type of the item, for the error message. + schema: The schema made from ``data``. + data: The item's decoded data, whose ``__short_read__``, if any, + names the field its octets ended inside and how many of that + field's octets there were. + + Returns: + The packed schema, without the octets past the cut. + + Raises: + ProtocolError: If ``__short_read__`` is not a field of ``schema`` + and a count of octets. + + """ + record = data.get('__short_read__') + if record is not None: + if not (isinstance(record, (tuple, list)) and len(record) == 2 and record[0] in schema.__fields__ + and isinstance(record[1], int) and not isinstance(record[1], bool) and record[1] >= 0): + raise ProtocolError(f'{self.alias}: invalid {tlv_type.name}.__short_read__: {record!r}') + # NOTE: ``Schema.pack`` cuts its buffers back to a record it holds. + schema.__dict__['__short_read__'] = tuple(record) + return schema.pack() def _make_vendor_data(self, field_type: 'Enum_FieldType', schema: 'Type[Schema]', entry: 'Mapping[str, Any]') -> 'tuple[bytes, bytes]': diff --git a/tests/protocols/link/test_radiotap_fields_2576_unit.py b/tests/protocols/link/test_radiotap_fields_2576_unit.py index 78ead84708..c198c354d1 100644 --- a/tests/protocols/link/test_radiotap_fields_2576_unit.py +++ b/tests/protocols/link/test_radiotap_fields_2576_unit.py @@ -377,8 +377,11 @@ def item(tlv_type: int, data: bytes) -> bytes: 'presence_type': 5, 'reserved': 0, 'data': b'\xaa\xbb'})) self.assertEqual(got[4], ('TSFT', 8, {'type': 0, 'mactime': 0x8887868584838281})) - # S1G cut short by its producer, and a padding item, stay raw - self.assertEqual(got[5:], [('S1G', 4, s1g[:4]), ('TLV_fields_in_radiotap', 0, b'')]) + # S1G cut short by its producer decodes with the zeros it left out (#2655), + # and a padding item stays raw + self.assertEqual(got[5:], [('S1G', 4, {'type': 32, 'known': 1, 'data1': 2, 'data2': 0, + '__short_read__': ('data2', 0)}), + ('TLV_fields_in_radiotap', 0, b'')]) self._assert_round_trips(parsed, raw) def test_a_tlv_past_the_region_is_unknown(self) -> None: @@ -580,8 +583,8 @@ def test_each_field_type_pairs_a_schema_with_its_methods(self) -> None: self.assertTrue(callable(getattr(Radiotap, f'_read_field_{name}', None))) self.assertTrue(callable(getattr(Radiotap, f'_make_field_{name}', None))) - def test_a_ragged_eht_tlv_stays_raw_and_quiet(self) -> None: - """An EHT item whose user words do not end on a word stays raw, and nothing is logged.""" + def test_a_ragged_eht_tlv_decodes_quietly(self) -> None: + """An EHT item whose last user word is cut short decodes it padded with zeros (#2655), and nothing is logged.""" import logging records = [] # type: list[logging.LogRecord] @@ -595,7 +598,9 @@ def test_a_ragged_eht_tlv_stays_raw_and_quiet(self) -> None: parsed = self._parse(raw) finally: logger.removeHandler(handler) - self.assertEqual([(item.type.name, item.data) for item in parsed.info.decoded[28].tlvs], [('EHT', data)]) + self.assertEqual([(item.type.name, item.data) for item in parsed.info.decoded[28].tlvs], + [('EHT', {'type': 34, 'known': 0, 'data': tuple(range(9)), 'user_info': (0xA2A1,), + '__short_read__': ('user_info', 2)})]) self.assertEqual(records, []) self._assert_round_trips(parsed, raw) diff --git a/tests/protocols/link/test_radiotap_short_tlv_2655_unit.py b/tests/protocols/link/test_radiotap_short_tlv_2655_unit.py new file mode 100644 index 0000000000..33077a9484 --- /dev/null +++ b/tests/protocols/link/test_radiotap_short_tlv_2655_unit.py @@ -0,0 +1,301 @@ +# -*- coding: utf-8 -*- +"""GitHub issue #2655: a radiotap TLV item a producer cut short is decoded. + +radiotap.org's TLV page has parsers take an item whose data stops before its +layout ends, since a producer may leave out the octets it does not fill at the +end, and those octets are zeros. :class:`~pcapkit.protocols.link.radiotap.Radiotap` +decodes such an item as its layout padded with the zeros, records the cut as +``__short_read__``, as ``Schema.unpack`` does, and writes the octets back as +captured, through ``from_data`` and through ``make()``. + +Each case cuts a U-SIG, S1G, EHT or vendor item at a field boundary or inside a +field. Its control is the same item with the zeros put back, which is whole, so +it records no cut and decodes to the same values. Every step is quiet, and an item +whose zeros the padding limits of :mod:`pcapkit.utilities.config` refuse stays +raw, quietly too. + +Each test imports :mod:`pcapkit` itself, after +:func:`~tests._support.reimport_once_per_class`. + +""" + +import contextlib +import logging +import struct +import sys +import threading +import unittest +import warnings +from typing import TYPE_CHECKING + +from tests._support import reimport_once_per_class + +if TYPE_CHECKING: + from typing import Iterator + +#: The presence word selecting the TLV fields, bit 28, alone. +PRESENT = 1 << 28 + +#: Each item whole, by name: its TLV type, and its data from the type's page on +#: radiotap.org, every subfield holding non-zero octets throughout. +WHOLE = { + 'U-SIG': (33, struct.pack(' bytes: + """A TLV item: its type, its length, its data, and padding to 4 octets.""" + return struct.pack(' bytes: + """A radiotap header whose only field is the TLV list holding ``item``.""" + return struct.pack(' 'dict': # type: ignore[no-untyped-def] + """The subfields of an item's decoded data, its cut left out.""" + return {name: value for name, value in data.items() if name not in ('type', '__short_read__')} + + +class TestRadiotapShortTLV(unittest.TestCase): + """Pin the decoding of a TLV item cut short, and its writing back.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _parse(self, raw: bytes): # type: ignore[no-untyped-def] + from pcapkit.protocols.link.radiotap import Radiotap + + parsed = Radiotap(raw, len(raw)) + self.assertEqual(parsed.data, raw) + return parsed + + def _cut(self, case: str) -> 'tuple[bytes, int, bytes]': + """The header holding ``case``'s item, its TLV type, and the item's data.""" + name, keep, _, _, _ = CUTS[case] + tlv_type, whole = WHOLE[name] + return _radiotap(_item(tlv_type, whole[:keep])), tlv_type, whole[:keep] + + def test_a_short_item_decodes_as_its_layout_padded_with_zeros(self) -> None: + """The cut item decodes as its control does, and only it records the cut.""" + for case, (name, keep, layout, record, values) in CUTS.items(): + with self.subTest(case=case): + raw, tlv_type, data = self._cut(case) + (item,) = self._parse(raw).info.decoded[28].tlvs + (control,) = self._parse(_radiotap(_item(tlv_type, data + bytes(layout - keep)))).info.decoded[28].tlvs + + self.assertEqual((item.type, item.length), (tlv_type, keep)) + self.assertIs(type(item.data), type(control.data)) + self.assertEqual(_subfields(item.data), values) + self.assertEqual(_subfields(control.data), values) + self.assertEqual(item.data['__short_read__'], record) + self.assertNotIn('__short_read__', control.data) + + for name, (tlv_type, whole) in WHOLE.items(): + with self.subTest(whole=name): + (item,) = self._parse(_radiotap(_item(tlv_type, whole))).info.decoded[28].tlvs + self.assertNotIsInstance(item.data, bytes) + self.assertNotIn('__short_read__', item.data) + + # data past a layout of fixed size stays raw + for name in ('U-SIG', 'S1G'): + with self.subTest(long=name): + tlv_type, whole = WHOLE[name] + (item,) = self._parse(_radiotap(_item(tlv_type, whole + b'\x01'))).info.decoded[28].tlvs + self.assertEqual(item.data, whole + b'\x01') + + def test_a_registered_layout_decodes_a_cut_item(self) -> None: + """A layout registered for a TLV type, its tail taking the rest, decodes an item cut short of its head.""" + from unittest import mock + + from pcapkit.const.radiotap.field_type import FieldType + from pcapkit.corekit.fields.collections import ListField + from pcapkit.corekit.fields.numbers import UInt16Field + from pcapkit.corekit.infoclass import info_final + from pcapkit.foundation.registry import register_radiotap_field + from pcapkit.protocols.data.link.radiotap import RadiotapField as DataField + from pcapkit.protocols.link.radiotap import Radiotap + from pcapkit.protocols.schema.link.radiotap import RadiotapField + from pcapkit.protocols.schema.schema import schema_final + + # a list of no item type is the octets it spans, so the tail is not of whole items + @schema_final + class TaggedSchema(RadiotapField): + tag = UInt16Field(byteorder='little') + body = ListField(length=lambda pkt: pkt['__length__']) + + @info_final + class Tagged(DataField): + tag: int + body: bytes + + def parser(field_type, schema): # type: ignore[no-untyped-def] + return Tagged(type=field_type, tag=schema.tag, body=schema.body) + + def constructor(field_type, entry): # type: ignore[no-untyped-def] + return TaggedSchema(tag=entry['tag'], body=entry['body']) + + #: The item's data, to its tag, body and cut. + cases = { + b'\x34': (0x34, b'', ('tag', 1)), + b'\x34\x12': (0x1234, b'', None), + b'\x34\x12\xab\xcd\xef': (0x1234, b'\xab\xcd\xef', None), + } + with mock.patch.dict(Radiotap.__field__), mock.patch.dict(RadiotapField.registry): + register_radiotap_field(FieldType.get(35), (parser, constructor), schema=TaggedSchema) + for data, (tag, body, record) in cases.items(): + with self.subTest(data=data.hex()): + raw = _radiotap(_item(35, data)) + parsed = self._parse(raw) + (item,) = parsed.info.decoded[28].tlvs + self.assertIsInstance(item.data, Tagged) + self.assertEqual((item.data.tag, item.data.body, item.data.get('__short_read__')), + (tag, body, record)) + self.assertEqual(Radiotap(present=(PRESENT,), fields=parsed.info.decoded).data.hex(), raw.hex()) + + def test_from_data_rebuilds_a_short_item_as_captured(self) -> None: + """``from_data`` of the info, and of its dict, writes the cut octets and decodes them again.""" + from pcapkit.protocols.link.radiotap import Radiotap + + for case, (_, _, _, record, _) in CUTS.items(): + raw, _, _ = self._cut(case) + parsed = self._parse(raw) + for info in (parsed.info, parsed.info.to_dict()): + with self.subTest(case=case, data=type(info).__name__): + rebuilt = Radiotap.from_data(info) + self.assertEqual(rebuilt.data.hex(), raw.hex()) + (item,) = rebuilt.info.decoded[28].tlvs + self.assertEqual(item.data['__short_read__'], record) + + def test_make_writes_a_short_item_as_captured(self) -> None: + """``make()`` writes the cut octets from ``decoded``, from an item's dict, and from a hand-written cut.""" + from pcapkit.protocols.link.radiotap import Radiotap + + for case, (_, _, _, record, values) in CUTS.items(): + raw, tlv_type, _ = self._cut(case) + parsed = self._parse(raw) + (item,) = parsed.info.decoded[28].tlvs + given = { + 'decoded': parsed.info.decoded, + 'dict': [(28, {'tlvs': [{'type': tlv_type, 'data': item.data.to_dict()}]})], + 'hand-written': [(28, {'tlvs': [{'type': tlv_type, + 'data': {**values, '__short_read__': record}}]})], + # a JSON round trip turns the record into a list + 'list record': [(28, {'tlvs': [{'type': tlv_type, + 'data': {**values, '__short_read__': list(record)}}]})], + } + for how, fields in given.items(): + with self.subTest(case=case, fields=how): + made = Radiotap(present=(PRESENT,), fields=fields) + self.assertEqual(made.data.hex(), raw.hex()) + + @contextlib.contextmanager + def _assert_quiet(self) -> 'Iterator[None]': + """Fail if the block warns, logs to ``pcapkit`` at any level, or replaces an exception hook.""" + logger = logging.getLogger('pcapkit') + records = [] # type: list[logging.LogRecord] + handler = logging.Handler(logging.DEBUG) + handler.emit = records.append # type: ignore[assignment,method-assign] + level, hooks = logger.level, (sys.excepthook, threading.excepthook) + logger.addHandler(handler) + logger.setLevel(logging.DEBUG) + try: + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + yield + finally: + logger.removeHandler(handler) + logger.setLevel(level) + self.assertEqual([str(warning.message) for warning in caught], []) + self.assertEqual([record.getMessage() for record in records], []) + self.assertEqual((sys.excepthook, threading.excepthook), hooks) + + def test_a_short_item_decodes_quietly(self) -> None: + """Parsing, rebuilding and making a cut item warns of nothing and logs nothing.""" + from pcapkit.protocols.link.radiotap import Radiotap + + for case in CUTS: + with self.subTest(case=case): + raw, _, _ = self._cut(case) + with self._assert_quiet(): + parsed = Radiotap(raw, len(raw)) + Radiotap.from_data(parsed.info) + Radiotap(present=(PRESENT,), fields=parsed.info.decoded) + (item,) = parsed.info.decoded[28].tlvs + self.assertNotIsInstance(item.data, bytes) + + def test_a_cut_the_padding_limit_refuses_stays_raw_quietly(self) -> None: + """Under a zero-padding limit too low for the layout the cut item stays raw, quietly; by default it decodes.""" + from unittest import mock + + from pcapkit.protocols.link.radiotap import Radiotap + from pcapkit.utilities import config + + for case in CUTS: + with self.subTest(case=case): + raw, _, data = self._cut(case) + with mock.patch.object(config, 'MAX_ZERO_PAD_LENGTH', 1), self._assert_quiet(): + parsed = Radiotap(raw, len(raw)) + self.assertEqual(Radiotap.from_data(parsed.info).data.hex(), raw.hex()) + self.assertEqual(Radiotap(present=(PRESENT,), fields=parsed.info.decoded).data.hex(), raw.hex()) + (item,) = parsed.info.decoded[28].tlvs + self.assertEqual(item.data, data) + + (item,) = self._parse(raw).info.decoded[28].tlvs + self.assertNotIsInstance(item.data, bytes) + + def test_make_refuses_a_malformed_cut(self) -> None: + """A ``__short_read__`` that names no field and a count, or a vendor TLV cut short before its data, raises.""" + from pcapkit.protocols.link.radiotap import Radiotap + from pcapkit.utilities.exceptions import ProtocolError + + usig = {'common': 1, 'value': 2, 'mask': 3} + for record in (('nosuch', 0), ('mask', -1), ('mask', True), ('mask', '0'), ('mask',), 'mask'): + with self.subTest(record=record): + with self.assertRaises(ProtocolError): + Radiotap(present=(PRESENT,), + fields=[(28, {'tlvs': [{'type': 33, 'data': {**usig, '__short_read__': record}}]})]) + + vendor = {'oui': bytes.fromhex('001374'), 'subtype': 1, 'presence_type': 5, 'data': b'\xaa', + '__short_read__': ('presence_type', 1)} + with self.assertRaises(ProtocolError): + Radiotap(present=(PRESENT,), fields=[(28, {'tlvs': [{'type': 30, 'data': vendor}]})]) + + +if __name__ == '__main__': + unittest.main()