OAuth2 в FastAPI: разбираем аутентификацию на реальном примере

Зачем вообще OAuth2 в FastAPI

FastAPI из коробки даёт модуль fastapi.security, где уже есть классы для OAuth2-потоков. Но между «импортировать класс» и «получить рабочую аутентификацию» лежит десяток решений: как хранить пароли, какой алгоритм подписи выбрать для токена, как разделить access- и refresh-токены, что возвращать при невалидном токене. Ниже — реализация потока Resource Owner Password Credentials (самый частый сценарий для внутренних API и SPA), собранная из модулей, которые легко заменить, когда требования изменятся.

Всё написано для FastAPI 0.110+ и Python 3.11+. Для JWT используется библиотека PyJWT 2.x, для хеширования — passlib[bcrypt].

Что нужно до старта

• Python 3.11 или новее.
• Виртуальное окружение (venv, Poetry — без разницы).
• Установленные пакеты: fastapi, uvicorn[standard], PyJWT, passlib[bcrypt], python-multipart (FastAPI требует его для формы логина).

pip install "fastapi>=0.110" "uvicorn[standard]" PyJWT "passlib[bcrypt]" python-multipart

Структура проекта

Для наглядности всё лежит в одном файле main.py. В продакшене вы разнесёте модели, зависимости и роутеры по пакетам — но логика останется той же.

Шаг 1. Настройки и утилиты для паролей

# main.py — часть 1: настройки
from datetime import datetime, timedelta, timezone
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from passlib.context import CryptContext
from pydantic import BaseModel
import jwt

# Секрет для подписи JWT. В проде — из переменной окружения, не хардкод.
SECRET_KEY = "YOUR_SECRET_KEY"  # замените на свой
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

app = FastAPI()


def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)


def hash_password(plain: str) -> str:
    return pwd_context.hash(plain)

OAuth2PasswordBearer — класс, который говорит FastAPI: «жди токен в заголовке Authorization: Bearer <token>». Параметр tokenUrl нужен для Swagger UI — он подставит этот путь в форму логина.

Шаг 2. Модели и «база» пользователей

# main.py — часть 2: модели
class Token(BaseModel):
    access_token: str
    token_type: str


class TokenData(BaseModel):
    username: str | None = None


class User(BaseModel):
    username: str
    disabled: bool = False


class UserInDB(User):
    hashed_password: str


# Имитация базы данных — словарь.
# В реальном проекте здесь SQLAlchemy / Tortoise / любой ORM.
fake_users_db: dict[str, dict] = {
    "alice": {
        "username": "alice",
        "hashed_password": hash_password("secret123"),
        "disabled": False,
    }
}


def get_user(db: dict, username: str) -> UserInDB | None:
    if username in db:
        return UserInDB(**db[username])
    return None

Шаг 3. Создание и проверка JWT

# main.py — часть 3: токены
def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)


async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Невалидный токен",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str | None = payload.get("sub")
        if username is None:
            raise credentials_exception
    except jwt.ExpiredSignatureError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Токен истёк",
            headers={"WWW-Authenticate": "Bearer"},
        )
    except jwt.InvalidTokenError:
        raise credentials_exception

    user = get_user(fake_users_db, username)
    if user is None:
        raise credentials_exception
    return user

Обратите внимание: jwt.decode выбрасывает разные исключения для истёкшего и повреждённого токена. Ловим их раздельно — клиент получит внятное сообщение вместо общего 401.

Секрет подписи JWT — это единственный барьер между атакующим и любым аккаунтом в системе. Генерируйте его криптографически стойким генератором (например, openssl rand -hex 32) и храните в переменной окружения, а не в коде.

Шаг 4. Эндпоинты

# main.py — часть 4: роуты
@app.post("/auth/token", response_model=Token)
async def login(form: Annotated[OAuth2PasswordRequestForm, Depends()]):
    user = get_user(fake_users_db, form.username)
    if not user or not verify_password(form.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Неверный логин или пароль",
            headers={"WWW-Authenticate": "Bearer"},
        )
    if user.disabled:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Аккаунт деактивирован",
        )
    token = create_access_token(
        data={"sub": user.username},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
    )
    return Token(access_token=token, token_type="bearer")


@app.get("/users/me", response_model=User)
async def read_users_me(current_user: Annotated[User, Depends(get_current_user)]):
    return current_user

Шаг 5. Запуск и проверка

1. Запустите сервер: uvicorn main:app --reload.
2. Откройте http://127.0.0.1:8000/docs — Swagger UI покажет кнопку Authorize.
3. Введите alice / secret123. После успешного логина Swagger сам подставит токен в заголовки.
4. Вызовите GET /users/me — в ответе увидите {"username": "alice", "disabled": false}.

Что здесь стоит менять под продакшен

^* Refresh-токен — долгоживущий токен, который не отправляется на каждый запрос, а используется только для получения нового access-токена. Хранится в httpOnly-cookie или в защищённом хранилище мобильного приложения.

Типичные ошибки и что с ними делать

• 422 Unprocessable Entity при логине — не установлен пакет python-multipart. FastAPI не может разобрать form-data без него. Решение: pip install python-multipart.
• ModuleNotFoundError: No module named 'jwt' — установлен пакет jwt вместо PyJWT. Это два разных пакета с конфликтующими именами. Удалите оба (pip uninstall jwt PyJWT) и поставьте заново только PyJWT.
• Токен принимается, но payload пустой — забыли положить данные в sub. Проверьте, что create_access_token получает {"sub": username}, а не пустой словарь.
• 401 сразу после логина — время на сервере расходится с UTC. jwt.decode сравнивает exp с текущим временем; если системные часы отстают, токен выглядит истёкшим. Проверьте timedatectl или включите NTP.

Где подход ломается

Password Credentials Grant (логин + пароль напрямую в API) подходит, когда клиент — ваше собственное приложение и вы контролируете обе стороны. Если вам нужно давать доступ сторонним приложениям — это уже Authorization Code Grant с редиректами и consent-экраном. FastAPI не запрещает его реализовать, но встроенных хелперов для серверного хранения authorization codes в фреймворке нет: понадобится либо писать руками, либо подключать Authlib.

Ещё один предел — отзыв токенов. JWT по природе stateless: пока exp не наступил, токен валиден. Если нужно мгновенно блокировать сессию (увольнение сотрудника, утечка), придётся вводить чёрный список токенов в Redis или укорачивать время жизни access-токена до нескольких минут и полагаться на refresh-поток.

Что делать дальше

Рабочий минимум готов. Следующие шаги, которые стоят внимания:

1. Добавить refresh-токены и хранить их в базе с возможностью отзыва.
2. Вынести зависимость get_current_user в отдельный модуль и добавить проверку ролей или scopes — FastAPI поддерживает Security с параметром scopes прямо в OAuth2PasswordBearer.
3. Поставить rate-limit на /auth/token — например, через middleware или reverse-proxy.
4. Покрыть поток тестами: httpx.AsyncClient позволяет тестировать FastAPI-приложения без запуска сервера.

Документация FastAPI по безопасности остаётся лучшим стартовым справочником — она покрывает и scopes, и кастомные схемы.