JWT-аутентификация в FastAPI: токены, ротация и тесты

Базовый пример аутентификации из официальной документации FastAPI обычно ограничивается выдачей одного долгоживущего JWT. Для учебных проектов этого достаточно, но в рабочей среде такой подход создает прямую угрозу безопасности: украденный токен нельзя отозвать до истечения срока действия без введения постоянных запросов в базу данных на каждый вызов API.

Полноценная система аутентификации строится на разделении токенов на две роли (access и refresh), серверном контроле сессий, ротации учетных данных и автоматическом реагировании на попытки повторного использования токенов.

Архитектура и модель угроз

Разделение ответственности токенов

Использование двух токенов решает фундаментальный конфликт между производительностью (stateless) и контролем доступа (stateful):

Access-токен (короткоживущий, 5–15 минут): передается в заголовке Authorization: Bearer <token>. Сервер валидирует его криптографическую подпись и claims локально на уровне FastAPI-зависимостей (Depends), не обращаясь к базе данных или распределенному кэшу. Если токен скомпрометирован, окно атаки жестко ограничено минутами.
Refresh-токен (долгоживущий, 7–30 дней): используется исключительно для отправки на специализированный эндпоинт /auth/refresh с целью получения новой пары токенов. При каждом обращении сервер проверяет его статус в базе данных сессий.

+--------+                               +---------+                               +--------+
| Client |                               | FastAPI |                               |   DB   |
+--------+                               +---------+                               +--------+
    |                                         |                                         |
    |  1. POST /auth/login                    |                                         |
    |---------------------------------------->|  Проверка пароля                        |
    |                                         |---------------------------------------->|
    |                                         |  Сохранение хеша refresh-токена         |
    |  2. Возврат Access + Refresh            |<----------------------------------------|
    |<----------------------------------------|                                         |
    |                                         |                                         |
    |  3. GET /api/v1/resource (Access Token) |                                         |
    |---------------------------------------->|  Локальная валидация JWT (без БД)       |
    |<----------------------------------------|  200 OK / Ответ с данными               |
    |                                         |                                         |
    |  4. POST /auth/refresh (Refresh Token)  |                                         |
    |---------------------------------------->|  Поиск сессии и проверка хеша токена    |
    |                                         |---------------------------------------->|
    |  5. Новая пара Access + Refresh         |  Атомарная ротация сессии               |
    |<----------------------------------------|<----------------------------------------|

Места хранения токенов на клиенте: XSS против CSRF

Хранение чувствительных данных в localStorage или sessionStorage браузера подвергает приложение риску: любая XSS-уязвимость (включая внедрение через сторонние npm-пакеты) позволяет злоумышленнику прочитать токен через JavaScript и отправить его на внешний сервер.

Рекомендованная схема для веб-интерфейсов:

Refresh-токен: передается сервером исключительно в заголовке Set-Cookie с флагами HttpOnly, Secure и SameSite=Lax (или Strict). JavaScript браузера не имеет доступа к содержимому этой cookie, что нивелирует риск кражи через XSS.
Access-токен: сохраняется в памяти JavaScript-приложения (in-memory state) или также отправляется через защищенную cookie с коротким TTL. При передаче через cookie требуется обязательная CSRF-защита (например, double-submit cookie или проверка origin/referer).
Мобильные клиенты и межсервисные вызовы: используют безопасные системные хранилища (iOS Keychain, Android Keystore) и передают токены через стандартные HTTP-заголовки.

Жизненный цикл токенов и структура JWT

Токены формируются в соответствии со спецификацией RFC 7519. Чтобы исключить неопределенность при валидации, каждый токен должен содержать четкий набор зарегистрированных claims (заявлений).

Обязательные claims

{
  "sub": "usr_9b1deb4d-3b7d-4bad-9bdd-2b0d7b3dcb6d",
  "type": "access",
  "jti": "tok_1f8a5e3c-8e2b-4c12-a7d1-ef594a112233",
  "iat": 1710000000,
  "exp": 1710000900,
  "iss": "https://auth.example.com",
  "aud": "https://api.example.com"
}

