FastAPI 8. Маршрут авторизации и JWT

Продолжаем писать наше сервис по сокращению ссылок - lkeep.

Мы подошли, пожалуй, к одному из самых сложных аспектов практически любого веб-приложения — авторизации.

Виды авторизации

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

Вот основные из них:

1. JWT (JSON Web Token) — используется зашифрованный токен, который содержит информацию о пользователе и срок действия. Этот токен передаётся между клиентом и сервером при каждом запросе.
2. OAuth2 — протокол авторизации, позволяющий пользователю входить через сторонние сервисы (например, Google, Яндекс). Часто используется в больших системах с микросервисной архитектурой.
3. Session-based — на сервере создаётся сессия, а клиенту в cookie передаётся session ID. При каждом запросе сервер сверяет этот ID с хранящейся сессией.
4. Basic Authentication — простой и устаревающий способ: логин и пароль передаются в заголовке HTTP-запроса, обычно в Base64. Используется крайне редко, в основном для тестов или во внутренних сервисах.
5. API Key — клиент передаёт заранее выданный уникальный ключ в каждом запросе. Часто используется для авторизации приложений или автоматизированных клиентов.
6. OpenID Connect — надстройка над OAuth2, добавляющая идентификацию пользователя и поддержку единого входа (SSO, Single Sign-On).

Мы воспользуемся одним из самых популярных вариантов — JWT.

Плюсы:

1. Без состояния (Stateless): Сервер не хранит информацию о сессиях, что упрощает масштабируемость и повышает производительность.
2. Гибкость: JWT может содержать дополнительные данные (например, роли, права доступа), что позволяет выполнять авторизацию и аутентификацию без дополнительных запросов к базе данных.
3. Производительность: Токены не требуют обращения к серверу для проверки состояния сессии, так как все нужные данные находятся внутри токена.
4. Безопасность: Если токены хранятся в HttpOnly cookie, они защищены от XSS-атак, так как не доступны через JavaScript.

Минусы:

1. Безопасность: Если токен скомпрометирован (например, через утечку или кражу cookie), злоумышленник может использовать его до истечения срока действия.
2. Размер: JWT может быть достаточно большим, что увеличивает нагрузку на сеть при каждом запросе.
3. Управление сроком действия: Хотя JWT имеет срок действия, если токен истёк, его нужно будет обновить. Без механизма обновления токенов (например, через refresh token) это может привести к неудобствам для пользователя.
4. Отсутствие механизма отзыва: Стандартные JWT не поддерживают механизм отзыва до истечения срока действия, что усложняет управление безопасностью в случае утечек.

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

1. HTTP-заголовок — токен передаётся в заголовке Authorization с префиксом Bearer. Клиент сам добавляет заголовок к каждому запросу, а сервер — проверяет его. Это популярный вариант при работе с внешними API.
2. Cookie — токен сохраняется в cookie, и браузер автоматически отправляет его при каждом запросе. На стороне сервера остаётся только валидация. Чаще всего применяется во фронт-бэк связке.
3. Body и Query — токен передаётся явно в теле (POST) или параметрах (GET) запроса. Это наименее безопасный вариант и используется крайне редко.

В рамках этой статьи мы выберем второй способ — хранение JWT в cookie, потому что:

1. Управление временем жизни — хотя срок действия уже зашит в сам JWT, он не удаляется автоматически. Если поместить его в cookie и указать max-age или expires, браузер сам удалит токен после истечения срока.
2. Увеличение безопасности — при передаче токена через заголовки или тело запроса он доступен из JavaScript, что делает его уязвимым для XSS-атак и утечек. Cookie можно пометить как HttpOnly, и тогда он будет доступен только для браузера и недоступен из JavaScript — даже из расширений.

Подготовка

Установка библиотеки PyJWT

Для работы с JWT, а именно — для создания (кодирования) и проверки (декодирования) токенов, мы воспользуемся библиотекой PyJWT.

