ObsiGate 2.84.0 : API d'administration des comptes machine-à-machine (HUB-13, fusion #127, #226) — routes /api/admin/machine/comptes protégées par DRACO_ADMIN_MACHINE_KEY, révocation des jti draco à la désactivation, must_change_password, journalisation d'audit

tests/test_comptes_machine.py

@@ -0,0 +1,377 @@
+# 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"]