guardian_panic.py

"""Dependency-free forensic capture for Python Guardian supervised children."""
from __future__ import annotations

import datetime as dt
import json
import os
import platform
import sys
import tempfile
import threading
import traceback
from pathlib import Path
from typing import Any


CRASH_DIR = Path(tempfile.gettempdir()) / "PythonGuardian" / "panic"
_ORIGINAL_EXCEPTHOOK = sys.excepthook
_SECRET_WORDS = (
    "token", "password", "passwd", "secret", "api_key", "apikey", "authorization",
    "cookie", "credential", "private_key", "session_key", "bearer",
)


def _safe_repr(name: str, value: Any, limit: int = 500) -> str:
    if any(word in name.casefold() for word in _SECRET_WORDS):
        return "<REDACTED>"
    try:
        rendered = repr(value)
    except Exception:
        return "<unrepresentable>"
    lowered = rendered.casefold()
    if "bearer " in lowered or any(f"{word}=" in lowered for word in _SECRET_WORDS):
        return "<REDACTED:value-pattern>"
    return rendered if len(rendered) <= limit else rendered[:limit] + "…"


def _locals_from_traceback(exc_tb) -> dict[str, str]:
    if exc_tb is None:
        return {}
    frame = exc_tb
    while frame.tb_next:
        frame = frame.tb_next
    result: dict[str, str] = {}
    for name, value in list(frame.tb_frame.f_locals.items())[:50]:
        if not name.startswith("_") and not isinstance(value, type(sys)):
            result[name] = _safe_repr(name, value)
    return result


def capture_state(exc_type=None, exc_value=None, exc_tb=None, *, reason: str | None = None,
                  agent_name: str = "main") -> dict[str, Any]:
    CRASH_DIR.mkdir(parents=True, exist_ok=True)
    now = dt.datetime.now(dt.timezone.utc)
    report = {
        "crash_report_version": "guardian-1.0",
        "timestamp": now.isoformat(),
        "agent_name": agent_name,
        "reason": reason,
        "exception_type": getattr(exc_type, "__name__", "Unknown"),
        "error_message": str(exc_value) if exc_value is not None else "",
        "traceback": [line.rstrip() for line in traceback.format_tb(exc_tb)] if exc_tb else [],
        "local_variables": _locals_from_traceback(exc_tb),
        "system_info": {
            "python": sys.version,
            "executable": sys.executable,
            "platform": platform.platform(),
            "pid": os.getpid(),
            "thread": threading.current_thread().name,
        },
    }
    stamp = now.strftime("%Y%m%d_%H%M%S_%f")
    destination = CRASH_DIR / f"crash_Global_{stamp}_{os.getpid()}.json"
    temporary = destination.with_suffix(".json.tmp")
    temporary.write_text(json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8")
    os.replace(temporary, destination)
    print(f"[PYTHON GUARDIAN] Crash report written: {destination}", file=sys.stderr)
    return report


def _hook(exc_type, exc_value, exc_tb) -> None:
    try:
        capture_state(exc_type, exc_value, exc_tb)
    except Exception as capture_error:
        print(f"[PYTHON GUARDIAN] Forensic capture failed: {capture_error}", file=sys.stderr)
    _ORIGINAL_EXCEPTHOOK(exc_type, exc_value, exc_tb)


def install() -> None:
    sys.excepthook = _hook