Добавим её в проект с помощью Poetry:

poetry add pyjwt

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

Теперь зададим параметр, определяющий срок действия токена. По его истечении токен станет недействительным, и пользователю придётся авторизоваться заново. Это помогает повысить безопасность приложения и избежать ситуации, когда кто-то получил доступ к старому токену.

.env

Добавим переменную окружения, указывающую срок жизни токена в секундах. Например, 3600 — это один час:

ACCESS_TOKEN_EXPIRE=3600

settings.py

В нашем классе настроек (Settings) добавим новое поле:

class Settings(BaseSettings):  
    # ... другие поля  
    access_token_expire: int

Генератор Redis-сессий

В нашем случае токен передаётся в cookie. Однако удаление cookie не гарантирует полной безопасности. Почему?

Представим, что токен попал в руки злоумышленника. Даже если пользователь выйдет из аккаунта (удалив cookie), сам токен останется действительным до окончания срока жизни. Это значит, что злоумышленник сможет продолжить использовать его, пока он не протухнет.

Чтобы избежать этой ситуации, мы будем сохранять все выданные токены в Redis и при каждом запросе проверять, есть ли этот токен в актуальном списке. Таким образом, при выходе пользователя из системы токен можно будет сразу удалить из Redis — и дальнейший доступ станет невозможным, даже если cookie осталась где-то в браузере.

💡 Если Redis тебе не знаком — это сверхбыстрая in-memory база данных, идеально подходящая (данные хранятся в оперативной памяти) для хранения временных данных, таких как токены, сессии, кеши и очереди задач.

Подключение Redis

В прошлой статье мы уже установили Redis и настроили подключение. Celery использует его "под капотом", но в этом случае нам нужно вручную управлять соединением, как мы делаем это с базой данных.

Для этого создадим отдельный генератор сессий Redis, по аналогии с генератором сессий базы данных.

Новый модуль redis_dependency.py

Внутри папки core_dependency создадим файл redis_dependency.py и добавим в него класс RedisDependency. Он будет содержать три метода:

1. __init__()

Конструктор класса, не принимает аргументов.

Он инициализирует два внутренних поля:

• self._url — URL для подключения к Redis, получаем из настроек;
• self._pool — объект класса ConnectionPool, который создаётся с помощью метода _init_pool().

def __init__(self) -> None:  
    self._url = settings.redis_settings.redis_url  
    self._pool: ConnectionPool = self._init_pool()

2. _init_pool()

Вспомогательный метод, создающий пул соединений с Redis.

Используем метод ConnectionPool.from_url() и передаём:

• url — адрес подключения;
• encoding — кодировку, указываем utf-8;
• decode_responses=True — автоматически декодировать полученные значения (например, строки вместо байтов).

def _init_pool(self) -> ConnectionPool:  
    return ConnectionPool.from_url(url=self._url, encoding="utf-8", decode_responses=True)

🔍 Что такое пул соединений? 
Это механизм повторного использования существующих соединений с Redis вместо постоянного создания новых. Он ускоряет работу и снижает нагрузку.

3. get_client()

Асинхронный контекстный менеджер, возвращающий подключение к Redis. Благодаря декоратору @asynccontextmanager он работает так же, как если бы мы описали dunder-методы __aenter__ и __aexit__.

@asynccontextmanager  
async def get_client(self) -> AsyncGenerator:  
    redis_client = Redis(connection_pool=self._pool)  
    try:  
        yield redis_client  
    finally:  
        await redis_client.aclose()

💡 Что делает yield в контекстном менеджере? 
Он "передаёт" управление наружу, позволяя использовать объект Redis внутри блока async with. После выхода из блока будет выполнен finally, закрывающий соединение.

Полный код redis_dependency.py

Обратите внимание: используем redis.asyncio для асинхронной работы, если импортировать из redis, то будут ошибки при асинхронном взаимодействии.

from collections.abc import AsyncGenerator  
from contextlib import asynccontextmanager  

