607 lines
22 KiB
Python
607 lines
22 KiB
Python
"""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}
|