diff --git a/pcapkit/protocols/misc/pcapng.py b/pcapkit/protocols/misc/pcapng.py index 8869335200..28721b441b 100644 --- a/pcapkit/protocols/misc/pcapng.py +++ b/pcapkit/protocols/misc/pcapng.py @@ -6833,6 +6833,7 @@ def _make_secrets_unknown(self, type: 'Enum_SecretsType', secrets: 'Optional[Dat def _make_secrets_tls(self, type: 'Enum_SecretsType', secrets: 'Optional[Data_TLSKeyLog]' = None, *, entries: 'Optional[dict[Enum_TLSKeyLabel, OrderedMultiDict[bytes, bytes]]]' = None, + data: 'Optional[str]' = None, **kwargs: 'Any') -> 'Schema_TLSKeyLog': """Make PCAP-NG TLS Key Log secrets. @@ -6843,37 +6844,48 @@ def _make_secrets_tls(self, type: 'Enum_SecretsType', secrets: 'Optional[Data_TL On a parsed model the text wins, so an edit to its ``entries`` alone is silently ignored. entries: TLS Key Log entries. + data: TLS key log text, written back verbatim in place of + ``entries``, as the text of ``secrets`` is. **kwargs: Arbitrary keyword arguments. Returns: Constructed secrets schema. + .. versionchanged:: 1.6.0 + + Takes the key log text as ``data``, so that the :obj:`dict` of a + parsed model rebuilds the log as captured, as the model itself does, + rather than as its entries under a generated comment + (:issue:`2698`). + """ from pcapkit import __version__ if secrets is not None: - text = secrets.get('data') - if isinstance(text, str): - return Schema_TLSKeyLog( - data=text, - ) + data = secrets.get('data') + if isinstance(data, str): + return Schema_TLSKeyLog( + data=data, + ) + if secrets is not None: # NOTE: each an OrderedMultiInfo, which is an OrderedMultiDict. entries = cast('dict[Enum_TLSKeyLabel, OrderedMultiDict[bytes, bytes]]', secrets.entries) if entries is None: entries = {} - data = [f'# generated by PyPCAPKit v{__version__}\n'] # type: list[str] + lines = [f'# generated by PyPCAPKit v{__version__}\n'] # type: list[str] for label, entry in entries.items(): for k, v in entry.items(multi=True): - data.append(f'{label.name} {k.hex()} {v.hex()}\n') + lines.append(f'{label.name} {k.hex()} {v.hex()}\n') return Schema_TLSKeyLog( - data=''.join(data), + data=''.join(lines), ) def _make_secrets_wireguard(self, type: 'Enum_SecretsType', secrets: 'Optional[Data_WireGuardKeyLog]' = None, *, entries: 'Optional[OrderedMultiDict[WireGuardKeyLabel, bytes]]' = None, + data: 'Optional[str]' = None, **kwargs: 'Any') -> 'Schema_WireGuardKeyLog': """Make PCAP-NG WireGuard secrets. @@ -6884,36 +6896,45 @@ def _make_secrets_wireguard(self, type: 'Enum_SecretsType', secrets: 'Optional[D On a parsed model the text wins, so an edit to its ``entries`` alone is silently ignored. entries: WireGuard Key Log entries. + data: WireGuard key log text, written back verbatim in place of + ``entries``, as the text of ``secrets`` is. **kwargs: Arbitrary keyword arguments. Returns: Constructed secrets schema. + .. versionchanged:: 1.6.0 + + Takes the key log text as ``data``, as + :meth:`_make_secrets_tls` does (:issue:`2698`). + """ from pcapkit import __version__ if secrets is not None: - text = secrets.get('data') - if isinstance(text, str): - return Schema_WireGuardKeyLog( - data=text, - ) + data = secrets.get('data') + if isinstance(data, str): + return Schema_WireGuardKeyLog( + data=data, + ) + if secrets is not None: entries = secrets.entries if entries is None: entries = OrderedMultiDict() - data = [f'# generated by PyPCAPKit v{__version__}\n'] # type: list[str] + lines = [f'# generated by PyPCAPKit v{__version__}\n'] # type: list[str] for label, value in entries.items(multi=True): - data.append(f'{label.name} = {base64.b64encode(value).decode()}\n') + lines.append(f'{label.name} = {base64.b64encode(value).decode()}\n') return Schema_WireGuardKeyLog( - data=''.join(data), + data=''.join(lines), ) def _make_secrets_zigbee_nwk(self, type: 'Enum_SecretsType', secrets: 'Optional[Data_ZigBeeNWKKey]' = None, *, nwk_key: 'bytes' = b'', pan_id: 'int' = 0, + data: 'Optional[bytes]' = None, **kwargs: 'Any') -> 'Schema_ZigBeeNWKKey': """Make PCAP-NG ZigBee NWK Key secrets. @@ -6927,16 +6948,25 @@ def _make_secrets_zigbee_nwk(self, type: 'Enum_SecretsType', secrets: 'Optional[ ignored. nwk_key: NWK key. pan_id: PAN ID. + data: Secrets Data octets, written back verbatim in place of + ``nwk_key`` and ``pan_id``, as the octets of ``secrets`` are. **kwargs: Arbitrary keyword arguments. Returns: Constructed secrets schema. + .. versionchanged:: 1.6.0 + + Takes the Secrets Data octets as ``data``, so that the :obj:`dict` + of a parsed model rebuilds the key as captured, as the model itself + does, rather than as its zero-filled fields (:issue:`2698`). + """ if secrets is not None: - held = secrets.get('data') - if isinstance(held, bytes): - return zigbee_key_held(Schema_ZigBeeNWKKey, held) + data = secrets.get('data') + if isinstance(data, bytes): + return zigbee_key_held(Schema_ZigBeeNWKKey, data) + if secrets is not None: nwk_key = secrets.nwk_key pan_id = secrets.pan_id @@ -6949,6 +6979,7 @@ def _make_secrets_zigbee_aps(self, type: 'Enum_SecretsType', secrets: 'Optional[ aps_key: 'bytes' = b'', pan_id: 'int' = 0, short_address: 'int' = 0, + data: 'Optional[bytes]' = None, **kwargs: 'Any') -> 'Schema_ZigBeeAPSKey': """Make PCAP-NG ZigBee APS Key secrets. @@ -6960,16 +6991,24 @@ def _make_secrets_zigbee_aps(self, type: 'Enum_SecretsType', secrets: 'Optional[ aps_key: APS key. pan_id: PAN ID. short_address: Short address. + data: Secrets Data octets, written back verbatim in place of the + fields, as the octets of ``secrets`` are. **kwargs: Arbitrary keyword arguments. Returns: Constructed secrets schema. + .. versionchanged:: 1.6.0 + + Takes the Secrets Data octets as ``data``, as + :meth:`_make_secrets_zigbee_nwk` does (:issue:`2698`). + """ if secrets is not None: - held = secrets.get('data') - if isinstance(held, bytes): - return zigbee_key_held(Schema_ZigBeeAPSKey, held) + data = secrets.get('data') + if isinstance(data, bytes): + return zigbee_key_held(Schema_ZigBeeAPSKey, data) + if secrets is not None: aps_key = secrets.aps_key pan_id = secrets.pan_id short_address = secrets.short_address diff --git a/tests/protocols/misc/test_pcapng_dsb_secrets_dict_2698_unit.py b/tests/protocols/misc/test_pcapng_dsb_secrets_dict_2698_unit.py new file mode 100644 index 0000000000..5cfe7d4603 --- /dev/null +++ b/tests/protocols/misc/test_pcapng_dsb_secrets_dict_2698_unit.py @@ -0,0 +1,187 @@ +# -*- coding: utf-8 -*- +"""GitHub issue #2698: a Decryption Secrets Block rebuilds as captured from its dict. + +:meth:`PCAPNG.from_data ` takes a +parsed block's data model or the :obj:`dict` that ``to_dict`` makes of it, and +both must give the captured block back. The dict's secrets come back from +:meth:`~pcapkit.corekit.infoclass.Info.from_dict` as a mapping rather than as +their data model, so the block maker spreads them into the secrets maker as its +keywords. A key log's text and a ZigBee key's Secrets Data octets are held +under ``data``, which the TLS, WireGuard and ZigBee makers read from a data +model alone: from the dict, a key log was written from its entries under a +generated comment, and a ZigBee key from its zero-filled fields. Each takes +``data`` as a keyword too, as the unknown-secrets maker does. + +Every registered secrets type, and an unregistered one, is swept whole, cut at +every octet, and past its fixed width where it has one. Each block is rebuilt +from its data model and from its dict, and both must give the captured octets +and report the same. :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 + +#: Secrets Types: an unregistered one, then the TLS and WireGuard key logs and +#: the ZigBee NWK and APS keys. +UNKNOWN = 0x12345678 +TLS = 0x544C534B +WIREGUARD = 0x57474B4C +ZIGBEE_NWK = 0x5A4E574B +ZIGBEE_APS = 0x5A415053 + +#: Each Secrets Type's Secrets Data, whole: the key logs with a comment, a +#: label in lower case and a line that does not parse, all of which the +#: entries drop and the text keeps, and the ZigBee keys at their fixed width. +WHOLE = { + UNKNOWN: bytes(range(1, 23)), + TLS: (b'# keys\nCLIENT_RANDOM 00112233 aabbccdd\n' + b'client_random 0011 aa\nCLIENT_RANDOM 00112233\n'), + WIREGUARD: (b'# keys\nLOCAL_STATIC_PRIVATE_KEY = ' + b'A' * 43 + b'=\n' + b'preshared_key = ' + b'B' * 43 + b'=\nPRESHARED_KEY\n'), + ZIGBEE_NWK: bytes(range(1, 19)), + ZIGBEE_APS: bytes(range(1, 23)), +} + +#: The Secrets Types read as a fixed width, past which more octets are kept. +FIXED = (UNKNOWN, ZIGBEE_NWK, ZIGBEE_APS) + +#: Octets past the whole Secrets Data that the sweep reads as well. +PAST = bytes(range(0xF0, 0xF4)) + + +def _dsb(secrets_type: 'int', secrets: 'bytes', secrets_length: 'int | None' = None) -> 'bytes': + """A little-endian Decryption Secrets Block over ``secrets``, padded with zeros.""" + declared = len(secrets) if secrets_length is None else secrets_length + body = struct.pack(' 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 a rebuild 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 TestPCAPNGDSBSecretsDict(PCAPNGTestCase): + """Pin a Decryption Secrets Block rebuilt from its data model and from its dict.""" + + def _secrets_types(self) -> 'Iterator[int]': + """Each Secrets Type swept, after checking that the sweep names every registered one.""" + from pcapkit.protocols.misc.pcapng import PCAPNG + + self.assertNotIn(UNKNOWN, PCAPNG.__secrets__) + self.assertEqual(set(WHOLE) - {UNKNOWN}, {int(code) for code in PCAPNG.__secrets__}) + yield from WHOLE + + def _assert_rebuilds(self, octets: 'bytes') -> None: + """Parse ``octets``, then rebuild them from the data model and from its dict.""" + from pcapkit.protocols.misc.pcapng import PCAPNG + + context = self._context() + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + parsed = PCAPNG(octets, len(octets), num=2, sct=1, ctx=context) + + reports = {} # type: dict[str, list[str]] + for path, data in (('model', parsed.info), ('dict', parsed.info.to_dict())): + with self.subTest(path=path): + with _reported() as messages: + rebuilt = PCAPNG.from_data(data, num=2, sct=1, ctx=context).data + reports[path] = messages + self.assertEqual(rebuilt.hex(), octets.hex()) + self.assertEqual(reports.get('dict'), reports.get('model')) + + def test_whole_secrets_rebuild(self) -> None: + """The issue's TLS line, and each Secrets Type whole, rebuild from either.""" + with self.subTest(secrets_type='issue'): + self._assert_rebuilds(_dsb(TLS, b'CLIENT_RANDOM 00112233 aabbccdd\n')) + for secrets_type in self._secrets_types(): + with self.subTest(secrets_type=hex(secrets_type)): + self._assert_rebuilds(_dsb(secrets_type, WHOLE[secrets_type])) + + def test_cut_secrets_rebuild(self) -> None: + """Secrets Data cut at every octet rebuild from either, as captured.""" + for secrets_type in self._secrets_types(): + whole = WHOLE[secrets_type] + for count in range(len(whole)): + with self.subTest(secrets_type=hex(secrets_type), count=count): + self._assert_rebuilds(_dsb(secrets_type, whole[:count])) + + def test_secrets_past_the_fixed_width_rebuild(self) -> None: + """Octets past a fixed width rebuild from either, as captured.""" + for secrets_type in self._secrets_types(): + if secrets_type not in FIXED: + continue + for count in range(1, len(PAST) + 1): + with self.subTest(secrets_type=hex(secrets_type), count=count): + self._assert_rebuilds(_dsb(secrets_type, WHOLE[secrets_type] + PAST[:count])) + + def test_secrets_length_past_the_block_rebuilds(self) -> None: + """A Secrets Length past the block rebuilds from either, as declared (#1980).""" + for secrets_type in self._secrets_types(): + whole = WHOLE[secrets_type] + for count in (0, len(whole) // 2, len(whole)): + with self.subTest(secrets_type=hex(secrets_type), count=count): + self._assert_rebuilds(_dsb(secrets_type, whole[:count], 0xFFFF_FFF0)) + + def test_data_keyword_is_written_verbatim(self) -> None: + """Made from keywords, ``data`` is written as given, over the decoded fields.""" + from pcapkit.const.pcapng.block_type import BlockType + from pcapkit.const.pcapng.tls_key_label import TLSKeyLabel + from pcapkit.corekit.multidict import OrderedMultiDict + from pcapkit.protocols.misc.pcapng import PCAPNG, WireGuardKeyLabel + + decoded = { + UNKNOWN: {}, + TLS: {'entries': {TLSKeyLabel.CLIENT_RANDOM: OrderedMultiDict([(b'\x01', b'\x02')])}}, + WIREGUARD: {'entries': OrderedMultiDict([(WireGuardKeyLabel.PRESHARED_KEY, bytes(32))])}, + ZIGBEE_NWK: {'nwk_key': bytes(16), 'pan_id': 0x1234}, + ZIGBEE_APS: {'aps_key': bytes(16), 'pan_id': 0x1234, 'short_address': 0x56789ABC}, + } # type: dict[int, dict[str, Any]] + for secrets_type in self._secrets_types(): + whole = WHOLE[secrets_type] + data = whole.decode('ascii') if secrets_type in (TLS, WIREGUARD) else whole + with self.subTest(secrets_type=hex(secrets_type)), warnings.catch_warnings(): + # the made block is parsed back, which reports the malformed lines + warnings.simplefilter('ignore') + made = PCAPNG(num=2, sct=1, ctx=self._context(), type=BlockType.Decryption_Secrets_Block, + block={'secrets_type': secrets_type, + 'secrets_data': dict(decoded[secrets_type], data=data)}).data + self.assertEqual(made.hex(), _dsb(secrets_type, whole).hex()) + + +if __name__ == '__main__': + unittest.main()