sub (Subject): постоянный внутренний идентификатор пользователя (UUID или целочисленный ID). Не используйте изменяемые поля (например, email).
type: кастомный claim (access или refresh), блокирующий попытку отправить refresh-токен в бизнес-эндпоинты вместо access-токена.
jti (JWT ID): уникальный идентификатор токена (UUID v4), необходимый для связывания токена с записью сессии и отслеживания повторного использования.
iat (Issued At): временная метка выпуска (Unix timestamp).
exp (Expiration Time): время истечения срока действия.
iss / aud: валидация издателя и целевой аудитории API для предотвращения межсервисной подмены контекста.

Серверный учет сессий и ротация refresh-токенов (RFC 9700)

Stateless-подход для refresh-токенов неприменим, так как сервер должен иметь возможность принудительно аннулировать доступ пользователя.

Схема хранения сессий в БД

В базе данных категорически запрещено хранить refresh-токены в открытом виде. В случае утечки дампа базы злоумышленник получит доступ ко всем активным сессиям пользователей. Сохраняется только криптографический хеш токена (SHA-256):

CREATE TABLE user_sessions (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    refresh_token_hash VARCHAR(64) NOT NULL UNIQUE,
    jti VARCHAR(36) NOT NULL UNIQUE,
    parent_jti VARCHAR(36) NULL,
    user_agent VARCHAR(255),
    ip_address INET,
    is_revoked BOOLEAN NOT NULL DEFAULT FALSE,
    expires_at TIMESTAMPTZ NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_user_sessions_user_id ON user_sessions(user_id);
CREATE INDEX idx_user_sessions_jti ON user_sessions(jti);

Алгоритм ротации и обнаружение компрометации (Reuse Detection)

В соответствии с RFC 9700 (Security Best Current Practice for OAuth 2.0), при каждом обновлении access-токена старый refresh-токен однократно инвалидируется, а клиенту выдается полностью новая пара (Access, Refresh).

Цепочка ротации:
Token A (использован) -> Token B (использован) -> Token C (активен)

Сценарий атаки:
Злоумышленник перехватил Token A и пытается применить его после того, как легитимный клиент уже получил Token B.
Сервер видит запрос с Token A (jti помечен как revoked/замененный) -> Срабатывает Reuse Detection.
Действие сервера: Все токены цепочки (A, B, C) пользователя аннулируются.

Логика обработки на сервере:

Клиент присылает refresh-токен.
Сервер декодирует токен, валидирует подпись и срок exp.
По jti находится запись в базе данных.
Случай 1 (Успех): Запись активна (is_revoked == False), и хеш совпадает. Сервер атомарно помечает текущую сессию как использованную (или обновляет хеш и jti на новые) и возвращает новую пару.
Случай 2 (Атака / Повторное использование): Запись уже помечена как отозванная (is_revoked == True). Это сигнал, что токен был скопирован и применен повторно. Сервер немедленно отзывает все активные сессии данного пользователя (UPDATE user_sessions SET is_revoked = TRUE WHERE user_id = :user_id) и возвращает HTTP 401 Unauthorized.

Реализация механизма в FastAPI

Ниже приведена серверная логика на базе Python, PyJWT и стандартных инструментов FastAPI.

Утилиты безопасности и генерация токенов

import hashlib
from datetime import datetime, timedelta, timezone
import uuid
import jwt
from fastapi import HTTPException, status

SECRET_KEY = "CHANGE_THIS_IN_PRODUCTION_TO_A_SECURE_KEY"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 15
REFRESH_TOKEN_EXPIRE_DAYS = 7

def hash_token(token: str) -> str:
    return hashlib.sha256(token.encode("utf-8")).hexdigest()

def create_jwt_token(
    user_id: str,
    token_type: str,
    expires_delta: timedelta,
    jti: str | None = None
) -> tuple[str, str]:
    token_jti = jti or str(uuid.uuid4())
    now = datetime.now(timezone.utc)
    payload = {
        "sub": str(user_id),
        "type": token_type,
        "jti": token_jti,
        "iat": now,
        "exp": now + expires_delta,
        "iss": "fastapi-auth-app"
    }
    encoded_jwt = jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt, token_jti

def decode_and_validate_token(token: str, expected_type: str) -> dict:
    try:
        payload = jwt.decode(
            token,
            SECRET_KEY,
            algorithms=[ALGORITHM],
            issuer="fastapi-auth-app",
            options={"require": ["exp", "iat", "sub", "jti", "type"]}
        )
        if payload.get("type") != expected_type:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail=f"Недопустимый тип токена. Ожидался '{expected_type}'",
                headers={"WWW-Authenticate": "Bearer"},
            )
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Срок действия токена истек",
            headers={"WWW-Authenticate": "Bearer"},
        )
    except jwt.PyJWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Не удалось проверить учетные данные",
            headers={"WWW-Authenticate": "Bearer"},
        )

