feat: HUB-22 — API machine des sessions draco : liste et révocation des sessions (2.85.0, #227)

This commit is contained in:
Spark
2026-11-06 11:07:22 -05:00
parent 5127510509
commit 222cd0b562
+50
View File
@@ -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]: