api/auth_guard.py

"""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