From 40b3e9c5dd9b78d6235357846abe7cd5014208c8 Mon Sep 17 00:00:00 2001 From: bruno Date: Tue, 27 Oct 2026 21:45:40 -0400 Subject: [PATCH] [Coffre Secrets] Phase 1b : routeur /api/coffres, autorisations, octrois, distribution --- backend/routers/coffres.py | 606 +++++++++++++++++++++++++++++++++++++ 1 file changed, 606 insertions(+) create mode 100644 backend/routers/coffres.py diff --git a/backend/routers/coffres.py b/backend/routers/coffres.py new file mode 100644 index 0000000..b3aa91c --- /dev/null +++ b/backend/routers/coffres.py @@ -0,0 +1,606 @@ +"""API du coffre Secrets ObsiGate (Phase 1b, spec §7, §10, §11). + +Monté dans ``backend/main.py`` à côté de ``config_router`` — JAMAIS dans +``backend/routers/vaults.py`` : un coffre n'est pas un vault de notes, il +ne passe ni par l'indexeur, ni par la recherche, ni par le graphe, et +aucun outil MCP de secret n'existe. + +Règles d'or, appliquées côté serveur à chaque requête : +- Toutes les lectures renvoient des métadonnées seules (``has_value``, + version, dates, octrois actifs) : jamais de valeur, jamais de texte + chiffré, jamais de nonce. +- Seul ``POST /api/coffres/distribuer`` rend une valeur, et uniquement à + un consommateur authentifié par jeton (hash + jti, jamais la session + utilisateur), avec un octroi actif et un contexte obligatoire. +- Autorisations (spec §6) : le propriétaire seul sur son coffre + personnel ; l'admin gère les coffres d'instance, les consommateurs et + la KEK, jamais le coffre personnel d'autrui ; l'ACL « * » des vaults + ne donne AUCUN droit sur un coffre (``check_vault_access`` n'est pas + utilisé ici, volontairement). +- Feature flag ``OBSIGATE_SECRETS_ENABLED`` (défaut désactivé) : tant + qu'il n'est pas activé, toutes les routes répondent 503. +""" + +from __future__ import annotations + +import logging +import os + +from fastapi import APIRouter, Body, Depends, HTTPException, Request + +from backend.audit import get_recent_entries, log_coffre_event +from backend.auth.middleware import require_admin, require_auth +from backend.coffre_crypto import ( + CryptoIndisponibleError, + KEKAbsenteError, + generate_kek_b64, +) +from backend.coffre_store import ( + CoffreExisteError, + IntrouvableError, + SecretExisteError, + SecretExpireError, + ValidationError, + compter_octrois_actifs, + compter_secrets, + create_coffre, + create_consommateur, + create_octroi, + create_secret, + delete_secret, + distribuer_valeur, + get_coffre, + get_coffre_by_id, + get_coffre_or_404, + get_octroi, + get_secret, + list_coffres, + list_consommateurs, + list_octrois, + list_secrets, + revoquer_consommateur, + revoquer_octroi, + rotate_dek, + rotate_kek, + rotate_secret, + secret_metadata, + secrets_accessibles_utilisateur, + statut_octroi_consommateur, + toucher_utilisation, + verifier_jeton_consommateur, +) + +logger = logging.getLogger("obsigate.coffres") + +router = APIRouter(tags=["Coffres Secrets"]) + +CODES_ERREUR = { + "octroi_absent", + "octroi_revoque", + "octroi_expire", + "secret_expire", + "kek_absente", + "contexte_manquant", + "consommateur_invalide", + "consommateur_revoque", + "crypto_indisponible", + "secrets_desactives", +} + + +def secrets_actives() -> bool: + """Feature flag : défaut désactivé, activation explicite par l'opérateur.""" + return os.environ.get("OBSIGATE_SECRETS_ENABLED", "false").lower() in ( + "1", + "true", + "yes", + "on", + ) + + +def _exiger_actives() -> None: + if not secrets_actives(): + raise HTTPException(status_code=503, detail="secrets_desactives") + + +def _username(user: dict) -> str: + return str(user.get("username", "")) + + +def _est_admin(user: dict) -> bool: + return user.get("role") == "admin" + + +def _peut_voir_coffre(user: dict, coffre: dict) -> bool: + """Voir les métadonnées d'un coffre.""" + if coffre["proprietaire"] == _username(user): + return True + if _est_admin(user) and coffre["perimetre"] == "instance": + return True + # Octroi utilisateur (coffre-wide ou par secret) : accès en lecture. + accessibles = secrets_accessibles_utilisateur(_username(user), coffre) + if accessibles is None: + return True + return len(accessibles) > 0 + + +def _peut_gerer_coffre(user: dict, coffre: dict) -> bool: + """Créer / roter / supprimer des secrets et gérer les octrois.""" + if coffre["proprietaire"] == _username(user): + return True + return _est_admin(user) and coffre["perimetre"] == "instance" + + +def _exiger_voir(user: dict, coffre: dict) -> None: + if not _peut_voir_coffre(user, coffre): + raise HTTPException(status_code=403, detail="Accès refusé à ce coffre") + + +def _exiger_gerer(user: dict, coffre: dict) -> None: + if not _peut_gerer_coffre(user, coffre): + raise HTTPException(status_code=403, detail="Accès refusé à ce coffre") + + +def _coffre_or_404(slug: str) -> dict: + try: + return get_coffre_or_404(slug) + except IntrouvableError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + except ValidationError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + +def _secret_or_404(coffre: dict, nom: str) -> dict: + try: + secret = get_secret(coffre["id"], nom) + except ValidationError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + if secret is None: + raise HTTPException(status_code=404, detail=f"Secret introuvable : {nom}") + return secret + + +def _metadata_coffre(coffre: dict) -> dict: + return { + "slug": coffre["slug"], + "nom": coffre["nom"], + "description": coffre.get("description", ""), + "proprietaire": coffre["proprietaire"], + "perimetre": coffre["perimetre"], + "statut": coffre["statut"], + "kek_configuree": coffre.get("dek_chiffree") is not None, + "nb_secrets": compter_secrets(coffre["id"]), + "cree_le": coffre["cree_le"], + } + + +def _metadata_secret(coffre: dict, secret: dict) -> dict: + return secret_metadata(secret, compter_octrois_actifs(secret["id"])) + + +def _secrets_visibles(user: dict, coffre: dict) -> list: + """Secrets visibles par l'utilisateur (métadonnées seules).""" + if coffre["proprietaire"] == _username(user) or ( + _est_admin(user) and coffre["perimetre"] == "instance" + ): + return [_metadata_secret(coffre, s) for s in list_secrets(coffre["id"])] + accessibles = secrets_accessibles_utilisateur(_username(user), coffre) + if accessibles is None: + return [_metadata_secret(coffre, s) for s in list_secrets(coffre["id"])] + return [ + _metadata_secret(coffre, s) + for s in list_secrets(coffre["id"]) + if s["id"] in accessibles + ] + + +def _audit(user: str, action: str, **kwargs) -> None: + try: + log_coffre_event(user, action, **kwargs) + except Exception as exc: # l'audit ne doit jamais casser l'opération + logger.warning("Échec audit coffre (%s) : %s", action, exc) + + +# --------------------------------------------------------------------------- +# Coffres +# --------------------------------------------------------------------------- + +@router.get("/api/coffres") +async def api_list_coffres(current_user=Depends(require_auth)): + _exiger_actives() + visibles = [c for c in list_coffres() if _peut_voir_coffre(current_user, c)] + return [_metadata_coffre(c) for c in visibles] + + +@router.post("/api/coffres") +async def api_create_coffre(body: dict = Body(...), current_user=Depends(require_auth)): + _exiger_actives() + slug = body.get("slug", "") + perimetre = body.get("perimetre", "personnel") + if perimetre == "instance" and not _est_admin(current_user): + raise HTTPException( + status_code=403, detail="Seul un admin peut créer un coffre d'instance" + ) + try: + coffre = create_coffre( + slug=slug, + nom=body.get("nom", ""), + proprietaire=_username(current_user), + perimetre=perimetre, + description=body.get("description", ""), + ) + except CoffreExisteError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + except ValidationError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + _audit( + _username(current_user), + "coffre_cree", + coffre=coffre["slug"], + resultat="ok", + ) + return _metadata_coffre(coffre) + + +# --------------------------------------------------------------------------- +# Consommateurs (admin seulement, jeton rendu une seule fois) +# --------------------------------------------------------------------------- + +@router.get("/api/coffres/consommateurs") +async def api_list_consommateurs(current_user=Depends(require_admin)): + _exiger_actives() + return list_consommateurs() + + +@router.post("/api/coffres/consommateurs") +async def api_create_consommateur( + body: dict = Body(...), current_user=Depends(require_admin) +): + _exiger_actives() + try: + consommateur, jeton = create_consommateur( + nom=body.get("nom", ""), cree_par=_username(current_user) + ) + except ValidationError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + _audit( + _username(current_user), + "consommateur_cree", + consommateur=consommateur["nom"], + resultat="ok", + ) + # Le jeton n'est rendu qu'ici, jamais stocké en clair, jamais relisté. + return {**consommateur, "jeton": jeton} + + +@router.delete("/api/coffres/consommateurs/{consommateur_id}") +async def api_revoke_consommateur( + consommateur_id: str, current_user=Depends(require_admin) +): + _exiger_actives() + try: + consommateur = revoquer_consommateur(consommateur_id) + except IntrouvableError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + _audit( + _username(current_user), + "consommateur_revoque", + consommateur=consommateur["nom"], + resultat="ok", + ) + return consommateur + + +# --------------------------------------------------------------------------- +# Distribution (serveur-à-serveur, hors session utilisateur, hors LLM) +# --------------------------------------------------------------------------- + +@router.post("/api/coffres/distribuer") +async def api_distribuer(request: Request, body: dict = Body(...)): + _exiger_actives() + reference = (body.get("reference") or "").strip() + contexte_type = (body.get("contexte_type") or "").strip() + contexte_id = (body.get("contexte_id") or "").strip() + if not contexte_type or not contexte_id: + raise HTTPException(status_code=400, detail="contexte_manquant") + if not reference.startswith("coffre:") or "/" not in reference[7:]: + raise HTTPException(status_code=400, detail="reference_invalide") + slug, nom = reference[7:].split("/", 1) + # Authentification consommateur : Bearer, hash + statut, jamais un JWT + # de session utilisateur. + auth = request.headers.get("Authorization", "") + jeton = auth[7:] if auth.startswith("Bearer ") else "" + consommateur = verifier_jeton_consommateur(jeton) + if consommateur is None: + _audit("consommateur", "distribution_refusee", secret_ref=reference, + resultat="consommateur_invalide", contexte_id=contexte_id) + raise HTTPException(status_code=401, detail="consommateur_invalide") + if consommateur["statut"] != "actif": + _audit("consommateur", "distribution_refusee", secret_ref=reference, + consommateur=consommateur["nom"], + resultat="consommateur_revoque", contexte_id=contexte_id) + raise HTTPException(status_code=403, detail="consommateur_revoque") + try: + coffre = get_coffre(slug) + secret = get_secret(coffre["id"], nom) if coffre else None + except ValidationError: + raise HTTPException(status_code=400, detail="reference_invalide") from None + if coffre is None or secret is None or coffre["statut"] != "actif": + # Anti-énumération : un secret inexistant se comporte comme un + # octroi absent, il ne révèle pas l'existence du coffre/secret. + _audit("consommateur", "distribution_refusee", secret_ref=reference, + consommateur=consommateur["nom"], + resultat="octroi_absent", contexte_id=contexte_id) + raise HTTPException(status_code=403, detail="octroi_absent") + statut, _octroi = statut_octroi_consommateur( + consommateur["id"], coffre["id"], secret["id"] + ) + if statut != "ok": + code = { + "revoque": "octroi_revoque", + "expire": "octroi_expire", + "absent": "octroi_absent", + }[statut] + _audit("consommateur", "distribution_refusee", coffre=coffre["slug"], + secret_ref=reference, consommateur=consommateur["nom"], + resultat=code, contexte_id=contexte_id) + raise HTTPException(status_code=403, detail=code) + try: + valeur = distribuer_valeur(coffre, secret) + except SecretExpireError: + _audit("consommateur", "distribution_refusee", coffre=coffre["slug"], + secret_ref=reference, consommateur=consommateur["nom"], + resultat="secret_expire", contexte_id=contexte_id) + raise HTTPException(status_code=410, detail="secret_expire") from None + except KEKAbsenteError: + _audit("consommateur", "distribution_refusee", coffre=coffre["slug"], + secret_ref=reference, consommateur=consommateur["nom"], + resultat="kek_absente", contexte_id=contexte_id) + raise HTTPException(status_code=503, detail="kek_absente") from None + except CryptoIndisponibleError: + raise HTTPException(status_code=503, detail="crypto_indisponible") from None + toucher_utilisation(secret["id"]) + _audit("consommateur", "distribution_ok", coffre=coffre["slug"], + secret_ref=reference, consommateur=consommateur["nom"], + resultat="ok", contexte_id=contexte_id) + return { + "reference": reference, + "valeur": valeur, + "version": secret["version_courante"], + } + + +# --------------------------------------------------------------------------- +# Un coffre, ses secrets, ses octrois, son journal +# --------------------------------------------------------------------------- + +@router.get("/api/coffres/{slug}") +async def api_get_coffre(slug: str, current_user=Depends(require_auth)): + _exiger_actives() + coffre = _coffre_or_404(slug) + _exiger_voir(current_user, coffre) + return _metadata_coffre(coffre) + + +@router.get("/api/coffres/{slug}/secrets") +async def api_list_secrets(slug: str, current_user=Depends(require_auth)): + _exiger_actives() + coffre = _coffre_or_404(slug) + _exiger_voir(current_user, coffre) + return _secrets_visibles(current_user, coffre) + + +@router.post("/api/coffres/{slug}/secrets") +async def api_create_secret( + slug: str, body: dict = Body(...), current_user=Depends(require_auth) +): + _exiger_actives() + coffre = _coffre_or_404(slug) + _exiger_gerer(current_user, coffre) + try: + secret = create_secret( + coffre=coffre, + nom=body.get("nom", ""), + type_secret=body.get("type", "autre"), + valeur=body.get("valeur", ""), + service=body.get("service", ""), + expiration=body.get("expiration"), + note=body.get("note", ""), + ) + except SecretExisteError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + except ValidationError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + except KEKAbsenteError: + raise HTTPException(status_code=503, detail="kek_absente") from None + except CryptoIndisponibleError: + raise HTTPException(status_code=503, detail="crypto_indisponible") from None + _audit( + _username(current_user), + "secret_cree", + coffre=coffre["slug"], + secret_ref=f"coffre:{coffre['slug']}/{secret['nom']}", + resultat="ok", + ) + return _metadata_secret(coffre, secret) + + +@router.post("/api/coffres/{slug}/secrets/{nom}/rotation") +async def api_rotate_secret( + slug: str, nom: str, body: dict = Body(...), current_user=Depends(require_auth) +): + _exiger_actives() + coffre = _coffre_or_404(slug) + _exiger_gerer(current_user, coffre) + try: + secret = rotate_secret(coffre, nom, body.get("valeur", "")) + except IntrouvableError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + except ValidationError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + except KEKAbsenteError: + raise HTTPException(status_code=503, detail="kek_absente") from None + except CryptoIndisponibleError: + raise HTTPException(status_code=503, detail="crypto_indisponible") from None + _audit( + _username(current_user), + "secret_rotation", + coffre=coffre["slug"], + secret_ref=f"coffre:{coffre['slug']}/{secret['nom']}", + resultat="ok", + ) + return _metadata_secret(coffre, secret) + + +@router.delete("/api/coffres/{slug}/secrets/{nom}") +async def api_delete_secret( + slug: str, nom: str, current_user=Depends(require_auth) +): + _exiger_actives() + coffre = _coffre_or_404(slug) + _exiger_gerer(current_user, coffre) + try: + secret, _octrois = delete_secret(coffre, nom) + except IntrouvableError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + _audit( + _username(current_user), + "secret_supprime", + coffre=coffre["slug"], + secret_ref=f"coffre:{coffre['slug']}/{secret['nom']}", + resultat="ok", + ) + return {"status": "deleted", "nom": secret["nom"]} + + +@router.get("/api/coffres/{slug}/octrois") +async def api_list_octrois(slug: str, current_user=Depends(require_auth)): + _exiger_actives() + coffre = _coffre_or_404(slug) + _exiger_voir(current_user, coffre) + return list_octrois(coffre["id"]) + + +@router.post("/api/coffres/{slug}/octrois") +async def api_create_octroi( + slug: str, body: dict = Body(...), current_user=Depends(require_auth) +): + _exiger_actives() + coffre = _coffre_or_404(slug) + _exiger_gerer(current_user, coffre) + secret_id = None + if body.get("secret_nom"): + secret = _secret_or_404(coffre, body["secret_nom"]) + secret_id = secret["id"] + try: + octroi = create_octroi( + coffre=coffre, + cible_type=body.get("cible_type", ""), + cible_id=body.get("cible_id", ""), + secret_id=secret_id, + cree_par=_username(current_user), + expiration=body.get("expiration"), + ) + except IntrouvableError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + except ValidationError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + _audit( + _username(current_user), + "octroi_cree", + coffre=coffre["slug"], + secret_ref=( + f"coffre:{coffre['slug']}/{body.get('secret_nom', '*')}" + ), + resultat="ok", + ) + return octroi + + +@router.delete("/api/coffres/octrois/{octroi_id}") +async def api_revoke_octroi(octroi_id: str, current_user=Depends(require_auth)): + _exiger_actives() + # Le gestionnaire du coffre concerné (ou l'admin d'instance) seul peut + # révoquer : on résout le coffre depuis l'octroi avant toute action. + octroi = get_octroi(octroi_id) + if octroi is None: + raise HTTPException(status_code=404, detail="Octroi introuvable") + coffre = get_coffre_by_id(octroi["coffre_id"]) + if coffre is None: + raise HTTPException(status_code=404, detail="Coffre introuvable") + _exiger_gerer(current_user, coffre) + try: + octroi = revoquer_octroi(octroi_id) + except IntrouvableError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + _audit( + _username(current_user), + "octroi_revoque", + coffre=coffre["slug"], + resultat="ok", + ) + return octroi + + +@router.get("/api/coffres/{slug}/journal") +async def api_journal(slug: str, current_user=Depends(require_auth)): + _exiger_actives() + coffre = _coffre_or_404(slug) + _exiger_voir(current_user, coffre) + # Défense en profondeur : même si une entrée contenait un jour un champ + # sensible, il ne sort jamais par cette route. + nettoyees = [] + for entry in get_recent_entries(limit=100): + if entry.get("coffre") != slug: + continue + nettoyees.append( + { + k: v + for k, v in entry.items() + if k not in ("valeur", "ciphertext", "nonce", "jeton", "dek_chiffree") + } + ) + return nettoyees + + +# --------------------------------------------------------------------------- +# Administration KEK / rotations (admin seulement) +# --------------------------------------------------------------------------- + +@router.post("/api/coffres/admin/generer-kek") +async def api_generer_kek(current_user=Depends(require_admin)): + _exiger_actives() + # La KEK générée n'est rendue qu'une fois, ici, et jamais stockée par + # ObsiGate : l'opérateur doit la sauvegarder hors dépôt immédiatement. + return {"kek": generate_kek_b64()} + + +@router.post("/api/coffres/{slug}/rotation-dek") +async def api_rotate_dek(slug: str, current_user=Depends(require_auth)): + _exiger_actives() + coffre = _coffre_or_404(slug) + _exiger_gerer(current_user, coffre) + try: + coffre = rotate_dek(slug) + except KEKAbsenteError: + raise HTTPException(status_code=503, detail="kek_absente") from None + except CryptoIndisponibleError: + raise HTTPException(status_code=503, detail="crypto_indisponible") from None + _audit(_username(current_user), "dek_rotation", coffre=coffre["slug"], + resultat="ok") + return _metadata_coffre(coffre) + + +@router.post("/api/coffres/admin/rotation-kek") +async def api_rotate_kek( + body: dict = Body(...), current_user=Depends(require_admin) +): + _exiger_actives() + try: + count = rotate_kek(body.get("nouvelle_kek", "")) + except KEKAbsenteError: + raise HTTPException(status_code=503, detail="kek_absente") from None + except Exception as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + _audit(_username(current_user), "kek_rotation", resultat="ok") + return {"status": "ok", "coffres_rechiffres": count}