"""Tests du coffre Secrets ObsiGate (Phase 1a / 1b, issues #17 et #18). Couverture exigée par les issues : - chiffrement/déchiffrement, tag invalide refusé, rotations (secret, DEK, KEK), mode dégradé sans KEK ; - matrice d'autorisation, y compris admin contre coffre personnel d'autrui et ACL « * » qui ne donne aucun droit sur un coffre ; - contexte manquant refusé, octroi révoqué refusé au prochain appel, expiration respectée, audit complet sans valeur ; - lectures = métadonnées seules, jamais de valeur ni de texte chiffré. Toutes les valeurs de secret utilisées ici sont fictives et ne servent qu'aux tests. Le store est isolé par test via OBSIGATE_DATA_DIR et le journal d'audit via backend.audit.AUDIT_LOG_FILE. """ from __future__ import annotations import base64 import json import os import shutil from datetime import datetime, timedelta, timezone from pathlib import Path import pytest from fastapi import FastAPI, HTTPException from fastapi.testclient import TestClient from backend import audit as audit_module from backend.auth.middleware import require_admin, require_auth from backend.coffre_crypto import ( KEKAbsenteError, KEKInvalideError, TagInvalideError, decrypt_secret_value, encrypt_secret_value, generate_dek, generate_kek_b64, load_kek, parse_kek, unwrap_dek, wrap_dek, ) from backend.coffre_store import ( CoffreExisteError, create_octroi, list_secrets, SecretExpireError, ValidationError, create_coffre, create_secret, distribuer_valeur, get_coffre_or_404, rotate_dek, rotate_kek, rotate_secret, ) from backend.routers.coffres import router as coffres_router VALEUR_TEST = "VALEUR-TEST-FICTIVE-7f3a9c-ne-pas-utiliser" VALEUR_TEST_2 = "VALEUR-TEST-FICTIVE-2-b4d8e1-ne-pas-utiliser" # Utilisateur courant injecté dans les routes via dependency_overrides. _COURANT = {"user": None} def _fake_auth(): if _COURANT["user"] is None: raise HTTPException(status_code=401, detail="Authentification requise") return _COURANT["user"] def _fake_admin(): user = _fake_auth() if user.get("role") != "admin": raise HTTPException(status_code=403, detail="Accès admin requis") return user def _user(username: str, role: str = "user", vaults: list | None = None) -> dict: return { "username": username, "role": role, "vaults": vaults if vaults is not None else [f"home-{username}"], "active": True, } @pytest.fixture def env_coffre(tmp_path, monkeypatch): """Environnement isolé : store, audit, KEK et feature flag.""" monkeypatch.setenv("OBSIGATE_SECRETS_ENABLED", "1") monkeypatch.setenv("OBSIGATE_DATA_DIR", str(tmp_path)) monkeypatch.setenv("OBSIGATE_KEK", generate_kek_b64()) monkeypatch.delenv("OBSIGATE_KEK_FILE", raising=False) monkeypatch.setattr(audit_module, "AUDIT_LOG_FILE", tmp_path / "audit.log") _COURANT["user"] = _user("bruno", role="admin") yield tmp_path _COURANT["user"] = None @pytest.fixture def client(env_coffre): app = FastAPI() app.include_router(coffres_router) app.dependency_overrides[require_auth] = _fake_auth app.dependency_overrides[require_admin] = _fake_admin with TestClient(app, raise_server_exceptions=False) as test_client: yield test_client def _connecter(client, username: str, role: str = "user", vaults=None): _COURANT["user"] = _user(username, role=role, vaults=vaults) def _creer_coffre_secret(client, slug="personnel", nom="gitea-token", valeur=VALEUR_TEST): resp = client.post( "/api/coffres", json={"slug": slug, "nom": "Personnel", "perimetre": "personnel"}, ) assert resp.status_code == 200, resp.text resp = client.post( f"/api/coffres/{slug}/secrets", json={ "nom": nom, "type": "token", "service": "Gitea", "valeur": valeur, }, ) assert resp.status_code == 200, resp.text 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 consommateur = resp.json() jeton = consommateur["jeton"] resp = client.post( f"/api/coffres/{slug}/octrois", json={ "cible_type": "consommateur", "cible_id": consommateur["id"], "secret_nom": nom, }, ) assert resp.status_code == 200, resp.text return consommateur, jeton, resp.json() def _distribuer(client, jeton, reference="coffre:personnel/gitea-token", contexte_type="mission", contexte_id="mission-test-1"): return client.post( "/api/coffres/distribuer", json={ "reference": reference, "contexte_type": contexte_type, "contexte_id": contexte_id, }, headers={"Authorization": f"Bearer {jeton}"}, ) def _fichiers_sans_valeur(tmp_path, valeur=VALEUR_TEST): for path in tmp_path.iterdir(): if path.is_file() and path.suffix in (".json", ".log"): assert valeur not in path.read_text(encoding="utf-8"), path.name # --------------------------------------------------------------------------- # Crypto (Phase 1a) # --------------------------------------------------------------------------- class TestCrypto: def test_aller_retour_aes_gcm(self): kek = bytes(range(32)) dek = generate_dek() wrapped = wrap_dek(dek, kek, "coffre-test") assert unwrap_dek(wrapped, kek, "coffre-test") == dek chiffre = encrypt_secret_value(dek, "secret-1", 1, VALEUR_TEST) assert decrypt_secret_value(dek, "secret-1", 1, chiffre["nonce"], chiffre["ciphertext"]) == VALEUR_TEST def test_tag_invalide_refuse(self): dek = generate_dek() chiffre = encrypt_secret_value(dek, "secret-1", 1, VALEUR_TEST) alt = bytearray(base64.b64decode(chiffre["ciphertext"])) alt[-1] ^= 1 chiffre_alt = { "nonce": chiffre["nonce"], "ciphertext": base64.b64encode(bytes(alt)).decode("ascii"), } with pytest.raises(TagInvalideError): decrypt_secret_value(dek, "secret-1", 1, chiffre_alt["nonce"], chiffre_alt["ciphertext"]) def test_aad_incorrect_refuse(self): dek = generate_dek() chiffre = encrypt_secret_value(dek, "secret-1", 1, VALEUR_TEST) with pytest.raises(TagInvalideError): decrypt_secret_value(dek, "secret-2", 1, chiffre["nonce"], chiffre["ciphertext"]) def test_parse_kek_invalide(self, monkeypatch): with pytest.raises(KEKInvalideError): parse_kek("trop-court") assert len(parse_kek(generate_kek_b64())) == 32 def test_kek_absente(self, tmp_path, monkeypatch): monkeypatch.delenv("OBSIGATE_KEK", raising=False) monkeypatch.delenv("OBSIGATE_KEK_FILE", raising=False) with pytest.raises(KEKAbsenteError): load_kek() def test_kek_fichier_0600_ok_et_trop_permissif_refuse(self, tmp_path, monkeypatch): kek_file = tmp_path / "kek.txt" kek_file.write_text(generate_kek_b64(), encoding="utf-8") os.chmod(kek_file, 0o600) monkeypatch.delenv("OBSIGATE_KEK", raising=False) monkeypatch.setenv("OBSIGATE_KEK_FILE", str(kek_file)) assert len(load_kek()) == 32 os.chmod(kek_file, 0o644) with pytest.raises(KEKInvalideError): load_kek() # --------------------------------------------------------------------------- # Store (Phase 1a) # --------------------------------------------------------------------------- class TestStore: def test_roundtrip_et_aucun_clair_au_repos(self, env_coffre): coffre = create_coffre("personnel", "Personnel", "bruno") secret = create_secret(coffre, "gitea-token", "token", VALEUR_TEST, service="Gitea") assert distribuer_valeur(coffre, secret) == VALEUR_TEST _fichiers_sans_valeur(env_coffre) def test_rotation_secret(self, env_coffre): coffre = create_coffre("personnel", "Personnel", "bruno") secret = create_secret(coffre, "gitea-token", "token", VALEUR_TEST) secret = rotate_secret(coffre, "gitea-token", VALEUR_TEST_2) assert secret["version_courante"] == 2 assert distribuer_valeur(coffre, secret) == VALEUR_TEST_2 def test_rotation_dek_puis_kek(self, env_coffre, monkeypatch): coffre = create_coffre("personnel", "Personnel", "bruno") secret = create_secret(coffre, "gitea-token", "token", VALEUR_TEST) rotate_dek("personnel") assert distribuer_valeur(get_coffre_or_404("personnel"), secret) == VALEUR_TEST nouvelle_kek = generate_kek_b64() assert rotate_kek(nouvelle_kek) == 1 monkeypatch.setenv("OBSIGATE_KEK", nouvelle_kek) assert distribuer_valeur(get_coffre_or_404("personnel"), secret) == VALEUR_TEST def test_mode_degrade_sans_kek(self, env_coffre, monkeypatch): monkeypatch.delenv("OBSIGATE_KEK", raising=False) coffre = create_coffre("personnel", "Personnel", "bruno") assert coffre["dek_chiffree"] is None with pytest.raises(KEKAbsenteError): create_secret(coffre, "gitea-token", "token", VALEUR_TEST) def test_validation_et_doublons(self, env_coffre): coffre = create_coffre("personnel", "Personnel", "bruno") with pytest.raises(ValidationError): create_coffre("MAUVAIS SLUG !!", "X", "bruno") with pytest.raises(CoffreExisteError): create_coffre("personnel", "Autre", "bruno") create_secret(coffre, "gitea-token", "token", VALEUR_TEST) with pytest.raises(ValidationError): create_secret( coffre, "autre-secret", "token", VALEUR_TEST, note="api-key: ceci-ressemble-a-un-secret", ) def test_secret_expire_refuse_a_la_distribution(self, env_coffre): coffre = create_coffre("personnel", "Personnel", "bruno") hier = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat() secret = create_secret( coffre, "vieux-jeton", "token", VALEUR_TEST, expiration=hier ) with pytest.raises(SecretExpireError): distribuer_valeur(coffre, secret) # --------------------------------------------------------------------------- # API et autorisations (Phase 1b) # --------------------------------------------------------------------------- class TestApiAutorisations: def test_flag_desactive_repond_503(self, client, monkeypatch): monkeypatch.setenv("OBSIGATE_SECRETS_ENABLED", "0") assert client.get("/api/coffres").status_code == 503 def test_lectures_metadonnees_seules(self, client, env_coffre): _creer_coffre_secret(client) resp = client.get("/api/coffres/personnel/secrets") assert resp.status_code == 200 corps = resp.text assert VALEUR_TEST not in corps assert "ciphertext" not in corps assert "nonce" not in corps item = resp.json()[0] assert item["has_value"] is True assert item["reference"] == "coffre:personnel/gitea-token" assert "valeur" not in item def test_admin_ne_voit_pas_le_coffre_personnel_d_autrui(self, client): _creer_coffre_secret(client) # propriétaire : bruno (admin de test) _connecter(client, "alice", role="user") assert client.get("/api/coffres/personnel").status_code == 403 assert client.get("/api/coffres/personnel/secrets").status_code == 403 # Même un admin avec vaults=["*"] n'a aucun droit sur un coffre # personnel dont il n'est pas propriétaire. _connecter(client, "admin2", role="admin", vaults=["*"]) assert client.get("/api/coffres/personnel").status_code == 403 def test_acl_etoile_ne_donne_aucun_droit_coffre(self, client): _creer_coffre_secret(client) _connecter(client, "mallory", role="user", vaults=["*"]) assert client.get("/api/coffres/personnel/secrets").status_code == 403 def test_coffre_instance_reserve_admin(self, client): _connecter(client, "bruno", role="user") resp = client.post( "/api/coffres", json={"slug": "instance", "nom": "Instance", "perimetre": "instance"}, ) assert resp.status_code == 403 _connecter(client, "bruno", role="admin") resp = client.post( "/api/coffres", json={"slug": "instance", "nom": "Instance", "perimetre": "instance"}, ) assert resp.status_code == 200 _connecter(client, "alice", role="user") assert client.get("/api/coffres/instance").status_code == 403 def test_creation_secret_sans_kek_repond_503(self, client, monkeypatch): monkeypatch.delenv("OBSIGATE_KEK", raising=False) resp = client.post("/api/coffres", json={"slug": "personnel", "nom": "Personnel"}) assert resp.status_code == 200 # métadonnées créées en mode dégradé resp = client.post( "/api/coffres/personnel/secrets", json={"nom": "gitea-token", "type": "token", "valeur": VALEUR_TEST}, ) assert resp.status_code == 503 assert resp.json()["detail"] == "kek_absente" class TestDistribution: def test_distribution_ok_puis_revoquee_puis_consommateur_revoque(self, client, env_coffre): _creer_coffre_secret(client) consommateur, jeton, octroi = _creer_consommateur_octroi(client) resp = _distribuer(client, jeton) assert resp.status_code == 200, resp.text assert resp.json()["valeur"] == VALEUR_TEST assert resp.json()["version"] == 1 # Le jeton complet n'est jamais relisté, ni son hash (seul le # préfixe d'affichage, non exploitable, apparaît). liste = client.get("/api/coffres/consommateurs").json() assert jeton not in json.dumps(liste) assert "jeton_hash" not in json.dumps(liste) assert all(set(item.keys()) == {"id", "nom", "jeton_prefixe", "statut", "cree_par", "cree_le"} for item in liste) # Révocation de l'octroi : refusé dès le prochain appel. resp = client.delete(f"/api/coffres/octrois/{octroi['id']}") assert resp.status_code == 200 resp = _distribuer(client, jeton) assert resp.status_code == 403 assert resp.json()["detail"] == "octroi_revoque" # Révocation du consommateur : tous ses octrois tombent. resp = client.post( "/api/coffres/personnel/octrois", json={"cible_type": "consommateur", "cible_id": consommateur["id"], "secret_nom": "gitea-token"}, ) assert resp.status_code == 200 resp = client.delete(f"/api/coffres/consommateurs/{consommateur['id']}") assert resp.status_code == 200 resp = _distribuer(client, jeton) assert resp.status_code == 403 assert resp.json()["detail"] == "consommateur_revoque" def test_contexte_manquant_refuse(self, client): _creer_coffre_secret(client) _, jeton, _ = _creer_consommateur_octroi(client) resp = client.post( "/api/coffres/distribuer", json={"reference": "coffre:personnel/gitea-token"}, headers={"Authorization": f"Bearer {jeton}"}, ) assert resp.status_code == 400 assert resp.json()["detail"] == "contexte_manquant" def test_session_utilisateur_ne_distribue_pas(self, client): _creer_coffre_secret(client) _creer_consommateur_octroi(client) resp = _distribuer(client, "jeton-session-utilisateur-fictif") assert resp.status_code == 401 assert resp.json()["detail"] == "consommateur_invalide" def test_octroi_absent_et_secret_expire(self, client): _creer_coffre_secret(client) resp = client.post("/api/coffres/consommateurs", json={"nom": "sans-octroi"}) jeton = resp.json()["jeton"] resp = _distribuer(client, jeton) assert resp.status_code == 403 assert resp.json()["detail"] == "octroi_absent" # Secret expiré : 410, même avec un octroi actif. _connecter(client, "bruno", role="admin") hier = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat() resp = client.post( "/api/coffres/personnel/secrets", json={"nom": "vieux-jeton", "type": "token", "valeur": VALEUR_TEST, "expiration": hier}, ) assert resp.status_code == 200 _, jeton2, _ = _creer_consommateur_octroi(client, nom="vieux-jeton") resp = _distribuer(client, jeton2, reference="coffre:personnel/vieux-jeton") assert resp.status_code == 410 assert resp.json()["detail"] == "secret_expire" def test_rotation_change_la_valeur_distribuee(self, client): _creer_coffre_secret(client) _, jeton, _ = _creer_consommateur_octroi(client) resp = client.post( "/api/coffres/personnel/secrets/gitea-token/rotation", json={"valeur": VALEUR_TEST_2}, ) assert resp.status_code == 200 assert resp.json()["version"] == 2 resp = _distribuer(client, jeton) assert resp.json()["valeur"] == VALEUR_TEST_2 def test_suppression_casse_les_octrois(self, client): _creer_coffre_secret(client) _, jeton, _ = _creer_consommateur_octroi(client) resp = client.delete("/api/coffres/personnel/secrets/gitea-token") assert resp.status_code == 200 resp = _distribuer(client, jeton) assert resp.status_code == 403 assert resp.json()["detail"] == "octroi_absent" def test_octroi_par_defaut_90_jours(self, client): _creer_coffre_secret(client) _, _, octroi = _creer_consommateur_octroi(client) expiration = datetime.fromisoformat(octroi["expiration"]) delta = expiration - datetime.now(timezone.utc) assert 89 <= delta.days <= 91 def test_audit_sans_valeur_et_journal(self, client, env_coffre): _creer_coffre_secret(client) _, jeton, _ = _creer_consommateur_octroi(client) assert _distribuer(client, jeton).status_code == 200 _fichiers_sans_valeur(env_coffre) audit = (env_coffre / "audit.log").read_text(encoding="utf-8") assert "distribution_ok" in audit assert VALEUR_TEST not in audit resp = client.get("/api/coffres/personnel/journal") assert resp.status_code == 200 actions = [e["action"] for e in resp.json()] assert "secret_cree" in actions assert "distribution_ok" in actions assert "valeur" not in json.dumps(resp.json()) # --------------------------------------------------------------------------- # Phase 2 — Migration api_keys.json (issue #20) # --------------------------------------------------------------------------- class TestMigrationApiKeys: def _ecrire_api_keys(self, tmp_path): (tmp_path / "api_keys.json").write_text( json.dumps( { "DEEPSEEK_API_KEY": VALEUR_TEST, "OBSIGATE_GITEA_TOKEN": VALEUR_TEST_2, "OBSIGATE_GITEA_URL": "https://example.invalid", "MISTRAL_API_KEY": "", } ), encoding="utf-8", ) def test_apercu_import_purge(self, client, env_coffre): self._ecrire_api_keys(env_coffre) resp = client.get("/api/coffres/migration/api-keys") assert resp.status_code == 200, resp.text # Aperçu = métadonnées seules, jamais une valeur. assert VALEUR_TEST not in resp.text assert VALEUR_TEST_2 not in resp.text candidats = {c["nom"]: c for c in resp.json()["candidats"]} assert candidats["deepseek_api_key"]["has_value"] is True assert candidats["mistral_api_key"]["has_value"] is False assert "obsigate_gitea_url" not in candidats # URL, pas un secret # Import : ne supprime rien dans api_keys.json. resp = client.post("/api/coffres/migration/api-keys/import", json={}) assert resp.status_code == 200, resp.text resultat = resp.json() assert resultat["coffre"] == "obsigate-interne" assert "deepseek_api_key" in resultat["migres"] assert "obsigate_gitea_token" in resultat["migres"] assert "mistral_api_key" in resultat["sans_valeur"] cles = json.loads((env_coffre / "api_keys.json").read_text(encoding="utf-8")) assert cles["DEEPSEEK_API_KEY"] == VALEUR_TEST # Les secrets migrés sont visibles en métadonnées et distribuables. resp = client.get("/api/coffres/obsigate-interne/secrets") assert resp.status_code == 200 assert VALEUR_TEST not in resp.text assert any(s["nom"] == "deepseek_api_key" and s["has_value"] for s in resp.json()) _, jeton, _ = _creer_consommateur_octroi( client, slug="obsigate-interne", nom="deepseek_api_key" ) resp = _distribuer( client, jeton, reference="coffre:obsigate-interne/deepseek_api_key" ) assert resp.status_code == 200, resp.text assert resp.json()["valeur"] == VALEUR_TEST # Purge sans confirmation : refusée, fichier intact. resp = client.post( "/api/coffres/migration/api-keys/purger", json={"confirmer": False, "noms": ["deepseek_api_key"]}, ) assert resp.status_code == 400 cles = json.loads((env_coffre / "api_keys.json").read_text(encoding="utf-8")) assert "DEEPSEEK_API_KEY" in cles # Purge confirmée : seules les clés listées disparaissent. resp = client.post( "/api/coffres/migration/api-keys/purger", json={"confirmer": True, "noms": ["deepseek_api_key", "obsigate_gitea_token"]}, ) assert resp.status_code == 200, resp.text assert set(resp.json()["supprimes"]) == { "deepseek_api_key", "obsigate_gitea_token", } cles = json.loads((env_coffre / "api_keys.json").read_text(encoding="utf-8")) assert "DEEPSEEK_API_KEY" not in cles assert "OBSIGATE_GITEA_TOKEN" not in cles assert cles["OBSIGATE_GITEA_URL"] == "https://example.invalid" # Audit sans valeur. _fichiers_sans_valeur(env_coffre) audit = (env_coffre / "audit.log").read_text(encoding="utf-8") assert "migration_api_keys" in audit assert "migration_api_keys_purge" in audit def test_import_ne_ecrase_pas_existant(self, client, env_coffre): self._ecrire_api_keys(env_coffre) assert client.post("/api/coffres/migration/api-keys/import", json={}).status_code == 200 (env_coffre / "api_keys.json").write_text( json.dumps({"DEEPSEEK_API_KEY": VALEUR_TEST_2}), encoding="utf-8" ) resp = client.post("/api/coffres/migration/api-keys/import", json={}) assert resp.status_code == 200 assert "deepseek_api_key" in resp.json()["deja_existants"] _, jeton, _ = _creer_consommateur_octroi( client, slug="obsigate-interne", nom="deepseek_api_key" ) resp = _distribuer( client, jeton, reference="coffre:obsigate-interne/deepseek_api_key" ) assert resp.json()["valeur"] == VALEUR_TEST # valeur d'origine conservée def test_import_sans_kek_repond_503(self, client, env_coffre, monkeypatch): # Une clé à migrer est nécessaire : sans valeur à chiffrer, # l'import réussit trivialement même sans KEK (mode dégradé). self._ecrire_api_keys(env_coffre) monkeypatch.delenv("OBSIGATE_KEK", raising=False) resp = client.post("/api/coffres/migration/api-keys/import", json={}) assert resp.status_code == 503 assert resp.json()["detail"] == "kek_absente" # --------------------------------------------------------------------------- # Phase 2 — Restauration sur instance vierge (issue #20) # --------------------------------------------------------------------------- class TestRestaurationVierge: def test_restauration_meme_kek_distribue(self, env_coffre, tmp_path): coffre = create_coffre("personnel", "Personnel", "bruno") secret = create_secret(coffre, "gitea-token", "token", VALEUR_TEST) assert distribuer_valeur(coffre, secret) == VALEUR_TEST # Simulation d'une instance vierge : seuls les deux fichiers sont # copiés, la KEK (variable d'environnement) est inchangée. restaure = tmp_path / "restaure" restaure.mkdir() for nom in ("coffres.json", "coffres_valeurs.json"): shutil.copy(env_coffre / nom, restaure / nom) coffre_r = get_coffre_or_404("personnel", restaure) assert distribuer_valeur(coffre_r, secret, restaure) == VALEUR_TEST for nom in ("coffres.json", "coffres_valeurs.json"): assert VALEUR_TEST not in (restaure / nom).read_text(encoding="utf-8") def test_restauration_mauvaise_kek_refuse(self, env_coffre, tmp_path, monkeypatch): coffre = create_coffre("personnel", "Personnel", "bruno") secret = create_secret(coffre, "gitea-token", "token", VALEUR_TEST) restaure = tmp_path / "restaure" restaure.mkdir() for nom in ("coffres.json", "coffres_valeurs.json"): shutil.copy(env_coffre / nom, restaure / nom) # Une autre KEK ne déchiffre rien : tag GCM invalide, pas de valeur. monkeypatch.setenv("OBSIGATE_KEK", generate_kek_b64()) coffre_r = get_coffre_or_404("personnel", restaure) with pytest.raises(TagInvalideError): distribuer_valeur(coffre_r, secret, restaure) # --------------------------------------------------------------------------- # Phase 3 — Contexte Spark imbriqué (issue #21) # --------------------------------------------------------------------------- class TestDistributionSpark: def test_contexte_nested_mission(self, client): _creer_coffre_secret(client) _, jeton, _ = _creer_consommateur_octroi(client) resp = client.post( "/api/coffres/distribuer", json={ "reference": "coffre:personnel/gitea-token", "contexte": {"type": "mission", "id": "mission-test-2"}, }, headers={"Authorization": f"Bearer {jeton}"}, ) assert resp.status_code == 200, resp.text assert resp.json()["valeur"] == VALEUR_TEST def test_contexte_nested_sans_octroi_acces_requis(self, client): _creer_coffre_secret(client) resp = client.post("/api/coffres/consommateurs", json={"nom": "spark"}) jeton = resp.json()["jeton"] resp = client.post( "/api/coffres/distribuer", json={ "reference": "coffre:personnel/gitea-token", "contexte": {"type": "mission", "id": "mission-test-3"}, }, headers={"Authorization": f"Bearer {jeton}"}, ) # Côté Spark, ce code se traduit par l'état acces_requis. assert resp.status_code == 403 assert resp.json()["detail"] == "octroi_absent" # --------------------------------------------------------------------------- # Phase 4 — Consommateur + octrois en un appel (issue #22) # --------------------------------------------------------------------------- class TestConsommateurAvecOctrois: def test_avec_octrois_distribue(self, client, env_coffre): _creer_coffre_secret(client) resp = client.post( "/api/coffres/consommateurs/avec-octrois", json={ "nom": "tiktak", "octrois": [{"coffre_slug": "personnel", "secret_nom": "gitea-token"}], }, ) assert resp.status_code == 200, resp.text corps = resp.json() assert corps["nom"] == "tiktak" assert corps["jeton"].startswith("csk_") assert len(corps["octrois"]) == 1 assert corps["erreurs"] == [] resp = _distribuer(client, corps["jeton"]) assert resp.status_code == 200 assert resp.json()["valeur"] == VALEUR_TEST liste = client.get("/api/coffres/consommateurs").json() assert corps["jeton"] not in json.dumps(liste) assert "jeton_hash" not in json.dumps(liste) def test_avec_octrois_erreur_signalee_sans_annuler(self, client): _creer_coffre_secret(client) resp = client.post( "/api/coffres/consommateurs/avec-octrois", json={ "nom": "flowdeck", "octrois": [ {"coffre_slug": "personnel", "secret_nom": "gitea-token"}, {"coffre_slug": "inexistant", "secret_nom": "gitea-token"}, ], }, ) assert resp.status_code == 200, resp.text 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