# tests/test_comptes_machine.py — HUB-13 (fusion #127) # API d'administration des comptes machine-à-machine # (/api/admin/machine/comptes), protégée par la clé dédiée # DRACO_ADMIN_MACHINE_KEY : lecture/écriture, désactivation avec # révocation des sessions, réinitialisations du mot de passe (avec # changement forcé) et du MFA, suppression, journalisation d'audit. import json import os import shutil import tempfile from pathlib import Path import pytest MACHINE_KEY = "cle-admin-machine-test" MACHINE_HEADERS = {"X-Draco-Machine-Key": MACHINE_KEY, "X-Draco-Actor": "hub-test"} BASE = "/api/admin/machine/comptes" @pytest.fixture def machine_client(): """TestClient avec auth active, données isolées, clés machine posées.""" tmp = Path(tempfile.mkdtemp()) data_dir = tmp / "data" data_dir.mkdir() from backend.auth.mfa import generate_secret from backend.auth.password import hash_password users = { "version": 1, "users": { "admin": { "id": "admin-1", "username": "admin", "display_name": "admin", "password_hash": hash_password("chab30"), "role": "admin", "vaults": ["*"], "active": True, "created_at": "2026-01-01T00:00:00", }, "mfauser": { "id": "user-mfa", "username": "mfauser", "display_name": "Utilisateur MFA", "password_hash": hash_password("motdepasse-mfa"), "role": "user", "vaults": ["TestVault"], "active": True, "created_at": "2026-01-01T00:00:00", "mfa_enabled": True, "mfa_secret": generate_secret(), "mfa_method": "totp", "mfa_recovery_codes": ["hash-fictif"], }, }, } (data_dir / "users.json").write_text(json.dumps(users), encoding="utf-8") src_secret = Path("data/secret.key") if src_secret.exists(): shutil.copy2(str(src_secret), str(data_dir / "secret.key")) orig_cwd = os.getcwd() test_vault_path = os.path.abspath("test-vault") os.chdir(str(tmp)) os.environ["VAULT_1_NAME"] = "TestVault" os.environ["VAULT_1_PATH"] = test_vault_path os.environ["OBSIGATE_AUTH_ENABLED"] = "true" os.environ["OBSIGATE_WATCHER_ENABLED"] = "false" os.environ["DRACO_ADMIN_MACHINE_KEY"] = MACHINE_KEY os.environ["DRACO_SSO_MACHINE_KEY"] = "cle-sso-test" os.environ["DRACO_SESSION_SECRET"] = "secret-draco-test" from backend import draco_session draco_session._revoked = None draco_session._issued = None import backend.main backend.main._load_config = lambda: {"watcher_enabled": False} from backend.main import app from backend.indexer import build_index, index import asyncio for key in list(index.keys()): del index[key] loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) loop.run_until_complete(build_index()) from backend.search import init_inverted_index init_inverted_index() from fastapi.testclient import TestClient client = TestClient(app) yield client if hasattr(client, "close"): client.close() loop.run_until_complete(asyncio.sleep(0)) draco_session._revoked = None draco_session._issued = None os.chdir(orig_cwd) shutil.rmtree(str(tmp), ignore_errors=True) for k in ["VAULT_1_NAME", "VAULT_1_PATH", "OBSIGATE_AUTH_ENABLED", "OBSIGATE_WATCHER_ENABLED", "DRACO_ADMIN_MACHINE_KEY", "DRACO_SSO_MACHINE_KEY", "DRACO_SESSION_SECRET"]: os.environ.pop(k, None) def _login(client, username, password): return client.post("/api/auth/login", json={"username": username, "password": password}) def _creer(client, username="nouveau", password="motdepasse-initial", **kw): corps = {"username": username, "password": password, "display_name": kw.pop("display_name", "Nouveau Compte"), "role": kw.pop("role", "user")} corps.update(kw) return client.post(BASE, headers=MACHINE_HEADERS, json=corps) # ── Protection par clé machine ────────────────────────────────────────── class TestProtection: def test_sans_cle_refuse(self, machine_client): assert machine_client.get(BASE).status_code == 401 assert machine_client.post( BASE, json={"username": "x1", "password": "motdepasse"} ).status_code == 401 def test_mauvaise_cle_refusee(self, machine_client): resp = machine_client.get( BASE, headers={"X-Draco-Machine-Key": "mauvaise-cle"}) assert resp.status_code == 401 def test_cle_sso_ne_suffit_pas(self, machine_client): """La clé des révocations (lecture) n'ouvre pas l'administration.""" resp = machine_client.get( BASE, headers={"X-Draco-Machine-Key": "cle-sso-test"}) assert resp.status_code == 401 def test_session_admin_ne_suffit_pas(self, machine_client): """Même une session administratrice ordinaire est refusée.""" token = _login(machine_client, "admin", "chab30").json()["access_token"] resp = machine_client.get( BASE, headers={"Authorization": f"Bearer {token}"}) assert resp.status_code == 401 def test_non_configuree_inerte(self, machine_client, monkeypatch): monkeypatch.delenv("DRACO_ADMIN_MACHINE_KEY") resp = machine_client.get(BASE, headers=MACHINE_HEADERS) assert resp.status_code == 404 # ── Lecture ───────────────────────────────────────────────────────────── class TestLecture: def test_liste(self, machine_client): resp = machine_client.get(BASE, headers=MACHINE_HEADERS) assert resp.status_code == 200 data = resp.json() assert data["total"] >= 2 noms = [c["username"] for c in data["comptes"]] assert "admin" in noms and "mfauser" in noms def test_profil_etendu_sans_secret(self, machine_client): resp = machine_client.get(f"{BASE}/mfauser", headers=MACHINE_HEADERS) assert resp.status_code == 200 profil = resp.json() assert profil["username"] == "mfauser" assert profil["display_name"] == "Utilisateur MFA" assert profil["language"] == "fr" assert profil["avatar_present"] is False assert profil["role"] == "user" assert profil["active"] is True assert profil["mfa_enabled"] is True # Jamais de secret : ni le haché du mot de passe, ni le TOTP. brut = resp.text assert "password_hash" not in brut assert "mfa_secret" not in brut assert "mfa_recovery" not in brut def test_compte_inconnu_404(self, machine_client): resp = machine_client.get(f"{BASE}/personne", headers=MACHINE_HEADERS) assert resp.status_code == 404 # ── Création / modification ───────────────────────────────────────────── class TestCreationModification: def test_creation_puis_lecture(self, machine_client): resp = _creer(machine_client) assert resp.status_code == 201, resp.text profil = resp.json() assert profil["username"] == "nouveau" assert profil["must_change_password"] is True lu = machine_client.get(f"{BASE}/nouveau", headers=MACHINE_HEADERS) assert lu.status_code == 200 assert lu.json()["display_name"] == "Nouveau Compte" # Le compte créé se connecte, avec changement forcé signalé. login = _login(machine_client, "nouveau", "motdepasse-initial") assert login.status_code == 200 assert login.json()["user"]["must_change_password"] is True def test_creation_doublon_409(self, machine_client): assert _creer(machine_client).status_code == 201 assert _creer(machine_client).status_code == 409 assert _creer(machine_client, username="admin").status_code == 409 def test_modification(self, machine_client): assert _creer(machine_client).status_code == 201 resp = machine_client.patch( f"{BASE}/nouveau", headers=MACHINE_HEADERS, json={"display_name": "Nom Modifié", "role": "admin"}) assert resp.status_code == 200 assert resp.json()["display_name"] == "Nom Modifié" assert resp.json()["role"] == "admin" def test_modification_inconnue_404(self, machine_client): resp = machine_client.patch( f"{BASE}/personne", headers=MACHINE_HEADERS, json={"display_name": "X"}) assert resp.status_code == 404 # ── Désactivation / réactivation ──────────────────────────────────────── class TestDesactivation: def test_desactivation_revoque_puis_reactivation(self, machine_client): from backend import draco_session assert _creer(machine_client).status_code == 201 login = _login(machine_client, "nouveau", "motdepasse-initial") assert login.status_code == 200 token_natif = login.json()["access_token"] profil = machine_client.get( f"{BASE}/nouveau", headers=MACHINE_HEADERS).json() token_draco, jti, _exp = draco_session.issue_token( {"id": profil["id"], "username": "nouveau", "display_name": "Nouveau Compte"}) assert draco_session.verify_token(token_draco) is not None resp = machine_client.post(f"{BASE}/nouveau/desactiver", headers=MACHINE_HEADERS) assert resp.status_code == 200 assert resp.json()["active"] is False # Connexion refusée, ancienne session native refusée, # jti draco révoqué et publié dans la liste des révocations. assert _login(machine_client, "nouveau", "motdepasse-initial").status_code == 401 me = machine_client.get( "/api/auth/me", headers={"Authorization": f"Bearer {token_natif}"}) assert me.status_code == 401 assert draco_session.verify_token(token_draco) is None revocations = machine_client.get( "/api/sso/revocations", headers={"X-Draco-Machine-Key": "cle-sso-test"}) assert revocations.status_code == 200 assert jti in [r["jti"] for r in revocations.json()["revoked"]] resp = machine_client.post(f"{BASE}/nouveau/reactiver", headers=MACHINE_HEADERS) assert resp.status_code == 200 assert resp.json()["active"] is True # Réactivation : nouvelle connexion OK, ancien jti toujours révoqué. assert _login(machine_client, "nouveau", "motdepasse-initial").status_code == 200 assert draco_session.verify_token(token_draco) is None # ── Réinitialisation du mot de passe ──────────────────────────────────── class TestReinitialisationMotDePasse: def test_ancien_refuse_nouveau_accepte_changement_force( self, machine_client): assert _creer(machine_client).status_code == 201 resp = machine_client.post( f"{BASE}/nouveau/reinitialiser-mot-de-passe", headers=MACHINE_HEADERS, json={"password": "nouveau-mot-de-passe"}) assert resp.status_code == 200 assert resp.json()["must_change_password"] is True assert _login(machine_client, "nouveau", "motdepasse-initial").status_code == 401 login = _login(machine_client, "nouveau", "nouveau-mot-de-passe") assert login.status_code == 200 assert login.json()["user"]["must_change_password"] is True # Le changement effectif par l'utilisateur efface le marqueur. token = login.json()["access_token"] changement = machine_client.post( "/api/auth/change-password", headers={"Authorization": f"Bearer {token}"}, json={"current_password": "nouveau-mot-de-passe", "new_password": "mot-de-passe-choisi"}) assert changement.status_code == 200, changement.text login2 = _login(machine_client, "nouveau", "mot-de-passe-choisi") assert login2.status_code == 200 assert login2.json()["user"]["must_change_password"] is False # ── Réinitialisation du MFA ───────────────────────────────────────────── class TestReinitialisationMfa: def test_mfa_reinitialise(self, machine_client): # Avant : la connexion exige le second facteur. login = _login(machine_client, "mfauser", "motdepasse-mfa") assert login.status_code == 200 assert login.json().get("mfa_required") is True resp = machine_client.post(f"{BASE}/mfauser/reinitialiser-mfa", headers=MACHINE_HEADERS) assert resp.status_code == 200 assert resp.json()["mfa_enabled"] is False # Après : connexion directe, sans second facteur. login = _login(machine_client, "mfauser", "motdepasse-mfa") assert login.status_code == 200 assert "access_token" in login.json() # ── Suppression ───────────────────────────────────────────────────────── class TestSuppression: def test_suppression(self, machine_client): assert _creer(machine_client).status_code == 201 resp = machine_client.delete(f"{BASE}/nouveau", headers=MACHINE_HEADERS) assert resp.status_code == 200 assert machine_client.get( f"{BASE}/nouveau", headers=MACHINE_HEADERS).status_code == 404 assert _login(machine_client, "nouveau", "motdepasse-initial").status_code == 401 def test_suppression_inconnue_404(self, machine_client): resp = machine_client.delete(f"{BASE}/personne", headers=MACHINE_HEADERS) assert resp.status_code == 404 # ── Journalisation ────────────────────────────────────────────────────── class TestJournalisation: def test_actions_consignees(self, machine_client): assert _creer(machine_client).status_code == 201 machine_client.post(f"{BASE}/nouveau/desactiver", headers=MACHINE_HEADERS) audit_file = Path("data/audit.log") assert audit_file.exists() entrees = [json.loads(ligne) for ligne in audit_file.read_text( encoding="utf-8").splitlines() if ligne.strip()] actions = {(e["action"], e["cible"]) for e in entrees if e.get("cible") == "nouveau"} assert ("compte_cree", "nouveau") in actions assert ("compte_desactive", "nouveau") in actions for e in entrees: if e.get("cible") == "nouveau": assert e["acteur"] == "hub-test" assert e["timestamp"]