ObsiGate/backend/draco_session.py

# backend/draco_session.py
# Fusion ACC-3.1 (#216) — Émettrice SSO « draco_session » (option 2a, C2).
#
# ObsiGate est l'unique émettrice du cookie de session partagé
# ``draco_session`` (Domain=.dracodev.net) que les autres applications
# de l'écosystème (Flowdeck, TikTak, Kanbante) valideront localement
# (ACC-3.2, leurs propres tickets). Design : architecture Fusion §5.2 —
# JWS compact HS256, claims minimaux (sub, name, iat, exp 12 h, jti),
# aucun rôle dans le jeton, renouvellement glissant par l'émettrice,
# révocation par jti exposée aux autres apps.
#
# Configuration (environnement) :
#   DRACO_SESSION_SECRET        secret de signature partagé (obligatoire
#                               pour activer la fonction ; sans lui,
#                               aucun cookie n'est émis et rien ne change)
#   DRACO_SESSION_SECRET_PREV   ancien secret accepté en vérification
#                               pendant la fenêtre de rotation (24 h)
#   DRACO_SESSION_COOKIE_DOMAIN domaine du cookie (défaut .dracodev.net ;
#                               vide = cookie lié à l'hôte, utile en dev)
#   DRACO_SSO_MACHINE_KEY       clé machine protégeant la lecture de la
#                               liste de révocation (GET /api/sso/revocations)

from __future__ import annotations

import json
import logging
import os
import threading
import time
import uuid
from pathlib import Path
from urllib.parse import urlparse

from jose import JWTError, jwt

logger = logging.getLogger("obsigate.draco_session")

COOKIE_NAME = "draco_session"
ALGORITHM = "HS256"
TOKEN_TTL_SECONDS = 12 * 3600          # durée de vie du jeton (design §5.2)
RENEWAL_WINDOW_SECONDS = 2 * 3600      # réémettre si exp - now < 2 h
REVOKED_FILE = Path("data/draco_revoked.json")
ALLOWED_RETURN_SUFFIX = "dracodev.net"

_lock = threading.Lock()
_revoked: dict[str, int] | None = None  # jti -> exp (chargé paresseusement)


def _secret() -> str:
    return os.environ.get("DRACO_SESSION_SECRET", "")


def _secret_prev() -> str:
    return os.environ.get("DRACO_SESSION_SECRET_PREV", "")


def _cookie_domain() -> str:
    return os.environ.get("DRACO_SESSION_COOKIE_DOMAIN", ".dracodev.net")


def enabled() -> bool:
    """La fonction est active seulement si le secret partagé est configuré."""
    return bool(_secret())


# ── Révocation (jti) ────────────────────────────────────────────────────

def _load_revoked() -> dict[str, int]:
    global _revoked
    if _revoked is None:
        try:
            data = json.loads(REVOKED_FILE.read_text(encoding="utf-8"))
            _revoked = {str(k): int(v) for k, v in data.items()} \
                if isinstance(data, dict) else {}
        except (OSError, ValueError):
            _revoked = {}
    return _revoked


def _save_revoked() -> None:
    REVOKED_FILE.parent.mkdir(parents=True, exist_ok=True)
    tmp = REVOKED_FILE.with_suffix(".tmp")
    tmp.write_text(json.dumps(_revoked or {}), encoding="utf-8")
    tmp.replace(REVOKED_FILE)


def revoke(jti: str, exp: int) -> None:
    """Révoque un jeton par son jti (déconnexion globale)."""
    with _lock:
        store = _load_revoked()
        store[jti] = int(exp)
        now = int(time.time())
        for k in [k for k, v in store.items() if v <= now]:
            del store[k]                     # purge des expirés
        _save_revoked()


def is_revoked(jti: str) -> bool:
    with _lock:
        return jti in _load_revoked()


def revoked_list() -> list[dict]:
    """Liste des révocations actives (pour les autres apps, cache 60 s)."""
    with _lock:
        now = int(time.time())
        return [{"jti": jti, "exp": exp}
                for jti, exp in _load_revoked().items() if exp > now]


# ── Émission / vérification ─────────────────────────────────────────────

def issue_token(user: dict) -> tuple[str, str, int]:
    """Émet un JWS pour l'utilisateur. Retourne (jeton, jti, exp)."""
    now = int(time.time())
    exp = now + TOKEN_TTL_SECONDS
    jti = uuid.uuid4().hex
    claims = {
        "sub": str(user.get("id") or user.get("username") or ""),
        "name": str(user.get("display_name") or user.get("username") or ""),
        "iat": now,
        "exp": exp,
        "jti": jti,
    }
    return jwt.encode(claims, _secret(), algorithm=ALGORITHM), jti, exp


