From cc0a0f38d13e57980a3ce223245cfeecfdcfa972 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Tue, 20 Oct 2026 13:24:30 -0400 Subject: [PATCH 1/3] fix(pcapng): warn on a malformed TLS key-log line instead of aborting (#1985) A TLS or WireGuard key log line that did not parse raised FieldValueError from post_process, which aborted the whole extraction, wherever the line sat in the log. It now raises a ProtocolWarning ("...; line kept as captured"), is left out of the entries, and the well-formed entries around it are kept. The log text is already kept verbatim on the data model, so the block still rebuilds octet for octet. A last line the capture or the block cut is still left out silently, as #1983 does. Tests that pinned the old refusal now pin the warning. Closes #1985 --- pcapkit/protocols/schema/misc/pcapng.py | 29 ++- .../misc/test_misc_short_read_1470_runtime.py | 30 ++-- .../misc/test_pcapng_roundtrip2_unit.py | 14 +- ...t_pcapng_tls_keylog_malformed_1985_unit.py | 166 ++++++++++++++++++ tests/protocols/misc/test_pcapng_unit.py | 9 +- 5 files changed, 224 insertions(+), 24 deletions(-) create mode 100644 tests/protocols/misc/test_pcapng_tls_keylog_malformed_1985_unit.py diff --git a/pcapkit/protocols/schema/misc/pcapng.py b/pcapkit/protocols/schema/misc/pcapng.py index ff61179147..669b8bd6d7 100644 --- a/pcapkit/protocols/schema/misc/pcapng.py +++ b/pcapkit/protocols/schema/misc/pcapng.py @@ -2302,6 +2302,17 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': Returns: Revised schema. + A line that does not parse -- an unknown label, a field that is not + hexadecimal, a wrong number of fields, or the first octets of an entry + cut inside the block's padding -- is left out of :attr:`entries` and + reported as a :class:`~pcapkit.utilities.warnings.ProtocolWarning`, + wherever it sits in the log (:issue:`1985`). Refusing it would abort + the whole extraction rather than this one line, as :func:`nonnegative` + explains, and :attr:`data` keeps the log text verbatim, so the line is + still kept as captured and the block rebuilds octet for octet. A last line the + capture or the block cut is left out silently (see + :func:`_cut_last_line`), since the cut is reported already. + """ from pcapkit.const.pcapng.tls_key_label import TLSKeyLabel as Enum_TLSKeyLabel @@ -2316,10 +2327,11 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': label_enum = Enum_TLSKeyLabel(label.upper()) entries[label_enum].add(bytes.fromhex(random), bytes.fromhex(secret)) - except ValueError as error: + except ValueError: if _cut_last_line(self, index, lines, packet): break - raise FieldValueError(f'invalid TLS key log format: {line!r}') from error + warn(f'PCAP-NG: invalid TLS key log format: {line!r}; line kept as captured', + ProtocolWarning, stacklevel=stacklevel()) self.entries = entries return self @@ -2347,6 +2359,10 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': Returns: Revised schema. + A line that does not parse is left out of :attr:`entries`, reported and + kept as captured in :attr:`data`, as :meth:`TLSKeyLog.post_process` + describes (:issue:`1985`). + """ from pcapkit.protocols.misc.pcapng import WireGuardKeyLabel @@ -2359,16 +2375,15 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': fields = line.strip().split() try: if len(fields) != 3 or fields[1] != '=': - raise FieldValueError(f'invalid WireGuard key log format: {line!r}') + raise ValueError(line) label, _, secret = fields label_enum = WireGuardKeyLabel(label.upper()) entries.add(label_enum, base64.b64decode(secret)) - except ValueError as error: + except ValueError: if _cut_last_line(self, index, lines, packet): break - if isinstance(error, FieldValueError): - raise - raise FieldValueError(f'invalid WireGuard key log format: {line!r}') from error + warn(f'PCAP-NG: invalid WireGuard key log format: {line!r}; line kept as captured', + ProtocolWarning, stacklevel=stacklevel()) self.entries = entries return self diff --git a/tests/protocols/misc/test_misc_short_read_1470_runtime.py b/tests/protocols/misc/test_misc_short_read_1470_runtime.py index 252d21d1e8..54b592120d 100644 --- a/tests/protocols/misc/test_misc_short_read_1470_runtime.py +++ b/tests/protocols/misc/test_misc_short_read_1470_runtime.py @@ -209,17 +209,27 @@ def test_cut_last_line_is_left_out_of_entries(self) -> None: self.assertTrue(schema.data.startswith(text.decode()), schema.data) self.assertEqual(len(schema.entries), 1) - def test_malformed_line_of_a_whole_log_is_refused(self) -> None: - from pcapkit.utilities.exceptions import FieldValueError - - text = b'CLIENT_RANDOM abcd\n' - with self.assertRaisesRegex(FieldValueError, r"^invalid TLS key log format: 'CLIENT_RANDOM abcd'$"): - self._unpack('TLSKeyLog', text, len(text)) + def test_malformed_line_of_a_whole_log_is_reported(self) -> None: + # a line that does not parse warns and is kept as captured, rather than + # aborting the extraction (#1985) + import importlib - text = b'LOCAL_STATIC_PRIVATE_KEY = Q\n' - with self.assertRaisesRegex(FieldValueError, - r"^invalid WireGuard key log format: 'LOCAL_STATIC_PRIVATE_KEY = Q'$"): - self._unpack('WireGuardKeyLog', text, len(text)) + from pcapkit.utilities.warnings import ProtocolWarning + + module = importlib.import_module('pcapkit.protocols.schema.misc.pcapng') + for schema, text, message in ( + ('TLSKeyLog', b'CLIENT_RANDOM abcd\n', + "PCAP-NG: invalid TLS key log format: 'CLIENT_RANDOM abcd'; line kept as captured"), + ('WireGuardKeyLog', b'LOCAL_STATIC_PRIVATE_KEY = Q\n', + "PCAP-NG: invalid WireGuard key log format: 'LOCAL_STATIC_PRIVATE_KEY = Q'; " + 'line kept as captured'), + ): + with self.subTest(schema=schema): + with self.assertWarns(ProtocolWarning) as caught: + parsed = getattr(module, schema).unpack(text, len(text)) + self.assertEqual(str(caught.warning), message) + self.assertEqual(len(parsed.entries), 0) + self.assertEqual(parsed.data, text.decode()) if __name__ == '__main__': diff --git a/tests/protocols/misc/test_pcapng_roundtrip2_unit.py b/tests/protocols/misc/test_pcapng_roundtrip2_unit.py index 619fa40fec..1ef735f347 100644 --- a/tests/protocols/misc/test_pcapng_roundtrip2_unit.py +++ b/tests/protocols/misc/test_pcapng_roundtrip2_unit.py @@ -296,8 +296,10 @@ def test_repeated_epb_verdict_rebuilds(self) -> None: self.assertRebuilds(_epb_with_options(verdict + verdict)) def test_wireguard_error_names_the_line(self) -> None: - """#1286: the error message carries the offending line.""" - from pcapkit.utilities.exceptions import FieldValueError + """#1286: the report carries the offending line, which no longer aborts the parse (#1985).""" + import warnings + + from pcapkit.utilities.warnings import ProtocolWarning body = b'a b c\n\x00\x00' octets = b''.join(( @@ -305,8 +307,12 @@ def test_wireguard_error_names_the_line(self) -> None: (0x57474b4c).to_bytes(4, 'little'), (6).to_bytes(4, 'little'), body, (28).to_bytes(4, 'little'), )) - with self.assertRaisesRegex(FieldValueError, r"format: 'a b c'"): - self._parse(octets) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + with self.assertWarnsRegex(ProtocolWarning, r"format: 'a b c'; line kept as captured"): + parsed = self._parse(octets) + self.assertEqual(len(parsed.info.secrets_data.entries), 0) + self.assertRebuilds(octets) if __name__ == '__main__': diff --git a/tests/protocols/misc/test_pcapng_tls_keylog_malformed_1985_unit.py b/tests/protocols/misc/test_pcapng_tls_keylog_malformed_1985_unit.py new file mode 100644 index 0000000000..9cd4ec9ca1 --- /dev/null +++ b/tests/protocols/misc/test_pcapng_tls_keylog_malformed_1985_unit.py @@ -0,0 +1,166 @@ +# -*- coding: utf-8 -*- +"""GitHub issue #1985: a malformed key log line no longer aborts the extraction. + +A well-sized Decryption Secrets Block whose TLS key log held a line that did not +parse -- an unknown label, fields that are not hexadecimal, a garbage line, or +the first octets of an entry cut inside the block's padding -- raised +:exc:`~pcapkit.utilities.exceptions.FieldValueError` from +:meth:`TLSKeyLog.post_process `, +which ended the whole extraction. The WireGuard key log did the same. Such a line +now raises a :class:`~pcapkit.utilities.warnings.ProtocolWarning` and is kept as +captured in the log text, the well-formed entries around it are kept, and the +block rebuilds octet for octet. + +:mod:`pcapkit` is imported inside each test, after +:func:`~tests._support.reimport_once_per_class`. + +""" + +import base64 +import unittest +import warnings + +from tests.protocols.misc.test_pcapng_dsb_secrets_overrun_1980_unit import (IDB, SHB, + TLS_KEY_LOG, _dsb, + _epb) +from tests.protocols.misc.test_pcapng_roundtrip2_unit import PCAPNGTestCase + +FRAME = bytes(range(1, 61)) + +#: Secrets Type of a WireGuard Key Log. +WIREGUARD_KEY_LOG = 0x57474B4C + +#: Two well-formed TLS key log lines, under different labels. +TLS_LINE = b'CLIENT_RANDOM ' + b'ab' * 32 + b' ' + b'cd' * 48 + b'\n' +TLS_LINE2 = b'SERVER_HANDSHAKE_TRAFFIC_SECRET ' + b'12' * 32 + b' ' + b'34' * 32 + b'\n' + +#: One well-formed WireGuard key log line. +WG_KEY = bytes(range(32)) +WG_LINE = b'LOCAL_STATIC_PRIVATE_KEY = ' + base64.b64encode(WG_KEY) + b'\n' + +#: Malformed TLS key log lines, each followed by a newline. +TLS_MALFORMED = ( + b'NOT_A_LABEL aa bb\n', + b'CLIENT_RANDOM zz yy\n', + b'garbage\n', + b'CLIENT_RANDOM ' + b'ab' * 32 + b'\n', + b'\xff\xfe junk\n', +) + + +class TestPCAPNGKeyLogMalformedLine(PCAPNGTestCase): + """Pin how a key log line that does not parse is treated.""" + + def _extract(self, block: 'bytes') -> 'tuple[list[bytes], list[str]]': + """Extract a file holding ``block`` between two Enhanced Packet Blocks.""" + import io + + import pcapkit + from pcapkit.utilities.warnings import ProtocolWarning + + data = SHB + IDB + _epb() + block + _epb() + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + extraction = pcapkit.extract(fin=io.BytesIO(data), nofile=True, store=True, + engine='default') + return ([bytes(frame.info.packet) for frame in extraction.frame], + [str(item.message) for item in caught if issubclass(item.category, ProtocolWarning)]) + + def _parse_quietly(self, octets: 'bytes') -> 'object': + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + return self._parse(octets) + + def assertRebuildsQuietly(self, octets: 'bytes') -> None: # pylint: disable=invalid-name + """Assert ``octets`` rebuild byte-exactly from both the schema and the data model.""" + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + self.assertEqual(self._parse(octets).__header__.pack().hex(), octets.hex()) # type: ignore[attr-defined] + self.assertRebuilds(octets) + + def assertKeyLogLine(self, messages: 'list[str]', kind: 'str', line: 'bytes') -> None: # pylint: disable=invalid-name + """Assert ``messages`` report ``line`` of a ``kind`` key log kept as captured.""" + text = line.rstrip(b'\n').decode('ascii', errors='replace') + self.assertIn(f'PCAP-NG: invalid {kind} key log format: {text!r}; line kept as captured', + messages) + + def _tls_entries(self, octets: 'bytes') -> 'list[tuple[str, bytes, bytes]]': + secrets = self._parse_quietly(octets).info.secrets_data # type: ignore[attr-defined] + return [(label.name, random, secret) + for label, entry in secrets.entries.items() + for random, secret in entry.items(multi=True)] + + def test_malformed_first_line(self) -> None: + for line in TLS_MALFORMED: + with self.subTest(line=line): + secrets = line + TLS_LINE + octets = _dsb(TLS_KEY_LOG, len(secrets), secrets) + frames, messages = self._extract(octets) + self.assertEqual(frames, [FRAME, FRAME]) + self.assertKeyLogLine(messages, 'TLS', line) + self.assertEqual(self._tls_entries(octets), + [('CLIENT_RANDOM', b'\xab' * 32, b'\xcd' * 48)]) + self.assertRebuildsQuietly(octets) + + def test_malformed_middle_line(self) -> None: + for line in TLS_MALFORMED: + with self.subTest(line=line): + secrets = TLS_LINE + line + TLS_LINE2 + octets = _dsb(TLS_KEY_LOG, len(secrets), secrets) + frames, messages = self._extract(octets) + self.assertEqual(frames, [FRAME, FRAME]) + self.assertKeyLogLine(messages, 'TLS', line) + self.assertEqual(self._tls_entries(octets), + [('CLIENT_RANDOM', b'\xab' * 32, b'\xcd' * 48), + ('SERVER_HANDSHAKE_TRAFFIC_SECRET', b'\x12' * 32, b'\x34' * 32)]) + self.assertRebuildsQuietly(octets) + + def test_cut_last_line_inside_the_padding(self) -> None: + """A last line cut inside the block's padding is no overrun, and is reported as malformed.""" + secrets = TLS_LINE + b'CLIENT_RA' + # NOTE: one octet past the secrets, which the block's padding still holds. + octets = _dsb(TLS_KEY_LOG, len(secrets) + 1, secrets) + frames, messages = self._extract(octets) + self.assertEqual(frames, [FRAME, FRAME]) + self.assertFalse(any('runs past the block' in text for text in messages), messages) + self.assertKeyLogLine(messages, 'TLS', b'CLIENT_RA\x00') + self.assertEqual(self._tls_entries(octets), [('CLIENT_RANDOM', b'\xab' * 32, b'\xcd' * 48)]) + self.assertEqual(self._parse_quietly(octets).info.secrets_length, len(secrets) + 1) # type: ignore[attr-defined] + self.assertRebuildsQuietly(octets) + + def test_malformed_wireguard_line(self) -> None: + from pcapkit.protocols.misc.pcapng import WireGuardKeyLabel + + for line in (b'BOGUS = AAAA\n', b'LOCAL_STATIC_PRIVATE_KEY AAAA\n', b'LOCAL_STATIC_PRIVATE_KEY = Q\n'): + for secrets in (line + WG_LINE, WG_LINE + line + WG_LINE): + with self.subTest(secrets=secrets): + octets = _dsb(WIREGUARD_KEY_LOG, len(secrets), secrets) + frames, messages = self._extract(octets) + self.assertEqual(frames, [FRAME, FRAME]) + self.assertKeyLogLine(messages, 'WireGuard', line) + entries = self._parse_quietly(octets).info.secrets_data.entries # type: ignore[attr-defined] + self.assertEqual(list(entries.items(multi=True)), + [(WireGuardKeyLabel.LOCAL_STATIC_PRIVATE_KEY, WG_KEY)] + * secrets.count(WG_LINE)) + self.assertRebuildsQuietly(octets) + + def test_well_formed_logs_are_unchanged(self) -> None: + cases = ( + (TLS_KEY_LOG, TLS_LINE + b'# a comment\n\n' + TLS_LINE2), + (TLS_KEY_LOG, TLS_LINE), + (WIREGUARD_KEY_LOG, b'# a comment\n' + WG_LINE), + ) + for secrets_type, secrets in cases: + with self.subTest(secrets_type=hex(secrets_type), secrets=secrets): + octets = _dsb(secrets_type, len(secrets), secrets) + frames, messages = self._extract(octets) + self.assertEqual(frames, [FRAME, FRAME]) + self.assertFalse(any('key log format' in text for text in messages), messages) + self.assertRebuildsQuietly(octets) + self.assertEqual(self._tls_entries(_dsb(TLS_KEY_LOG, len(cases[0][1]), cases[0][1])), + [('CLIENT_RANDOM', b'\xab' * 32, b'\xcd' * 48), + ('SERVER_HANDSHAKE_TRAFFIC_SECRET', b'\x12' * 32, b'\x34' * 32)]) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/misc/test_pcapng_unit.py b/tests/protocols/misc/test_pcapng_unit.py index 23f0a920c5..021f32eb1f 100644 --- a/tests/protocols/misc/test_pcapng_unit.py +++ b/tests/protocols/misc/test_pcapng_unit.py @@ -1447,7 +1447,8 @@ def test_pcapng_schema_helpers_and_post_process_branches(self) -> None: from pcapkit.corekit.fields.numbers import UInt16Field, UInt32Field from pcapkit.protocols.misc.pcapng import WireGuardKeyLabel from pcapkit.protocols.schema.misc import pcapng as schema_pcapng - from pcapkit.utilities.exceptions import FieldValueError, ProtocolError + from pcapkit.utilities.exceptions import ProtocolError + from pcapkit.utilities.warnings import ProtocolWarning field = UInt16Field() schema_pcapng.byteorder_callback(field, {'__packet__': {'byteorder': 'little'}}) @@ -1622,8 +1623,10 @@ class LocalIterableOption(schema_pcapng.Option, wg = schema_pcapng.WireGuardKeyLog(data='# comment\n\nPRESHARED_KEY = a2V5') wg.post_process({}) self.assertEqual(wg.entries.getlist(WireGuardKeyLabel.PRESHARED_KEY), [b'key']) - with self.assertRaises(FieldValueError): - schema_pcapng.WireGuardKeyLog(data='PRESHARED_KEY != a2V5').post_process({}) + with self.assertWarns(ProtocolWarning): + wg = schema_pcapng.WireGuardKeyLog(data='PRESHARED_KEY != a2V5') + wg.post_process({}) + self.assertEqual(len(wg.entries), 0) def test_pcapng_block_readers_cover_container_blocks(self) -> None: from pcapkit.const.pcapng.block_type import BlockType From 13d72b971c77f5f9d9f9abb0811c0c5244931447 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Tue, 20 Oct 2026 13:39:58 -0400 Subject: [PATCH 2/3] fix(pcapng): keep a malformed key-log line out of entries entirely (#1985) TLSKeyLog.post_process indexed entries[label] before bytes.fromhex ran, so a line with a known label but fields that do not decode left an empty bucket for that label. Both values are decoded first and entries is touched only once they are. WireGuardKeyLog decodes before it adds, so it needs no change. Also drops timed wording from the test docstrings and rewraps the post_process docstring. Refs #1985 --- pcapkit/protocols/schema/misc/pcapng.py | 11 +++-- .../misc/test_pcapng_roundtrip2_unit.py | 2 +- ...t_pcapng_tls_keylog_malformed_1985_unit.py | 46 +++++++++++++++---- 3 files changed, 44 insertions(+), 15 deletions(-) diff --git a/pcapkit/protocols/schema/misc/pcapng.py b/pcapkit/protocols/schema/misc/pcapng.py index 669b8bd6d7..3c7358fda3 100644 --- a/pcapkit/protocols/schema/misc/pcapng.py +++ b/pcapkit/protocols/schema/misc/pcapng.py @@ -2309,8 +2309,8 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': wherever it sits in the log (:issue:`1985`). Refusing it would abort the whole extraction rather than this one line, as :func:`nonnegative` explains, and :attr:`data` keeps the log text verbatim, so the line is - still kept as captured and the block rebuilds octet for octet. A last line the - capture or the block cut is left out silently (see + still kept as captured and the block rebuilds octet for octet. A last + line the capture or the block cut is left out silently (see :func:`_cut_last_line`), since the cut is reported already. """ @@ -2325,13 +2325,16 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': try: label, random, secret = line.strip().split() label_enum = Enum_TLSKeyLabel(label.upper()) - entries[label_enum].add(bytes.fromhex(random), - bytes.fromhex(secret)) + # NOTE: both values are decoded before ``entries`` is touched, so + # a line that does not parse leaves no empty bucket behind (#1985). + random_octets, secret_octets = bytes.fromhex(random), bytes.fromhex(secret) except ValueError: if _cut_last_line(self, index, lines, packet): break warn(f'PCAP-NG: invalid TLS key log format: {line!r}; line kept as captured', ProtocolWarning, stacklevel=stacklevel()) + continue + entries[label_enum].add(random_octets, secret_octets) self.entries = entries return self diff --git a/tests/protocols/misc/test_pcapng_roundtrip2_unit.py b/tests/protocols/misc/test_pcapng_roundtrip2_unit.py index 1ef735f347..209d3e4eae 100644 --- a/tests/protocols/misc/test_pcapng_roundtrip2_unit.py +++ b/tests/protocols/misc/test_pcapng_roundtrip2_unit.py @@ -296,7 +296,7 @@ def test_repeated_epb_verdict_rebuilds(self) -> None: self.assertRebuilds(_epb_with_options(verdict + verdict)) def test_wireguard_error_names_the_line(self) -> None: - """#1286: the report carries the offending line, which no longer aborts the parse (#1985).""" + """#1286: the report carries the offending line; #1985: it warns rather than aborting the parse.""" import warnings from pcapkit.utilities.warnings import ProtocolWarning diff --git a/tests/protocols/misc/test_pcapng_tls_keylog_malformed_1985_unit.py b/tests/protocols/misc/test_pcapng_tls_keylog_malformed_1985_unit.py index 9cd4ec9ca1..3dc9837424 100644 --- a/tests/protocols/misc/test_pcapng_tls_keylog_malformed_1985_unit.py +++ b/tests/protocols/misc/test_pcapng_tls_keylog_malformed_1985_unit.py @@ -1,14 +1,14 @@ # -*- coding: utf-8 -*- -"""GitHub issue #1985: a malformed key log line no longer aborts the extraction. - -A well-sized Decryption Secrets Block whose TLS key log held a line that did not -parse -- an unknown label, fields that are not hexadecimal, a garbage line, or -the first octets of an entry cut inside the block's padding -- raised -:exc:`~pcapkit.utilities.exceptions.FieldValueError` from -:meth:`TLSKeyLog.post_process `, -which ended the whole extraction. The WireGuard key log did the same. Such a line -now raises a :class:`~pcapkit.utilities.warnings.ProtocolWarning` and is kept as -captured in the log text, the well-formed entries around it are kept, and the +"""GitHub issue #1985: a malformed key log line is reported, not refused. + +A line of a TLS or WireGuard key log that does not parse -- an unknown label, +fields that are not hexadecimal, a garbage line, or the first octets of an entry +cut inside the block's padding -- raises a +:class:`~pcapkit.utilities.warnings.ProtocolWarning` from +:meth:`TLSKeyLog.post_process ` +rather than a :exc:`~pcapkit.utilities.exceptions.FieldValueError`, which would +end the whole extraction. The line is kept as captured in the log text and leaves +no trace in the entries, the well-formed entries around it are kept, and the block rebuilds octet for octet. :mod:`pcapkit` is imported inside each test, after @@ -144,6 +144,32 @@ def test_malformed_wireguard_line(self) -> None: * secrets.count(WG_LINE)) self.assertRebuildsQuietly(octets) + def test_malformed_line_leaves_no_empty_bucket(self) -> None: + """A line with a known label but fields that do not decode adds no bucket for that label.""" + from pcapkit.const.pcapng.tls_key_label import TLSKeyLabel + from pcapkit.protocols.schema.misc.pcapng import TLSKeyLog, WireGuardKeyLog + + for secrets in (b'CLIENT_RANDOM zz yy\n', b'CLIENT_RANDOM zz yy\n' + TLS_LINE2, + b'CLIENT_RANDOM ' + b'ab' * 32 + b' zz\n'): + with self.subTest(secrets=secrets): + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + schema = TLSKeyLog.unpack(secrets, len(secrets)) + self.assertNotIn(TLSKeyLabel.CLIENT_RANDOM, schema.entries) + self.assertTrue(all(len(entry) for entry in schema.entries.values()), schema.entries) + + octets = _dsb(TLS_KEY_LOG, len(secrets), secrets) + entries = self._parse_quietly(octets).info.secrets_data.entries # type: ignore[attr-defined] + self.assertNotIn(TLSKeyLabel.CLIENT_RANDOM, entries) + self.assertTrue(all(len(entry) for entry in entries.values()), entries) + self.assertRebuildsQuietly(octets) + + secrets = b'LOCAL_STATIC_PRIVATE_KEY = Q\n' + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + schema = WireGuardKeyLog.unpack(secrets, len(secrets)) + self.assertEqual(len(schema.entries), 0) + def test_well_formed_logs_are_unchanged(self) -> None: cases = ( (TLS_KEY_LOG, TLS_LINE + b'# a comment\n\n' + TLS_LINE2), From 1476720ea0cc6547c218c37708c797c3599a3d6a Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Tue, 20 Oct 2026 13:52:17 -0400 Subject: [PATCH 3/3] fix(pcapng): elide key-log values from the malformed-line warning (#1985) The malformed-line warning quoted the whole key-log line, which carries key material. _malformed_line_message names only the label, when the key log knows it, and elides the values. An unrecognised label is not quoted either, since it may be key material too. The WireGuard parser's internal ValueError no longer carries the line. Tests assert the new text and that no warning quotes a hex secret or base64 key from their input. Refs #1985 --- pcapkit/protocols/schema/misc/pcapng.py | 43 ++++++++++-- .../misc/test_misc_short_read_1470_runtime.py | 10 +-- .../misc/test_pcapng_roundtrip2_unit.py | 6 +- ...t_pcapng_tls_keylog_malformed_1985_unit.py | 67 +++++++++++++------ tests/protocols/misc/test_pcapng_unit.py | 5 +- 5 files changed, 98 insertions(+), 33 deletions(-) diff --git a/pcapkit/protocols/schema/misc/pcapng.py b/pcapkit/protocols/schema/misc/pcapng.py index 3c7358fda3..2537cf2eb9 100644 --- a/pcapkit/protocols/schema/misc/pcapng.py +++ b/pcapkit/protocols/schema/misc/pcapng.py @@ -627,6 +627,34 @@ def _cut_last_line(schema: 'Schema', index: 'int', lines: 'list[str]', return '__short_read__' in schema.__dict__ or (packet is not None and _secrets_cut(packet)) +def _malformed_line_message(kind: 'str', labels: 'Callable[[str], Any]', line: 'str') -> 'str': + """Report a key log line that does not parse, without quoting what it holds. + + Args: + kind: Name of the key log, e.g. ``'TLS'``. + labels: Enumeration of the key log's labels, called on the upper-cased + first field of the line. + line: The line that does not parse. + + Returns: + A warning message naming the line's label, when that is one ``labels`` + knows, and eliding its values. The line itself is never quoted, since + it carries key material (:issue:`1985`); nor is an unrecognised label, + which may be key material too. + + """ + fields = line.split() + try: + name = labels(fields[0].upper()).name if fields else None + except ValueError: + name = None + if name is None: + return (f'PCAP-NG: invalid {kind} key log line with an unrecognised label; ' + 'line elided, kept as captured') + return (f'PCAP-NG: invalid {kind} key log line for label {name!r}; ' + 'values elided, line kept as captured') + + def dsb_secrets_selector(packet: 'dict[str, Any]') -> 'Field': """Selector function for :attr:`DecryptionSecretsBlock.secrets_data` field. @@ -2305,10 +2333,11 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': A line that does not parse -- an unknown label, a field that is not hexadecimal, a wrong number of fields, or the first octets of an entry cut inside the block's padding -- is left out of :attr:`entries` and - reported as a :class:`~pcapkit.utilities.warnings.ProtocolWarning`, - wherever it sits in the log (:issue:`1985`). Refusing it would abort - the whole extraction rather than this one line, as :func:`nonnegative` - explains, and :attr:`data` keeps the log text verbatim, so the line is + reported as a :class:`~pcapkit.utilities.warnings.ProtocolWarning` + that names its label and elides its values (see + :func:`_malformed_line_message`), wherever it sits in the log + (:issue:`1985`). Refusing it would abort the whole extraction rather + than this one line, as :func:`nonnegative` explains, and :attr:`data` keeps the log text verbatim, so the line is still kept as captured and the block rebuilds octet for octet. A last line the capture or the block cut is left out silently (see :func:`_cut_last_line`), since the cut is reported already. @@ -2331,7 +2360,7 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': except ValueError: if _cut_last_line(self, index, lines, packet): break - warn(f'PCAP-NG: invalid TLS key log format: {line!r}; line kept as captured', + warn(_malformed_line_message('TLS', Enum_TLSKeyLabel, line), ProtocolWarning, stacklevel=stacklevel()) continue entries[label_enum].add(random_octets, secret_octets) @@ -2378,14 +2407,14 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': fields = line.strip().split() try: if len(fields) != 3 or fields[1] != '=': - raise ValueError(line) + raise ValueError('not a WireGuard key log line') label, _, secret = fields label_enum = WireGuardKeyLabel(label.upper()) entries.add(label_enum, base64.b64decode(secret)) except ValueError: if _cut_last_line(self, index, lines, packet): break - warn(f'PCAP-NG: invalid WireGuard key log format: {line!r}; line kept as captured', + warn(_malformed_line_message('WireGuard', WireGuardKeyLabel, line), ProtocolWarning, stacklevel=stacklevel()) self.entries = entries diff --git a/tests/protocols/misc/test_misc_short_read_1470_runtime.py b/tests/protocols/misc/test_misc_short_read_1470_runtime.py index 54b592120d..a15bd88eaa 100644 --- a/tests/protocols/misc/test_misc_short_read_1470_runtime.py +++ b/tests/protocols/misc/test_misc_short_read_1470_runtime.py @@ -219,15 +219,17 @@ def test_malformed_line_of_a_whole_log_is_reported(self) -> None: module = importlib.import_module('pcapkit.protocols.schema.misc.pcapng') for schema, text, message in ( ('TLSKeyLog', b'CLIENT_RANDOM abcd\n', - "PCAP-NG: invalid TLS key log format: 'CLIENT_RANDOM abcd'; line kept as captured"), - ('WireGuardKeyLog', b'LOCAL_STATIC_PRIVATE_KEY = Q\n', - "PCAP-NG: invalid WireGuard key log format: 'LOCAL_STATIC_PRIVATE_KEY = Q'; " - 'line kept as captured'), + "PCAP-NG: invalid TLS key log line for label 'CLIENT_RANDOM'; " + 'values elided, line kept as captured'), + ('WireGuardKeyLog', b'LOCAL_STATIC_PRIVATE_KEY = QUFBQ\n', + "PCAP-NG: invalid WireGuard key log line for label 'LOCAL_STATIC_PRIVATE_KEY'; " + 'values elided, line kept as captured'), ): with self.subTest(schema=schema): with self.assertWarns(ProtocolWarning) as caught: parsed = getattr(module, schema).unpack(text, len(text)) self.assertEqual(str(caught.warning), message) + self.assertNotIn(text.split()[-1].decode(), message) self.assertEqual(len(parsed.entries), 0) self.assertEqual(parsed.data, text.decode()) diff --git a/tests/protocols/misc/test_pcapng_roundtrip2_unit.py b/tests/protocols/misc/test_pcapng_roundtrip2_unit.py index 209d3e4eae..8ac45f378d 100644 --- a/tests/protocols/misc/test_pcapng_roundtrip2_unit.py +++ b/tests/protocols/misc/test_pcapng_roundtrip2_unit.py @@ -296,7 +296,7 @@ def test_repeated_epb_verdict_rebuilds(self) -> None: self.assertRebuilds(_epb_with_options(verdict + verdict)) def test_wireguard_error_names_the_line(self) -> None: - """#1286: the report carries the offending line; #1985: it warns rather than aborting the parse.""" + """#1286, #1985: a malformed line warns, eliding what it holds, rather than aborting the parse.""" import warnings from pcapkit.utilities.warnings import ProtocolWarning @@ -309,8 +309,10 @@ def test_wireguard_error_names_the_line(self) -> None: )) with warnings.catch_warnings(): warnings.simplefilter('ignore') - with self.assertWarnsRegex(ProtocolWarning, r"format: 'a b c'; line kept as captured"): + with self.assertWarns(ProtocolWarning) as caught: parsed = self._parse(octets) + self.assertEqual(str(caught.warning), 'PCAP-NG: invalid WireGuard key log line with an ' + 'unrecognised label; line elided, kept as captured') self.assertEqual(len(parsed.info.secrets_data.entries), 0) self.assertRebuilds(octets) diff --git a/tests/protocols/misc/test_pcapng_tls_keylog_malformed_1985_unit.py b/tests/protocols/misc/test_pcapng_tls_keylog_malformed_1985_unit.py index 3dc9837424..8064ca99b2 100644 --- a/tests/protocols/misc/test_pcapng_tls_keylog_malformed_1985_unit.py +++ b/tests/protocols/misc/test_pcapng_tls_keylog_malformed_1985_unit.py @@ -38,15 +38,30 @@ WG_KEY = bytes(range(32)) WG_LINE = b'LOCAL_STATIC_PRIVATE_KEY = ' + base64.b64encode(WG_KEY) + b'\n' -#: Malformed TLS key log lines, each followed by a newline. +#: Malformed TLS key log lines, each followed by a newline, and the label the +#: warning names for each, if any. TLS_MALFORMED = ( - b'NOT_A_LABEL aa bb\n', - b'CLIENT_RANDOM zz yy\n', - b'garbage\n', - b'CLIENT_RANDOM ' + b'ab' * 32 + b'\n', - b'\xff\xfe junk\n', + (b'NOT_A_LABEL aa bb\n', None), + (b'CLIENT_RANDOM zz yy\n', 'CLIENT_RANDOM'), + (b'garbage\n', None), + (b'CLIENT_RANDOM ' + b'ab' * 32 + b'\n', 'CLIENT_RANDOM'), + (b'CLIENT_RANDOM zz ' + b'cd' * 48 + b'\n', 'CLIENT_RANDOM'), + (b'client_random ' + b'ab' * 32 + b' ' + b'cd' * 48 + b' extra\n', 'CLIENT_RANDOM'), + (b'\xff\xfe junk\n', None), ) +#: Malformed WireGuard key log lines, and the label the warning names for each, if any. +WG_MALFORMED = ( + (b'BOGUS = AAAA\n', None), + (b'LOCAL_STATIC_PRIVATE_KEY AAAA\n', 'LOCAL_STATIC_PRIVATE_KEY'), + (b'LOCAL_STATIC_PRIVATE_KEY = Q\n', 'LOCAL_STATIC_PRIVATE_KEY'), + (b'LOCAL_STATIC_PRIVATE_KEY == ' + base64.b64encode(WG_KEY) + b'\n', 'LOCAL_STATIC_PRIVATE_KEY'), + (b'NOT_A_KEY = ' + base64.b64encode(WG_KEY) + b'\n', None), +) + +#: Key material the test inputs carry, which no warning may quote. +KEY_MATERIAL = ('ab' * 32, 'cd' * 48, '12' * 32, '34' * 32, base64.b64encode(WG_KEY).decode()) + class TestPCAPNGKeyLogMalformedLine(PCAPNGTestCase): """Pin how a key log line that does not parse is treated.""" @@ -78,11 +93,25 @@ def assertRebuildsQuietly(self, octets: 'bytes') -> None: # pylint: disable=inv self.assertEqual(self._parse(octets).__header__.pack().hex(), octets.hex()) # type: ignore[attr-defined] self.assertRebuilds(octets) - def assertKeyLogLine(self, messages: 'list[str]', kind: 'str', line: 'bytes') -> None: # pylint: disable=invalid-name - """Assert ``messages`` report ``line`` of a ``kind`` key log kept as captured.""" - text = line.rstrip(b'\n').decode('ascii', errors='replace') - self.assertIn(f'PCAP-NG: invalid {kind} key log format: {text!r}; line kept as captured', - messages) + def assertKeyLogLine(self, messages: 'list[str]', kind: 'str', # pylint: disable=invalid-name + label: 'str | None', line: 'bytes') -> None: + """Assert ``messages`` report ``line`` of a ``kind`` key log by ``label`` alone.""" + if label is None: + expected = (f'PCAP-NG: invalid {kind} key log line with an unrecognised label; ' + 'line elided, kept as captured') + else: + expected = (f'PCAP-NG: invalid {kind} key log line for label {label!r}; ' + 'values elided, line kept as captured') + self.assertIn(expected, messages) + # NOTE: neither the line's values, nor a label the key log does not + # know, nor any key material of the log is quoted. + fields = line.decode('ascii', errors='replace').split() + if label is not None: + fields = fields[1:] + for text in messages: + for value in (*fields, *KEY_MATERIAL): + if len(value) >= 4: + self.assertNotIn(value, text) def _tls_entries(self, octets: 'bytes') -> 'list[tuple[str, bytes, bytes]]': secrets = self._parse_quietly(octets).info.secrets_data # type: ignore[attr-defined] @@ -91,25 +120,25 @@ def _tls_entries(self, octets: 'bytes') -> 'list[tuple[str, bytes, bytes]]': for random, secret in entry.items(multi=True)] def test_malformed_first_line(self) -> None: - for line in TLS_MALFORMED: + for line, label in TLS_MALFORMED: with self.subTest(line=line): secrets = line + TLS_LINE octets = _dsb(TLS_KEY_LOG, len(secrets), secrets) frames, messages = self._extract(octets) self.assertEqual(frames, [FRAME, FRAME]) - self.assertKeyLogLine(messages, 'TLS', line) + self.assertKeyLogLine(messages, 'TLS', label, line) self.assertEqual(self._tls_entries(octets), [('CLIENT_RANDOM', b'\xab' * 32, b'\xcd' * 48)]) self.assertRebuildsQuietly(octets) def test_malformed_middle_line(self) -> None: - for line in TLS_MALFORMED: + for line, label in TLS_MALFORMED: with self.subTest(line=line): secrets = TLS_LINE + line + TLS_LINE2 octets = _dsb(TLS_KEY_LOG, len(secrets), secrets) frames, messages = self._extract(octets) self.assertEqual(frames, [FRAME, FRAME]) - self.assertKeyLogLine(messages, 'TLS', line) + self.assertKeyLogLine(messages, 'TLS', label, line) self.assertEqual(self._tls_entries(octets), [('CLIENT_RANDOM', b'\xab' * 32, b'\xcd' * 48), ('SERVER_HANDSHAKE_TRAFFIC_SECRET', b'\x12' * 32, b'\x34' * 32)]) @@ -123,7 +152,7 @@ def test_cut_last_line_inside_the_padding(self) -> None: frames, messages = self._extract(octets) self.assertEqual(frames, [FRAME, FRAME]) self.assertFalse(any('runs past the block' in text for text in messages), messages) - self.assertKeyLogLine(messages, 'TLS', b'CLIENT_RA\x00') + self.assertKeyLogLine(messages, 'TLS', None, b'CLIENT_RA\x00') self.assertEqual(self._tls_entries(octets), [('CLIENT_RANDOM', b'\xab' * 32, b'\xcd' * 48)]) self.assertEqual(self._parse_quietly(octets).info.secrets_length, len(secrets) + 1) # type: ignore[attr-defined] self.assertRebuildsQuietly(octets) @@ -131,13 +160,13 @@ def test_cut_last_line_inside_the_padding(self) -> None: def test_malformed_wireguard_line(self) -> None: from pcapkit.protocols.misc.pcapng import WireGuardKeyLabel - for line in (b'BOGUS = AAAA\n', b'LOCAL_STATIC_PRIVATE_KEY AAAA\n', b'LOCAL_STATIC_PRIVATE_KEY = Q\n'): + for line, label in WG_MALFORMED: for secrets in (line + WG_LINE, WG_LINE + line + WG_LINE): with self.subTest(secrets=secrets): octets = _dsb(WIREGUARD_KEY_LOG, len(secrets), secrets) frames, messages = self._extract(octets) self.assertEqual(frames, [FRAME, FRAME]) - self.assertKeyLogLine(messages, 'WireGuard', line) + self.assertKeyLogLine(messages, 'WireGuard', label, line) entries = self._parse_quietly(octets).info.secrets_data.entries # type: ignore[attr-defined] self.assertEqual(list(entries.items(multi=True)), [(WireGuardKeyLabel.LOCAL_STATIC_PRIVATE_KEY, WG_KEY)] @@ -181,7 +210,7 @@ def test_well_formed_logs_are_unchanged(self) -> None: octets = _dsb(secrets_type, len(secrets), secrets) frames, messages = self._extract(octets) self.assertEqual(frames, [FRAME, FRAME]) - self.assertFalse(any('key log format' in text for text in messages), messages) + self.assertFalse(any('key log line' in text for text in messages), messages) self.assertRebuildsQuietly(octets) self.assertEqual(self._tls_entries(_dsb(TLS_KEY_LOG, len(cases[0][1]), cases[0][1])), [('CLIENT_RANDOM', b'\xab' * 32, b'\xcd' * 48), diff --git a/tests/protocols/misc/test_pcapng_unit.py b/tests/protocols/misc/test_pcapng_unit.py index 021f32eb1f..32ee92f4c0 100644 --- a/tests/protocols/misc/test_pcapng_unit.py +++ b/tests/protocols/misc/test_pcapng_unit.py @@ -1623,10 +1623,13 @@ class LocalIterableOption(schema_pcapng.Option, wg = schema_pcapng.WireGuardKeyLog(data='# comment\n\nPRESHARED_KEY = a2V5') wg.post_process({}) self.assertEqual(wg.entries.getlist(WireGuardKeyLabel.PRESHARED_KEY), [b'key']) - with self.assertWarns(ProtocolWarning): + with self.assertWarns(ProtocolWarning) as caught: wg = schema_pcapng.WireGuardKeyLog(data='PRESHARED_KEY != a2V5') wg.post_process({}) self.assertEqual(len(wg.entries), 0) + self.assertEqual(str(caught.warning), "PCAP-NG: invalid WireGuard key log line for label " + "'PRESHARED_KEY'; values elided, line kept as captured") + self.assertNotIn('a2V5', str(caught.warning)) def test_pcapng_block_readers_cover_container_blocks(self) -> None: from pcapkit.const.pcapng.block_type import BlockType