NotBigGhostandClaude Opus 4.8
3f8667b561
logout отзывает предъявленный токен по jti (in-memory denylist до exp); смена и сброс пароля инкрементят users.token_version (claim ver в JWT, сверка в auth/deps) — все прежние сессии отзываются. Своё устройство при смене пароля остаётся в сессии (перевыдача cookie). Миграция 0013. #57 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XfTsytzT6TojfmprRDKiV6
54 lines
2.2 KiB
Python
54 lines
2.2 KiB
Python
"""Точечный отзыв отдельных JWT по `jti` — для logout (выход именно этого токена).
|
||
|
||
Список живёт в памяти процесса, как throttle и SSE-шина: рассчитан на один воркер uvicorn.
|
||
Рестарт очищает список — это приемлемо: записи и так живут лишь до `exp` токена, а на новый
|
||
процесс приходят уже свежие cookie. Для «выйти со всех устройств» и отзыва при смене пароля
|
||
используется `token_version` у пользователя (см. models.User, auth/deps), а не этот список.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import threading
|
||
import time
|
||
|
||
# Выше этого числа записей при отзыве вычищаем протухшие, чтобы поток logout'ов не копил память.
|
||
_PRUNE_ABOVE = 10_000
|
||
|
||
|
||
class RevokedTokens:
|
||
"""Множество отозванных `jti` с временем истечения (unix-время, как `exp` в JWT)."""
|
||
|
||
def __init__(self) -> None:
|
||
self._revoked: dict[str, float] = {}
|
||
self._lock = threading.Lock()
|
||
|
||
def revoke(self, jti: str | None, expires_at: float) -> None:
|
||
if not jti:
|
||
return
|
||
now = time.time()
|
||
with self._lock:
|
||
if len(self._revoked) > _PRUNE_ABOVE:
|
||
for key, exp in list(self._revoked.items()):
|
||
if exp <= now:
|
||
del self._revoked[key]
|
||
self._revoked[jti] = expires_at
|
||
|
||
def is_revoked(self, jti: str | None) -> bool:
|
||
if not jti:
|
||
return False
|
||
now = time.time()
|
||
with self._lock:
|
||
exp = self._revoked.get(jti)
|
||
if exp is None:
|
||
return False
|
||
if exp <= now:
|
||
del self._revoked[jti] # протухла — заодно вычищаем
|
||
return False
|
||
return True
|
||
|
||
def clear(self) -> None:
|
||
with self._lock:
|
||
self._revoked.clear()
|
||
|
||
|
||
revoked_tokens = RevokedTokens()
|