From f078fbd96345f244587341d6239ed21c629fb73c Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Fri, 9 Oct 2026 00:21:08 -0400 Subject: [PATCH] fix(pcap,pcapng)!: a cut PCAP header or PCAP-NG block rebuilds as captured (#1470) - PCAP Header calls #1465's keep_short_read/replay_short_read, so a header cut short rebuilds from its info as the octets captured, not all 24. - A PCAP-NG block the capture ends inside keeps its octets as __truncated_raw__; a rebuild parses them again instead of writing the block at its declared length (which wrote zeros, or raised ProtocolError). - A cut block whose own parser refuses the zero-filled fields is kept as an UnknownBlock rather than raising. - TLS and WireGuard key logs skip a last line the capture cut, and raise FieldValueError for any other malformed line (was a bare ValueError). Adds tests/protocols/misc/test_misc_short_read_1470_runtime.py. --- pcapkit/protocols/data/misc/pcapng.py | 6 + pcapkit/protocols/misc/pcap/header.py | 25 +- pcapkit/protocols/misc/pcapng.py | 77 +++++- pcapkit/protocols/schema/misc/pcapng.py | 55 ++++- .../misc/test_misc_short_read_1470_runtime.py | 223 ++++++++++++++++++ 5 files changed, 371 insertions(+), 15 deletions(-) create mode 100644 tests/protocols/misc/test_misc_short_read_1470_runtime.py diff --git a/pcapkit/protocols/data/misc/pcapng.py b/pcapkit/protocols/data/misc/pcapng.py index fcd16ae9a1..5c6e4f367e 100644 --- a/pcapkit/protocols/data/misc/pcapng.py +++ b/pcapkit/protocols/data/misc/pcapng.py @@ -69,6 +69,12 @@ class PCAPNG(Protocol): #: Block total length. length: 'int' + if TYPE_CHECKING: + #: The block's octets as captured, when the capture ended inside it; + #: present only then. A rebuild parses them again rather than writing + #: the block at its declared length (:issue:`1470`). + __truncated_raw__: 'bytes' + @info_final class UnknownBlock(PCAPNG): diff --git a/pcapkit/protocols/misc/pcap/header.py b/pcapkit/protocols/misc/pcap/header.py index 66e38787c4..80bd185eec 100644 --- a/pcapkit/protocols/misc/pcap/header.py +++ b/pcapkit/protocols/misc/pcap/header.py @@ -35,6 +35,7 @@ from pcapkit.protocols.data.misc.pcap.header import MagicNumber as Data_MagicNumber from pcapkit.protocols.protocol import ProtocolBase from pcapkit.protocols.schema.misc.pcap.header import Header as Schema_Header +from pcapkit.protocols.schema.schema import keep_short_read, replay_short_read from pcapkit.utilities.exceptions import EndianError, FileError, UnsupportedCall if TYPE_CHECKING: @@ -42,7 +43,7 @@ from typing import IO, Any, NoReturn, Optional, Type from aenum import IntEnum as AenumEnum - from typing_extensions import Literal + from typing_extensions import Literal, Self __all__ = ['Header'] @@ -256,6 +257,11 @@ def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None, See Also: For construction argument, please refer to :meth:`make`. + Notes: + A header the data ends inside keeps that in its :attr:`info`, so + that :meth:`from_data` rebuilds only what was captured + (:issue:`1470`). + """ if file is None: _data = self.pack(**kwargs) @@ -270,6 +276,7 @@ def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None, self._file.name = file.name #: pcapkit.corekit.infoclass.Info: Parsed packet data. self._info = self.unpack(length, **kwargs) + keep_short_read(self) def __len__(self) -> 'Literal[24]': """Total length of corresponding protocol.""" @@ -279,6 +286,22 @@ def __length_hint__(self) -> 'Literal[24]': """Return an estimated length for the object.""" return 24 + @classmethod + def from_data(cls, data: 'Data_Header | dict[str, Any]', **kwargs: 'Any') -> 'Self': + """Create protocol instance from data. + + Args: + data: Protocol data. + **kwargs: Construction keywords, as for :meth:`ProtocolBase.from_data + `. + + Returns: + Protocol instance, cut back to what was captured when the header + it was parsed from was cut short (:issue:`1470`). + + """ + return replay_short_read(super().from_data(data, **kwargs), data) + @classmethod def __index__(cls) -> 'NoReturn': """Numeral registry index of the protocol. diff --git a/pcapkit/protocols/misc/pcapng.py b/pcapkit/protocols/misc/pcapng.py index ffa68313bc..d071f5d6d9 100644 --- a/pcapkit/protocols/misc/pcapng.py +++ b/pcapkit/protocols/misc/pcapng.py @@ -15,6 +15,7 @@ import base64 import builtins import collections +import collections.abc import datetime import decimal import enum @@ -178,8 +179,9 @@ from pcapkit.protocols.schema.misc.pcapng import ZigBeeNWKKey as Schema_ZigBeeNWKKey from pcapkit.protocols.schema.schema import Schema from pcapkit.utilities.compat import StrEnum, localcontext -from pcapkit.utilities.exceptions import (EnumValueError, FormatError, ProtocolError, RegistryError, - StreamEOFError, UnsupportedCall, stacklevel) +from pcapkit.utilities.exceptions import (BaseError, EnumValueError, FormatError, ProtocolError, + RegistryError, StreamEOFError, UnsupportedCall, + stacklevel) from pcapkit.utilities.warnings import (AttributeWarning, DeprecatedFormatWarning, ProtocolWarning, RegistryWarning, warn) @@ -1264,7 +1266,15 @@ def read(self, length: 'Optional[int]' = None, *, _read: 'bool' = True, getattr(self, meth_name, self._read_block_unknown)) else: meth = name[0] - block = meth(schema.block, header=schema) + try: + block = meth(schema.block, header=schema) + except BaseError as error: + # NOTE: Fields past the end of a block the capture ended inside are + # read as zeros, and a value its parser refuses is an artefact of + # that, not of the block; it is kept as captured instead. + if not _read or self._captured(_seek_set) >= schema.block.length: + raise + block = self._read_block_cut(schema, _seek_set, error) # NOTE: An option or record area holding a malformed entry is kept as # the octets captured, as one :obj:`bytes` item (see @@ -1314,6 +1324,14 @@ def read(self, length: 'Optional[int]' = None, *, _read: 'bool' = True, stacklevel=stacklevel()) seek_cur = min(seek_cur, _seek_set + read) + # NOTE: The fields read past the end of the capture are zeros, + # and an option, record or packet the end cut through is only + # its first octets, so the block cannot be rebuilt from them at + # the length it declares. The octets captured are kept as + # ``__truncated_raw__`` instead, which :meth:`__post_init__` parses + # again in place of building the block (:issue:`1470`). + block.__update__([('__truncated_raw__', self._data)]) + # move backward to the beginning of next block self._file.seek(seek_cur, io.SEEK_SET) @@ -1413,6 +1431,13 @@ def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None, length: 'Opt #: collections.Counter: Counter for option types. self._opt = collections.Counter() # type: Counter[Enum_OptionType] + # NOTE: A block the capture ended inside rebuilds as the octets that + # were captured, parsed again as they were the first time (see + # :meth:`read`), rather than at the length it declares (:issue:`1470`). + block = kwargs.get('block') + if file is None and isinstance(block, collections.abc.Mapping) and '__truncated_raw__' in block: + file = bytes(block['__truncated_raw__']) + if file is None: _read = False #: bytes: Raw packet data. @@ -1521,6 +1546,52 @@ def _check_block_floor(self, length: 'Optional[int]') -> 'None': raise StreamEOFError(f'PCAP-NG: block truncated: {length} octet(s) left, ' 'fewer than the 12 a block needs', quiet=True) + def _captured(self, offset: 'int') -> 'int': + """Count the octets left in the stream from ``offset`` on. + + Args: + offset: File offset of the block. + + Returns: + Octets from ``offset`` to the end of + :attr:`self._file `. + + """ + current = self._file.tell() + end = self._file.seek(0, io.SEEK_END) + self._file.seek(current, io.SEEK_SET) + return end - offset + + def _read_block_cut(self, header: 'Schema_PCAPNG', offset: 'int', + error: 'BaseError') -> 'Data_UnknownBlock': + """Read a PCAP-NG block the capture ended inside, whose parser refused it. + + Args: + header: Parsed PCAP-NG header schema. + offset: File offset of the block. + error: What the block's parser raised. + + Returns: + Parsed packet data, with the octets captured after the Block Total + Length as its body. :meth:`read` keeps the block's octets as + ``__truncated_raw__`` too, which a rebuild parses again, so that it + rebuilds as captured (:issue:`1470`). + + """ + warn(f'PCAP-NG: [Block {header.type}] block truncated; kept as captured: {error}', + ProtocolWarning, stacklevel=stacklevel()) + + current = self._file.tell() + self._file.seek(offset + 8, io.SEEK_SET) + body = self._read_fileng(header.block.length - 12) + self._file.seek(current, io.SEEK_SET) + + return Data_UnknownBlock( + type=header.type, + length=header.block.length, + body=body, + ) + def _get_payload(self) -> 'bytes': """Get payload of :attr:`self.__header__ `. diff --git a/pcapkit/protocols/schema/misc/pcapng.py b/pcapkit/protocols/schema/misc/pcapng.py index 7de523d197..dcb4acdb48 100644 --- a/pcapkit/protocols/schema/misc/pcapng.py +++ b/pcapkit/protocols/schema/misc/pcapng.py @@ -522,6 +522,25 @@ 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': + """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. + + 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`). + + """ + return index == len(lines) - 1 and '__short_read__' in schema.__dict__ + + def dsb_secrets_selector(packet: 'dict[str, Any]') -> 'Field': """Selector function for :attr:`DecryptionSecretsBlock.secrets_data` field. @@ -2194,14 +2213,20 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': from pcapkit.protocols.misc.pcapng import TLSKeyLabel entries = collections.defaultdict(OrderedMultiDict) # type: dict[TLSKeyLabel, OrderedMultiDict[bytes, bytes]] - for line in self.data.splitlines(): + lines = self.data.splitlines() + for index, line in enumerate(lines): if not line or line.startswith('#'): continue - label, random, secret = line.strip().split() - label_enum = TLSKeyLabel(label.upper()) - entries[label_enum].add(bytes.fromhex(random), - bytes.fromhex(secret)) + try: + label, random, secret = line.strip().split() + label_enum = TLSKeyLabel(label.upper()) + entries[label_enum].add(bytes.fromhex(random), + bytes.fromhex(secret)) + except ValueError as error: + if _cut_last_line(self, index, lines): + break + raise FieldValueError(f'invalid TLS key log format: {line!r}') from error self.entries = entries return self @@ -2233,16 +2258,24 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': from pcapkit.protocols.misc.pcapng import WireGuardKeyLabel entries = OrderedMultiDict() # type: OrderedMultiDict[WireGuardKeyLabel, bytes] - for line in self.data.splitlines(): + lines = self.data.splitlines() + for index, line in enumerate(lines): if not line or line.startswith('#'): continue fields = line.strip().split() - if len(fields) != 3 or fields[1] != '=': - raise FieldValueError(f'invalid WireGuard key log format: {line!r}') - label, _, secret = fields - label_enum = WireGuardKeyLabel(label.upper()) - entries.add(label_enum, base64.b64decode(secret)) + try: + if len(fields) != 3 or fields[1] != '=': + raise FieldValueError(f'invalid WireGuard key log format: {line!r}') + label, _, secret = fields + label_enum = WireGuardKeyLabel(label.upper()) + entries.add(label_enum, base64.b64decode(secret)) + except ValueError as error: + if _cut_last_line(self, index, lines): + break + if isinstance(error, FieldValueError): + raise + raise FieldValueError(f'invalid WireGuard key log format: {line!r}') from error 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 new file mode 100644 index 0000000000..c81dc5e999 --- /dev/null +++ b/tests/protocols/misc/test_misc_short_read_1470_runtime.py @@ -0,0 +1,223 @@ +# -*- coding: utf-8 -*- +"""A PCAP global header or PCAP-NG block cut short rebuilds as captured. + +GitHub issue #1470. The two classes under :mod:`pcapkit.protocols.misc` that +read a file's own framing do not inherit the layer base classes, so the short +read #1458 records did not reach them. ``Header(raw[:4])`` rebuilt from its +``info`` as all 24 octets, and a PCAP-NG block the capture ended inside rebuilt +at the length it declares, or raised -- :exc:`ProtocolError` on a rebuilt Block +Total Length that was not a multiple of four, and a bare :exc:`ValueError` on a +Decryption Secrets Block cut inside a TLS key log line. + +Each header and each block type found in the sample captures is cut at every +offset. A PCAP header must be refused with an in-library exception, or rebuild +as the octets cut. A PCAP-NG block of fewer than the twelve octets any block +needs is end-of-stream; one of twelve or more must parse, and rebuild as the +octets cut. Every rebuild is checked from both ``info`` and ``info.to_dict()``. + +The module reads generated captures, so it belongs to the fixture-dependent tier. +Classes are imported inside each test, after +:func:`~tests._support.reimport_once_per_class`. + +""" + +import importlib.util +import struct +import unittest +import warnings +from typing import TYPE_CHECKING + +from tests._support import reimport_once_per_class, time_limit +from tests._tiers import SAMPLE_ROOT + +if TYPE_CHECKING: + from typing import Any, Iterator + +RUNTIME_DEPS = ('aenum', 'chardet') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + +#: Whole seconds one sweep may take. Each takes well under a minute. +SWEEP_TIMEOUT = 300 + +#: Section Header Block type, the same read in either byte order. +SHB = b'\x0a\x0d\x0d\x0a' + + +def _captures(suffix: 'str') -> 'list[str]': + """Every sample capture whose name ends in ``suffix``.""" + return sorted(path.name for path in SAMPLE_ROOT.iterdir() if path.name.endswith(suffix)) + + +def _blocks(data: 'bytes') -> 'Iterator[tuple[int, bytes]]': + """Every block of a PCAP-NG file, as its type and its octets.""" + offset, endian = 0, '<' + while offset + 12 <= len(data): + if data[offset:offset + 4] == SHB: + endian = '<' if data[offset + 8:offset + 12] == b'\x4d\x3c\x2b\x1a' else '>' + type_, length = struct.unpack(f'{endian}II', data[offset:offset + 8]) + yield type_, data[offset:offset + length] + offset += length + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class TestMiscShortRead(unittest.TestCase): + """Pin byte-exact rebuilds of PCAP headers and PCAP-NG blocks cut short.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + catcher = warnings.catch_warnings() + catcher.__enter__() + self.addCleanup(catcher.__exit__, None, None, None) + # a cut block warns that its length runs past the data, which is the point + warnings.simplefilter('ignore') + + def test_pcap_header_cut_at_every_offset(self) -> None: + from pcapkit.protocols.misc.pcap.header import Header + from pcapkit.utilities.exceptions import BaseError + + names = _captures('.pcap') + self.assertGreater(len(names), 2, f'no generated captures under {SAMPLE_ROOT}') + with time_limit(SWEEP_TIMEOUT): + for name in names: + raw = (SAMPLE_ROOT / name).read_bytes()[:24] + for keep in range(1, 25): + cut = raw[:keep] + with self.subTest(capture=name, keep=keep): + try: + header = Header(cut) + except BaseError: + # no magic number to tell the file format by + self.assertLess(keep, 4) + continue + self.assertEqual(header.data, cut) + self.assertEqual(Header.from_data(header.info).data.hex(), cut.hex()) + self.assertEqual(Header.from_data(header.info.to_dict()).data.hex(), cut.hex()) + + def test_issue_1470_repro(self) -> None: + from pcapkit.protocols.misc.pcap.header import Header + + raw = (SAMPLE_ROOT / _captures('.pcap')[0]).read_bytes()[:24] + header = Header(raw[:4]) + self.assertEqual(len(header.data), 4) + self.assertEqual(len(Header.from_data(header.info).data), 4) + # a header captured whole is untouched + self.assertEqual(Header.from_data(Header(raw).info).data, raw) + + def _each_block(self) -> 'Iterator[tuple[str, str, bytes, dict[str, Any]]]': + """The first block of each type in each PCAP-NG capture, with its keywords.""" + from pcapkit.const.pcapng.block_type import BlockType + from pcapkit.foundation.engines.pcapng import Context + from pcapkit.protocols.misc.pcapng import PCAPNG + + names = _captures('.pcapng') + self.assertGreater(len(names), 2, f'no generated captures under {SAMPLE_ROOT}') + for name in names: + ctx, num, sct, seen = None, 0, 0, set() + for type_, raw in _blocks((SAMPLE_ROOT / name).read_bytes()): + if raw[:4] == SHB: + sct += 1 + kwargs = {'num': 0, 'sct': sct, 'ctx': None} # type: dict[str, Any] + else: + num += 1 + kwargs = {'num': num, 'sct': sct, 'ctx': ctx} + whole = PCAPNG(raw, len(raw), **kwargs) + if type_ not in seen: + seen.add(type_) + yield name, BlockType.get(type_).name, raw, kwargs + if raw[:4] == SHB: + ctx = Context(whole.info) + whole._ctx = ctx # pylint: disable=protected-access + elif type_ == BlockType.Interface_Description_Block: + ctx.interfaces.append(whole.info) + + def test_pcapng_block_cut_at_every_offset(self) -> None: + from pcapkit.protocols.misc.pcapng import PCAPNG + from pcapkit.utilities.exceptions import StreamEOFError + + types = set() + with time_limit(SWEEP_TIMEOUT): + for name, block, raw, kwargs in self._each_block(): + types.add(block) + for keep in range(1, len(raw)): + cut = raw[:keep] + with self.subTest(capture=name, block=block, keep=keep): + if keep < 12: + with self.assertRaises(StreamEOFError): + PCAPNG(cut, len(cut), **kwargs) + continue + parsed = PCAPNG(cut, len(cut), **kwargs) + self.assertEqual(parsed.data, cut) + self.assertEqual(parsed.info.__truncated_raw__, cut) + self.assertEqual(PCAPNG.from_data(parsed.info, **kwargs).data.hex(), cut.hex()) + self.assertEqual(PCAPNG.from_data(parsed.info.to_dict(), **kwargs).data.hex(), cut.hex()) + # every block type the samples hold was swept + self.assertGreaterEqual(len(types), 8, sorted(types)) + + def test_pcapng_whole_block_is_untouched(self) -> None: + from pcapkit.protocols.misc.pcapng import PCAPNG + + for name, block, raw, kwargs in self._each_block(): + with self.subTest(capture=name, block=block): + parsed = PCAPNG(raw, len(raw), **kwargs) + self.assertNotIn('__truncated_raw__', parsed.info) + self.assertEqual(PCAPNG.from_data(parsed.info, **kwargs).data.hex(), raw.hex()) + + def test_cut_block_keeps_its_fields(self) -> None: + # the fields the capture holds still parse, so a cut packet keeps its packet + from pcapkit.const.pcapng.block_type import BlockType + from pcapkit.protocols.misc.pcapng import PCAPNG + + for name, block, raw, kwargs in self._each_block(): + if block != BlockType.Enhanced_Packet_Block.name: + continue + with self.subTest(capture=name): + whole = PCAPNG(raw, len(raw), **kwargs).info + cut = PCAPNG(raw[:40], 40, **kwargs).info + self.assertEqual(cut.length, whole.length) + self.assertEqual(cut.captured_len, whole.captured_len) + self.assertEqual(cut.timestamp_epoch, whole.timestamp_epoch) + self.assertEqual(bytes(cut.packet), raw[28:40]) + + +class TestKeyLogLines(unittest.TestCase): + """Pin how the TLS and WireGuard key logs treat a line that does not parse.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _unpack(self, schema: 'str', text: 'bytes', length: 'int') -> 'Any': + import importlib + + cls = getattr(importlib.import_module('pcapkit.protocols.schema.misc.pcapng'), schema) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + return cls.unpack(text, length) + + def test_cut_last_line_is_left_out_of_entries(self) -> None: + line = b'CLIENT_RANDOM ' + b'ab' * 32 + b' ' + b'cd' * 48 + b'\n' + text = line + line[:20] + schema = self._unpack('TLSKeyLog', text, len(line) * 2) + self.assertTrue(schema.data.startswith(text.decode()), schema.data) + self.assertEqual(sum(len(value) for value in schema.entries.values()), 1) + + line = b'LOCAL_STATIC_PRIVATE_KEY = ' + b'QUFB' * 11 + b'QUE=\n' + text = line + line[:10] + schema = self._unpack('WireGuardKeyLog', text, len(line) * 2) + 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)) + + 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)) + + +if __name__ == '__main__': + unittest.main()