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
    ~/.claude-discord/approvals/grants/<thread_id>.json aprobari valabile pe fir

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"
# Decizie de buton, nu stare pe disc: "permite si nu ma mai intreba in firul asta".
ALLOW_SESSION = "allow_session"

# Cat traieste o aprobare de fir daca nimeni nu o revoca. Ziua de lucru, nu vesnicia.
GRANT_MAX_AGE_S = 12 * 3600.0

_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 grants_dir() -> pathlib.Path:
    """Aprobarile valabile pentru tot firul, cate un fisier per fir Discord."""
    return approvals_dir() / "grants"


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)
    grants_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 read_scope(request_id: str) -> str:
    """`thread` daca s-a apasat "Allow (fir)", altfel `once`. Eroare => `once`."""
    try:
        req = _read(request_path(request_id))
    except Exception:
        return "once"
    if not isinstance(req, dict):
        return "once"
    return "thread" if req.get("scope") == "thread" else "once"


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


# ------------------------------------------------- aprobari valabile pe fir
# Butonul "Allow (fir)" scrie aici. Hook-ul citeste inainte sa mai deranjeze
# omul: acelasi tipar de operatiune, in acelasi fir, nu se mai intreaba.
#
#     ~/.claude-discord/approvals/grants/<thread_id>.json
#
# Cheia NU e comanda, ci perechea (regula, motiv) produsa de clasificator:
# "host_productie|comanda catre hostul de productie 10.0.20.201". Asta face
# aprobarea utila (orice comanda catre acel host trece) fara sa fie oarba
# (alt host, alt serviciu => cerere noua).


def grants_enabled() -> bool:
    """`CLAUDE_DISCORD_SESSION_GRANTS=off` dezactiveaza complet mecanismul."""
    raw = (os.environ.get("CLAUDE_DISCORD_SESSION_GRANTS") or "").strip().lower()
    return raw not in ("off", "0", "false", "no", "nu")


def grant_max_age_s() -> float:
    raw = os.environ.get("CLAUDE_DISCORD_GRANT_TTL")
    try:
        if raw:
            return max(0.0, float(raw))
    except (TypeError, ValueError):
        pass
    return GRANT_MAX_AGE_S


def _safe_scope(scope_id: str | None) -> str:
    """Id de fir folosit ca nume de fisier. Orice suspiciune => ValueError."""
    sid = str(scope_id or "").strip()
    if not sid or len(sid) > 128:
        raise ValueError(f"scope invalid: {scope_id!r}")
    if not all(c.isalnum() or c in "._-" for c in sid) or sid.startswith("."):
        raise ValueError(f"scope invalid: {scope_id!r}")
    return sid


def grant_path(scope_id: str) -> pathlib.Path:
    return grants_dir() / f"{_safe_scope(scope_id)}.json"


def grant_key(rule: str, reason: str) -> str:
    return f"{str(rule or '').strip()}|{str(reason or '').strip()}"


def _load_grants(scope_id: str) -> dict:
    """Fisierul de aprobari al unui fir. Lipsa sau corupt => dict gol."""
    doc = _read(grant_path(scope_id))
    if not isinstance(doc, dict):
        return {}
    grants = doc.get("grants")
    return grants if isinstance(grants, dict) else {}


def has_grant(scope_id: str | None, rule: str, reason: str) -> bool:
    """Exista o aprobare de fir, inca valabila, pentru acest tipar?

    FAIL-CLOSED: fara scope, cu mecanismul oprit sau la orice eroare => False,
    adica se cere confirmarea in Discord ca pana acum.
    """
    if not grants_enabled():
        return False
    try:
        entry = _load_grants(_safe_scope(scope_id)).get(grant_key(rule, reason))
        if not isinstance(entry, dict):
            return False
        granted_at = float(entry.get("granted_at") or 0.0)
    except Exception:
        return False
    ttl = grant_max_age_s()
    return bool(granted_at) and (time.time() - granted_at) <= ttl


def add_grant(
    scope_id: str | None,
    rule: str,
    reason: str,
    *,
    session_id: str | None = None,
    granted_by: str | None = None,
) -> bool:
    """Memoreaza tiparul pentru tot firul. False daca nu s-a putut scrie."""
    if not grants_enabled():
        return False
    try:
        sid = _safe_scope(scope_id)
        ensure_dirs()
        path = grant_path(sid)
        doc = _read(path)
        if not isinstance(doc, dict) or not isinstance(doc.get("grants"), dict):
            doc = {"thread_id": sid, "created_at": time.time(), "grants": {}}
        doc["thread_id"] = sid
        doc["updated_at"] = time.time()
        doc["grants"][grant_key(rule, reason)] = {
            "rule": rule,
            "reason": reason,
            "granted_at": time.time(),
            "granted_by": granted_by,
            "session_id": session_id,
        }
        _write_atomic(path, doc)
        return True
    except Exception:
        return False


def list_grants(scope_id: str | None) -> list[dict]:
    """Aprobarile inca valabile ale unui fir, cele mai noi intai."""
    try:
        grants = _load_grants(_safe_scope(scope_id))
    except Exception:
        return []
    ttl = grant_max_age_s()
    now = time.time()
    out = [
        g for g in grants.values()
        if isinstance(g, dict) and 0 <= (now - float(g.get("granted_at") or 0.0)) <= ttl
    ]
    out.sort(key=lambda g: float(g.get("granted_at") or 0.0), reverse=True)
    return out


def clear_grants(scope_id: str | None) -> int:
    """Revoca tot ce a fost aprobat pentru fir. Returneaza cate a sters."""
    try:
        n = len(_load_grants(_safe_scope(scope_id)))
        grant_path(str(scope_id)).unlink()
        return n
    except FileNotFoundError:
        return 0
    except Exception:
        return 0


def cleanup_grants(max_age_s: float | None = None) -> int:
    """Sterge fisierele de aprobari expirate. Returneaza cate a sters."""
    ttl = grant_max_age_s() if max_age_s is None else float(max_age_s)
    n = 0
    now = time.time()
    try:
        for p in grants_dir().glob("*.json"):
            try:
                if now - p.stat().st_mtime > ttl:
                    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, ALLOW_SESSION)
    # `allow_session` este tot un `allow` pe disc; deosebirea sta in `scope`,
    # pe care hook-ul o citeste ca sa memoreze tiparul pentru tot firul.
    req["status"] = ALLOW if decision == ALLOW_SESSION else (decision if valid else DENY)
    req["decision"] = decision if valid else DENY
    req["scope"] = "thread" if decision == ALLOW_SESSION else "once"
    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)