diff --git a/docs/source/pcapkit/protocols/misc/pcapng.rst b/docs/source/pcapkit/protocols/misc/pcapng.rst index 3a371fc82b..39ffaf1ae9 100644 --- a/docs/source/pcapkit/protocols/misc/pcapng.rst +++ b/docs/source/pcapkit/protocols/misc/pcapng.rst @@ -580,6 +580,7 @@ shared with the PCAP frame header and documented in .. autofunction:: pcapkit.protocols.schema.misc.pcapng.bounded_option .. autofunction:: pcapkit.protocols.schema.misc.pcapng.bounded_area .. autofunction:: pcapkit.protocols.schema.misc.pcapng.captured_area +.. autofunction:: pcapkit.protocols.schema.misc.pcapng.secrets_area .. autofunction:: pcapkit.protocols.schema.misc.pcapng._spb_data_length Data Models diff --git a/pcapkit/protocols/misc/pcapng.py b/pcapkit/protocols/misc/pcapng.py index 3cd4e5144f..7cbada2a7b 100644 --- a/pcapkit/protocols/misc/pcapng.py +++ b/pcapkit/protocols/misc/pcapng.py @@ -4733,9 +4733,11 @@ def _make_block_dsb(self, block: 'Optional[Data_DecryptionSecretsBlock]' = None, Constructed block schema. """ + declared = None # type: Optional[int] if block is not None: secrets_type_val = block.secrets_type secrets_data = block.secrets_data + declared = block.secrets_length options = self._block_area(block, 'options') else: secrets_type_val = self._make_index(secrets_type, secrets_type_default, namespace=secrets_type_namespace, # type: ignore[assignment] @@ -4762,8 +4764,29 @@ def _make_block_dsb(self, block: 'Optional[Data_DecryptionSecretsBlock]' = None, raise ProtocolError(f'PCAP-NG: [DSB] secrets {secrets_type_val} invalid format') # NOTE: Secrets Length excludes the padding to the next 32-bit boundary, # which ``padding_data`` writes and the block length has to include. - secrets_length = len(secrets_data_val) - secrets_padded = secrets_length + -secrets_length % 4 + if block is not None and isinstance(secrets_data_val, Schema): + # NOTE: Packed in the context of the block it was parsed from, so + # that a key log the block cut is still read as cut (#1980). + secrets_held = len(secrets_data_val.pack({'__packet__': { + 'length': block.length, 'secrets_length': block.secrets_length, + }})) + else: + secrets_held = len(secrets_data_val) + secrets_padded = secrets_held + -secrets_held % 4 + # NOTE: A Secrets Length running past the block is kept as declared, as + # an EPB keeps its captured length, so the block rebuilds as captured; + # the block length is still sized from the secrets written (#1980). + # Only a true overrun is kept -- the one the parse side bounds and + # warns on, measured by ``secrets_area`` -- so a length that fits the + # block, or one an edit has outdated, follows the secrets written. + from pcapkit.protocols.schema.misc.pcapng import secrets_area + + if block is not None and declared is not None and declared > secrets_area({ + 'length': block.length, 'secrets_length': declared, + }): + secrets_length = declared + else: + secrets_length = secrets_held if options is not None: options_value, total_length = self._make_pcapng_options(options, namespace='dsb') diff --git a/pcapkit/protocols/schema/misc/pcapng.py b/pcapkit/protocols/schema/misc/pcapng.py index 2e8d235e2f..ff61179147 100644 --- a/pcapkit/protocols/schema/misc/pcapng.py +++ b/pcapkit/protocols/schema/misc/pcapng.py @@ -498,10 +498,73 @@ def captured_area(name: 'str') -> 'Callable[[dict[str, Any]], int]': framing = BLOCK_MINIMUM[Enum_BlockType.Enhanced_Packet_Block] def callback(pkt: 'dict[str, Any]') -> 'int': - return max(0, min(pkt[name], pkt['length'] - framing)) + return _held_octets(pkt[name], pkt['length'], framing) return callback +def _held_octets(declared: 'int', length: 'int', framing: 'int') -> 'int': + """Bound a declared length to the octets its block holds. + + Args: + declared: Length a field of the block declares, e.g. ``captured_len``. + length: The block's own Block Total Length. + framing: Octets the block's fixed fields span, both copies of the Block + Total Length included. + + Returns: + ``declared``, never past ``length`` less ``framing``, and never below + zero. Shared by :func:`captured_area` and :func:`secrets_area`. + + """ + return max(0, min(declared, length - framing)) + + +def secrets_area(packet: 'dict[str, Any]') -> 'int': + """Bound a Decryption Secrets Block's Secrets Data to the octets the block holds. + + Args: + packet: Unpacked data so far, with ``length`` and ``secrets_length``. + + Returns: + The declared ``secrets_length``, never past the block's Block Total + Length less its 20 framing octets, and never below zero. + + ``secrets_length`` is a 32-bit wire field. Read as declared, a length past + the block read the trailing Block Total Length and the blocks after it as + secrets, and past :data:`~pcapkit.utilities.config.MAX_ZERO_PAD_LENGTH` the + read raised :exc:`~pcapkit.utilities.exceptions.FieldValueError` from inside + the block's schema, which ended the whole extraction (:issue:`1980`). This + is the Secrets Data counterpart of :func:`captured_area`: the block keeps its + declared ``secrets_length``, the secrets are the octets the block holds, and + the padding and option area are sized from those octets. + :meth:`DecryptionSecretsBlock.post_process` reports the overrun. + + """ + return _held_octets(packet['secrets_length'], packet['length'], + BLOCK_MINIMUM[Enum_BlockType.Decryption_Secrets_Block]) + + +def _secrets_cut(packet: 'dict[str, Any]') -> 'bool': + """Tell whether a key log's enclosing block holds less than it declares. + + Args: + packet: Unpacked data of the key log secrets schema, whose ``__packet__`` + is the enclosing Decryption Secrets Block's. + + Returns: + Whether that block's ``secrets_length`` runs past the block, so that + :func:`secrets_area` cut the key log at the block's end (:issue:`1980`). + + """ + block = packet.get('__packet__') + if not isinstance(block, dict): + return False + declared, length = block.get('secrets_length'), block.get('length') + if not isinstance(declared, int) or not isinstance(length, int): + return False + return declared > secrets_area(block) + + def pcapng_block_selector(packet: 'dict[str, Any]') -> 'Field': """Selector function for :attr:`PCAPNG.block` field. @@ -539,23 +602,29 @@ def pcapng_block_selector(packet: 'dict[str, Any]') -> 'Field': return SchemaField(length=max(packet['__length__'], 0), schema=schema) -def _cut_last_line(schema: 'Schema', index: 'int', lines: 'list[str]') -> 'bool': +def _cut_last_line(schema: 'Schema', index: 'int', lines: 'list[str]', + packet: 'Optional[dict[str, Any]]' = None) -> 'bool': """Tell whether a key log line that did not parse is one the capture cut. Args: schema: Key log secrets schema, just unpacked. index: Index of the line in ``lines``. lines: Lines of the key log. + packet: Unpacked data of ``schema``, if any. Returns: Whether ``lines[index]`` is the last line of a key log the data ended inside, which :meth:`Schema.unpack ` records as - ``__short_read__``. Such a line is only the first octets of an entry, - and the log text, kept verbatim, still rebuilds it (:issue:`1470`). + ``__short_read__``, or that the enclosing block ended inside, as + :func:`secrets_area` bounds it (:issue:`1980`). Such a line is only the + first octets of an entry, and the log text, kept verbatim, still + rebuilds it (:issue:`1470`). """ - return index == len(lines) - 1 and '__short_read__' in schema.__dict__ + if index != len(lines) - 1: + return False + return '__short_read__' in schema.__dict__ or (packet is not None and _secrets_cut(packet)) def dsb_secrets_selector(packet: 'dict[str, Any]') -> 'Field': @@ -568,7 +637,9 @@ def dsb_secrets_selector(packet: 'dict[str, Any]') -> 'Field': A :class:`~pcapkit.corekit.fields.misc.SchemaField` wrapped instance of the :class:`~pcapkit.protocols.schema.misc.pcapng.DSBSecrets` subclass registered for ``secrets_type``, or of - :class:`~pcapkit.protocols.schema.misc.pcapng.UnknownSecrets` if none is. + :class:`~pcapkit.protocols.schema.misc.pcapng.UnknownSecrets` if none is, + spanning the octets of ``secrets_length`` that the block holds (see + :func:`secrets_area`). See Also: * :class:`pcapkit.const.pcapng.secrets_type.SecretsType` @@ -577,7 +648,7 @@ def dsb_secrets_selector(packet: 'dict[str, Any]') -> 'Field': """ secrets_type = packet['secrets_type'] # type: int schema = DSBSecrets.registry[secrets_type] - return SchemaField(length=packet['secrets_length'], schema=schema) + return SchemaField(length=secrets_area(packet), schema=schema) class FlagsField(UInt32Field): @@ -2246,7 +2317,7 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': entries[label_enum].add(bytes.fromhex(random), bytes.fromhex(secret)) except ValueError as error: - if _cut_last_line(self, index, lines): + if _cut_last_line(self, index, lines, packet): break raise FieldValueError(f'invalid TLS key log format: {line!r}') from error @@ -2293,7 +2364,7 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': label_enum = WireGuardKeyLabel(label.upper()) entries.add(label_enum, base64.b64decode(secret)) except ValueError as error: - if _cut_last_line(self, index, lines): + if _cut_last_line(self, index, lines, packet): break if isinstance(error, FieldValueError): raise @@ -2365,18 +2436,18 @@ class DecryptionSecretsBlock(BlockType, code=Enum_BlockType.Decryption_Secrets_B secrets_type: 'Enum_SecretsType' = EnumField(length=4, namespace=Enum_SecretsType, callback=byteorder_callback) #: Secrets length. secrets_length: 'int' = UInt32Field(callback=byteorder_callback) - #: Secrets data. + #: Secrets data, bounded by the block (see :func:`secrets_area`). secrets_data: 'DSBSecrets' = SwitchField( selector=dsb_secrets_selector, ) #: Padding. - padding_data: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['secrets_length'] % 4) % 4) + padding_data: 'bytes' = PaddingField(length=lambda pkt: (4 - secrets_area(pkt) % 4) % 4) #: Options. options: 'list[Option]' = OptionAreaField( # NOTE: see EnhancedPacketBlock.options on why the padding is recomputed # here instead of being read back from ``padding_data``. length=nonnegative(lambda pkt: pkt['length'] - BLOCK_MINIMUM[Enum_BlockType.Decryption_Secrets_Block] - - pkt['secrets_length'] - (4 - pkt['secrets_length'] % 4) % 4), + - (secrets_area(pkt) + 3) // 4 * 4), base_schema=_DSB_Option, type_name='type', registry=Option.registry['dsb'], @@ -2388,6 +2459,37 @@ class DecryptionSecretsBlock(BlockType, code=Enum_BlockType.Decryption_Secrets_B #: Block total length. length2: 'int' = UInt32Field(callback=byteorder_callback) + def post_process(self, packet: 'dict[str, Any]') -> 'Schema': + """Revise ``schema`` data after unpacking process. + + Args: + packet: Unpacked data. + + Returns: + Revised schema. + + A ``secrets_length`` running past the block is kept as declared, and + the secrets are the octets the block holds (see :func:`secrets_area`), + so this reports the overrun as a + :class:`~pcapkit.utilities.warnings.ProtocolWarning` and carries on to + the next block (:issue:`1980`). + + The report is made on the unpacking path only, where ``__length__`` is + what is left of the stream. :meth:`Schema.pack + ` seeds it as ``-1``, and + rebuilding a block already reported is no new overrun. The lengths are + read from the schema rather than from ``packet``, since packing the + option area rebinds ``packet['length']`` to each option's own length. + + """ + held = _held_octets(self.secrets_length, self.length, + BLOCK_MINIMUM[Enum_BlockType.Decryption_Secrets_Block]) + if self.secrets_length > held and packet.get('__length__', -1) >= 0: + warn(f'PCAP-NG: [Block {Enum_BlockType.Decryption_Secrets_Block}] secrets length ' + f'{self.secrets_length} runs past the block, which holds {held} octet(s); ' + 'kept as captured', ProtocolWarning, stacklevel=stacklevel()) + return super().post_process(packet) + if TYPE_CHECKING: def __init__(self, length: 'int', secrets_type: 'Enum_SecretsType', secrets_length: 'int', secrets_data: 'DSBSecrets | bytes', diff --git a/tests/protocols/misc/test_pcapng_dsb_secrets_overrun_1980_unit.py b/tests/protocols/misc/test_pcapng_dsb_secrets_overrun_1980_unit.py new file mode 100644 index 0000000000..2f24b4d047 --- /dev/null +++ b/tests/protocols/misc/test_pcapng_dsb_secrets_overrun_1980_unit.py @@ -0,0 +1,179 @@ +# -*- coding: utf-8 -*- +"""GitHub issue #1980: a Decryption Secrets Block's secrets are bounded by the block. + +A Decryption Secrets Block whose ``secrets_length`` ran far past the block read +that many octets as secrets: the trailing Block Total Length and the blocks after +it went with them, and past the zero-pad ceiling the read raised +:exc:`~pcapkit.utilities.exceptions.FieldValueError` from inside the block's +schema, which ended the whole extraction. The secrets are now bounded by the +block, as an Enhanced Packet Block's packet data is (#1405): the declared +``secrets_length`` is kept, a :class:`~pcapkit.utilities.warnings.ProtocolWarning` +reports the overrun, and extraction carries on to the next block. + +:mod:`pcapkit` is imported inside each test, after +:func:`~tests._support.reimport_once_per_class`. + +""" + +import io +import struct +import unittest +import warnings + +from tests.protocols.misc.test_pcapng_roundtrip2_unit import PCAPNGTestCase + +FRAME = bytes(range(1, 61)) + +#: Secrets Type of a TLS Key Log, of a ZigBee NWK Key, and one no registry names. +TLS_KEY_LOG = 0x544C534B +ZIGBEE_NWK_KEY = 0x5A4E574B +UNKNOWN_SECRETS = 0x12345678 + +#: One well-formed TLS key log line. +KEY_LOG = b'CLIENT_RANDOM ' + b'ab' * 32 + b' ' + b'cd' * 48 + b'\n' + + +def _block(btype: 'int', body: 'bytes') -> 'bytes': + """A little-endian block of ``btype`` around ``body``, padded to 32 bits.""" + body += bytes(-len(body) % 4) + total = 12 + len(body) + return struct.pack(' 'bytes': + """An Enhanced Packet Block on interface 0 holding :data:`FRAME`.""" + return _block(6, struct.pack(' 'bytes': + """A Decryption Secrets Block declaring ``secrets_length`` over ``secrets``.""" + return _block(10, struct.pack(' 'tuple[list[bytes], list[str]]': + """Extract a file holding ``block`` between two Enhanced Packet Blocks.""" + import pcapkit + + 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]) + + def test_frames_around_an_overrun_survive(self) -> None: + for secrets_type in (TLS_KEY_LOG, UNKNOWN_SECRETS, ZIGBEE_NWK_KEY): + for secrets_length in (0xFFFF_FFF0, 5): + with self.subTest(secrets_type=hex(secrets_type), secrets_length=secrets_length): + frames, messages = self._extract(_dsb(secrets_type, secrets_length, b'abcd')) + self.assertEqual(frames, [FRAME, FRAME]) + self.assertIn(f'PCAP-NG: [Block 10] secrets length {secrets_length} runs past ' + 'the block, which holds 4 octet(s); kept as captured', messages) + self.assertFalse(any('block length mismatch' in text for text in messages), + messages) + + def test_overrun_keeps_the_declared_length(self) -> None: + octets = _dsb(UNKNOWN_SECRETS, 0xFFFF_FFF0, b'abcd') + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + parsed = self._parse(octets) + self.assertEqual(parsed.info.secrets_length, 0xFFFF_FFF0) # type: ignore[attr-defined] + self.assertEqual(parsed.info.secrets_data.data, b'abcd') # type: ignore[attr-defined] + self.assertEqual(parsed.__header__.pack().hex(), octets.hex()) # type: ignore[attr-defined] + + def test_overrun_rebuilds(self) -> None: + """The data model rebuilds the declared ``secrets_length``, as an EPB does its ``captured_len``.""" + # NOTE: no ZigBee key here: one shorter than its fixed width rebuilds at + # full width whatever its secrets_length says, overrun or not. + for secrets_type in (TLS_KEY_LOG, UNKNOWN_SECRETS): + for secrets_length in (0xFFFF_FFF0, 5): + with self.subTest(secrets_type=hex(secrets_type), secrets_length=secrets_length): + octets = _dsb(secrets_type, secrets_length, b'abcd') + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + self.assertEqual(self._parse(octets).__header__.pack().hex(), octets.hex()) + self.assertRebuilds(octets) + + def test_cut_key_log_keeps_its_entries(self) -> None: + """A key log the block ends inside keeps the lines before the cut one.""" + secrets = KEY_LOG + b'CLIENT_RA' + # NOTE: short of the octets that follow, so the read is not a short one. + octets = _dsb(TLS_KEY_LOG, len(secrets) + 60, secrets) + frames, messages = self._extract(octets) + self.assertEqual(frames, [FRAME, FRAME]) + self.assertTrue(any('runs past the block' in text for text in messages), messages) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + parsed = self._parse(octets) + self.assertEqual(sum(len(entries) for entries in parsed.info.secrets_data.entries.values()), # type: ignore[attr-defined] + 1) + self.assertEqual(parsed.__header__.pack().hex(), octets.hex()) # type: ignore[attr-defined] + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + self.assertRebuilds(octets) + + def assertValidBlock(self, octets: 'bytes', secrets_length: 'int') -> None: # pylint: disable=invalid-name + """Assert ``octets`` is one DSB as long as both its Block Total Lengths say.""" + length, = struct.unpack(' None: + """A length that fits the block is no overrun, so the rebuild keeps its padding.""" + key = bytes(range(16)) + b'\x34\x12' + octets = _dsb(ZIGBEE_NWK_KEY, 20, key) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + rebuilt = self._rebuild(octets) + self.assertValidBlock(rebuilt, 18) + self.assertEqual(rebuilt[16:34], key) + + def test_edited_secrets_take_their_own_length(self) -> None: + from pcapkit.protocols.misc.pcapng import PCAPNG + + info = self._parse(_dsb(UNKNOWN_SECRETS, 5, b'abcde')).info + fields = {key: value for key, value in dict(info).items() if key != 'packet'} + fields['secrets_data'] = info.secrets_data.__class__(data=b'ab') + context = self._context() + rebuilt = PCAPNG.from_data(info.__class__(**fields), num=2, sct=1, ctx=context).data + self.assertValidBlock(rebuilt, 2) + self.assertEqual(rebuilt[16:20], b'ab\x00\x00') + + def test_well_formed_blocks_are_unchanged(self) -> None: + cases = ( + (TLS_KEY_LOG, KEY_LOG), + (UNKNOWN_SECRETS, b'abcde'), + (UNKNOWN_SECRETS, b''), + (ZIGBEE_NWK_KEY, bytes(range(16)) + b'\x34\x12'), + ) + 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('runs past the block' in text for text in messages), messages) + self.assertEqual(self._parse(octets).info.secrets_length, len(secrets)) # type: ignore[attr-defined] + self.assertRebuilds(octets) + + def test_well_formed_block_with_options_is_unchanged(self) -> None: + from pcapkit.const.pcapng.option_type import OptionType + + octets = _dsb(UNKNOWN_SECRETS, 5, b'abcde\x00\x00\x00' + b'\x01\x00\x03\x00abc\x00\x00\x00\x00\x00') + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + parsed = self._parse(octets) + self.assertFalse(any('runs past the block' in str(item.message) for item in caught)) + self.assertEqual(parsed.info.options[OptionType.opt_comment].comment, 'abc') # type: ignore[attr-defined] + self.assertRebuilds(octets) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/misc/test_pcapng_unit.py b/tests/protocols/misc/test_pcapng_unit.py index a2fa28f7bc..1712137217 100644 --- a/tests/protocols/misc/test_pcapng_unit.py +++ b/tests/protocols/misc/test_pcapng_unit.py @@ -1471,12 +1471,20 @@ def test_pcapng_schema_helpers_and_post_process_branches(self) -> None: }) self.assertIs(isb_block_field.schema, schema_pcapng.InterfaceStatisticsBlock) + # NOTE: 20 framing octets and 17 of secrets padded to 20 (#1980). secrets_field = schema_pcapng.dsb_secrets_selector({ + 'length': 40, 'secrets_type': SecretsType.TLS_Key_Log, 'secrets_length': 17, }) self.assertIs(secrets_field.schema, schema_pcapng.TLSKeyLog) self.assertEqual(secrets_field.length, 17) + overrun_field = schema_pcapng.dsb_secrets_selector({ + 'length': 24, + 'secrets_type': SecretsType.TLS_Key_Log, + 'secrets_length': 17, + }) + self.assertEqual(overrun_field.length, 4) option_field = schema_pcapng.OptionEnumField(length=2, namespace='opt') option_field.name = 'type'