""" backend/api/policy.py — Policy Engine (ARCH-K2.4) Gestisce centralmente Authorization, Budget, Quota, Sandbox, Retry e Timeout per ogni tool call / task submission. Il Kernel (ARCH-K2.1) consulta questo modulo prima di eseguire qualsiasi operazione. Endpoints (auth: MACHINE): GET /api/policy/rules — lista regole policy per risk level GET /api/policy/budget — stato budget provider (reale, da memoria) POST /api/policy/budget/record — registra utilizzo provider (chiamato dal loop LLM) POST /api/policy/check — valuta se un tool/task è autorizzato GET /api/policy/quota/{sid} — stato quota per sessione POST /api/policy/quota/reset — reset quota sessione (OPERATOR) Invarianti rispettate: - Budget check fail-open: se Supabase non risponde, non blocca (log warning) - Quota sliding window: 60s — senza stato persistente non bloccante - Timeout per risk level: safe=30s, medium=90s, risky=180s, dangerous=300s - Retry per risk level: safe=3, medium=2, risky=1, dangerous=0 - Tool "dangerous" richiede sempre conferma esplicita (caller_confirmed=True) """ from __future__ import annotations import logging import time from collections import defaultdict, deque from typing import Any, Deque, Dict, List, Optional from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel, Field from .auth_guard import AuthRole, require_role from .state import sb _logger = logging.getLogger("api.policy") # Safety cap per provider sconosciuto: evita che una riga budget assente # venga interpretata come accesso illimitato o blocco permanente. _DEFAULT_PROVIDER_LIMIT_USD = 1.0 # ── Router ───────────────────────────────────────────────────────────────────── router = APIRouter( prefix="/api/policy", tags=["policy"], dependencies=[Depends(require_role(AuthRole.MACHINE))], ) # ── Risk levels e policy statiche ───────────────────────────────────────────── RISK_TIMEOUT_S: Dict[str, int] = { "safe": 30, "medium": 90, "risky": 180, "dangerous": 300, } RISK_MAX_RETRY: Dict[str, int] = { "safe": 3, "medium": 2, "risky": 1, "dangerous": 0, # nessun retry automatico su azioni distruttive } RISK_SANDBOX: Dict[str, bool] = { "safe": False, # no sandbox necessario "medium": False, "risky": True, # sandboxed execution "dangerous": True, } POLICY_RULES: List[Dict[str, Any]] = [ # ── Safe ────────────────────────────────────────────────────────────────── {"tool": "web_search", "risk": "safe", "label": "Ricerca web", "description": "Solo lettura"}, {"tool": "read_page", "risk": "safe", "label": "Leggi pagina web", "description": "Fetch URL"}, {"tool": "recall", "risk": "safe", "label": "Recupera memoria", "description": "Lettura memoria"}, {"tool": "read_file", "risk": "safe", "label": "Leggi file", "description": "VFS read-only"}, {"tool": "search_github", "risk": "safe", "label": "Cerca GitHub", "description": "API GitHub read"}, {"tool": "get_weather", "risk": "safe", "label": "Meteo", "description": "API meteo"}, {"tool": "get_currency", "risk": "safe", "label": "Cambio valuta", "description": "API valuta"}, {"tool": "get_news", "risk": "safe", "label": "Notizie", "description": "API news"}, {"tool": "search_wikipedia", "risk": "safe", "label": "Wikipedia", "description": "Lettura"}, {"tool": "run_code", "risk": "safe", "label": "Esegui codice", "description": "Sandbox browser"}, {"tool": "list_files", "risk": "safe", "label": "Lista file", "description": "VFS dir listing"}, # ── Medium ──────────────────────────────────────────────────────────────── {"tool": "write_file", "risk": "medium", "label": "Scrivi file", "description": "VFS write"}, {"tool": "remember", "risk": "medium", "label": "Salva in memoria", "description": "Aggiorna memoria"}, {"tool": "pip_install", "risk": "medium", "label": "Installa pacchetti", "description": "pip install"}, {"tool": "propose_action", "risk": "medium", "label": "Proposta azione", "description": "UI only"}, {"tool": "send_email", "risk": "medium", "label": "Invia email", "description": "SMTP"}, {"tool": "api_call", "risk": "medium", "label": "Chiamata API", "description": "HTTP request"}, # ── Risky ───────────────────────────────────────────────────────────────── {"tool": "execute_shell", "risk": "risky", "label": "Esegui shell", "description": "Comando backend"}, {"tool": "push_github", "risk": "risky", "label": "Push GitHub", "description": "Git push"}, {"tool": "deploy", "risk": "risky", "label": "Deploy", "description": "Deploy produzione"}, {"tool": "install_package", "risk": "risky", "label": "Installa sistema", "description": "apt/brew"}, {"tool": "modify_config", "risk": "risky", "label": "Modifica config", "description": "File configurazione"}, # ── Dangerous ───────────────────────────────────────────────────────────── {"tool": "delete_file", "risk": "dangerous", "label": "Elimina file", "description": "rm irreversibile"}, {"tool": "drop_table", "risk": "dangerous", "label": "Drop tabella DB", "description": "DDL distruttivo"}, {"tool": "purge_memory", "risk": "dangerous", "label": "Svuota memoria", "description": "Reset totale"}, {"tool": "overwrite_file", "risk": "dangerous", "label": "Sovrascrivi file", "description": "Sovrascrittura"}, {"tool": "reset_session", "risk": "dangerous", "label": "Reset sessione", "description": "Dati sessione persi"}, ] _RULE_MAP: Dict[str, Dict[str, Any]] = {r["tool"]: r for r in POLICY_RULES} _DEFAULT_RULE: Dict[str, Any] = { "tool": "_unknown", "risk": "risky", "label": "Azione sconosciuta", "description": "Tool non registrato — trattato come risky per sicurezza", } # ── Budget store (in-memory, aggiornato da /budget/record) ─────────────────── # Struttura: { provider: { limit: float, used: float, currency: str } } _BUDGET: Dict[str, Dict[str, Any]] = { "openai": {"limit": 10.0, "used": 0.0, "currency": "USD"}, "groq": {"limit": 0.0, "used": 0.0, "currency": "USD"}, # free tier "openrouter": {"limit": 10.0, "used": 0.0, "currency": "USD"}, "anthropic": {"limit": 10.0, "used": 0.0, "currency": "USD"}, "gemini": {"limit": 0.0, "used": 0.0, "currency": "USD"}, # free tier "sambanova": {"limit": 0.0, "used": 0.0, "currency": "USD"}, # free tier "cerebras": {"limit": 0.0, "used": 0.0, "currency": "USD"}, # free tier } # ── Quota store — sliding window 60s per (session_id, tool) ────────────────── # Struttura: { (session_id, tool): deque[ts, ...] } _QUOTA_WINDOW_S = 60 _QUOTA_LIMITS: Dict[str, int] = { "safe": 60, # max 60 chiamate/min "medium": 20, "risky": 5, "dangerous": 1, } _quota_store: Dict[tuple, Deque[float]] = defaultdict(deque) # ── Pydantic models ─────────────────────────────────────────────────────────── class ToolPolicy(BaseModel): tool: str risk: str label: str description: str timeout_s: int max_retry: int sandbox: bool class BudgetStatus(BaseModel): provider: str limit: float used: float remaining: float exhausted: bool currency: str = "USD" class BudgetRecordRequest(BaseModel): provider: str cost_usd: float = Field(ge=0.0) model: Optional[str] = None tokens: Optional[int] = None class PolicyCheckRequest(BaseModel): tool: str args: Dict[str, Any] = {} session_id: str = "default" caller_confirmed: bool = False # True se l'utente ha confermato esplicitamente class PolicyCheckResult(BaseModel): tool: str risk: str label: str allowed: bool requires_confirm: bool reason: Optional[str] = None timeout_s: int max_retry: int sandbox: bool quota_remaining: int budget_ok: bool class QuotaStatus(BaseModel): session_id: str calls: Dict[str, int] # tool → calls in window limits: Dict[str, int] # risk → limit # ── Helpers ─────────────────────────────────────────────────────────────────── def _get_rule(tool: str) -> Dict[str, Any]: return _RULE_MAP.get(tool, _DEFAULT_RULE) def _quota_check(session_id: str, tool: str, risk: str) -> tuple[bool, int]: """ Sliding window quota check. Ritorna (allowed, remaining_in_window). """ key = (session_id, tool) now = time.time() dq = _quota_store[key] limit = _QUOTA_LIMITS.get(risk, 5) # Rimuovi timestamp fuori dalla finestra while dq and dq[0] < now - _QUOTA_WINDOW_S: dq.popleft() remaining = max(0, limit - len(dq)) return remaining > 0, remaining def _quota_consume(session_id: str, tool: str) -> None: _quota_store[(session_id, tool)].append(time.time()) def _budget_ok(tool: str) -> bool: """ True se nessun provider con limite >0 è esaurito. Fail-open: se non ci sono provider con limite impostato → OK. """ for info in _BUDGET.values(): if info["limit"] > 0 and info["used"] >= info["limit"]: return False return True async def _sync_budget_from_supabase() -> None: """Carica usage da Supabase all'avvio (best-effort, silenzioso in caso di errore).""" try: client = sb() res = client.table("provider_budget") \ .select("provider,used,limit,currency") \ .execute() if res.data: for row in res.data: p = row.get("provider", "") if p in _BUDGET: _BUDGET[p]["used"] = float(row.get("used", 0)) # Supabase può contenere vecchie righe con limit=0: non # deve disattivare accidentalmente i limiti paid hardcoded. stored_limit = float(row.get("limit", 0)) if stored_limit > 0 or _BUDGET[p]["limit"] <= 0: _BUDGET[p]["limit"] = stored_limit _BUDGET[p]["currency"] = str(row.get("currency", "USD")) except Exception as exc: _logger.debug("[policy] Sync budget Supabase fallito (non bloccante): %s", exc) # ── Endpoints ───────────────────────────────────────────────────────────────── @router.get("/rules", response_model=List[ToolPolicy]) async def get_policy_rules() -> List[ToolPolicy]: """Lista completa delle regole policy con timeout/retry/sandbox per ogni tool.""" return [ ToolPolicy( tool=r["tool"], risk=r["risk"], label=r["label"], description=r["description"], timeout_s=RISK_TIMEOUT_S.get(r["risk"], 60), max_retry=RISK_MAX_RETRY.get(r["risk"], 1), sandbox=RISK_SANDBOX.get(r["risk"], False), ) for r in POLICY_RULES ] @router.get("/budget", response_model=List[BudgetStatus]) async def get_budget_status() -> List[BudgetStatus]: """Stato budget provider aggiornato (in-memory, sincronizzato con Supabase al boot).""" await _sync_budget_from_supabase() return [ BudgetStatus( provider=provider, limit=info["limit"], used=round(info["used"], 6), remaining=round(max(0.0, info["limit"] - info["used"]), 6), exhausted=(info["limit"] > 0 and info["used"] >= info["limit"]), currency=info.get("currency", "USD"), ) for provider, info in _BUDGET.items() ] @router.post("/budget/record", status_code=200) async def record_budget_usage(req: BudgetRecordRequest) -> Dict[str, Any]: """ Registra utilizzo provider dopo una chiamata LLM. Aggiorna budget in-memory e persiste su Supabase fire-and-forget. Chiamato dal loop LLM / providerBridge dopo ogni risposta. """ provider = req.provider.lower() if provider not in _BUDGET: _BUDGET[provider] = {"limit": _DEFAULT_PROVIDER_LIMIT_USD, "used": 0.0, "currency": "USD"} _BUDGET[provider]["used"] = round(_BUDGET[provider]["used"] + req.cost_usd, 6) new_used = _BUDGET[provider]["used"] # Persisti su Supabase (fire-and-forget) try: client = sb() client.table("provider_budget").upsert({ "provider": provider, "used": new_used, "limit": _BUDGET[provider]["limit"], "currency": _BUDGET[provider].get("currency", "USD"), "updated_at": time.time(), }, on_conflict="provider").execute() except Exception as exc: _logger.debug("[policy] Budget persist Supabase fallito (non bloccante): %s", exc) return { "provider": provider, "cost_usd": req.cost_usd, "total_used": new_used, "exhausted": (_BUDGET[provider]["limit"] > 0 and new_used >= _BUDGET[provider]["limit"]), } @router.post("/check", response_model=PolicyCheckResult) async def check_tool_call(req: PolicyCheckRequest) -> PolicyCheckResult: """ Valuta se un tool call è autorizzato secondo Authorization, Budget, Quota. Il Kernel chiama questo endpoint prima di ogni task submission (ARCH-K2.4). Logica: 1. Authorization: tool "dangerous" richiede caller_confirmed=True 2. Budget: se qualsiasi provider con limite ha used >= limit → blocca 3. Quota: sliding window 60s per (session_id, tool) """ rule = _get_rule(req.tool) risk = rule["risk"] timeout = RISK_TIMEOUT_S.get(risk, 60) retry = RISK_MAX_RETRY.get(risk, 1) sandbox = RISK_SANDBOX.get(risk, False) # 1. Authorization check — dangerous richiede conferma esplicita if risk == "dangerous" and not req.caller_confirmed: return PolicyCheckResult( tool=req.tool, risk=risk, label=rule["label"], allowed=False, requires_confirm=True, reason="Azione dangerous: richiede caller_confirmed=True (conferma utente esplicita)", timeout_s=timeout, max_retry=retry, sandbox=sandbox, quota_remaining=0, budget_ok=True, ) # 2. Budget check (fail-open: se errore DB → allowed) budget_ok = _budget_ok(req.tool) if not budget_ok: return PolicyCheckResult( tool=req.tool, risk=risk, label=rule["label"], allowed=False, requires_confirm=False, reason="Budget LLM esaurito — aggiorna i limiti in /api/policy/budget", timeout_s=timeout, max_retry=retry, sandbox=sandbox, quota_remaining=0, budget_ok=False, ) # 3. Quota check quota_ok, remaining = _quota_check(req.session_id, req.tool, risk) if not quota_ok: return PolicyCheckResult( tool=req.tool, risk=risk, label=rule["label"], allowed=False, requires_confirm=False, reason=f"Quota sessione esaurita — max {_QUOTA_LIMITS.get(risk, 5)} chiamate/min per tool '{req.tool}'", timeout_s=timeout, max_retry=retry, sandbox=sandbox, quota_remaining=0, budget_ok=True, ) # ✅ Autorizzato — consuma quota e ritorna policy _quota_consume(req.session_id, req.tool) return PolicyCheckResult( tool=req.tool, risk=risk, label=rule["label"], allowed=True, requires_confirm=(risk in ("risky", "dangerous")), reason=None, timeout_s=timeout, max_retry=retry, sandbox=sandbox, quota_remaining=remaining - 1, budget_ok=True, ) @router.get("/quota/{session_id}", response_model=QuotaStatus) async def get_quota_status(session_id: str) -> QuotaStatus: """Stato quota sliding-window per una sessione.""" now = time.time() calls = {} for (sid, tool), dq in _quota_store.items(): if sid != session_id: continue active = sum(1 for ts in dq if ts >= now - _QUOTA_WINDOW_S) if active > 0: calls[tool] = active return QuotaStatus( session_id=session_id, calls=calls, limits={risk: lim for risk, lim in _QUOTA_LIMITS.items()}, ) @router.post( "/quota/reset", dependencies=[Depends(require_role(AuthRole.OPERATOR))], status_code=200, ) async def reset_quota(session_id: str) -> Dict[str, Any]: """Reset quota sliding-window per una sessione (OPERATOR only).""" keys_removed = [k for k in list(_quota_store.keys()) if k[0] == session_id] for k in keys_removed: del _quota_store[k] return {"session_id": session_id, "cleared_tools": len(keys_removed)}