ROMFASTSQL/proxmox/lxc171-claude-agent/discord-bridge/security/approvals.py

"""Canal de aprobari pe disc intre hook-ul PreToolUse si botul Discord.

Hook-ul `confirm_hook.py` ruleaza in alt proces decat botul (il porneste CLI-ul
`claude`), deci canalul dintre ele este un director de cereri:

    ~/.claude-discord/approvals/<request_id>.json     cerere in asteptare
    ~/.claude-discord/approvals/done/<request_id>.json  cerere incheiata

Regula de baza: FAIL-CLOSED. Orice eroare, timeout, fisier corupt sau director
lipsa inseamna "deny". Modulul nu atinge reteaua si nu stie nimic despre Discord.
"""

from __future__ import annotations

import asyncio
import json
import os
import pathlib
import time
import uuid

# config.py apartine Lane A si poate lipsi in unele contexte (hook rulat izolat).
# Import tolerant, cu acelasi fallback pe ~/.claude-discord.
try:  # pragma: no cover - depinde de contextul de import
    from .. import config as _config  # type: ignore
except Exception:  # pragma: no cover
    try:
        import config as _config  # type: ignore
    except Exception:
        _config = None  # type: ignore

ALLOW = "allow"
DENY = "deny"
PENDING = "pending"

_POLL_S = 0.2  # cat de des verificam decizia pe disc
_WATCH_S = 0.5  # cat de des verificam cereri noi pentru bot


# ---------------------------------------------------------------- cai pe disc

def state_dir() -> pathlib.Path:
    """~/.claude-discord, cu CLAUDE_DISCORD_DIR ca override (folosit in teste)."""
    override = os.environ.get("CLAUDE_DISCORD_DIR")
    if override:
        return pathlib.Path(override)
    if _config is not None:
        try:
            return pathlib.Path(_config.STATE_DIR)
        except Exception:
            pass
    return pathlib.Path.home() / ".claude-discord"


def approvals_dir() -> pathlib.Path:
    return state_dir() / "approvals"


def done_dir() -> pathlib.Path:
    return approvals_dir() / "done"


def log_dir() -> pathlib.Path:
    return state_dir() / "logs"


def ensure_dirs() -> None:
    """Creeaza subdirectoarele de aprobari.

    Nu creeaza directorul de baza: daca ~/.claude-discord lipseste inseamna ca
    puntea nu e instalata, iar hook-ul trebuie sa refuze (fail-closed), nu sa
    improvizeze un director nou.
    """
    base = state_dir()
    if not base.is_dir():
        raise FileNotFoundError(f"directorul de stare lipseste: {base}")
    approvals_dir().mkdir(parents=True, exist_ok=True)
    done_dir().mkdir(parents=True, exist_ok=True)


# ------------------------------------------------------------ scriere atomica

def _write_atomic(path: pathlib.Path, payload: dict) -> None:
    """tmp + os.replace, ca un cititor sa nu vada niciodata JSON pe jumatate."""
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.parent / f".{path.name}.{os.getpid()}.{uuid.uuid4().hex[:8]}.tmp"
    data = json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True)
    try:
        with open(tmp, "w", encoding="utf-8") as fh:
            fh.write(data)
            fh.flush()
            os.fsync(fh.fileno())
        os.replace(tmp, path)
    finally:
        try:
            tmp.unlink()
        except OSError:
            pass


def _read(path: pathlib.Path) -> dict | None:
    """Citeste o cerere. JSON corupt sau fisier disparut => None."""
    try:
        raw = path.read_text(encoding="utf-8")
    except OSError:
        return None
    try:
        obj = json.loads(raw)
    except (ValueError, TypeError):
        return None
    return obj if isinstance(obj, dict) else None


# --------------------------------------------------------------- API interna
# (folosita de confirm_hook.py; botul nu are nevoie de ea)

def new_request_id() -> str:
    return uuid.uuid4().hex[:16]


def create_request(
    *,
    tool_name: str,
    command: str,
    reason: str = "",
    rule: str = "",
    thread_id: str | None = None,
    session_id: str | None = None,
    cwd: str | None = None,
    timeout: float = 300.0,
    request_id: str | None = None,
) -> dict:
    """Scrie o cerere de confirmare si o returneaza. Arunca daca nu poate scrie."""
    ensure_dirs()
    rid = request_id or new_request_id()
    now = time.time()
    req = {
        "request_id": rid,
        "thread_id": thread_id,
        "session_id": session_id,
        "tool_name": tool_name,
        "command": command,
        "rule": rule,
        "reason": reason,
        "cwd": cwd,
        "created_at": now,
        "expires_at": now + float(timeout),
        "status": PENDING,
        "decision": None,
        "decided_at": None,
        "decided_by": None,
    }
    _write_atomic(approvals_dir() / f"{rid}.json", req)
    return req


def request_path(request_id: str) -> pathlib.Path:
    return approvals_dir() / f"{_safe_id(request_id)}.json"


def _safe_id(request_id: str) -> str:
    """Nu lasam un id sa evadeze din director prin `../`."""
    rid = str(request_id)
    if not rid or "/" in rid or "\\" in rid or rid.startswith("."):
        raise ValueError(f"request_id invalid: {rid!r}")
    return rid


