Аутентификация JWT в FastAPI

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

В этой статье мы разберём:

Настройку JWT в FastAPI
Регистрацию пользователей
Логин и выдачу токена
Защиту эндпоинтов
Refresh-токены
Роли и права доступа

Содержание

Что нужно знать перед началом
Основная часть
Установка зависимостей
Настройка безопасности
Модели данных
Заглушка базы данных
Эндпоинты аутентификации
Запуск и тестирование
Задачи для закрепления
Нюансы и подводные камни
Хранение секретов
Срок жизни токена
JWT не шифрует данные
Обработка ошибок
Частые ошибки и как их избежать
Ошибка 1: Хранение SECRET_KEY в коде
Ошибка 2: Слишком долгий срок жизни access-токена
Ошибка 3: Отсутствие проверки type у refresh-токена
Ошибка 4: Необработанные исключения JWT
Шпаргалка
Заключение
КВИЗ
Что дальше?

Что нужно знать перед началом

Для этого урока тебе понадобится:

Установленный FastAPI и Uvicorn
Базовое понимание асинхронности

Совет: JWT — это компактный способ передачи информации между сторонами, который можно проверить и которому можно доверять, потому что он подписан цифровой подписью.

Основная часть

Установка зависимостей

pip install fastapi uvicorn python-jose[cryptography] passlib[bcrypt] python-multipart

python-jose — для создания и проверки JWT
passlib — для хеширования паролей
python-multipart — для обработки OAuth2 форм

Настройка безопасности

security.py:

from datetime import datetime, timedelta, timezone
from jose import jwt
from jose.exceptions import JWTError
from passlib.context import CryptContext
from fastapi.security import OAuth2PasswordBearer
from fastapi import HTTPException, status

# Секретный ключ (в продакшене используй env)
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
REFRESH_TOKEN_EXPIRE_DAYS = 7

# Хеширование паролей
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# OAuth2 схема
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")

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:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire, "type": "access"})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def create_refresh_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
    to_encode.update({"exp": expire, "type": "refresh"})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def decode_token(token: str):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        return None

def get_current_user(token: str = Depends(oauth2_scheme)):
    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
    
    username: str = payload.get("sub")
    if username is None:
        raise credentials_exception
    
    # Здесь нужно получить пользователя из БД
    # user = get_user_from_db(username)
    # if user is None:
    #     raise credentials_exception
    # return user
    
    # Для примера — возвращаем заглушку
    return {"username": username}

def get_current_active_user(current_user: dict = Depends(get_current_user)):
    # Проверка, активен ли пользователь
    # if current_user.disabled:
    #     raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

def require_roles(*required_roles):
    def role_checker(current_user: dict = Depends(get_current_active_user)):
        user_roles = current_user.get("roles", [])
        if not any(role in user_roles for role in required_roles):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Insufficient permissions"
            )
        return current_user
    return role_checker

Модели данных

schemas.py:

from pydantic import BaseModel, EmailStr, Field
from typing import Optional, List
from datetime import datetime

class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr
    password: str = Field(..., min_length=6)

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

class UserResponse(BaseModel):
    id: int
    username: str
    email: str
    is_active: bool
    created_at: datetime
    roles: List[str] = []

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

class TokenRefresh(BaseModel):
    refresh_token: str

class ChangePassword(BaseModel):
    old_password: str
    new_password: str = Field(..., min_length=6)

Заглушка базы данных

database.py:

from security import get_password_hash
from datetime import datetime

# Заглушка БД
users_db = {}
next_id = 1

def get_user_by_username(username: str):
    return users_db.get(username)

def get_user_by_email(email: str):
    for user in users_db.values():
        if user["email"] == email:
            return user
    return None

def create_user(username: str, email: str, hashed_password: str):
    global next_id
    user = {
        "id": next_id,
        "username": username,
        "email": email,
        "hashed_password": hashed_password,
        "is_active": True,
        "roles": ["user"],
        "created_at": datetime.now()
    }
    users_db[username] = user
    next_id += 1
    return user

def update_user(username: str, **kwargs):
    user = users_db.get(username)
    if user:
        user.update(kwargs)
    return user

Эндпоинты аутентификации

main.py:

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from datetime import timedelta
from security import (
    verify_password, get_password_hash,
    create_access_token, create_refresh_token,
    decode_token, get_current_active_user,
    require_roles, ACCESS_TOKEN_EXPIRE_MINUTES
)
from schemas import UserCreate, UserLogin, Token, TokenRefresh, UserResponse, ChangePassword
from database import get_user_by_username, get_user_by_email, create_user, update_user

app = FastAPI(title="JWT Authentication API")

# ======== Регистрация ========
@app.post("/auth/register", response_model=UserResponse, status_code=201)
async def register(user: UserCreate):
    if get_user_by_username(user.username):
        raise HTTPException(status_code=400, detail="Username already exists")
    if get_user_by_email(user.email):
        raise HTTPException(status_code=400, detail="Email already exists")
    
    hashed_password = get_password_hash(user.password)
    db_user = create_user(user.username, user.email, hashed_password)
    
    return UserResponse(
        id=db_user["id"],
        username=db_user["username"],
        email=db_user["email"],
        is_active=db_user["is_active"],
        created_at=db_user["created_at"],
        roles=db_user["roles"]
    )

