- PVSM.RU - https://www.pvsm.ru -
При разработке современных веб-приложений и API вопрос безопасности и аутентификации пользователей встаёт одним из первых. Как сделать так, чтобы пользователь мог войти один раз и получать доступ к защищённым ресурсам без постоянного ввода пароля? Как организовать систему, которая легко масштабируется и не требует хранения состояния сессии на сервере?
В этой статье я разберу подход, основанный на JWT (JSON Web Tokens), и покажу, как реализовать полноценную авторизацию в FastAPI — одном из самых быстрых и современных фреймворков для Python. Мы пройдём путь от архитектуры приложения до готового кода, который можно использовать в реальных проектах.
JWT (JSON Web Token) — это компактный и самодостаточный способ передачи информации между сторонами в виде JSON-объекта. Токен подписан цифровой подписью, что гарантирует его подлинность.
Токен состоит из трёх частей, разделённых точками:
header.payload.signature
Header (заголовок) — содержит тип токена и алгоритм подписи:
{
"alg": "HS256",
"typ": "JWT"
}
Payload (полезная нагрузка) — содержит утверждения (claims): информация о пользователе, время выпуска, срок действия и другие данные:
{
"sub": "user_id_123",
"exp": 1735689600,
"iat": 1735603200
}
Signature (подпись) — создаётся путём шифрования header и payload с секретным ключом.
Stateless (отсутствие состояния) — сервер не хранит информацию о сессиях, что упрощает горизонтальное масштабирование.
Самодостаточность — токен содержит всю необходимую информацию о пользователе.
Кросс-платформенность — работает одинаково для веба, мобильных приложений и микросервисов.
В реальных проектах редко ограничиваются одним токеном. Обычно используют пару:
Access Token — короткоживущий (15–30 минут), используется для доступа к защищённым ресурсам.
Refresh Token — долгоживущий (до 30 дней), служит для получения нового access токена без повторной аутентификации.
Этот подход повышает безопасность: если access токен скомпрометирован, он будет действителен недолго, а refresh токен можно хранить в защищённом месте (например, HttpOnly cookie).
Прежде чем переходить к коду, важно понять, как организовать проект. Грамотное разделение на слои делает код поддерживаемым и тестируемым.
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 — конфигурация, утилиты, зависимости.
Начнём с файла core/config.py [1], где будут храниться чувствительные настройки:
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()
В core/security.py [2] создадим функции создания и проверки токенов:
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
Определим схемы для входящих и исходящих данных.
schemas/user.py [3]:
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 [4]:
from pydantic import BaseModel
class Token(BaseModel):
access_token: str
refresh_token: str
token_type: str = "bearer"
class TokenData(BaseModel):
user_id: int
Это ключевой компонент, который будет использоваться для защиты эндпоинтов.
core/dependencies.py [5]:
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
routers/auth.py [6]:
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"
}
routers/users.py [7]:
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
main.py [8]:
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"}
Чтобы наглядно представить, как все компоненты взаимодействуют, ниже приведена схема двух основных процессов.


После запуска приложения (uvicorn app.main:app --reload) документация будет доступна по адресу http://localhost:8000/docs [9].
Проверка работы:
Регистрация — POST /api/auth/register с JSON {"email": "user@example.com [10]", "password": "secret", "full_name": "Иван Иванов"}.
Логин — POST /api/auth/login с form-data username и password. В ответе получаем токены.
Запрос защищённого ресурса — в Swagger нажмите кнопку "Authorize" и введите Bearer <access_token>. Затем выполните GET /api/users/me.
Обновление токена — POST /api/auth/refresh с body {"refresh_token": "..."}.
При развёртывании приложения обязательно:
Используйте надёжный SECRET_KEY — не храните его в коде, используйте переменные окружения или менеджеры секретов (например, HashiCorp Vault).
Переключитесь на HTTPS — чтобы токены не передавались в открытом виде.
Установите короткое время жизни access токенов — 15–30 минут оптимально.
Храните refresh токены в HttpOnly cookies — это защищает от XSS-атак.
Внедрите логирование и мониторинг — чтобы отслеживать подозрительную активность.
Мы разобрали, как устроена JWT-авторизация, начиная от теории и заканчивая практической реализацией на FastAPI. Ключевые выводы:
JWT позволяет создавать stateless-приложения, которые легко масштабировать.
Разделение на access и refresh токены повышает безопасность.
FastAPI предоставляет удобные инструменты для работы с JWT через зависимости и OAuth2PasswordBearer.
Грамотная архитектура (разделение на routers, schemas, core) делает код поддерживаемым.
Теперь у вас есть готовая основа, которую можно расширять: добавлять ролевую модель, интеграцию с социальными сетями, двухфакторную аутентификацию. Экспериментируйте и адаптируйте под свои задачи!
Автор: valtureso
Источник [11]
Сайт-источник PVSM.RU: https://www.pvsm.ru
Путь до страницы источника: https://www.pvsm.ru/razrabotka/447739
Ссылки в тексте:
[1] config.py: http://config.py
[2] security.py: http://security.py
[3] user.py: http://user.py
[4] token.py: http://token.py
[5] dependencies.py: http://dependencies.py
[6] auth.py: http://auth.py
[7] users.py: http://users.py
[8] main.py: http://main.py
[9] http://localhost:8000/docs: http://localhost:8000/docs
[10] user@example.com: mailto:user@example.com
[11] Источник: https://habr.com/ru/articles/1015148/?utm_source=habrahabr&utm_medium=rss&utm_campaign=1015148
Нажмите здесь для печати.