diff --git a/pcapkit/protocols/schema/misc/pcapng.py b/pcapkit/protocols/schema/misc/pcapng.py index ff61179147..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. @@ -2302,6 +2330,18 @@ 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` + 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. + """ from pcapkit.const.pcapng.tls_key_label import TLSKeyLabel as Enum_TLSKeyLabel @@ -2314,12 +2354,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)) - except ValueError as error: + # 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 - raise FieldValueError(f'invalid TLS key log format: {line!r}') from error + warn(_malformed_line_message('TLS', Enum_TLSKeyLabel, line), + ProtocolWarning, stacklevel=stacklevel()) + continue + entries[label_enum].add(random_octets, secret_octets) self.entries = entries return self @@ -2347,6 +2391,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 +2407,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('not a WireGuard key log 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(_malformed_line_message('WireGuard', WireGuardKeyLabel, line), + 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..a15bd88eaa 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,29 @@ 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 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()) 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..8ac45f378d 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, #1985: a malformed line warns, eliding what it holds, rather than aborting the parse.""" + import warnings + + from pcapkit.utilities.warnings import ProtocolWarning body = b'a b c\n\x00\x00' octets = b''.join(( @@ -305,8 +307,14 @@ 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.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) 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..8064ca99b2 --- /dev/null +++ b/tests/protocols/misc/test_pcapng_tls_keylog_malformed_1985_unit.py @@ -0,0 +1,221 @@ +# -*- coding: utf-8 -*- +"""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 +: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, and the label the +#: warning names for each, if any. +TLS_MALFORMED = ( + (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.""" + + 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', # 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] + 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, 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', 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, 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', 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)]) + 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', 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) + + def test_malformed_wireguard_line(self) -> None: + from pcapkit.protocols.misc.pcapng import WireGuardKeyLabel + + 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', 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)] + * 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), + (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 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), + ('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..32ee92f4c0 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,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.assertRaises(FieldValueError): - schema_pcapng.WireGuardKeyLog(data='PRESHARED_KEY != a2V5').post_process({}) + 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