# ======== Логин ========
@app.post("/auth/login", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = get_user_by_username(form_data.username)
    if not user or not verify_password(form_data.password, user["hashed_password"]):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    access_token = create_access_token(
        data={"sub": user["username"], "roles": user["roles"]},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    refresh_token = create_refresh_token(data={"sub": user["username"]})
    
    return {"access_token": access_token, "refresh_token": refresh_token}

# ======== Refresh токен ========
@app.post("/auth/refresh", response_model=Token)
async def refresh_token(refresh_data: TokenRefresh):
    payload = decode_token(refresh_data.refresh_token)
    if payload is None:
        raise HTTPException(status_code=401, detail="Invalid refresh token")
    
    if payload.get("type") != "refresh":
        raise HTTPException(status_code=401, detail="Invalid token type")
    
    username = payload.get("sub")
    if not username:
        raise HTTPException(status_code=401, detail="Invalid token")
    
    user = get_user_by_username(username)
    if not user:
        raise HTTPException(status_code=401, detail="User not found")
    
    access_token = create_access_token(
        data={"sub": username, "roles": user["roles"]},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    refresh_token = create_refresh_token(data={"sub": username})
    
    return {"access_token": access_token, "refresh_token": refresh_token}

# ======== Текущий пользователь ========
@app.get("/auth/me", response_model=UserResponse)
async def get_me(current_user: dict = Depends(get_current_active_user)):
    user = get_user_by_username(current_user["username"])
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    
    return UserResponse(
        id=user["id"],
        username=user["username"],
        email=user["email"],
        is_active=user["is_active"],
        created_at=user["created_at"],
        roles=user["roles"]
    )

# ======== Смена пароля ========
@app.post("/auth/change-password")
async def change_password(
    data: ChangePassword,
    current_user: dict = Depends(get_current_active_user)
):
    user = get_user_by_username(current_user["username"])
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    
    if not verify_password(data.old_password, user["hashed_password"]):
        raise HTTPException(status_code=400, detail="Incorrect old password")
    
    new_hashed = get_password_hash(data.new_password)
    update_user(current_user["username"], hashed_password=new_hashed)
    
    return {"message": "Password changed successfully"}

# ======== Защищённый эндпоинт ========
@app.get("/protected")
async def protected(current_user: dict = Depends(get_current_active_user)):
    return {"message": f"Hello, {current_user['username']}!"}

# ======== Только для администраторов ========
@app.get("/admin")
async def admin_only(current_user: dict = Depends(require_roles("admin"))):
    return {"message": f"Hello, admin {current_user['username']}!"}

Запуск и тестирование

Запуск сервера:

uvicorn main:app --reload

Тестирование через Swagger:

Открой http://localhost:8000/docs

Зарегистрируйся: POST /auth/register
Войди: POST /auth/login (используй кнопку «Authorize» для установки токена)
Получи данные: GET /auth/me
Попробуй защищённые эндпоинты: GET /protected, GET /admin

Задачи для закрепления

Задача 1. Добавь проверку на существование пользователя при регистрации.

Задача 2. Реализуй logout (удаление токена на клиенте).

Задача 3. Добавь роль «admin» и эндпоинт /admin.

Задача 4. Реализуй обновление профиля пользователя.

Задача 5. Добавь проверку срока действия токена и автоматический выход.

Ответы:

Задача 1.

def register(user: UserCreate):
    if get_user_by_username(user.username):
        raise HTTPException(status_code=400, detail="Username already exists")
    # ...

Задача 2.

# Logout реализуется на клиенте удалением токена.
# Сервер не хранит токены, поэтому logout — это клиентская операция.

Задача 3.

@app.post("/admin")
async def admin_only(current_user: dict = Depends(require_roles("admin"))):
    return {"message": "Admin access"}

Задача 4.

@app.put("/auth/profile")
async def update_profile(
    username: str = None,
    email: str = None,
    current_user: dict = Depends(get_current_active_user)
):
    user = get_user_by_username(current_user["username"])
    if not user:
        raise HTTPException(status_code=404)
    if username:
        user["username"] = username
    if email:
        user["email"] = email
    update_user(current_user["username"], username=username, email=email)
    return {"message": "Profile updated"}

Задача 5.

# В get_current_user проверяем exp:
payload = decode_token(token)
if payload is None or payload.get("exp") < datetime.now().timestamp():
    raise credentials_exception

Нюансы и подводные камни

Хранение секретов

Никогда не храни SECRET_KEY в коде. Используй переменные окружения.

Срок жизни токена

Устанавливай короткий срок жизни access-токена (15–30 минут). Используй refresh-токен для продления.

JWT не шифрует данные

Не храни пароли и другую чувствительную информацию в JWT.

Обработка ошибок

Всегда возвращай понятные сообщения об ошибках.

Частые ошибки и как их избежать

Ошибка 1: Хранение SECRET_KEY в коде

Используй .env файл и os.getenv().

Ошибка 2: Слишком долгий срок жизни access-токена

15–30 минут — оптимальный срок.

Ошибка 3: Отсутствие проверки `type` у refresh-токена

Добавь "type": "refresh" в payload и проверяй.

Ошибка 4: Необработанные исключения JWT

Оборачивай jwt.decode() в try-except.

Шпаргалка

Компонент	Назначение
`SECRET_KEY`	Секрет для подписи токенов
`ALGORITHM`	Алгоритм подписи (HS256)
`create_access_token()`	Создание access-токена
`create_refresh_token()`	Создание refresh-токена
`decode_token()`	Проверка и декодирование
`get_current_user()`	Получение текущего пользователя
`require_roles()`	Проверка прав доступа