[Coffre Secrets] Phase 1a/1b : tests crypto, store, autorisations, distribution
This commit is contained in:
@@ -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())
|
||||
Reference in New Issue
Block a user