JWT авторизация в FastAPI: от теории до работающего кода

в 5:22, , рубрики: access token, api, fastapi, JSON Web Token, python, авторизация, аутентификация, бэкенд, разработка, токены

При разработке современных веб-приложений и API вопрос безопасности и аутентификации пользователей встаёт одним из первых. Как сделать так, чтобы пользователь мог войти один раз и получать доступ к защищённым ресурсам без постоянного ввода пароля? Как организовать систему, которая легко масштабируется и не требует хранения состояния сессии на сервере?

В этой статье я разберу подход, основанный на JWT (JSON Web Tokens), и покажу, как реализовать полноценную авторизацию в FastAPI — одном из самых быстрых и современных фреймворков для Python. Мы пройдём путь от архитектуры приложения до готового кода, который можно использовать в реальных проектах.

Что такое JWT и зачем он нужен?

JWT (JSON Web Token) — это компактный и самодостаточный способ передачи информации между сторонами в виде JSON-объекта. Токен подписан цифровой подписью, что гарантирует его подлинность.

Структура JWT

Токен состоит из трёх частей, разделённых точками:

header.payload.signature

Header (заголовок) — содержит тип токена и алгоритм подписи:

{
  "alg": "HS256",
  "typ": "JWT"
}

Payload (полезная нагрузка) — содержит утверждения (claims): информация о пользователе, время выпуска, срок действия и другие данные:

{
  "sub": "user_id_123",
  "exp": 1735689600,
  "iat": 1735603200
}

Signature (подпись) — создаётся путём шифрования header и payload с секретным ключом.

Почему JWT популярен?

  • Stateless (отсутствие состояния) — сервер не хранит информацию о сессиях, что упрощает горизонтальное масштабирование.

  • Самодостаточность — токен содержит всю необходимую информацию о пользователе.

  • Кросс-платформенность — работает одинаково для веба, мобильных приложений и микросервисов.

Access и Refresh токены

В реальных проектах редко ограничиваются одним токеном. Обычно используют пару:

  • Access Token — короткоживущий (15–30 минут), используется для доступа к защищённым ресурсам.

  • Refresh Token — долгоживущий (до 30 дней), служит для получения нового access токена без повторной аутентификации.

Этот подход повышает безопасность: если access токен скомпрометирован, он будет действителен недолго, а refresh токен можно хранить в защищённом месте (например, HttpOnly cookie).

Архитектура приложения FastAPI

Прежде чем переходить к коду, важно понять, как организовать проект. Грамотное разделение на слои делает код поддерживаемым и тестируемым. 

project/
├── app/
│   ├── core/
│   │   ├── config.py          # Настройки приложения
│   │   ├── security.py        # Функции для работы с JWT
│   │   └── dependencies.py    # Зависимости FastAPI
│   ├── models/
│   │   └── user.py            # Модели базы данных
│   ├── schemas/
│   │   ├── user.py            # Pydantic-схемы для пользователя
│   │   └── token.py           # Pydantic-схемы для токенов
│   ├── routers/
│   │   ├── auth.py            # Эндпоинты аутентификации
│   │   └── users.py           # Защищённые эндпоинты пользователя
│   ├── services/
│   │   └── user.py            # Бизнес-логика
│   └── main.py                # Точка входа
├── requirements.txt
└── .env

Зачем такое разделение?

  • Routers — отвечают только за маршрутизацию и HTTP-ответы.

  • Schemas — валидация входящих данных и сериализация ответов.

  • Models — описание таблиц в базе данных.

  • Services — бизнес-логика, которая может переиспользоваться.

  • Core — конфигурация, утилиты, зависимости.

Реализация JWT авторизации

Шаг 1. Настройка конфигурации

Начнём с файла core/config.py, где будут храниться чувствительные настройки:

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    SECRET_KEY: str = "your-secret-key-change-in-production"
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    REFRESH_TOKEN_EXPIRE_DAYS: int = 7
    
    class Config:
        env_file = ".env"

settings = Settings()

Шаг 2. Функции для работы с JWT

В core/security.py создадим функции создания и проверки токенов:

from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from .config import settings

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain_password: str, hashed_password: str) -> bool:
    """Проверка пароля"""
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    """Хэширование пароля"""
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: timedelta = None) -> str:
    """Создание access токена"""
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
    return encoded_jwt

def create_refresh_token(data: dict) -> str:
    """Создание refresh токена"""
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
    return encoded_jwt

def decode_token(token: str) -> dict:
    """Декодирование токена"""
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        return payload
    except JWTError:
        return None

Шаг 3. Pydantic-схемы

Определим схемы для входящих и исходящих данных.

schemas/user.py:

from pydantic import BaseModel, EmailStr

class UserCreate(BaseModel):
    email: EmailStr
    password: str
    full_name: str

class UserResponse(BaseModel):
    id: int
    email: EmailStr
    full_name: str
    
    class Config:
        from_attributes = True

schemas/token.py:

from pydantic import BaseModel

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

class TokenData(BaseModel):
    user_id: int

Шаг 4. Зависимость для получения текущего пользователя

Это ключевой компонент, который будет использоваться для защиты эндпоинтов.

core/dependencies.py:

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError
from sqlalchemy.orm import Session
from ..database import get_db
from ..models.user import User
from ..schemas.token import TokenData
from .security import decode_token

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

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db)
) -> User:
    """
    Зависимость, которая извлекает текущего пользователя из токена.
    Используется для защиты эндпоинтов.
    """
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    
    payload = decode_token(token)
    if payload is None:
        raise credentials_exception
    
    user_id = payload.get("sub")
    if user_id is None:
        raise credentials_exception
    
    token_data = TokenData(user_id=int(user_id))
    
    user = db.query(User).filter(User.id == token_data.user_id).first()
    if user is None:
        raise credentials_exception
    
    return user

