From 983a5c6df27d0232ad3dbe771c518f7da6e129ee Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Mon, 2 Nov 2026 16:38:17 -0500 Subject: [PATCH] fix(misc): let the built-in record a held ZigBee key's short read (#2688) zigbee_key_held zero-filled the Secrets Data to the key's fixed width, unpacked that, and wrote __short_read__ by hand, duplicating Schema._note_short_read. Its buffers then spanned the fixed width until packed, so a Decryption Secrets Block made from a parsed short key was sized from them and written corrupt. Unpack the octets as held, with a declared length so that empty Secrets Data is not the end of the stream, and let Schema.unpack record the cut. The parse already reported the short read, so its warnings are dropped through hold_warnings with discard set. --- pcapkit/protocols/schema/misc/pcapng.py | 47 +++--- .../test_pcapng_zigbee_key_held_2688_unit.py | 148 ++++++++++++++++++ 2 files changed, 172 insertions(+), 23 deletions(-) create mode 100644 tests/protocols/misc/test_pcapng_zigbee_key_held_2688_unit.py diff --git a/pcapkit/protocols/schema/misc/pcapng.py b/pcapkit/protocols/schema/misc/pcapng.py index e5fe5139f1..4e96bc93a6 100644 --- a/pcapkit/protocols/schema/misc/pcapng.py +++ b/pcapkit/protocols/schema/misc/pcapng.py @@ -31,7 +31,8 @@ from pcapkit.protocols.schema.schema import EnumSchema, Schema, schema_final from pcapkit.utilities.exceptions import FieldValueError, ProtocolError, stacklevel from pcapkit.utilities.logging import SPHINX_TYPE_CHECKING -from pcapkit.utilities.warnings import ProtocolWarning, RegistryWarning, SchemaWarning, warn +from pcapkit.utilities.warnings import (ProtocolWarning, RegistryWarning, SchemaWarning, + hold_warnings, warn) __all__ = [ 'PCAPNG', @@ -663,30 +664,30 @@ def zigbee_key_held(schema: 'Type[_ZigBeeKey]', data: 'bytes') -> '_ZigBeeKey': data: The Secrets Data octets, as the block held them. Returns: - A ``schema`` instance that packs back to exactly ``data``. - - The fixed fields are read from ``data``, and any octets past them are kept - as the schema's ``trailing`` octets. A ``data`` short of the fixed width is - read as though zero-filled and recorded as a short read, so that packing - cuts it back to the octets held, as :meth:`Schema._cut_short_read - ` describes. A key - shorter than its fixed width, or a Secrets Length longer than it, then - rebuilds byte for byte (:issue:`1986`). + A ``schema`` instance that holds, and packs back to, exactly ``data``. + + ``data`` is unpacked as the block's Secrets Data is, with any octets past + the fixed fields kept as the schema's ``trailing`` octets. A ``data`` short + of the fixed width is a short read, which :meth:`Schema.unpack + ` records for + :meth:`Schema._cut_short_read + `. A key shorter + than its fixed width, or a Secrets Length longer than it, then rebuilds + byte for byte (:issue:`1986`). The length is declared, so an empty ``data`` + is an empty key rather than the end of the stream. The short read's + warnings are dropped, since the parse that held ``data`` reported them. + + .. versionchanged:: 1.6.0 + + The schema's buffers hold ``data`` as given, rather than zero-filled + to the fixed width, so its :func:`bytes` and :func:`len` are those of + ``data`` and a block sized from it is written as captured + (:issue:`2688`). """ - fixed = [(name, field.length) for (name, field) in schema.__fields__.items() - if name != 'trailing'] - width = sum(length for (_, length) in fixed) - padded = data.ljust(width, b'\x00') - value = cast('_ZigBeeKey', schema.unpack(padded)) # type: ignore[call-arg,misc] - - offset = 0 - for (name, length) in fixed: - if len(data) < offset + length: - value.__dict__['__short_read__'] = (name, len(data) - offset) - break - offset += length - return value + with hold_warnings() as held: + held.discard = True + return cast('_ZigBeeKey', schema.unpack(data, len(data))) # type: ignore[call-arg,misc] def pcapng_block_selector(packet: 'dict[str, Any]') -> 'Field': diff --git a/tests/protocols/misc/test_pcapng_zigbee_key_held_2688_unit.py b/tests/protocols/misc/test_pcapng_zigbee_key_held_2688_unit.py new file mode 100644 index 0000000000..12fe789bed --- /dev/null +++ b/tests/protocols/misc/test_pcapng_zigbee_key_held_2688_unit.py @@ -0,0 +1,148 @@ +# -*- coding: utf-8 -*- +"""GitHub issue #2688: a held ZigBee key is an ordinary short read. + +:func:`~pcapkit.protocols.schema.misc.pcapng.zigbee_key_held` rebuilds a ZigBee +NWK or APS key schema from the Secrets Data octets a parsed block held. Without +the fix it zero-fills those octets to the key's fixed width and records the cut +by hand, so the schema's buffers span the fixed width until it is packed: +:func:`bytes` and :func:`len` of it are wrong, and a Decryption Secrets Block +made from a parsed short key is sized from them and written corrupt. With it, +the octets are unpacked as they are, :meth:`Schema.unpack` records the cut, and +the warning the parse reported for that cut is not reported again. + +Each key is cut at every offset, and read past its fixed width as well. +:mod:`pcapkit` is imported inside each test, after +:func:`~tests._support.reimport_once_per_class`. + +""" + +import contextlib +import logging +import struct +import unittest +import warnings +from typing import TYPE_CHECKING + +from tests.protocols.misc.test_pcapng_roundtrip2_unit import PCAPNGTestCase + +if TYPE_CHECKING: + from typing import Any, Iterator, Optional + +#: Each key's Secrets Type, with its fields and their widths in octets as the +#: PCAP-NG specification lays the Secrets Data out. +FIELDS = { + 0x5A4E574B: (('key', 16), ('panid', 2)), # ZigBee NWK Key + 0x5A415053: (('key', 16), ('panid', 2), ('addr_low', 2), ('addr_high', 2)), # ZigBee APS Key +} + +#: Octets past the fixed width the sweep reads as well, kept as ``trailing``. +PAST = 3 + + +def _secrets(count: 'int') -> 'bytes': + """``count`` non-zero octets, so a zero-filled rebuild cannot pass for them.""" + return bytes(range(1, count + 1)) + + +def _dsb(secrets_type: 'int', secrets: 'bytes') -> 'bytes': + """A little-endian Decryption Secrets Block declaring exactly ``secrets``.""" + body = struct.pack(' 'Optional[tuple[str, int]]': + """The field ``count`` octets end inside, and how many of its octets they hold.""" + offset = 0 + for name, width in fields: + if count < offset + width: + return (name, count - offset) + offset += width + return None + + +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: + self.messages.append(f'logged: {record.getMessage()}') + + +@contextlib.contextmanager +def _reported() -> 'Iterator[list[str]]': + """Collect what the block reports, on the :mod:`warnings` channel and the logger.""" + handler = _Logged() + logger = logging.getLogger('pcapkit') + logger.addHandler(handler) + try: + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + yield handler.messages + finally: + logger.removeHandler(handler) + handler.messages.extend(f'warned: {item.message}' for item in caught) + + +class TestPCAPNGZigBeeKeyHeld(PCAPNGTestCase): + """Pin a held ZigBee key cut at every offset, at schema and at protocol level.""" + + def _cases(self) -> 'Iterator[tuple[Any, int, tuple[tuple[str, int], ...], int]]': + """Each key's schema, Secrets Type and fields, with every count of octets swept.""" + from pcapkit.protocols.schema.misc.pcapng import ZigBeeAPSKey, ZigBeeNWKKey + + schemas = {0x5A4E574B: ZigBeeNWKKey, 0x5A415053: ZigBeeAPSKey} + for secrets_type, fields in FIELDS.items(): + schema = schemas[secrets_type] + self.assertEqual(tuple((name, field.length) for name, field in schema.__fields__.items() + if name != 'trailing'), fields) + for count in range(sum(width for _, width in fields) + PAST + 1): + yield schema, secrets_type, fields, count + + def test_schema_holds_the_octets_held(self) -> None: + """The schema holds exactly the octets given, records their cut, and reports nothing.""" + from pcapkit.protocols.schema.misc.pcapng import zigbee_key_held + + for schema, secrets_type, fields, count in self._cases(): + with self.subTest(secrets_type=hex(secrets_type), count=count): + data = _secrets(count) + with _reported() as messages: + held = zigbee_key_held(schema, data) + self.assertEqual(messages, []) + # read before packing, which cuts the buffers back itself + self.assertEqual(bytes(held).hex(), data.hex()) + self.assertEqual(len(held), count) + self.assertEqual(held.__dict__.get('__short_read__'), _cut(fields, count)) + self.assertEqual(held.key, data[:16].ljust(16, b'\x00')) + self.assertEqual(held.panid, int.from_bytes(data[16:18].ljust(2, b'\x00'), 'little')) + width = sum(width for _, width in fields) + self.assertEqual(held.trailing, data[width:] if count > width else None) + self.assertEqual(held.pack().hex(), data.hex()) + + def test_made_block_is_the_captured_block(self) -> None: + """Made from the parsed key, or rebuilt from the block, it is as captured and quiet.""" + from pcapkit.const.pcapng.block_type import BlockType + from pcapkit.protocols.misc.pcapng import PCAPNG + + for _, secrets_type, _, count in self._cases(): + with self.subTest(secrets_type=hex(secrets_type), count=count): + octets = _dsb(secrets_type, _secrets(count)) + context = self._context() + with _reported(): + parsed = PCAPNG(octets, len(octets), num=2, sct=1, ctx=context) + with _reported() as messages: + made = PCAPNG(num=2, sct=1, ctx=context, type=BlockType.Decryption_Secrets_Block, + block={'secrets_type': parsed.info.secrets_type, # type: ignore[attr-defined] + 'secrets_data': parsed.info.secrets_data}).data # type: ignore[attr-defined] + rebuilt = PCAPNG.from_data(parsed.info, num=2, sct=1, ctx=context).data + self.assertEqual(made.hex(), octets.hex()) + self.assertEqual(rebuilt.hex(), octets.hex()) + self.assertEqual(messages, []) + + +if __name__ == '__main__': + unittest.main()