From f7ceba2b9b0530dfbe4756bb929af7f620703395 Mon Sep 17 00:00:00 2001 From: m4ko Date: Tue, 15 Sep 2026 21:23:10 +0200 Subject: [PATCH] Added crash dumps class. --- src/ptlinux/diagnostics/crash_dumps.py | 291 +++++++++++++++++++++++++ 1 file changed, 291 insertions(+) create mode 100644 src/ptlinux/diagnostics/crash_dumps.py diff --git a/src/ptlinux/diagnostics/crash_dumps.py b/src/ptlinux/diagnostics/crash_dumps.py new file mode 100644 index 0000000..88d0f91 --- /dev/null +++ b/src/ptlinux/diagnostics/crash_dumps.py @@ -0,0 +1,291 @@ +"""Bounded, dependency-free crash signatures for Windows minidumps.""" + +from __future__ import annotations + +import base64 +from dataclasses import dataclass +import gzip +import hashlib +from pathlib import Path, PureWindowsPath +import struct + + +_MINIDUMP_SIGNATURE = b"MDMP" +_HEADER_SIZE = 32 +_DIRECTORY_ENTRY_SIZE = 12 +_MODULE_ENTRY_SIZE = 108 +_THREAD_ENTRY_SIZE = 48 +_EXCEPTION_STREAM_TYPE = 6 +_MODULE_LIST_STREAM_TYPE = 4 +_SYSTEM_INFO_STREAM_TYPE = 7 +_THREAD_LIST_STREAM_TYPE = 3 +_MAX_DUMP_BYTES = 32 * 1024 * 1024 +_MAX_STREAMS = 256 +_MAX_MODULES = 4096 +_MAX_THREADS = 4096 +_MAX_STACK_BYTES = 64 * 1024 +_MAX_STACK_SIGNATURES = 12 +MAX_CRASH_DUMP_UPLOAD_BYTES = 2 * 1024 * 1024 +MAX_CRASH_DUMP_COMPRESSED_BYTES = 30 * 1024 +MAX_CRASH_DUMP_ATTACHMENT_CHARACTERS = 40 * 1024 + +_ARCHITECTURE_NAMES = { + 0: "x86", + 5: "arm", + 6: "ia64", + 9: "x86_64", + 12: "arm64", +} +_EXCEPTION_NAMES = { + 0x80000003: "breakpoint", + 0xC0000005: "access_violation", + 0xC0000017: "no_memory", + 0xC0000094: "integer_divide_by_zero", + 0xC00000FD: "stack_overflow", + 0xC0000409: "stack_buffer_overrun", + 0xE06D7363: "msvc_cpp_exception", +} + + +@dataclass(frozen=True) +class _Module: + base: int + size: int + name: str + system: bool + + +def inspect_minidump(path: Path) -> dict[str, object] | None: + """Return a privacy-safe crash signature, or ``None`` for an unusable dump. + + The signature keeps stable exception and module offsets without absolute paths + or memory addresses. Full dump attachment is handled by the separate opt-in path. + """ + + try: + with path.open("rb") as handle: + data = handle.read(_MAX_DUMP_BYTES + 1) + except OSError: + return None + if len(data) < _HEADER_SIZE or len(data) > _MAX_DUMP_BYTES: + return None + try: + return _inspect_minidump_bytes(data) + except (OverflowError, struct.error, UnicodeError, ValueError): + return None + + +def prepare_minidump_attachment( + path: Path, +) -> tuple[str, dict[str, object] | None]: + """Compress one explicitly approved dump for bounded telemetry delivery.""" + + try: + with path.open("rb") as handle: + data = handle.read(MAX_CRASH_DUMP_UPLOAD_BYTES + 1) + except OSError: + return "unreadable", None + if len(data) > MAX_CRASH_DUMP_UPLOAD_BYTES: + return "source_too_large", None + try: + if _inspect_minidump_bytes(data) is None: + return "invalid_minidump", None + except (OverflowError, struct.error, UnicodeError, ValueError): + return "invalid_minidump", None + compressed = gzip.compress(data, compresslevel=9, mtime=0) + if len(compressed) > MAX_CRASH_DUMP_COMPRESSED_BYTES: + return "compressed_too_large", None + encoded = base64.urlsafe_b64encode(compressed).rstrip(b"=").decode("ascii") + if len(encoded) > MAX_CRASH_DUMP_ATTACHMENT_CHARACTERS: + return "encoded_too_large", None + return ( + "included", + { + "schema": 1, + "kind": "windows_minidump", + "encoding": "gzip+base64url", + "original_size": len(data), + "compressed_size": len(compressed), + "sha256": hashlib.sha256(data).hexdigest(), + "data_base64url": encoded, + }, + ) + + +def _inspect_minidump_bytes(data: bytes) -> dict[str, object] | None: + if len(data) < _HEADER_SIZE or data[:4] != _MINIDUMP_SIGNATURE: + return None + _stream_count, directory_rva = struct.unpack_from(" _MAX_STREAMS: + return None + _require_range(data, directory_rva, _stream_count * _DIRECTORY_ENTRY_SIZE) + streams: dict[int, tuple[int, int]] = {} + for index in range(_stream_count): + offset = directory_rva + index * _DIRECTORY_ENTRY_SIZE + stream_type, size, rva = struct.unpack_from(" tuple[_Module, ...]: + if stream is None or stream[1] < 4: + return () + rva, size = stream + count = _read_u32(data, rva) + if count > _MAX_MODULES or 4 + count * _MODULE_ENTRY_SIZE > size: + raise ValueError("invalid minidump module list") + modules: list[_Module] = [] + for index in range(count): + offset = rva + 4 + index * _MODULE_ENTRY_SIZE + base = _read_u64(data, offset) + image_size = _read_u32(data, offset + 8) + name_rva = _read_u32(data, offset + 20) + module_path = _read_utf16_string(data, name_rva) + name = PureWindowsPath(module_path).name or "unknown" + folded_path = module_path.replace("/", "\\").casefold() + modules.append( + _Module( + base=base, + size=image_size, + name=name, + system="\\windows\\" in folded_path, + ) + ) + return tuple(modules) + + +def _read_architecture( + data: bytes, + stream: tuple[int, int] | None, +) -> str: + if stream is None or stream[1] < 2: + return "unknown" + architecture = struct.unpack_from(" bytes | None: + if stream is None or stream[1] < 4: + return None + rva, size = stream + count = _read_u32(data, rva) + if count > _MAX_THREADS or 4 + count * _THREAD_ENTRY_SIZE > size: + raise ValueError("invalid minidump thread list") + for index in range(count): + offset = rva + 4 + index * _THREAD_ENTRY_SIZE + if _read_u32(data, offset) != thread_id: + continue + stack_size = min(_read_u32(data, offset + 32), _MAX_STACK_BYTES) + stack_rva = _read_u32(data, offset + 36) + _require_range(data, stack_rva, stack_size) + return data[stack_rva : stack_rva + stack_size] + return None + + +def _application_stack_signatures( + stack: bytes, + pointer_size: int, + modules: tuple[_Module, ...], +) -> list[dict[str, str]]: + result: list[dict[str, str]] = [] + seen: set[tuple[str, int]] = set() + format_code = "= _MAX_STACK_SIGNATURES: + break + return result + + +def _module_for_address( + modules: tuple[_Module, ...], + address: int, +) -> _Module | None: + for module in modules: + if module.base <= address < module.base + module.size: + return module + return None + + +def _read_utf16_string(data: bytes, rva: int) -> str: + byte_length = _read_u32(data, rva) + if byte_length > 64 * 1024 or byte_length % 2: + raise ValueError("invalid minidump string") + _require_range(data, rva + 4, byte_length) + return data[rva + 4 : rva + 4 + byte_length].decode("utf-16-le") + + +def _read_u32(data: bytes, offset: int) -> int: + _require_range(data, offset, 4) + return struct.unpack_from(" int: + _require_range(data, offset, 8) + return struct.unpack_from(" None: + if offset < 0 or size < 0 or offset > len(data) or size > len(data) - offset: + raise ValueError("minidump range is outside the file")