Зависимость проверки access-токена

Зависимость get_current_user осуществляет stateless-проверку без лишней нагрузки на базу данных:

from fastapi import Depends
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

class TokenData(BaseModel):
    user_id: str
    jti: str

async def get_current_user(token: str = Depends(oauth2_scheme)) -> TokenData:
    payload = decode_and_validate_token(token, expected_type="access")
    return TokenData(user_id=payload["sub"], jti=payload["jti"])

Эндпоинты авторизации, ротации и отзыва

from fastapi import APIRouter, Response, Request, HTTPException, status
from pydantic import BaseModel

router = APIRouter(prefix="/auth", tags=["auth"])

class LoginRequest(BaseModel):
    username: str
    password: str

class TokenResponse(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str = "bearer"

class RefreshRequest(BaseModel):
    refresh_token: str

# In-memory хранилище для демонстрации (в production используется БД/SQLAlchemy или Redis)
session_db: dict[str, dict] = {}

@router.post("/login", response_model=TokenResponse)
async def login(credentials: LoginRequest):
    if credentials.username != "admin" or credentials.password != "secret":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Неверное имя пользователя или пароль"
        )
    
    user_id = "usr_system_admin_01"
    
    # 1. Генерация токенов
    access_token, _ = create_jwt_token(
        user_id=user_id,
        token_type="access",
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    refresh_token, refresh_jti = create_jwt_token(
        user_id=user_id,
        token_type="refresh",
        expires_delta=timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
    )
    
    # 2. Сохранение метаданных сессии
    session_db[refresh_jti] = {
        "user_id": user_id,
        "token_hash": hash_token(refresh_token),
        "is_revoked": False,
        "expires_at": datetime.now(timezone.utc) + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
    }
    
    return TokenResponse(
        access_token=access_token,
        refresh_token=refresh_token
    )

@router.post("/refresh", response_model=TokenResponse)
async def refresh_tokens(payload_in: RefreshRequest):
    payload = decode_and_validate_token(payload_in.refresh_token, expected_type="refresh")
    old_jti = payload["jti"]
    user_id = payload["sub"]
    token_hash = hash_token(payload_in.refresh_token)
    
    session = session_db.get(old_jti)
    
    # Обнаружение повторного использования (Reuse Detection)
    if not session or session["is_revoked"]:
        for jti_key, s_data in list(session_db.items()):
            if s_data["user_id"] == user_id:
                s_data["is_revoked"] = True
                
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Обнаружено повторное использование токена. Все сессии аннулированы."
        )
        
    if session["token_hash"] != token_hash:
        session["is_revoked"] = True
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Несоответствие контрольной суммы токена"
        )
        
    # Инвалидация старой сессии
    session["is_revoked"] = True
    
    # Выпуск новой пары токенов
    new_access_token, _ = create_jwt_token(
        user_id=user_id,
        token_type="access",
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    new_refresh_token, new_refresh_jti = create_jwt_token(
        user_id=user_id,
        token_type="refresh",
        expires_delta=timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
    )
    
    session_db[new_refresh_jti] = {
        "user_id": user_id,
        "token_hash": hash_token(new_refresh_token),
        "is_revoked": False,
        "expires_at": datetime.now(timezone.utc) + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
    }
    
    return TokenResponse(
        access_token=new_access_token,
        refresh_token=new_refresh_token
    )

@router.post("/logout", status_code=status.HTTP_204_NO_CONTENT)
async def logout(payload_in: RefreshRequest):
    payload = decode_and_validate_token(payload_in.refresh_token, expected_type="refresh")
    jti = payload["jti"]
    if jti in session_db:
        session_db[jti]["is_revoked"] = True
    return Response(status_code=status.HTTP_204_NO_CONTENT)

@router.post("/logout-all", status_code=status.HTTP_204_NO_CONTENT)
async def logout_all(current_user: TokenData = Depends(get_current_user)):
    """Аннулирует все refresh-сессии текущего пользователя на всех устройствах."""
    for s_data in session_db.values():
        if s_data["user_id"] == current_user.user_id:
            s_data["is_revoked"] = True
    return Response(status_code=status.HTTP_204_NO_CONTENT)

Автоматизированное тестирование аутентификации

Система аутентификации требует обязательного покрытия негативными сценариями: валидация подписи, время жизни и поведение при повторном использовании токенов.

Ниже приведен набор тестов на базе pytest и httpx.AsyncClient.

import pytest
from httpx import AsyncClient, ASGITransport
from datetime import timedelta
import jwt
from main import app, session_db, create_jwt_token, SECRET_KEY, ALGORITHM

@pytest.fixture(autouse=True)
def clean_sessions():
    session_db.clear()
    yield
    session_db.clear()

@pytest.mark.asyncio
async def test_full_auth_and_rotation_flow():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        # 1. Успешный вход
        login_res = await ac.post("/auth/login", json={"username": "admin", "password": "secret"})
        assert login_res.status_code == 200
        tokens = login_res.json()
        access_token = tokens["access_token"]
        refresh_token = tokens["refresh_token"]

        # 2. Ротация токенов
        refresh_res = await ac.post("/auth/refresh", json={"refresh_token": refresh_token})
        assert refresh_res.status_code == 200
        new_tokens = refresh_res.json()
        assert new_tokens["access_token"] != access_token
        assert new_tokens["refresh_token"] != refresh_token

        # 3. Попытка повторного использования старого refresh-токена (Reuse Detection)
        reuse_res = await ac.post("/auth/refresh", json={"refresh_token": refresh_token})
        assert reuse_res.status_code == 401
        assert "аннулированы" in reuse_res.json()["detail"]

        # 4. Проверка, что новый токен также стал недействителен из-за срабатывания защиты
        fail_res = await ac.post("/auth/refresh", json={"refresh_token": new_tokens["refresh_token"]})
        assert fail_res.status_code == 401

@pytest.mark.asyncio
async def test_expired_access_token():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        # Генерация просроченного токена
        expired_token, _ = create_jwt_token(
            user_id="usr_123",
            token_type="access",
            expires_delta=timedelta(seconds=-1)
        )
        res = await ac.post("/auth/logout-all", headers={"Authorization": f"Bearer {expired_token}"})
        assert res.status_code == 401
        assert res.json()["detail"] == "Срок действия токена истек"

@pytest.mark.asyncio
async def test_invalid_signature():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        # Токен, подписанный сторонним ключом
        fake_token = jwt.encode(
            {"sub": "usr_123", "type": "access", "jti": "abc", "iss": "fastapi-auth-app"},
            "WRONG_SECRET_KEY",
            algorithm="HS256"
        )
        res = await ac.post("/auth/logout-all", headers={"Authorization": f"Bearer {fake_token}"})
        assert res.status_code == 401
        assert res.json()["detail"] == "Не удалось проверить учетные данные"

@pytest.mark.asyncio
async def test_token_type_confusion():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        # Попытка использовать refresh-токен в качестве access-токена
        refresh_token, _ = create_jwt_token(
            user_id="usr_123",
            token_type="refresh",
            expires_delta=timedelta(days=1)
        )
        res = await ac.post("/auth/logout-all", headers={"Authorization": f"Bearer {refresh_token}"})
        assert res.status_code == 401
        assert "Недопустимый тип токена" in res.json()["detail"]

Чек-лист безопасности перед деплоем в production

Алгоритм подписи: явно зафиксируйте разрешенные алгоритмы при декодировании (algorithms=["HS256"] или асимметричный RS256 / EdDSA). Не допускайте алгоритм "none".
Асимметричные ключи для распределенной архитектуры: если проверку токенов выполняют несколько независимых сервисов, используйте RS256/ES256. Приватный ключ для выпуска остается у Auth-сервиса, публичный ключ распространяется по остальным сервисам.
Хранение в БД только хешей: убедитесь, что refresh-токены хешируются (SHA-256) до записи в базу данных.
Очистка просроченных сессий: настройте периодическую фоновую задачу (Celery / cron / pg_cron) для удаления записей, у которых expires_at < NOW(), чтобы таблица сессий не разрасталась бесконтрольно.
Rate Limiting: обязательно ограничьте количество запросов к эндпоинтам /auth/login и /auth/refresh (через Redis Token Bucket или Nginx), чтобы предотвратить перебор паролей и DoS-атаки на механизм ротации.

Часто задаваемые вопросы (FAQ)

Зачем хранить refresh-токены в базе данных, если JWT задумывался как stateless?

JWT как stateless-инструмент эффективен для access-токенов: они имеют короткий срок жизни и валидируются без обращения к хранилищу. Однако для долговременного доступа необходим механизм принудительного отзыва (при краже устройства, смене пароля, блокировке аккаунта). Хранение метаданных refresh-токенов на сервере возвращает контроль над жизненным циклом сессий, сохраняя высокую скорость обработки основных API-запросов.

Что происходит при одновременной отправке одного refresh-токена легитимным клиентом и злоумышленником?

При реализации ротации по RFC 9700 сервер принимает только первый пришедший запрос, успешно выдает новую пару токенов и помечает старый токен как использованный. Когда поступает второй запрос с тем же токеном, сервер фиксирует повторное использование (Reuse Detection), мгновенно аннулирует всю цепочку сессий пользователя и принудительно разлогинивает все устройства.

В чем разница между хранением токена в Authorization: Bearer и в HttpOnly Cookie?

При передаче через заголовок Authorization: Bearer токен сохраняется на клиенте средствами JavaScript (в памяти или хранилище браузера), что делает его уязвимым к XSS-атакам. HttpOnly Secure Cookie блокирует доступ к токену со стороны клиентских скриптов, защищая его от кражи при XSS, однако требует обязательной защиты эндпоинтов от CSRF-атак.

Как мгновенно отозвать доступ пользователя, если его access-токен все еще валиден?

Поскольку access-токен проверяется локально без БД, он остается валидным до истечения срока exp (обычно 5–15 минут). Если требуется мгновенная блокировка (например, при компрометации администратора), применяют распределенный черный список токенов (Token Blacklist / Revocation List) в Redis, где сохраняются отозванные jti на время их оставшейся жизни, либо уменьшают TTL access-токена до 1–2 минут.

Вывод

Безопасная аутентификация в FastAPI требует выхода за рамки стандартных примеров с одиночными токенами. Разделение ответственности между короткоживущими access-токенами и контролируемыми сервером refresh-токенами обеспечивает баланс между производительностью и безопасностью. Реализация ротации по стандарту RFC 9700, хеширование данных сессий в базе и автоматическое реагирование на попытки повторного использования токенов формируют надежный фундамент для защиты пользовательских данных.

JWT-аутентификация в FastAPI: access/refresh-токены, ротация сессий, отзыв доступа и тестирование