diff --git a/backend/routers/coffres.py b/backend/routers/coffres.py index c6b0bd3..34d0ab4 100644 --- a/backend/routers/coffres.py +++ b/backend/routers/coffres.py @@ -57,6 +57,7 @@ from backend.coffre_store import ( SecretExpireError, ValidationError, compter_octrois_actifs, + est_expire, compter_secrets, create_coffre, create_consommateur, @@ -495,6 +496,50 @@ async def api_distribuer(request: Request, body: dict = Body(...)): } +@router.get("/api/coffres/distribuables") +async def api_distribuables(request: Request): + """Secrets distribuables pour un consommateur (issue #43). + + Authentification par jeton consommateur (la même que + /distribuer), jamais par session utilisateur : c'est le + serveur agent qui appelle, pour alimenter la liste déroulante + de ses fournisseurs LLM. Retour = métadonnées seules des + secrets pour lesquels ce consommateur a un octroi ACTIF + (coffre actif, secret non expiré, octroi ni révoqué ni + expiré, y compris les octrois coffre-wide). Un consommateur + sans octroi reçoit une liste vide — jamais d'erreur + révélatrice, jamais de valeur. + """ + _exiger_actives() + auth = request.headers.get("Authorization", "") + jeton = auth[7:] if auth.startswith("Bearer ") else "" + consommateur = verifier_jeton_consommateur(jeton) + if consommateur is None: + raise HTTPException(status_code=401, + detail="consommateur_invalide") + if consommateur["statut"] != "actif": + raise HTTPException(status_code=403, + detail="consommateur_revoque") + sortie = [] + for coffre in list_coffres(): + if coffre["statut"] != "actif": + continue + for secret in list_secrets(coffre["id"]): + if est_expire(secret.get("expiration")): + continue + statut, _octroi = statut_octroi_consommateur( + consommateur["id"], coffre["id"], secret["id"]) + if statut != "ok": + continue + sortie.append({ + "reference": f"coffre:{coffre['slug']}/{secret['nom']}", + "coffre": coffre["slug"], + "nom": secret["nom"], + "version": secret["version_courante"], + }) + return sortie + + # --------------------------------------------------------------------------- # Un coffre, ses secrets, ses octrois, son journal # --------------------------------------------------------------------------- diff --git a/tests/test_coffre_secrets.py b/tests/test_coffre_secrets.py index 6d07fc6..68391b4 100644 --- a/tests/test_coffre_secrets.py +++ b/tests/test_coffre_secrets.py @@ -44,6 +44,8 @@ from backend.coffre_crypto import ( ) from backend.coffre_store import ( CoffreExisteError, + create_octroi, + list_secrets, SecretExpireError, ValidationError, create_coffre, @@ -131,6 +133,17 @@ def _creer_coffre_secret(client, slug="personnel", nom="gitea-token", valeur=VAL return resp.json() +def _creer_secret(client, slug="personnel", nom="second", + valeur=VALEUR_TEST_2): + resp = client.post( + f"/api/coffres/{slug}/secrets", + json={"nom": nom, "type": "token", "service": "Test", + "valeur": valeur}, + ) + assert resp.status_code == 200, resp.text + return resp.json() + + def _creer_consommateur_octroi(client, slug="personnel", nom="gitea-token"): resp = client.post("/api/coffres/consommateurs", json={"nom": "spark"}) assert resp.status_code == 200, resp.text @@ -678,3 +691,90 @@ class TestConsommateurAvecOctrois: assert len(resp.json()["octrois"]) == 1 assert len(resp.json()["erreurs"]) == 1 assert resp.json()["erreurs"][0]["coffre_slug"] == "inexistant" + + +# --------------------------------------------------------------------------- +# Liste des distribuables pour un consommateur (issue #43) +# --------------------------------------------------------------------------- + +class TestDistribuables: + def _lister(self, client, jeton=None): + headers = {} + if jeton is not None: + headers["Authorization"] = f"Bearer {jeton}" + return client.get("/api/coffres/distribuables", headers=headers) + + def test_ne_liste_que_les_octrois_actifs_metadonnees_seules( + self, client, env_coffre): + _creer_coffre_secret(client) # personnel/gitea-token + _creer_secret(client, nom="sans-octroi") + consommateur, jeton, octroi = _creer_consommateur_octroi(client) + resp = self._lister(client, jeton) + assert resp.status_code == 200, resp.text + assert resp.json() == [{ + "reference": "coffre:personnel/gitea-token", + "coffre": "personnel", + "nom": "gitea-token", + "version": 1, + }] + assert VALEUR_TEST not in resp.text + assert VALEUR_TEST_2 not in resp.text + assert "valeur" not in resp.text + # Octroi révoqué : le secret disparaît, sans erreur. + client.delete(f"/api/coffres/octrois/{octroi['id']}") + assert self._lister(client, jeton).json() == [] + + def test_octroi_expire_exclu_et_coffre_wide_inclus( + self, client, env_coffre): + from datetime import datetime, timedelta, timezone + _creer_coffre_secret(client) + _creer_secret(client, nom="second") + consommateur, jeton, _octroi = _creer_consommateur_octroi(client) + # Octroi expiré (hier) sur « second » : exclu. + coffre = get_coffre_or_404("personnel") + second = [x for x in list_secrets(coffre["id"]) + if x["nom"] == "second"][0] + create_octroi(coffre, "consommateur", consommateur["id"], + second["id"], "bruno", + expiration=(datetime.now(timezone.utc) + - timedelta(days=1)).isoformat()) + # Octroi coffre-wide actif : couvre tous les secrets. + resp = client.post( + "/api/coffres/personnel/octrois", + json={"cible_type": "consommateur", + "cible_id": consommateur["id"], + "secret_nom": None}, + ) + # Si l'API refuse le coffre-wide sans nom, on le crée direct. + if resp.status_code != 200: + create_octroi(coffre, "consommateur", consommateur["id"], + None, "bruno") + refs = {x["reference"] + for x in self._lister(client, jeton).json()} + assert "coffre:personnel/gitea-token" in refs + assert "coffre:personnel/second" in refs + + def test_auth_consommateur_uniquement(self, client, env_coffre): + _creer_coffre_secret(client) + _creer_consommateur_octroi(client) + # Sans jeton : 401 (la session utilisateur ne suffit pas). + assert self._lister(client).status_code == 401 + # Jeton inconnu : 401, sans révéler d'inventaire. + assert self._lister(client, "jeton-inconnu").status_code == 401 + # Consommateur sans octroi : liste vide, pas d'erreur. + resp = client.post("/api/coffres/consommateurs", + json={"nom": "autre-app"}) + autre_jeton = resp.json()["jeton"] + assert self._lister(client, autre_jeton).json() == [] + + def test_consommateur_revoque_et_flag_desactive( + self, client, env_coffre, monkeypatch): + _creer_coffre_secret(client) + consommateur, jeton, _octroi = _creer_consommateur_octroi(client) + client.delete(f"/api/coffres/consommateurs/{consommateur['id']}") + resp = self._lister(client, jeton) + assert resp.status_code == 403 + assert resp.json()["detail"] == "consommateur_revoque" + # Flag désactivé : 503 comme les autres routes. + monkeypatch.setenv("OBSIGATE_SECRETS_ENABLED", "0") + assert self._lister(client, jeton).status_code == 503