PythonGuardian/guardian_common.py at main · gaoptimize/PythonGuardian · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
"""Shared paths, identity, and persistence helpers for Python Guardian."""
from __future__ import annotations

import hashlib
import json
import os
import secrets
import tempfile
from pathlib import Path
from typing import Any


BASE_DIR = Path(__file__).resolve().parent
STATE_ROOT = Path(os.environ.get("LOCALAPPDATA", tempfile.gettempdir())) / "PythonGuardian"
JOBS_DIR = STATE_ROOT / "jobs"
DB_PATH = STATE_ROOT / "guardian.db"
AUTH_PATH = STATE_ROOT / "auth.key"
LOG_PATH = STATE_ROOT / "guardian.log"
PIPE_ADDRESS = r"\\.\pipe\PythonGuardianCodex"
MUTEX_NAME = r"Local\PythonGuardianCodexBroker"
RUNTIME_DIR = BASE_DIR / "guardian_venv"
RUNTIME_PYTHON = RUNTIME_DIR / "Scripts" / "python.exe"
RUNTIME_PYTHONW = RUNTIME_DIR / "Scripts" / "pythonw.exe"
BROKER_PATH = BASE_DIR / "python_guardian_broker.pyw"
RUNNER_PATH = BASE_DIR / "python_console_runner.py"
REPORT_VIEWER_PATH = BASE_DIR / "guardian_report_viewer.pyw"
CODEX_CLI_PATH = RUNTIME_DIR / "Lib" / "site-packages" / "codex_cli_bin" / "bin" / "codex.exe"
INJECT_DIR = BASE_DIR / "guardian_inject"
PANIC_DIR = BASE_DIR


def ensure_state_dirs() -> None:
    STATE_ROOT.mkdir(parents=True, exist_ok=True)
    JOBS_DIR.mkdir(parents=True, exist_ok=True)


def auth_key() -> bytes:
    """Return a durable per-user IPC key, handling first-start races."""
    ensure_state_dirs()
    try:
        return bytes.fromhex(AUTH_PATH.read_text(encoding="ascii").strip())
    except (OSError, ValueError):
        value = secrets.token_bytes(32)
        try:
            fd = os.open(str(AUTH_PATH), os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
            with os.fdopen(fd, "w", encoding="ascii") as stream:
                stream.write(value.hex())
            return value
        except FileExistsError:
            return bytes.fromhex(AUTH_PATH.read_text(encoding="ascii").strip())


def file_sha256(path: str | Path) -> str:
    digest = hashlib.sha256()
    with Path(path).open("rb") as stream:
        for block in iter(lambda: stream.read(1024 * 1024), b""):
            digest.update(block)
    return digest.hexdigest()


def project_root(script: str | Path) -> Path:
    path = Path(script).resolve().parent
    markers = (".git", "AGENTS.md", "pyproject.toml", "setup.cfg", "setup.py")
    for candidate in (path, *path.parents):
        if any((candidate / marker).exists() for marker in markers):
            return candidate
    return path


def atomic_json(path: str | Path, value: Any) -> None:
    destination = Path(path)
    destination.parent.mkdir(parents=True, exist_ok=True)
    temporary = destination.with_suffix(destination.suffix + ".tmp")
    temporary.write_text(json.dumps(value, indent=2, ensure_ascii=False), encoding="utf-8")
    os.replace(temporary, destination)


def truncate(text: str, limit: int) -> str:
    if len(text) <= limit:
        return text
    half = max(1, (limit - 80) // 2)
    return text[:half] + "\n… [middle truncated by Python Guardian] …\n" + text[-half:]