458 lines
18 KiB
Python
458 lines
18 KiB
Python
"""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())
|