Отзыв JWT при выходе и смене пароля
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
This commit is contained in:
@@ -9,6 +9,7 @@ import jwt
|
||||
from fastapi import Request, Response
|
||||
|
||||
from app.core.config import settings
|
||||
from app.core.token_revocation import revoked_tokens
|
||||
|
||||
USER_COOKIE = "fs_session"
|
||||
ADMIN_COOKIE = "fs_admin"
|
||||
@@ -37,7 +38,13 @@ def verify_password(password: str, password_hash: str) -> bool:
|
||||
|
||||
# ─── JWT ─────────────────────────────────────────────────────────────────────
|
||||
|
||||
def create_token(subject: str | int, audience: str, ttl_minutes: int, provider: str = "") -> str:
|
||||
def create_token(
|
||||
subject: str | int,
|
||||
audience: str,
|
||||
ttl_minutes: int,
|
||||
provider: str = "",
|
||||
token_version: int = 0,
|
||||
) -> str:
|
||||
now = datetime.now(timezone.utc)
|
||||
payload = {
|
||||
"sub": str(subject),
|
||||
@@ -46,6 +53,9 @@ def create_token(subject: str | int, audience: str, ttl_minutes: int, provider:
|
||||
"exp": int((now + timedelta(minutes=ttl_minutes)).timestamp()),
|
||||
"jti": secrets.token_hex(8),
|
||||
"provider": provider,
|
||||
# Версия сессий владельца: при её росте (смена/сброс пароля) старые токены
|
||||
# с меньшим `ver` отклоняются в auth/deps — отзыв всех прежних сессий (#57).
|
||||
"ver": token_version,
|
||||
}
|
||||
return jwt.encode(payload, settings.secret_key, algorithm=settings.jwt_algorithm)
|
||||
|
||||
@@ -95,9 +105,9 @@ def fresh_csrf_set_cookie() -> tuple[bytes, bytes]:
|
||||
return next((k, v) for k, v in carrier.raw_headers if k == b"set-cookie")
|
||||
|
||||
|
||||
def set_user_session(response: Response, user_id: int, provider: str) -> None:
|
||||
def set_user_session(response: Response, user_id: int, provider: str, token_version: int = 0) -> None:
|
||||
ttl = settings.jwt_user_ttl_minutes
|
||||
token = create_token(user_id, AUDIENCE_USER, ttl, provider)
|
||||
token = create_token(user_id, AUDIENCE_USER, ttl, provider, token_version)
|
||||
response.set_cookie(
|
||||
key=USER_COOKIE,
|
||||
value=token,
|
||||
@@ -111,9 +121,9 @@ def set_user_session(response: Response, user_id: int, provider: str) -> None:
|
||||
_set_csrf_cookie(response)
|
||||
|
||||
|
||||
def set_admin_session(response: Response, admin_id: int) -> None:
|
||||
def set_admin_session(response: Response, admin_id: int, token_version: int = 0) -> None:
|
||||
ttl = settings.jwt_admin_ttl_minutes
|
||||
token = create_token(admin_id, AUDIENCE_ADMIN, ttl, "local")
|
||||
token = create_token(admin_id, AUDIENCE_ADMIN, ttl, "local", token_version)
|
||||
response.set_cookie(
|
||||
key=ADMIN_COOKIE,
|
||||
value=token,
|
||||
@@ -135,6 +145,33 @@ def clear_admin_session(response: Response) -> None:
|
||||
response.delete_cookie(ADMIN_COOKIE, path=_ADMIN_PATH, domain=settings.cookie_domain_value)
|
||||
|
||||
|
||||
def revoke_session_token(request: Request, cookie_name: str, audience: str) -> None:
|
||||
"""Отзывает предъявленный в cookie токен (по `jti`) до его `exp` — точечный logout.
|
||||
|
||||
Убивает именно этот токен (украденный/оставленный), не трогая другие устройства.
|
||||
Некорректный/просроченный токен отзывать нечего — молча выходим."""
|
||||
token = request.cookies.get(cookie_name)
|
||||
if not token:
|
||||
return
|
||||
try:
|
||||
payload = jwt.decode(
|
||||
token,
|
||||
settings.secret_key,
|
||||
algorithms=[settings.jwt_algorithm],
|
||||
audience=audience,
|
||||
)
|
||||
except jwt.PyJWTError:
|
||||
return
|
||||
exp = payload.get("exp")
|
||||
if exp is not None:
|
||||
revoked_tokens.revoke(payload.get("jti"), float(exp))
|
||||
|
||||
|
||||
def is_session_revoked(payload: dict) -> bool:
|
||||
"""Отозван ли этот токен точечно (через logout)."""
|
||||
return revoked_tokens.is_revoked(payload.get("jti"))
|
||||
|
||||
|
||||
def client_ip(request: Request) -> str | None:
|
||||
"""IP клиента для журнала аудита.
|
||||
|
||||
|
||||
Reference in New Issue
Block a user