from redis.asyncio import ConnectionPool, Redis  

from lkeep.core.settings import settings  


class RedisDependency:  
    def __init__(self) -> None:  
        self._url = settings.redis_settings.redis_url  
        self._pool: ConnectionPool = self._init_pool()  

    def _init_pool(self) -> ConnectionPool:  
        return ConnectionPool.from_url(url=self._url, encoding="utf-8", decode_responses=True)  

    @asynccontextmanager  
    async def get_client(self) -> AsyncGenerator:  
        redis_client = Redis(connection_pool=self._pool)  
        try:  
            yield redis_client  
        finally:  
            await redis_client.aclose()

Pydantic-схемы

В предыдущей статье — 
👉 FastAPI 5. Приложение аутентификации и Pydantic схемы — 
мы создавали основные Pydantic-схемы, в том числе RegisterUser с полями email и password.

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

Переименование схемы

Переименуем RegisterUser → AuthUser. 
Обратите внимание: нужно обновить все места, где использовался старый класс!

Как упростить замену в PyCharm:

Если вы работаете в PyCharm, можно сделать это за пару секунд с помощью инструмента рефакторинга:

1. Выделите имя класса RegisterUser;
2. Нажмите Shift+F6;
3. Введите новое имя: AuthUser;
4. Нажмите Enter.

PyCharm сам найдёт и заменит все использования класса в проекте.

Новая схема GetUserWithIDAndEmail

Добавим ещё одну схему: GetUserWithIDAndEmail. 
Она будет использоваться в ситуациях, когда нам нужно получить пользователя с тремя полями:

• id — идентификатор пользователя,
• email — почта,
• hashed_password — хэш пароля (например, при проверке токена).

Вместо того чтобы дублировать эти поля вручную, мы унаследуемся сразу от двух существующих схем: GetUserByID и CreateUser.

class GetUserWithIDAndEmail(GetUserByID, CreateUser):  
    pass

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

Именованный кортеж (NamedTuple)

Во многих случаях в Python функции возвращают несколько значений в виде обычного кортежа (tuple). 
Это удобно, но не всегда очевидно: при обращении к элементам приходится помнить их позицию (например, [0], [1]), а не суть значения. Это снижает читаемость и может привести к ошибкам.

Чтобы сделать возвращаемые значения более явными и самодокументируемыми, мы используем именованные кортежи — NamedTuple.

В нашем проекте понадобится вернуть из функции зашифрованный JWT и ID сессии. Вместо обычного (str, str), создадим специальный именованный кортеж:

auth/named_tuples.py

from typing import NamedTuple  


class CreateTokenTuple(NamedTuple):  
    encoded_jwt: str  
    session_id: str

Пример использования:

В функции:

def create_token(...) -> CreateTokenTuple:
    ...
    return CreateTokenTuple(encoded_jwt=token, session_id=session_id)

При получении результата:

result = create_token(...)
print(result.encoded_jwt)
print(result.session_id)

Преимущества NamedTuple:

✅ Явность — возвращается объект с понятными именами полей, а не просто (value1, value2).
✅ Подсказки типов — IDE понимает, что именно возвращается, подсказывает названия и типы.
✅ Непривязанность к ORM/Pydantic — легче и быстрее, чем dataclass или BaseModel, когда не нужна валидация.
✅ Иммутабельность — значения нельзя изменить после создания (как у tuple).

Маршрут авторизации

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

Создаём маршрут авторизации

Откроем файл routes.py в пакете auth.

Добавим в конце файла новую асинхронную функцию login, которая принимает:

• user: объект схемы AuthUser — данные пользователя, автоматически сериализуются из тела запроса;
• service: объект UserService, получаемый через Depends.

Возвращаемое значение — объект JSONResponse.

Маршрут оборачиваем в декоратор @auth_router.post, указывая путь и статус-код:

from fastapi import Depends  
from starlette import status  
from starlette.responses import JSONResponse