def read_decision(request_id: str) -> str:
    """`allow` / `deny` / `pending`. Orice problema => `deny` (fail-closed)."""
    try:
        req = _read(request_path(request_id))
    except Exception:
        return DENY
    if req is None:
        return DENY
    status = req.get("status")
    if status == ALLOW:
        return ALLOW
    if status == PENDING:
        return PENDING
    return DENY


def finish_request(request_id: str, status: str, note: str = "") -> None:
    """Muta cererea in `done/`, ca `pending_requests()` sa nu o mai vada."""
    try:
        src = request_path(request_id)
        req = _read(src) or {"request_id": request_id}
        req["status"] = status if status in (ALLOW, DENY) else DENY
        req["finished_at"] = time.time()
        if note:
            req["note"] = note
        _write_atomic(done_dir() / f"{_safe_id(request_id)}.json", req)
        try:
            src.unlink()
        except OSError:
            pass
    except Exception:
        pass


def cleanup_stale(max_age_s: float = 86400.0) -> int:
    """Sterge cereri incheiate mai vechi de o zi. Returneaza cate a sters."""
    n = 0
    now = time.time()
    try:
        for p in done_dir().glob("*.json"):
            try:
                if now - p.stat().st_mtime > max_age_s:
                    p.unlink()
                    n += 1
            except OSError:
                pass
    except Exception:
        pass
    return n


# ---------------------------------------------------- API publica (Lane A/bot)

async def wait_for_decision(request_id: str, timeout: float) -> str:
    """Asteapta decizia pentru o cerere. La timeout sau eroare returneaza `deny`."""
    deadline = time.monotonic() + max(0.0, float(timeout or 0))
    while True:
        decision = read_decision(request_id)
        if decision in (ALLOW, DENY):
            return decision
        if time.monotonic() >= deadline:
            return DENY
        await asyncio.sleep(_POLL_S)


def wait_for_decision_sync(request_id: str, timeout: float) -> str:
    """Varianta blocanta, pentru hook (proces separat, fara bucla asyncio)."""
    deadline = time.monotonic() + max(0.0, float(timeout or 0))
    while True:
        decision = read_decision(request_id)
        if decision in (ALLOW, DENY):
            return decision
        if time.monotonic() >= deadline:
            return DENY
        time.sleep(_POLL_S)


def submit_decision(request_id: str, decision: str) -> bool:
    """Apelata de bot.py cand utilizatorul apasa butonul.

    True daca cererea exista si a fost marcata. O decizie nerecunoscuta este
    tratata ca `deny` si returneaza False.
    """
    try:
        path = request_path(request_id)
    except Exception:
        return False
    req = _read(path)
    if req is None:
        return False
    valid = decision in (ALLOW, DENY)
    req["status"] = decision if valid else DENY
    req["decision"] = req["status"]
    req["decided_at"] = time.time()
    try:
        _write_atomic(path, req)
    except Exception:
        return False
    return valid


async def pending_requests() -> list[dict]:
    """Cererile inca in asteptare, cele mai vechi intai. Nu arunca niciodata."""
    out: list[dict] = []
    try:
        paths = sorted(approvals_dir().glob("*.json"))
    except Exception:
        return out
    for p in paths:
        req = _read(p)
        if not req or req.get("status") != PENDING:
            continue
        out.append(
            {
                "request_id": req.get("request_id") or p.stem,
                "thread_id": req.get("thread_id"),
                "tool_name": req.get("tool_name") or "",
                "command": req.get("command") or "",
                "created_at": req.get("created_at") or 0.0,
                "reason": req.get("reason") or "",
            }
        )
    out.sort(key=lambda r: r["created_at"])
    return out


_on_request = None
_watch_task = None
_seen: set[str] = set()


def set_on_request(callback) -> None:
    """Inregistreaza un callback async apelat cand apare o cerere noua.

    Botul posteaza atunci butoanele in firul Discord. Un callback `None`
    opreste urmarirea.
    """
    global _on_request, _watch_task
    _on_request = callback
    if callback is None:
        stop_watcher()
        return
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        return  # fara bucla activa nu pornim nimic; se reapeleaza din bot
    if _watch_task is None or _watch_task.done():
        _watch_task = loop.create_task(_watch_loop())


def stop_watcher() -> None:
    global _watch_task
    if _watch_task is not None and not _watch_task.done():
        _watch_task.cancel()
    _watch_task = None


async def _watch_loop() -> None:
    """Urmareste directorul de cereri si anunta botul o singura data per cerere."""
    while True:
        try:
            for req in await pending_requests():
                rid = req["request_id"]
                if rid in _seen:
                    continue
                _seen.add(rid)
                cb = _on_request
                if cb is None:
                    continue
                try:
                    res = cb(req)
                    if asyncio.iscoroutine(res):
                        await res
                except Exception:
                    pass  # un callback care crapa nu are voie sa opreasca botul
            if len(_seen) > 5000:
                _seen.clear()
        except asyncio.CancelledError:
            raise
        except Exception:
            pass
        await asyncio.sleep(_WATCH_S)