535 lines
22 KiB
Python
535 lines
22 KiB
Python
# 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
|
|
|
|
|
|
# ── Sessions draco (HUB-22) ─────────────────────────────────────────────
|
|
|
|
def _emettre(client, username, nombre=1):
|
|
"""Émet ``nombre`` jetons draco pour un compte ; retourne
|
|
[(jeton, jti, exp), …]."""
|
|
from backend import draco_session
|
|
|
|
profil = client.get(f"{BASE}/{username}",
|
|
headers=MACHINE_HEADERS).json()
|
|
return [draco_session.issue_token(
|
|
{"id": profil["id"], "username": username,
|
|
"display_name": profil["display_name"]})
|
|
for _ in range(nombre)]
|
|
|
|
|
|
class TestSessions:
|
|
def test_liste_sessions(self, machine_client):
|
|
from datetime import datetime
|
|
|
|
assert _creer(machine_client).status_code == 201
|
|
emis = _emettre(machine_client, "nouveau", 2)
|
|
|
|
resp = machine_client.get(f"{BASE}/nouveau/sessions",
|
|
headers=MACHINE_HEADERS)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["username"] == "nouveau"
|
|
assert len(data["sessions"]) == 2
|
|
jtis = [s["jti"] for s in data["sessions"]]
|
|
assert jtis == [j for _t, j, _e in emis] # exp croissant
|
|
for session, (_t, _j, exp) in zip(data["sessions"], emis):
|
|
assert session["exp"] == exp
|
|
assert session["revoquee"] is False
|
|
echeance = datetime.fromisoformat(session["expire_le"])
|
|
assert echeance.tzinfo is not None
|
|
assert int(echeance.timestamp()) == exp
|
|
|
|
def test_liste_sessions_vide_et_compte_inconnu(self, machine_client):
|
|
resp = machine_client.get(f"{BASE}/mfauser/sessions",
|
|
headers=MACHINE_HEADERS)
|
|
assert resp.status_code == 200
|
|
assert resp.json() == {"username": "mfauser", "sessions": []}
|
|
resp = machine_client.get(f"{BASE}/personne/sessions",
|
|
headers=MACHINE_HEADERS)
|
|
assert resp.status_code == 404
|
|
|
|
def test_revocation_unitaire(self, machine_client):
|
|
from backend import draco_session
|
|
|
|
assert _creer(machine_client).status_code == 201
|
|
(t1, jti1, _e1), (t2, _jti2, _e2) = _emettre(
|
|
machine_client, "nouveau", 2)
|
|
|
|
resp = machine_client.post(
|
|
f"{BASE}/nouveau/sessions/{jti1}/revoquer",
|
|
headers=MACHINE_HEADERS)
|
|
assert resp.status_code == 200
|
|
assert resp.json() == {"jti": jti1, "revoquee": True}
|
|
|
|
# La session révoquée est morte, l'autre vit ; la liste la
|
|
# signale comme révoquée.
|
|
assert draco_session.verify_token(t1) is None
|
|
assert draco_session.verify_token(t2) is not None
|
|
sessions = machine_client.get(
|
|
f"{BASE}/nouveau/sessions", headers=MACHINE_HEADERS).json()
|
|
etats = {s["jti"]: s["revoquee"] for s in sessions["sessions"]}
|
|
assert etats[jti1] is True
|
|
|
|
def test_revocation_cross_utilisateur_404(self, machine_client):
|
|
from backend import draco_session
|
|
|
|
assert _creer(machine_client).status_code == 201
|
|
(token, jti, _e), = _emettre(machine_client, "nouveau", 1)
|
|
|
|
# Le jti de « nouveau » demandé sous « mfauser » : 404, et la
|
|
# session de « nouveau » reste valide (jamais de révocation
|
|
# cross-utilisateur).
|
|
resp = machine_client.post(
|
|
f"{BASE}/mfauser/sessions/{jti}/revoquer",
|
|
headers=MACHINE_HEADERS)
|
|
assert resp.status_code == 404
|
|
assert draco_session.verify_token(token) is not None
|
|
|
|
# Jti inconnu et compte inconnu : 404 également.
|
|
resp = machine_client.post(
|
|
f"{BASE}/nouveau/sessions/jti-inconnu/revoquer",
|
|
headers=MACHINE_HEADERS)
|
|
assert resp.status_code == 404
|
|
resp = machine_client.post(
|
|
f"{BASE}/personne/sessions/{jti}/revoquer",
|
|
headers=MACHINE_HEADERS)
|
|
assert resp.status_code == 404
|
|
|
|
def test_revoquer_toutes_sans_desactiver(self, machine_client):
|
|
from backend import draco_session
|
|
|
|
assert _creer(machine_client).status_code == 201
|
|
emis = _emettre(machine_client, "nouveau", 2)
|
|
(autre_token, _j, _e), = _emettre(machine_client, "mfauser", 1)
|
|
|
|
resp = machine_client.post(
|
|
f"{BASE}/nouveau/sessions/revoquer-toutes",
|
|
headers=MACHINE_HEADERS)
|
|
assert resp.status_code == 200
|
|
assert resp.json() == {"revoquees": 2}
|
|
|
|
for token, _jti, _exp in emis:
|
|
assert draco_session.verify_token(token) is None
|
|
# Les sessions des autres comptes ne sont pas touchées.
|
|
assert draco_session.verify_token(autre_token) is not None
|
|
|
|
# Plus rien à révoquer : le registre a été vidé pour ce compte
|
|
# (avant toute nouvelle connexion, qui émettrait une session).
|
|
resp = machine_client.post(
|
|
f"{BASE}/nouveau/sessions/revoquer-toutes",
|
|
headers=MACHINE_HEADERS)
|
|
assert resp.json() == {"revoquees": 0}
|
|
|
|
# Déconnexion globale ≠ désactivation : le compte reste actif
|
|
# et une nouvelle connexion fonctionne.
|
|
profil = machine_client.get(f"{BASE}/nouveau",
|
|
headers=MACHINE_HEADERS).json()
|
|
assert profil["active"] is True
|
|
assert _login(machine_client, "nouveau",
|
|
"motdepasse-initial").status_code == 200
|
|
|
|
def test_sessions_protegees_par_cle_machine(self, machine_client,
|
|
monkeypatch):
|
|
assert machine_client.get(
|
|
f"{BASE}/nouveau/sessions").status_code == 401
|
|
assert machine_client.post(
|
|
f"{BASE}/nouveau/sessions/un-jti/revoquer").status_code == 401
|
|
assert machine_client.post(
|
|
f"{BASE}/nouveau/sessions/revoquer-toutes").status_code == 401
|
|
monkeypatch.delenv("DRACO_ADMIN_MACHINE_KEY")
|
|
assert machine_client.get(
|
|
f"{BASE}/nouveau/sessions",
|
|
headers=MACHINE_HEADERS).status_code == 404
|
|
|
|
def test_sessions_journalisees(self, machine_client):
|
|
assert _creer(machine_client).status_code == 201
|
|
(_t, jti, _e), = _emettre(machine_client, "nouveau", 1)
|
|
machine_client.post(f"{BASE}/nouveau/sessions/{jti}/revoquer",
|
|
headers=MACHINE_HEADERS)
|
|
machine_client.post(f"{BASE}/nouveau/sessions/revoquer-toutes",
|
|
headers=MACHINE_HEADERS)
|
|
|
|
audit_file = Path("data/audit.log")
|
|
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 ("session_revoquee", "nouveau") in actions
|
|
assert ("sessions_revoquees", "nouveau") in actions
|
|
|
|
|
|
# ── 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"]
|