Шаг 5. Роутер для аутентификации

routers/auth.py:

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from datetime import timedelta
from ..database import get_db
from ..models.user import User
from ..schemas.user import UserCreate, UserResponse
from ..schemas.token import Token
from ..core.security import (
    verify_password, 
    create_access_token, 
    create_refresh_token,
    get_password_hash
)
from ..core.config import settings

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

@router.post("/register", response_model=UserResponse)
def register(user_data: UserCreate, db: Session = Depends(get_db)):
    """Регистрация нового пользователя"""
    existing_user = db.query(User).filter(User.email == user_data.email).first()
    if existing_user:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Email already registered"
        )
    
    hashed_password = get_password_hash(user_data.password)
    new_user = User(
        email=user_data.email,
        hashed_password=hashed_password,
        full_name=user_data.full_name
    )
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user

@router.post("/login", response_model=Token)
def login(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db: Session = Depends(get_db)
):
    """
    Аутентификация пользователя.
    Возвращает access и refresh токены.
    """
    user = db.query(User).filter(User.email == form_data.username).first()
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect email or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    access_token = create_access_token(data={"sub": str(user.id)})
    refresh_token = create_refresh_token(data={"sub": str(user.id)})
    
    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "token_type": "bearer"
    }

@router.post("/refresh", response_model=Token)
def refresh_token(
    refresh_token: str,
    db: Session = Depends(get_db)
):
    """Обновление access токена с помощью refresh токена"""
    payload = decode_token(refresh_token)
    if payload is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid refresh token"
        )
    
    user_id = payload.get("sub")
    if user_id is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid refresh token"
        )
    
    user = db.query(User).filter(User.id == int(user_id)).first()
    if user is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="User not found"
        )
    
    new_access_token = create_access_token(data={"sub": str(user.id)})
    
    return {
        "access_token": new_access_token,
        "refresh_token": refresh_token,
        "token_type": "bearer"
    }

Шаг 6. Защищённый роутер

routers/users.py:

from fastapi import APIRouter, Depends
from ..schemas.user import UserResponse
from ..core.dependencies import get_current_user
from ..models.user import User

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

@router.get("/me", response_model=UserResponse)
def get_current_user_info(current_user: User = Depends(get_current_user)):
    """
    Получение информации о текущем пользователе.
    Эндпоинт защищён: требуется валидный access токен.
    """
    return current_user

Шаг 7. Точка входа приложения

main.py:

from fastapi import FastAPI
from .routers import auth, users

app = FastAPI(title="FastAPI JWT Auth", version="1.0.0")

app.include_router(auth.router)
app.include_router(users.router)

@app.get("/")
def root():
    return {"message": "Welcome to FastAPI JWT Auth API"}

Схема работы приложения

Чтобы наглядно представить, как все компоненты взаимодействуют, ниже приведена схема двух основных процессов.

Процесс 1: Логин и получение токена

JWT авторизация в FastAPI: от теории до работающего кода - 1

Процесс 2: Запрос защищённого ресурса

JWT авторизация в FastAPI: от теории до работающего кода - 2

Тестирование API

После запуска приложения (uvicorn app.main:app --reload) документация будет доступна по адресу http://localhost:8000/docs.

Проверка работы:

  1. Регистрация — POST /api/auth/register с JSON {"email": "user@example.com", "password": "secret", "full_name": "Иван Иванов"}.

  2. Логин — POST /api/auth/login с form-data username и password. В ответе получаем токены.

  3. Запрос защищённого ресурса — в Swagger нажмите кнопку "Authorize" и введите Bearer <access_token>. Затем выполните GET /api/users/me.

  4. Обновление токена — POST /api/auth/refresh с body {"refresh_token": "..."}.

Безопасность в production

При развёртывании приложения обязательно:

  1. Используйте надёжный SECRET_KEY — не храните его в коде, используйте переменные окружения или менеджеры секретов (например, HashiCorp Vault).

  2. Переключитесь на HTTPS — чтобы токены не передавались в открытом виде.

  3. Установите короткое время жизни access токенов — 15–30 минут оптимально.

  4. Храните refresh токены в HttpOnly cookies — это защищает от XSS-атак.

  5. Внедрите логирование и мониторинг — чтобы отслеживать подозрительную активность.

Заключение

Мы разобрали, как устроена JWT-авторизация, начиная от теории и заканчивая практической реализацией на FastAPI. Ключевые выводы:

  • JWT позволяет создавать stateless-приложения, которые легко масштабировать.

  • Разделение на access и refresh токены повышает безопасность.

  • FastAPI предоставляет удобные инструменты для работы с JWT через зависимости и OAuth2PasswordBearer.

  • Грамотная архитектура (разделение на routers, schemas, core) делает код поддерживаемым.

Теперь у вас есть готовая основа, которую можно расширять: добавлять ролевую модель, интеграцию с социальными сетями, двухфакторную аутентификацию. Экспериментируйте и адаптируйте под свои задачи!

Автор: valtureso

Источник

* - обязательные к заполнению поля

Главная   |  Архив новостей  |   Android  |   Google  |   Apple  |   Microsoft  |   Информационная безопасность  |   Веб – разработка
Публикации RSS  |  Комментарии RSS
https://ajax.googleapis.com/ajax/libs/jquery/3.4.1/jquery.min.js