Spaces:
Running
Running
File size: 971 Bytes
5aa2bd9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 | """JWT replay protection — jti denylist + per-user token version."""
import time
import uuid
from threading import Lock
_lock = Lock()
_revoked_jti: dict[str, float] = {} # jti -> expiry unix time
_TTL_BUFFER_SEC = 86400 * 2
def new_jti() -> str:
return str(uuid.uuid4())
def revoke_jti(jti: str, exp_unix: float | None = None) -> None:
if not jti:
return
until = exp_unix or (time.time() + _TTL_BUFFER_SEC)
with _lock:
_revoked_jti[jti] = until
_purge_expired_locked()
def is_jti_revoked(jti: str | None) -> bool:
if not jti:
return False
now = time.time()
with _lock:
_purge_expired_locked()
until = _revoked_jti.get(jti)
return until is not None and until > now
def _purge_expired_locked() -> None:
now = time.time()
expired = [k for k, v in _revoked_jti.items() if v <= now]
for k in expired:
del _revoked_jti[k]
|