diff --git a/tests/test_coffre_secrets.py b/tests/test_coffre_secrets.py index 3061ed1..6d07fc6 100644 --- a/tests/test_coffre_secrets.py +++ b/tests/test_coffre_secrets.py @@ -19,6 +19,7 @@ from __future__ import annotations import base64 import json import os +import shutil from datetime import datetime, timedelta, timezone from pathlib import Path @@ -455,3 +456,225 @@ class TestDistribution: 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"