from lkeep.apps.auth.schemas import AuthUser  
from lkeep.apps.auth.services import UserService


@auth_router.post(path="/login", status_code=status.HTTP_200_OK)  
async def login(user: AuthUser, service: UserService = Depends(UserService)) -> JSONResponse:  
    return await service.login_user(user=user)

Сервис авторизации

Откроем файл services.py в пакете auth и создадим новый метод login_user в классе UserService.

В этом методе будет располагаться вся логика маршрута авторизации. Некоторые вспомогательные методы пока не реализованы, но мы сразу добавим их вызовы, чтобы потом просто дописать недостающие части.

Получение пользователя

Для начала нам нужно получить объект пользователя, который пытается авторизоваться. Сделаем это с помощью метода .get_user_by_email у объекта self.manager, передав туда email из объекта user.

exist_user = await self.manager.get_user_by_email(email=user.email)

Проверка существования и пароля

После получения пользователя выполняем две проверки в одном if:

• exist_user is None — если пользователь с таким email не найден в базе данных;
• not await self.handler.verify_password(...) — если переданный пароль не совпадает с хешированным.

Для этого вызываем метод .verify_password у self.handler и передаём туда:

• hashed_password — хеш пароля из базы (exist_user.hashed_password);
• raw_password — пароль, введённый пользователем (user.password).

Всё это объединяем с помощью оператора or:

if exist_user is None or not await self.handler.verify_password(
    hashed_password=exist_user.hashed_password, raw_password=user.password
):
    raise HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Wrong email or password"
    )

💡 Как это работает: Python сначала проверяет левую часть (exist_user is None). Если она истинна, правая часть даже не будет вычисляться (оптимизация or). Если же пользователь существует, тогда проверяется корректность пароля.

Если хотя бы одно из условий выполнено — поднимаем исключение HTTPException с кодом 401 и сообщением о неправильном логине или пароле.

Генерация и сохранение токена

Если проверка прошла успешно, вызываем метод .create_access_token у self.handler, передавая user_id. Метод возвращает пару: token и session_id.

token, session_id = await self.handler.create_access_token(user_id=exist_user.id)

Далее сохраняем токен авторизации в Redis через метод .store_access_token, передав сам токен, ID пользователя и session_id.

await self.manager.store_access_token(
    token=token,
    user_id=exist_user.id,
    session_id=session_id
)

Формирование ответа

Теперь создадим объект ответа. Используем JSONResponse и передадим в него словарь с сообщением:

response = JSONResponse(content={"message": "Вход успешен"})

Добавим в ответ установку куки с токеном, вызвав метод .set_cookie:

response.set_cookie(
    key="Authorization",
    value=token,
    httponly=True,
    max_age=settings.access_token_expire,
)

Аргументы:

• key="Authorization" — имя куки;
• value=token — сам токен;
• httponly=True — защита от доступа к куке через JavaScript;
• max_age=settings.access_token_expire — время жизни куки (в секундах), после которого она будет автоматически удалена браузером.

После этого возвращаем response как результат работы метода.

Полный код метода

async def login_user(self, user: AuthUser) -> JSONResponse:
    exist_user = await self.manager.get_user_by_email(email=user.email)

    if exist_user is None or not await self.handler.verify_password(
        hashed_password=exist_user.hashed_password,
        raw_password=user.password
    ):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Wrong email or password"
        )

    token, session_id = await self.handler.create_access_token(user_id=exist_user.id)

    await self.manager.store_access_token(
        token=token,
        user_id=exist_user.id,
        session_id=session_id
    )

    response = JSONResponse(content={"message": "Вход успешен"})
    response.set_cookie(
        key="Authorization",
        value=token,
        httponly=True,
        max_age=settings.access_token_expire,
    )

    return response

Обработчик авторизации

В прошлом блоке мы использовали вызовы методов, которых пока не существует. Самое время их добавить.

Нам нужно реализовать два метода:

