diff --git a/tests/test_comptes_machine.py b/tests/test_comptes_machine.py index f49332d..9607b49 100644 --- a/tests/test_comptes_machine.py +++ b/tests/test_comptes_machine.py @@ -354,6 +354,163 @@ class TestSuppression: 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: