Test : distribuables (octrois actifs, métadonnées seules, auth consommateur)

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

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