• verify_password — проверка введённого пароля;
• create_access_token — генерация JWT-токена.

Откройте файл handlers.py в пакете auth.

Метод проверки пароля

Первым создадим метод verify_password, который принимает на вход два аргумента:

• raw_password — пароль, введённый пользователем;
• hashed_password — хэш пароля, полученный из базы данных.

Метод возвращает булево значение: True, если пароли совпадают, и False в противном случае.

Для сравнения используем встроенный метод .verify() у self.pwd_context. Он автоматически сравнит переданный пароль с хэшем.

async def verify_password(self, raw_password: str, hashed_password: str) -> bool:
    return self.pwd_context.verify(raw_password, hashed_password)

💡 Почему просто не сравниваем строки? 
Потому что пароли хэшируются — это односторонняя операция. Нельзя получить оригинальный пароль из хэша, но можно сравнить, совпадает ли хэш нового ввода с уже сохранённым.

Метод генерации JWT-токена

Теперь реализуем метод create_access_token, который принимает user_id и возвращает именованный кортеж CreateTokenTuple.

Сначала создаём дату и время истечения срока действия токена. Для этого к текущему времени (в UTC) прибавим timedelta на нужное количество секунд — значение берётся из настроек:

expire = datetime.datetime.now(datetime.UTC) + datetime.timedelta(seconds=settings.access_token_expire)

Далее генерируем уникальный session_id:

session_id = str(uuid.uuid4())

💡 Зачем нужен session_id?* 
Он позволяет отличать сессии одного пользователя — например, с телефона и с ноутбука. У каждой сессии будет свой токен, срок действия и возможность отозвать её отдельно.

Создаём словарь с полезной нагрузкой (payload) для токена:

data = {
    "exp": expire,  # срок действия токена
    "session_id": session_id,  # идентификатор сессии
    "user_id": str(user_id)  # идентификатор пользователя
}

Далее шифруем этот словарь в JWT-токен, используя секретный ключ и алгоритм HS256:

encoded_jwt = jwt.encode(payload=data, key=self.secret, algorithm="HS256")

Возвращаем результат в виде именованного кортежа:

return CreateTokenTuple(encoded_jwt=encoded_jwt, session_id=session_id)

Полный код метода:

async def create_access_token(self, user_id: uuid.UUID) -> CreateTokenTuple:
    expire = datetime.datetime.now(datetime.UTC) + datetime.timedelta(seconds=settings.access_token_expire)
    session_id = str(uuid.uuid4())

    data = {
        "exp": expire,
        "session_id": session_id,
        "user_id": str(user_id)
    }

    encoded_jwt = jwt.encode(payload=data, key=self.secret, algorithm="HS256")

    return CreateTokenTuple(encoded_jwt=encoded_jwt, session_id=session_id)

Менеджер авторизации

Осталось реализовать два метода:

• get_user_by_email — получение пользователя по email из базы данных;
• store_access_token — сохранение JWT-токена в Redis.

Откройте файл managers.py в пакете auth.

Обновление конструктора

На этапе подготовки мы добавляли класс RedisDependency, теперь подключим его в конструкторе UserManager — по аналогии с DBDependency.

Добавим redis в аргументы и сохраним в атрибуте self.redis:

def __init__(
    self, db: DBDependency = Depends(DBDependency), redis: RedisDependency = Depends(RedisDependency)
) -> None:
    self.db = db
    self.model = User
    self.redis = redis

Метод получения пользователя по email

Создадим метод get_user_by_email, принимающий email и возвращающий либо объект GetUserWithIDAndEmail, либо None, если пользователь не найден.

Используем уже знакомую схему работы с базой через async with:

async def get_user_by_email(self, email: str) -> GetUserWithIDAndEmail | None:
    async with self.db.db_session() as session:

Собираем SQL-запрос с помощью SQLAlchemy. Нам нужны только те поля, которые участвуют в авторизации: id, email и hashed_password.

