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)
#
# Registre des émissions (HUB-13, fusion #127) : chaque jti émis est
# consigné par `sub` dans data/draco_issued.json (jti -> exp, purgé à
# l'expiration) pour que l'administration des comptes puisse révoquer
# toutes les sessions draco d'un utilisateur (désactivation,
# suppression, réinitialisation du mot de passe) via revoke().
# HUB-22 (#227) ajoute la lecture de ce registre
# (issued_sessions_for_user) et la révocation unitaire contrôlée par
# appartenance (revoke_one_for_user), exposées par l'API machine des
# sessions (backend/routers/comptes_machine.py).
#
# Rôles du Hub (HUB-19, #228) : à l'ouverture de session draco,
# `user_for_draco` consulte en plus le registre des rôles du Hub
# (`backend/hub_roles.py`, contrat D3(a) posé par le pilote ShaarIt) :
# niveau connu ``aucun`` → la session draco est refusée (l'authenti-
# fication native d'ObsiGate, émettrice, n'est jamais touchée) ;
# niveau connu → il est joint au dict utilisateur (``_hub_role``)
# pour que les routes d'administration du canal draco exigent
# ``administrateur`` ; rôle inconnu (Hub injoignable) → comportement
# actuel, aucun verrouillage nouveau. Aucun rôle dans le jeton :
# validation locale, émission et révocations sont inchangées.

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")
ISSUED_FILE = Path("data/draco_issued.json")
ALLOWED_RETURN_SUFFIX = "dracodev.net"

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


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]


# ── Registre des émissions (par utilisateur) ────────────────────────────

def _load_issued() -> dict[str, dict[str, int]]:
    global _issued
    if _issued is None:
        try:
            data = json.loads(ISSUED_FILE.read_text(encoding="utf-8"))
            _issued = {
                str(sub): {str(j): int(e) for j, e in tokens.items()}
                for sub, tokens in data.items()
                if isinstance(tokens, dict)
            } if isinstance(data, dict) else {}
        except (OSError, ValueError):
            _issued = {}
    return _issued


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


def _register_issued(sub: str, jti: str, exp: int) -> None:
    """Consigne un jti émis pour ``sub`` (purge des expirés au passage)."""
    if not sub:
        return
    try:
        with _lock:
            store = _load_issued()
            now = int(time.time())
            for key in list(store.keys()):
                store[key] = {j: e for j, e in store[key].items() if e > now}
                if not store[key]:
                    del store[key]
            store.setdefault(sub, {})[jti] = int(exp)
            _save_issued()
    except Exception:  # le registre ne casse jamais une émission
        logger.warning("Registre des émissions draco non écrit", exc_info=True)


def revoke_all_for_user(sub: str) -> int:
    """Révoque tous les jti draco actifs émis pour ``sub``.

    Retourne le nombre de jti révoqués. Utilisé par l'administration
    des comptes (HUB-13) : désactivation, suppression, réinitialisation
    du mot de passe.
    """
    if not sub:
        return 0
    with _lock:
        store = _load_issued()
        tokens = store.pop(str(sub), {})
        now = int(time.time())
        actifs = {j: e for j, e in tokens.items() if e > now}
        if tokens:
            _save_issued()
    for jti, exp in actifs.items():
        revoke(jti, exp)
    return len(actifs)


def issued_sessions_for_user(sub: str) -> list[dict]:
    """Sessions draco actives émises pour ``sub`` (HUB-22, #227).

    Retourne ``[{jti, exp, revoquee}]`` triée par ``exp`` croissant,
    où ``revoquee`` indique si le ``jti`` figure déjà dans la liste des
    révocations. Les entrées expirées de cet utilisateur sont purgées
    du registre au passage (même patron que ``_register_issued``).
    """
    if not sub:
        return []
    with _lock:
        store = _load_issued()
        now = int(time.time())
        tokens = store.get(str(sub), {})
        actifs = {j: e for j, e in tokens.items() if e > now}
        if len(actifs) != len(tokens):
            if actifs:
                store[str(sub)] = actifs
            else:
                store.pop(str(sub), None)
            _save_issued()
        revoques = _load_revoked()
        entrees = sorted(actifs.items(), key=lambda kv: kv[1])
    return [{"jti": jti, "exp": exp, "revoquee": jti in revoques}
            for jti, exp in entrees]


def revoke_one_for_user(sub: str, jti: str) -> bool:
    """Révoque un seul ``jti`` draco de ``sub`` (HUB-22, #227).

    Ne révoque (via ``revoke``) que si le ``jti`` appartient bien à
    l'ensemble émis de ce ``sub`` et n'est pas expiré — jamais de
    révocation d'une session d'un autre utilisateur. Retourne ``True``
    si la révocation a eu lieu, ``False`` sinon.
    """
    if not sub or not jti:
        return False
    with _lock:
        store = _load_issued()
        exp = store.get(str(sub), {}).get(jti)
        if exp is None or exp <= int(time.time()):
            return False
    revoke(jti, exp)
    return True


# ── É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,
    }
    token = jwt.encode(claims, _secret(), algorithm=ALGORITHM)
    _register_issued(str(claims["sub"]), jti, exp)
    return token, 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)


# ── Validation locale (ACC-3.2, #9) ─────────────────────────────────────

def user_for_draco(request) -> dict | None:
    """Utilisateur ObsiGate prouvé par le cookie draco_session.

    ObsiGate est à la fois émettrice et consommatrice : le `sub` du
    jeton EST son identifiant d'utilisateur — la correspondance est
    directe, par id. Le garde-fou « jeton antérieur au dernier
    changement de mot de passe » (BUG-028 côté natif) est appliqué
    à l'identique via le claim `iat`.
    """
    claims = verify_token(request.cookies.get(COOKIE_NAME, ""))
    if not claims:
        return None
    sub = str(claims.get("sub", ""))
    if not sub:
        return None
    from backend.auth.user_store import get_all_users

    for candidate in get_all_users():
        if str(candidate.get("id")) != sub:
            continue
        if not candidate.get("active"):
            return None
        pca = candidate.get("password_changed_at")
        iat = claims.get("iat")
        if pca is not None and iat is not None:
            try:
                if int(iat) < int(float(pca)):
                    return None
            except (TypeError, ValueError):
                return None
        user = dict(candidate)
        user["_token_vaults"] = candidate.get("vaults", [])
        user["_token_scope"] = "full"
        user["_draco_sub"] = sub
        # HUB-19 (#228) : rôle du registre du Hub pour ce porteur.
        # Niveau connu « aucun » → pas de session établie par draco :
        # le refus est consigné sur la requête pour que `require_auth`
        # réponde un message explicite en français (et non un simple
        # 401 générique). Rôle connu → joint au dict (`_hub_role`) ;
        # rôle inconnu → `_hub_role` vaut None et le comportement
        # actuel s'applique, sans verrouillage nouveau.
        from backend import hub_roles

        cookie = ""
        try:
            cookie = request.headers.get("cookie", "") or ""
        except Exception:  # un objet requête minimal (tests) ne casse rien
            cookie = ""
        role = hub_roles.role_pour(sub, cookie)
        if role == "aucun":
            try:
                request.state.role_hub_refuse = "aucun"
            except Exception:  # pas d'état mutable : refus quand même
                pass
            return None
        user["_hub_role"] = role
        return user
    return None


# ── 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