"""backend/api/auth_guard.py — GAP-A6: Authorization model granulare. Definisce ruoli gerarchici + dependency FastAPI per proteggere gli endpoint. Ruoli (IntEnum, gerarchici): USER — nessuna auth (endpoint pubblici, chiamate frontend) MACHINE — X-Internal-Token header = INTERNAL_TOKEN env var (backend↔backend) OPERATOR — X-Operator-Token header = OPERATOR_TOKEN env var (monitoring, trigger) ADMIN — X-Admin-Token header = ADMIN_TOKEN env var (operazioni distruttive) Utilizzo: from api.auth_guard import require_role, AuthRole @router.delete("/tasks/{id}") async def delete_task(role: AuthRole = Depends(require_role(AuthRole.OPERATOR))): ... @router.post("/incidents/purge") async def purge(role: AuthRole = Depends(require_role(AuthRole.ADMIN))): ... Token env vars (Railway): INTERNAL_TOKEN — già usato in main.py (generato al boot se assente) OPERATOR_TOKEN — opzionale; se assente, endpoint OPERATOR bloccati ADMIN_TOKEN — opzionale; se assente, endpoint ADMIN bloccati NOTA: endpoint USER non richiedono alcun header. Se OPERATOR_TOKEN non è impostato, gli endpoint OPERATOR ritornano 503 invece di 403 (distinzione "non configurato" vs "token errato"). """ from __future__ import annotations import logging import os from enum import IntEnum from typing import Optional, Any from fastapi import Depends, Header, HTTPException, Request logger = logging.getLogger("agente_ai.auth_guard") # ── Rate limiter sliding window (RATE-LIMIT-FIX + GAP-AUTH-FIX) ────────────── # In-memory, per token hash (sicuro: non salva il token in chiaro). # Limiti per ruolo: USER=10/min, MACHINE=30/min, OPERATOR=60/min, ADMIN=illimitato. # Thread-safe in asyncio (GIL + nessun await interno). # # GAP-AUTH-FIX: per USER (role=0) la chiave usa client_ip invece di 'anonymous'. # Prima del fix: tutti gli utenti pubblici condividevano sha256("0:anonymous") → # un singolo client poteva svuotare il bucket per tutti gli utenti al mondo. # Fix: sha256("0:{client_ip}") — bucket separato per IP. # Fallback: 'unknown' se IP non rilevabile (raro su Railway/HF con proxy). import collections as _col # ── Redis-backed rate store (Upstash) — DoS-FIX: persiste cross-restart ──── # Fallback automatico a in-memory se UPSTASH_REDIS_URL non configurato. _upstash_url = os.getenv('UPSTASH_REDIS_URL', '') _upstash_token = os.getenv('UPSTASH_REDIS_TOKEN', '') _use_redis = bool(_upstash_url and _upstash_token) def _redis_rate_check(key: str, limit: int, window_s: int) -> tuple[bool, int]: """Rate check via Upstash Redis REST API (zero dipendenze extra).""" import urllib.request as _ur, json as _js, time as _rt now_s = int(_rt.time()) window_key = f'{key}:{now_s // window_s}' try: req = _ur.Request( f'{_upstash_url}/pipeline', data=_js.dumps([ ['INCR', window_key], ['EXPIRE', window_key, window_s * 2], ]).encode(), headers={'Authorization': f'Bearer {_upstash_token}', 'Content-Type': 'application/json'}, method='POST', ) with _ur.urlopen(req, timeout=1) as r: results = _js.loads(r.read()) count = results[0]['result'] if isinstance(results[0], dict) else results[0] if count > limit: return False, window_s return True, 0 except Exception as _exc: # SEC2-5: Redis irraggiungibile → fallback in-memory, non fail-open puro logger.warning('[auth_guard] Redis rate check fallito (%s), fallback in-memory', _exc) return _inmem_rate_check(key, limit, window_s) import time as _rl_time import hashlib as _rl_hash _RATE_LIMITS: dict[int, int] = { 0: 10, # USER 1: 30, # MACHINE 2: 60, # OPERATOR 3: -1, # ADMIN — illimitato } _RATE_WINDOW_S = 60 # finestra sliding 60s _rate_store: dict[str, _col.deque] = {} # token_hash → deque di timestamps # Lo store è usato anche quando Redis non è disponibile. Un client una tantum # lasciava una deque vuota nel dict per l'intera vita del processo. Eseguiamo uno # sweep ammortizzato: il lavoro resta O(1) per la quasi totalità delle richieste # e il numero di chiavi inattive rimane limitato al traffico tra due sweep. _RATE_STORE_SWEEP_EVERY = 128 _rate_store_checks = 0 def _prune_expired_rate_keys(now: float, window_s: float) -> None: """Rimuove bucket in-memory senza timestamp ancora nella finestra corrente.""" global _rate_store_checks _rate_store_checks += 1 if _rate_store_checks % _RATE_STORE_SWEEP_EVERY: return window_start = now - window_s stale_keys = [ stored_key for stored_key, timestamps in _rate_store.items() if not timestamps or timestamps[-1] < window_start ] for stored_key in stale_keys: _rate_store.pop(stored_key, None) def _rate_key(role: int, token_header: str | None, client_ip: str | None = None) -> str: """Chiave rate limiter: hash(role + discriminante) — non espone token né IP in chiaro. USER usa sempre l'IP come discriminante. MACHINE usa l'IP quando il proxy fidato lo inoltra: Cloudflare usa un unico token interno per tutti i browser, quindi il solo token renderebbe globale il limite di 30 richieste/minuto. In assenza di IP attestato, MACHINE conserva il fallback per-token. OPERATOR e ADMIN mantengono il bucket per-token. """ if role == 0: # USER: bucket per IP, mai globale condiviso. raw = f"0:{client_ip or 'unknown'}" elif role == 1 and client_ip: # MACHINE via proxy fidato: separa gli utenti dietro INTERNAL_TOKEN. raw = f"1:{client_ip}" else: # Chiamate server-to-server e ruoli elevati: bucket per token. raw = f"{role}:{token_header or 'anonymous'}" return _rl_hash.sha256(raw.encode()).hexdigest()[:16] def _inmem_rate_check(key: str, limit: int, window_s: float) -> tuple[bool, int]: """Rate check in-memory sliding window. SEC2-5: usata sia quando Redis non è configurato, sia come fallback quando Redis è temporaneamente irraggiungibile (prima era fail-open puro). Thread-safe in asyncio (GIL, nessun await interno). """ now = _rl_time.monotonic() window_start = now - window_s _prune_expired_rate_keys(now, window_s) if key not in _rate_store: _rate_store[key] = _col.deque() dq = _rate_store[key] while dq and dq[0] < window_start: dq.popleft() if len(dq) >= limit: retry_after = int(window_s - (now - dq[0])) + 1 return False, max(retry_after, 1) dq.append(now) return True, 0 def _check_rate_limit( role: int, token_header: str | None, client_ip: str | None = None, ) -> tuple[bool, int]: """Controlla il rate limit per ruolo. Ritorna (ok, retry_after_seconds). ok=True → richiesta consentita. ok=False → limite superato, retry_after = secondi alla prossima finestra. ADMIN (role=3) è sempre ok. """ limit = _RATE_LIMITS.get(role, 10) if limit < 0: return True, 0 # ADMIN — illimitato key = _rate_key(role, token_header, client_ip) # DoS-FIX: usa Redis se disponibile (persiste cross-restart HF Space) # SEC2-5: se Redis fallisce, _redis_rate_check fa fallback su _inmem_rate_check if _use_redis: return _redis_rate_check(key, limit, int(_RATE_WINDOW_S)) return _inmem_rate_check(key, limit, int(_RATE_WINDOW_S)) async def require_supabase_user(request: Request) -> dict[str, Any]: """Valida il Bearer JWT tramite Supabase Auth e restituisce il profilo minimo. La chiave Supabase resta server-side; il JWT arriva esclusivamente nell'header Authorization del chiamante e non viene scritto nei log. """ import httpx authorization = request.headers.get("Authorization", "") if not authorization.lower().startswith("bearer "): raise HTTPException(status_code=401, detail="Bearer token richiesto") jwt = authorization[7:].strip() if not jwt: raise HTTPException(status_code=401, detail="Bearer token non valido") supabase_url = os.getenv("SUPABASE_URL", "").rstrip("/") api_key = os.getenv("SUPABASE_SERVICE_ROLE_KEY") or os.getenv("SUPABASE_KEY", "") if not supabase_url or not api_key: raise HTTPException(status_code=503, detail="Autenticazione Supabase non configurata") try: async with httpx.AsyncClient(timeout=5) as client: response = await client.get( f"{supabase_url}/auth/v1/user", headers={ "apikey": api_key, "Authorization": f"Bearer {jwt}", "Accept": "application/json", }, ) except httpx.HTTPError as exc: logger.warning("supabase user validation unavailable: %s", type(exc).__name__) raise HTTPException(status_code=503, detail="Autenticazione temporaneamente non disponibile") from exc if response.status_code != 200: raise HTTPException(status_code=401, detail="Sessione Supabase non valida o scaduta") try: user = response.json() except ValueError as exc: raise HTTPException(status_code=401, detail="Risposta autenticazione non valida") from exc if not isinstance(user, dict) or not user.get("id"): raise HTTPException(status_code=401, detail="Utente Supabase non valido") return user async def require_admin_user(request: Request) -> dict[str, Any]: """Richiede un JWT Supabase con app_metadata.role=admin. app_metadata è server-controlled; user_metadata non viene mai considerato per autorizzare l’area amministrativa. """ user = await require_supabase_user(request) app_metadata = user.get("app_metadata") or {} roles = app_metadata.get("roles") or [] is_admin = app_metadata.get("role") == "admin" or "admin" in roles if not is_admin: raise HTTPException(status_code=403, detail="Membership amministrativa richiesta") return user class AuthRole(IntEnum): """Gerarchia ruoli: USER < MACHINE < OPERATOR < ADMIN.""" USER = 0 MACHINE = 1 OPERATOR = 2 ADMIN = 3 def _get_token(env_var: str) -> str: return os.getenv(env_var, "").strip() async def _resolve_role( x_internal_token: Optional[str] = Header(None, alias="X-Internal-Token"), x_machine_token: Optional[str] = Header(None, alias="X-Machine-Token"), x_operator_token: Optional[str] = Header(None, alias="X-Operator-Token"), x_admin_token: Optional[str] = Header(None, alias="X-Admin-Token"), ) -> AuthRole: """Risolve il ruolo del chiamante in base agli header presenti.""" import secrets as _sec_comp # ADMIN (massima priorità) admin_tok = _get_token("ADMIN_TOKEN") if admin_tok and x_admin_token and _sec_comp.compare_digest(x_admin_token, admin_tok): logger.debug("auth: ADMIN role granted") return AuthRole.ADMIN # OPERATOR op_tok = _get_token("OPERATOR_TOKEN") if op_tok and x_operator_token and _sec_comp.compare_digest(x_operator_token, op_tok): logger.debug("auth: OPERATOR role granted") return AuthRole.OPERATOR # MACHINE: supporta entrambi gli header per compatibilità tra runner e backend. # Il valore resta confrontato esclusivamente con il secret server-side. int_tok = _get_token("INTERNAL_TOKEN") or _get_token("MACHINE_TOKEN") machine_header = x_internal_token or x_machine_token if int_tok and machine_header and _sec_comp.compare_digest(machine_header, int_tok): logger.debug("auth: MACHINE role granted") return AuthRole.MACHINE # Nessun token valido → ruolo USER (minimo) return AuthRole.USER async def require_private_state_machine( request: 'Request', x_internal_token: Optional[str] = Header(None, alias="X-Internal-Token"), ) -> AuthRole: """Autorizza esclusivamente il proxy Pages dello stato privato. Usa un token dedicato per non ruotare o esporre ``INTERNAL_TOKEN``, da cui dipendono le integrazioni legacy del master B. Il token non conferisce un ruolo più ampio del canale MACHINE e resta soggetto allo stesso rate limit. """ import secrets as _sec_comp private_token = _get_token("PRIVATE_STATE_INTERNAL_TOKEN") if not private_token: raise HTTPException(status_code=503, detail="Canale stato privato non configurato") if not x_internal_token or not _sec_comp.compare_digest(x_internal_token, private_token): raise HTTPException(status_code=403, detail="Permessi insufficienti per lo stato privato") client_ip = ( request.headers.get('X-Forwarded-For', '').split(',')[0].strip() or request.headers.get('X-Real-IP', '') or (request.client.host if request.client else None) ) or None allowed, retry_after = _check_rate_limit(int(AuthRole.MACHINE), x_internal_token, client_ip) if not allowed: raise HTTPException( status_code=429, detail="Rate limit stato privato superato", headers={'Retry-After': str(retry_after)}, ) return AuthRole.MACHINE def require_role(min_role: AuthRole): """ FastAPI Depends factory per autorizzazione granulare. Esempio: @router.post("/trigger") async def trigger(role: AuthRole = Depends(require_role(AuthRole.OPERATOR))): ... Se il ruolo risolto < min_role → 403 Forbidden. Se il token richiesto non è configurato (env var assente) → 503 Service Unavailable. """ async def _check( request: 'Request', resolved: AuthRole = Depends(_resolve_role), ) -> AuthRole: # RATE-LIMIT-FIX + GAP-AUTH-FIX: estrai IP cliente per bucket USER per-IP _token_hdr = ( request.headers.get('X-Admin-Token') or request.headers.get('X-Operator-Token') or request.headers.get('X-Internal-Token') or request.headers.get('X-Machine-Token') ) # GAP-AUTH-FIX: estrai IP reale (Railway/HF dietro proxy → X-Forwarded-For) _client_ip: str | None = ( request.headers.get('X-Forwarded-For', '').split(',')[0].strip() or request.headers.get('X-Real-IP', '') or (request.client.host if request.client else None) ) or None _ok, _retry = _check_rate_limit(int(resolved), _token_hdr, _client_ip) if not _ok: raise HTTPException( status_code=429, detail={ 'error': 'Rate limit superato', 'role': resolved.name, 'limit': f'{_RATE_LIMITS.get(int(resolved), 10)}/min', 'retry_after_seconds': _retry, }, headers={'Retry-After': str(_retry)}, ) if resolved < min_role: # Distingue "token non configurato" (503) da "token errato" (403) if min_role == AuthRole.OPERATOR and not _get_token("OPERATOR_TOKEN"): raise HTTPException(503, { "error": "Endpoint non disponibile", "reason": "OPERATOR_TOKEN non configurato su Railway", "required": "AuthRole.OPERATOR", }) if min_role == AuthRole.ADMIN and not _get_token("ADMIN_TOKEN"): raise HTTPException(503, { "error": "Endpoint non disponibile", "reason": "ADMIN_TOKEN non configurato su Railway", "required": "AuthRole.ADMIN", }) raise HTTPException(403, { "error": "Permessi insufficienti", "required_role": min_role.name, "your_role": resolved.name, "hint": f"Fornire header X-{min_role.name.capitalize()}-Token con il token corretto", }) return resolved return _check # ── Convenienza: dependency per endpoint pubblici (nessun controllo) ───────── async def any_role(resolved: AuthRole = Depends(_resolve_role)) -> AuthRole: """Dependency che accetta qualsiasi ruolo (incluso USER senza token). Usare per endpoint pubblici che vogliono loggare il ruolo del chiamante.""" return resolved