query = select(
    self.model.id,
    self.model.email,
    self.model.hashed_password
).where(self.model.email  email)

Выполняем запрос и извлекаем первую запись:

result = await session.execute(query)
user = result.mappings().first()

Если пользователь найден — возвращаем его как объект схемы GetUserWithIDAndEmail. Иначе — None.

if user:
    return GetUserWithIDAndEmail(**user)

return None

Полный код метода

async def get_user_by_email(self, email: str) -> GetUserWithIDAndEmail | None:
    async with self.db.db_session() as session:
        query = select(
            self.model.id,
            self.model.email,
            self.model.hashed_password
        ).where(self.model.email  email)

        result = await session.execute(query)
        user = result.mappings().first()

        if user:
            return GetUserWithIDAndEmail(**user)

        return None

Метод сохранения JWT-токена в Redis

Создадим метод сохранения JTW-токена: store_access_token, принимающий token, user_id и session_id и ничего не возвращающий.

Внутри метода откроем асинхронный контекстный менеджер, но обратимся к другому полю, а именно к self.redis.get_client().

async with self.redis.get_client() as client:

Внутри контекстного менеджера вызовем метод .set() у объекта client, в который передадим два позиционных аргумента:

1. Ключ в виде строки. В нашем случае он формируется из user_id и session_id используя : как разделитель;
2. Значение аргумента token.

async def store_access_token(self, token: str, user_id: uuid.UUID, session_id: str) -> None:  
    async with self.redis.get_client() as client:  
        await client.set(f"{user_id}:{session_id}", token)

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

Для того, чтобы протестировать процесс авторизации, мы сперва запустим проект, выполнив poetry run app.

Есть несколько способов выполнить тестовый запрос:

cURL

• Что это: Консольная утилита для отправки HTTP-запросов.
• Пример:

 curl -X POST http://127.0.0.1:8000/api/v1/auth/login \
      -H "Content-Type: application/json" \
      -d '{"email":"user@example.com", "password":"string"}'

Postman

• Что это: GUI-клиент для тестирования API с поддержкой коллекций и сценариев.
• Как использовать:
  • Создайте запрос POST → укажите URL → в теле запроса (Body → raw → JSON) вставьте данные.

Swagger

• Что это: Встроенная интерактивная документация API.
• Плюсы: Тестирование через браузер, автоматическая подстановка параметров.

HTTPie

• Что это: Удобный инструмент для работы с HTTP-запросами с поддержкой GUI и CLI.
• Как использовать:
  • Создайте запрос POST → укажите URL → в теле запроса (Body → Text) вставьте данные.
• Плюсы:
  • Совмещает легкость CLI и удобство GUI.
  • Минималистичный дизайн, но мощный функционал.

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

Выбираем метод POST и прописываем путь http://127.0.0.1:8000/api/v1/auth/login

На вкладке "Body" выбираем "Text" и вписываем JSON с полями для авторизации.
В моём случае это стандартные данные:

{
  "email": "user@example.com",
  "password": "string"
}

Нажимаем на зелёную кнопку "Send" и в правой части появляется результат запроса:

Как видим, авторизация прошла успешно и в заголовках мы получили установку куки.

Заключение

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

В этой статье мы рассмотрели подход с использованием JWT, показали, как его удобно хранить в cookie для увеличения безопасности, и обсудили работу с Redis для более надежного управления токенами.

Знание всех деталей и особенностей авторизации поможет вам строить более защищенные веб-приложения. Помните, что безопасность всегда должна быть на первом месте, и даже если что-то кажется избыточным или сложным, это всегда оправдано для защиты данных пользователей.

В следующей статье напишем маршрут для выхода пользователя из системы (logout) и на этом закончим с приложением аутентификации.

Репозиторий проекта в GitHub: https://github.com/proDreams/lkeep 
Репозиторий проекта в "GIT на салфетке": https://git.pressanybutton.ru/proDream/lkeep