"""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("<II", data, 8)
	if _stream_count > _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("<III", data, offset)
		_require_range(data, rva, size)
		streams.setdefault(stream_type, (rva, size))

	exception_stream = streams.get(_EXCEPTION_STREAM_TYPE)
	if exception_stream is None or exception_stream[1] < 40:
		return None
	exception_rva, _exception_size = exception_stream
	thread_id = _read_u32(data, exception_rva)
	exception_code = _read_u32(data, exception_rva + 8)
	exception_flags = _read_u32(data, exception_rva + 12)
	exception_address = _read_u64(data, exception_rva + 24)
	exception_parameters = _read_exception_parameters(
		data,
		exception_rva,
		_exception_size,
	)
	modules = _read_modules(data, streams.get(_MODULE_LIST_STREAM_TYPE))
	exception_module = _module_for_address(modules, exception_address)
	architecture = _read_architecture(data, streams.get(_SYSTEM_INFO_STREAM_TYPE))
	result: dict[str, object] = {
		"format": "minidump",
		"architecture": architecture,
		"exception_code": f"0x{exception_code:08x}",
		"exception_kind": _EXCEPTION_NAMES.get(exception_code, "unknown"),
		"exception_flags": f"0x{exception_flags:08x}",
		"module_count": len(modules),
	}
	if exception_module is not None:
		result.update(
			{
				"exception_module": exception_module.name,
				"exception_module_offset": (
					f"0x{exception_address - exception_module.base:x}"
				),
			}
		)
	stack_pointer = _read_exception_stack_pointer(
		data,
		exception_stream,
		architecture,
	)
	stack = _read_exception_stack(
		data,
		streams.get(_THREAD_LIST_STREAM_TYPE),
		thread_id,
		stack_pointer=stack_pointer,
	)
	if stack is not None:
		pointer_size = 8 if architecture in {"x86_64", "arm64", "ia64"} else 4
		signatures = _application_stack_signatures(
			stack,
			pointer_size,
			modules,
			excluded_addresses=frozenset(exception_parameters),
		)
		if signatures:
			result["application_stack"] = signatures
	return result


def _read_modules(
	data: bytes,
	stream: tuple[int, int] | None,
) -> 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("<H", data, stream[0])[0]
	return _ARCHITECTURE_NAMES.get(architecture, f"architecture_{architecture}")


def _read_exception_stack(
	data: bytes,
	stream: tuple[int, int] | None,
	thread_id: int,
	*,
	stack_pointer: int | None = None,
) -> 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_start = _read_u64(data, offset + 24)
		reported_stack_size = _read_u32(data, offset + 32)
		stack_rva = _read_u32(data, offset + 36)
		stack_offset = 0
		if (
			stack_pointer is not None
			and stack_start <= stack_pointer < stack_start + reported_stack_size
		):
			stack_offset = stack_pointer - stack_start
		stack_size = min(
			reported_stack_size - stack_offset,
			_MAX_STACK_BYTES,
		)
		_require_range(data, stack_rva + stack_offset, stack_size)
		return data[stack_rva + stack_offset : stack_rva + stack_offset + stack_size]
	return None


def _read_exception_parameters(
	data: bytes,
	exception_rva: int,
	exception_size: int,
) -> tuple[int, ...]:
	parameter_count = _read_u32(data, exception_rva + 32)
	if parameter_count > 15 or 40 + parameter_count * 8 > exception_size:
		raise ValueError("invalid minidump exception parameters")
	return tuple(
		_read_u64(data, exception_rva + 40 + index * 8)
		for index in range(parameter_count)
	)


def _read_exception_stack_pointer(
	data: bytes,
	stream: tuple[int, int],
	architecture: str,
) -> int | None:
	rva, size = stream
	if size < 168:
		return None
	context_size = _read_u32(data, rva + 160)
	context_rva = _read_u32(data, rva + 164)
	if context_size == 0 or context_rva == 0:
		return None
	_require_range(data, context_rva, context_size)
	if architecture == "x86" and context_size >= 200:
		return _read_u32(data, context_rva + 196)
	if architecture == "x86_64" and context_size >= 160:
		return _read_u64(data, context_rva + 152)
	return None


def _application_stack_signatures(
	stack: bytes,
	pointer_size: int,
	modules: tuple[_Module, ...],
	*,
	excluded_addresses: frozenset[int] = frozenset(),
) -> list[dict[str, str]]:
	result: list[dict[str, str]] = []
	seen: set[tuple[str, int]] = set()
	format_code = "<Q" if pointer_size == 8 else "<I"
	for offset in range(0, len(stack) - pointer_size + 1, pointer_size):
		address = struct.unpack_from(format_code, stack, offset)[0]
		if address in excluded_addresses:
			continue
		module = _module_for_address(modules, address)
		if module is None or module.system:
			continue
		module_offset = address - module.base
		key = (module.name.casefold(), module_offset)
		if key in seen:
			continue
		seen.add(key)
		result.append(
			{
				"module": module.name,
				"offset": f"0x{module_offset:x}",
			}
		)
		if len(result) >= _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("<I", data, offset)[0]


def _read_u64(data: bytes, offset: int) -> int:
	_require_range(data, offset, 8)
	return struct.unpack_from("<Q", data, offset)[0]


def _require_range(data: bytes, offset: int, size: int) -> None:
	if offset < 0 or size < 0 or offset > len(data) or size > len(data) - offset:
		raise ValueError("minidump range is outside the file")