app/security/token_revocation.py

"""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]