При разработке современных веб-приложений и 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. Точка входа приложения
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: Логин и получение токена
Процесс 2: Запрос защищённого ресурса
Тестирование API
После запуска приложения (uvicorn app.main:app --reload) документация будет доступна по адресу http://localhost:8000/docs.
Проверка работы:
Регистрация — POST
/api/auth/registerс JSON{"email": "user@example.com", "password": "secret", "full_name": "Иван Иванов"}.Логин — POST
/api/auth/loginс form-datausernameиpassword. В ответе получаем токены.Запрос защищённого ресурса — в Swagger нажмите кнопку "Authorize" и введите
Bearer <access_token>. Затем выполните GET/api/users/me.Обновление токена — POST
/api/auth/refreshс body{"refresh_token": "..."}.
Безопасность в production
При развёртывании приложения обязательно:
Используйте надёжный SECRET_KEY — не храните его в коде, используйте переменные окружения или менеджеры секретов (например, HashiCorp Vault).
Переключитесь на HTTPS — чтобы токены не передавались в открытом виде.
Установите короткое время жизни access токенов — 15–30 минут оптимально.
Храните refresh токены в HttpOnly cookies — это защищает от XSS-атак.
Внедрите логирование и мониторинг — чтобы отслеживать подозрительную активность.
Заключение
Мы разобрали, как устроена JWT-авторизация, начиная от теории и заканчивая практической реализацией на FastAPI. Ключевые выводы:
JWT позволяет создавать stateless-приложения, которые легко масштабировать.
Разделение на access и refresh токены повышает безопасность.
FastAPI предоставляет удобные инструменты для работы с JWT через зависимости и
OAuth2PasswordBearer.Грамотная архитектура (разделение на routers, schemas, core) делает код поддерживаемым.
Теперь у вас есть готовая основа, которую можно расширять: добавлять ролевую модель, интеграцию с социальными сетями, двухфакторную аутентификацию. Экспериментируйте и адаптируйте под свои задачи!