def decode_claims(token: str) -> dict | None:
    """Vérifie signature et expiration (pas la révocation)."""
    if not token or not enabled():
        return None
    for secret in (_secret(), _secret_prev()):
        if not secret:
            continue
        try:
            return jwt.decode(token, secret, algorithms=[ALGORITHM])
        except JWTError:
            continue
    return None


def verify_token(token: str) -> dict | None:
    """decode_claims + contrôle de révocation du jti."""
    claims = decode_claims(token)
    if claims is None:
        return None
    if claims.get("jti") and is_revoked(str(claims["jti"])):
        return None
    return claims


def needs_renewal(claims: dict) -> bool:
    """Vrai si le jeton entre dans la fenêtre de renouvellement (< 2 h)."""
    try:
        return int(claims["exp"]) - int(time.time()) < RENEWAL_WINDOW_SECONDS
    except (KeyError, TypeError, ValueError):
        return True


def set_cookie(response, user: dict, secure: bool) -> None:
    """Pose le cookie draco_session sur une réponse (émission initiale)."""
    if not enabled():
        return
    token, _jti, _exp = issue_token(user)
    domain = _cookie_domain()
    extra = {"domain": domain} if domain else {}
    response.set_cookie(COOKIE_NAME, token, max_age=TOKEN_TTL_SECONDS,
                        httponly=True, samesite="lax", secure=secure,
                        path="/", **extra)


def clear_cookie(response) -> None:
    domain = _cookie_domain()
    extra = {"domain": domain} if domain else {}
    response.delete_cookie(COOKIE_NAME, path="/", **extra)


# ── Renouvellement glissant (middleware) ────────────────────────────────

class DracoRenewalMiddleware:
    """Réémet le cookie quand l'utilisateur est actif chez l'émettrice.

    Middleware ASGI pur (pas de BaseHTTPMiddleware : la pile d'ObsiGate
    streame des réponses SSE/WS que BaseHTTPMiddleware mettrait en
    tampon). On enveloppe l'envoi de la réponse : si la requête porte
    un cookie d'accès natif valide et que le cookie draco est absent,
    invalide ou dans sa fenêtre de renouvellement (< 2 h), on ajoute
    un Set-Cookie frais aux en-têtes de la réponse. La révocation
    ciblée d'un jti draco est ainsi « guérie » par la prochaine
    activité authentifiée locale — la déconnexion globale, elle,
    révoque aussi les jetons natifs, donc plus rien n'est réémis.
    """

    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        if scope.get("type") != "http" or not enabled():
            await self.app(scope, receive, send)
            return
        cookie_header = ""
        for name, value in scope.get("headers") or []:
            if name == b"cookie":
                cookie_header = value.decode("latin-1")
                break
        cookies = {}
        for part in cookie_header.split(";"):
            if "=" in part:
                k, v = part.strip().split("=", 1)
                cookies[k] = v
        user = self._user_from_native(cookies.get("access_token", ""))
        if user is None:
            await self.app(scope, receive, send)
            return
        claims = verify_token(cookies.get(COOKIE_NAME, "")) \
            if cookies.get(COOKIE_NAME) else None
        if claims is not None and not needs_renewal(claims):
            await self.app(scope, receive, send)
            return

        token, _jti, _exp = issue_token(user)
        domain = _cookie_domain()
        cookie = (f"{COOKIE_NAME}={token}; Max-Age={TOKEN_TTL_SECONDS}; "
                  f"Path=/; HttpOnly; SameSite=lax")
        if domain:
            cookie += f"; Domain={domain}"
        if scope.get("scheme") == "https":
            cookie += "; Secure"

        async def send_with_cookie(message):
            if message.get("type") == "http.response.start":
                headers = list(message.get("headers") or [])
                headers.append((b"set-cookie", cookie.encode("latin-1")))
                message = {**message, "headers": headers}
            await send(message)

        await self.app(scope, receive, send_with_cookie)

    @staticmethod
    def _user_from_native(access_token: str):
        """Utilisateur local prouvé par le cookie d'accès natif, ou None."""
        if not access_token:
            return None
        try:
            from backend.auth.jwt_handler import decode_token
            from backend.auth.user_store import get_user
            payload = decode_token(access_token)
            if not payload or payload.get("type") != "access":
                return None
            return get_user(str(payload.get("sub", "")))
        except Exception:  # le renouvellement ne casse jamais une requête
            return None


# ── return_to (anti open-redirect) ──────────────────────────────────────

def validate_return_to(url: str) -> str | None:
    """Accepte uniquement https vers dracodev.net ou un sous-domaine."""
    if not url:
        return None
    parsed = urlparse(url)
    host = (parsed.hostname or "").lower()
    if parsed.scheme != "https" or not host:
        return None
    if host == ALLOWED_RETURN_SUFFIX or host.endswith("." + ALLOWED_RETURN_SUFFIX):
        return url
    return None