diff --git a/tests/test_coffre_secrets.py b/tests/test_coffre_secrets.py new file mode 100644 index 0000000..3061ed1 --- /dev/null +++ b/tests/test_coffre_secrets.py @@ -0,0 +1,457 @@ +"""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 +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, + 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_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())