diff --git a/backend/draco_session.py b/backend/draco_session.py index 6b0e335..465ba6d 100644 --- a/backend/draco_session.py +++ b/backend/draco_session.py @@ -25,6 +25,10 @@ # l'expiration) pour que l'administration des comptes puisse révoquer # toutes les sessions draco d'un utilisateur (désactivation, # suppression, réinitialisation du mot de passe) via revoke(). +# HUB-22 (#227) ajoute la lecture de ce registre +# (issued_sessions_for_user) et la révocation unitaire contrôlée par +# appartenance (revoke_one_for_user), exposées par l'API machine des +# sessions (backend/routers/comptes_machine.py). from __future__ import annotations @@ -179,6 +183,52 @@ def revoke_all_for_user(sub: str) -> int: return len(actifs) +def issued_sessions_for_user(sub: str) -> list[dict]: + """Sessions draco actives émises pour ``sub`` (HUB-22, #227). + + Retourne ``[{jti, exp, revoquee}]`` triée par ``exp`` croissant, + où ``revoquee`` indique si le ``jti`` figure déjà dans la liste des + révocations. Les entrées expirées de cet utilisateur sont purgées + du registre au passage (même patron que ``_register_issued``). + """ + if not sub: + return [] + with _lock: + store = _load_issued() + now = int(time.time()) + tokens = store.get(str(sub), {}) + actifs = {j: e for j, e in tokens.items() if e > now} + if len(actifs) != len(tokens): + if actifs: + store[str(sub)] = actifs + else: + store.pop(str(sub), None) + _save_issued() + revoques = _load_revoked() + entrees = sorted(actifs.items(), key=lambda kv: kv[1]) + return [{"jti": jti, "exp": exp, "revoquee": jti in revoques} + for jti, exp in entrees] + + +def revoke_one_for_user(sub: str, jti: str) -> bool: + """Révoque un seul ``jti`` draco de ``sub`` (HUB-22, #227). + + Ne révoque (via ``revoke``) que si le ``jti`` appartient bien à + l'ensemble émis de ce ``sub`` et n'est pas expiré — jamais de + révocation d'une session d'un autre utilisateur. Retourne ``True`` + si la révocation a eu lieu, ``False`` sinon. + """ + if not sub or not jti: + return False + with _lock: + store = _load_issued() + exp = store.get(str(sub), {}).get(jti) + if exp is None or exp <= int(time.time()): + return False + revoke(jti, exp) + return True + + # ── Émission / vérification ───────────────────────────────────────────── def issue_token(user: dict) -> tuple[str, str, int]: