Аутентификация по JWT: защищаем эндпоинты

Сервис задач с базой данных из седьмого урока выложили на хостинг — и теперь любой прохожий может прочитать и удалить чужие задачи: у эндпоинтов нет ни дверей, ни замков. Пора заводить учётные записи. Сегодня строим аутентификацию по JWT (JSON Web Token) — схему, которая стоит за половиной API планеты: регистрация с хешированием пароля, выдача токена на /login и зависимость get_current_user, которая закрывает эндпоинты от анонимов.

Хорошая новость: ничего нового изобретать не придётся. Зависимости и Depends вы отработали в прошлом уроке, модели и Field — в четвёртом, HTTPException со статусами — в шестом. Сегодня эти три инструмента собираются в замок на входе сервиса.

Схема аутентификации: register, login, Bearer

1. POST /register — клиент присылает username и пароль, сервер сохраняет пользователя. Пароль уходит в базу только в виде хеша.
2. POST /login — сервер сверяет пару логин-пароль и выдаёт подписанный токен со сроком жизни.
3. Клиент хранит токен и прикладывает заголовок Authorization: Bearer <токен> к каждому запросу.
4. Сервер проверяет подпись и срок токена — и только потом выполняет эндпоинт.

Почему токен, а не привычные сессии с куками? Токен stateless: сервер не хранит «кто вошёл» — вся информация зашита в самом токене и защищена подписью. Ни общей памяти, ни таблицы сессий, ни sticky-сессий на балансировщике: хоть три экземпляра сервиса в трёх дата-центрах — проверка подписи везде одинаковая. Мобильному приложению куки вообще не родные, а заголовок Authorization — универсальный стандарт.

Пароли не хранят открыто: хеш и соль

Начнём с фундамента, на котором нельзя экономить: пароль в базе хранить в открытом виде нельзя. Не «нежелательно», а нельзя: базы утекают регулярно, и владелец утечки обязан быть уверенным, что пароли пользователей злоумышленник не прочитает. Вместо пароля хранят хеш — результат односторонней функции: из пароля в хеш дорога есть, обратно — нет. Плюс соль — случайная строка, примешиваемая к паролю перед хешированием. Оба механизма воспроизводятся стандартной библиотекой, и этот блок запускается прямо на странице:

import hashlib

def hash_password(password: str, salt: bytes) -> str:
    # В настоящем коде соль генерирует os.urandom(16) при регистрации;
    # здесь она фиксированная, чтобы вывод был воспроизводимым
    digest = hashlib.sha256(salt + password.encode()).hexdigest()
    return salt.hex() + "$" + digest

def verify_password(password: str, stored: str) -> bool:
    salt_hex, digest = stored.split("$")
    # Хешируем присланный пароль с той же солью и сравниваем
    return hash_password(password, bytes.fromhex(salt_hex)) == stored

SALT = bytes.fromhex("1a2b3c4d5e6f708192a3b4c5d6e7f809")
h1 = hash_password("кот-днём-спит", SALT)
h2 = hash_password("кот-днём-спит", bytes.fromhex("00ff00ff00ff00ff00ff00ff00ff00ff"))

print(h1)
print(h1 == h2)                          # пароль тот же, соль другая
print(verify_password("кот-днём-спит", h1))
print(verify_password("неверный", h1))

1a2b3c4d5e6f708192a3b4c5d6e7f809$5e1cfc9df3d855fa30243dc5ed76ceb11df17456e212b3e71385b3f5a48310ec
False
True
False

Запустите блок и посмотрите на формат хранения: соль$хеш — соль лежит открыто рядом с хешем, и это нормально, она не секрет. Секретность обеспечивают пароль и стойкость функции, а соль нужна, чтобы одинаковые пароли разных людей превращались в разные хеши: первый print показал хеш, второй — что при другой соли он уже не совпадёт. Проверка при логине симметрична: пароль пользователя хешируется с его же сохранённой солью, и результат сверяется с тем, что в базе.

Почему sha256 — только для демо, а в проде bcrypt и argon2

sha256 считает хеш за микросекунды — это отлично для подписей файлов, но плохо для паролей: видеокарта перебирает миллиарды вариантов в секунду, и даже с солью короткий пароль подбирается за часы. bcrypt и argon2 — функции, которые намеренно медленные: проверка пароля занимает около 100 миллисекунд, а «фактор стоимости» можно повышать по мере роста процессоров. Для вас разница невелика — API почти тот же:

from passlib.context import CryptContext

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

hash_olga = pwd_context.hash("кот-днём-спит")
print(hash_olga[:7], "...")
print(pwd_context.verify("кот-днём-спит", hash_olga))  # True
print(pwd_context.verify("неверный", hash_olga))       # False

$2b$12$ ...
True
False

Дальше по уроку мы пользуемся функциями hash_password и verify_password как абстракцией: какие именно они внутри — sha256 для песочницы или bcrypt для продакшена — остальной код не волнует. Это хороший стиль: механизм хеширования изолирован в двух функциях и меняется в одном месте.

Регистрация: POST /register

Добавим к сервису из седьмого урока таблицу пользователей и эндпоинт регистрации. Обратите внимание на три решения: пароль принимаем только с минимумом длины (Field из урока 4), имя делаем уникальным на уровне базы, а наружу отдаём только username — без всяких хешей:

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(30), unique=True)
    password_hash: Mapped[str] = mapped_column(String(200))


class UserCreate(BaseModel):
    username: str = Field(min_length=3, max_length=30)
    password: str = Field(min_length=8)


class UserPublic(BaseModel):
    username: str


@app.post("/register", status_code=201, response_model=UserPublic)
def register(payload: UserCreate, db: Session = Depends(get_db)):
    exists = db.scalars(
        select(User).where(User.username == payload.username)
    ).first()
    if exists is not None:
        raise HTTPException(status_code=409, detail="Имя уже занято")
    user = User(
        username=payload.username,
        password_hash=hash_password(payload.password),
    )
    db.add(user)
    db.commit()
    return user

{"username": "olga"}

Страховка от утечки — двойная. Первая: пароль превращается в хеш до того, как попадёт в ORM-объект, в колонке password_hash лежит уже необратимая строка. Вторая: response_model=UserPublic отрежет password_hash, даже если однажды вернёте полную модель по невнимательности — тот самый фильтр из шестого урока. Статус 409 вместо 400 — потому что конфликт не с форматом запроса, а с текущим состоянием базы: имя занято.

JWT: три строки, склеенные точками

Токен выглядит как абракадабра, но устроен примитивно: header.payload.signature — три куска base64, склеенные точками. header описывает алгоритм подписи, payload — данные (в терминах JWT — claims): кто пользователь, когда истекает токен. signature — подпись HMAC-SHA256 от первых двух частей секретным ключом сервера. Соберём такой токен руками — стандартный base64 и hmac доступны в песочнице:

import base64, hashlib, hmac, json

def b64url(data: bytes) -> str:
    # Вариант base64 для JWT: без переносов, без знаков = и +
    return base64.urlsafe_b64encode(data).rstrip(b"=").decode()

def sign(msg: bytes, secret: bytes) -> str:
    return b64url(hmac.new(secret, msg, hashlib.sha256).digest())

SECRET = b"dev-secret-change-me"

header = json.dumps({"alg": "HS256", "typ": "JWT"}, separators=(",", ":"))
payload = json.dumps(
    {"sub": "olga", "role": "user", "exp": 1752600000},
    separators=(",", ":"),
)

h = b64url(header.encode())
p = b64url(payload.encode())
s = sign((h + "." + p).encode(), SECRET)

print(h)
print(p)
print(s)
print(h + "." + p + "." + s)

eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9
eyJzdWIiOiJvbGdhIiwicm9sZSI6InVzZXIiLCJleHAiOjE3NTI2MDAwMDB9
gVtw5Frx6Zk9OOlcEXmu0bpuC4Bkfzr1AwkHlVaCUjc
eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJvbGdhIiwicm9sZSI6InVzZXIiLCJleHAiOjE3NTI2MDAwMDB9.gVtw5Frx6Zk9OOlcEXmu0bpuC4Bkfzr1AwkHlVaCUjc

Ключевые поля payload: sub (subject) — кто владелец токена, у нас имя пользователя; exp (expiration) — метка времени Unix, после которой токен недействителен. Проверьте сами: закодируйте первую строку вывода через base64.urlsafe_b64decode, дописав два знака ==, — и увидите JSON заголовка. Никакой магии, только кодировки и арифметика.

Подпись ловит подделку: пробуем заменить payload

Зачем вообще подпись? Злоумышленник видит структуру токена — и пробует выдать себя за админа: берёт payload, меняет sub и role на admin, подпись оставляет старую. Проверим, что из этого выйдет:

import base64, hashlib, hmac, json

def b64url(data: bytes) -> str:
    return base64.urlsafe_b64encode(data).rstrip(b"=").decode()

def sign(msg: bytes, secret: bytes) -> str:
    return b64url(hmac.new(secret, msg, hashlib.sha256).digest())

SECRET = b"dev-secret-change-me"

def make_token(payload: dict) -> str:
    head = b64url(json.dumps({"alg": "HS256", "typ": "JWT"}, separators=(",", ":")).encode())
    body = b64url(json.dumps(payload, separators=(",", ":")).encode())
    return head + "." + body + "." + sign((head + "." + body).encode(), SECRET)

def verify(token: str) -> bool:
    h, p, s = token.split(".")
    return hmac.compare_digest(sign((h + "." + p).encode(), SECRET), s)

good = make_token({"sub": "olga", "role": "user", "exp": 1752600000})
print(verify(good))  # свой токен

# Злоумышленник подменил payload, а подпись оставил старую
h, p, s = good.split(".")
bad_body = b64url(json.dumps(
    {"sub": "admin", "role": "admin", "exp": 1752600000},
    separators=(",", ":"),
).encode())
forged = h + "." + bad_body + "." + s
print(verify(forged))

True
False

Второй print вернул False — сервер отвергнет подделку со статусом 401, даже не глядя на содержимое. Обратите внимание на hmac.compare_digest: сравнение подписей делается через него, а не через ==, чтобы исключить тайминг-атаки, когда длина совпадения подбирается по времени ответа. Мелочь, а входит в привычку за один урок.

Логин: POST /login выдаёт токен

Регистрация есть, осталось пускать своих. Эндпоинт логина принимает пару логин-пароль, сверяет хеш и возвращает токен. За кодирование отвечает библиотека python-jose (вариант — PyJWT, API почти одинаков): она сама соберёт header, payload и подпись, как мы делали руками выше:

from datetime import datetime, timedelta, timezone

from fastapi import Depends, FastAPI, HTTPException
from fastapi.security import OAuth2PasswordRequestForm
from jose import jwt  # pip install "python-jose[cryptography]"

SECRET_KEY = "..."  # из переменных окружения, об этом ниже
ALGORITHM = "HS256"
TOKEN_EXPIRE_MINUTES = 30


def create_access_token(username: str) -> str:
    expires = datetime.now(timezone.utc) + timedelta(minutes=TOKEN_EXPIRE_MINUTES)
    return jwt.encode(
        {"sub": username, "exp": expires},
        SECRET_KEY,
        algorithm=ALGORITHM,
    )


@app.post("/login")
def login(form: OAuth2PasswordRequestForm = Depends(),
          db: Session = Depends(get_db)):
    user = db.scalars(
        select(User).where(User.username == form.username)
    ).first()
    if user is None or not verify_password(form.password, user.password_hash):
        raise HTTPException(status_code=401, detail="Неверный логин или пароль")
    return {
        "access_token": create_access_token(user.username),
        "token_type": "bearer",
    }

{"access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJvbGdhIiwiZXhwIjoxNzUyNjAzNjAwfQ.YN2mB884Fyxz7P_tkwzP0OxUOLJ_4wXfpPX5LnKBo8c", "token_type": "bearer"}

OAuth2PasswordRequestForm читает данные из form-data с полями username и password — тот же формат, что у обычной HTML-формы. Это не случайность: именно этот формат ждёт кнопка Authorize в Swagger, и она сама сходила бы в tokenUrl за токеном. Важна и деталь с exp: срок жизни вшит в токен, поэтому «выход из системы» не требует ничего стирать на сервере — истёкший токен просто перестаёт приниматься. Обратная сторона: отозвать живой токен до истечения сервер не может, об этом — в FAQ.

get_current_user: закрываем эндпоинты зависимостью

Последний кирпич — зависимость, которая из токена восстанавливает пользователя. Строится она слоями из тех же Depends: сначала OAuth2PasswordBearer достаёт заголовок Authorization, затем get_current_user проверяет подпись и срок, ищет пользователя в базе, и уже он подставляется в эндпоинт:

from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")


def get_current_user(token: str = Depends(oauth2_scheme),
                     db: Session = Depends(get_db)) -> User:
    error = HTTPException(
        status_code=401,
        detail="Недействительный токен",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    except JWTError:
        raise error  # кривая подпись, истёкший exp, битый base64
    username = payload.get("sub")
    if username is None:
        raise error
    user = db.scalars(
        select(User).where(User.username == username)
    ).first()
    if user is None:
        raise error
    return user


@app.get("/me")
def me(current_user: User = Depends(get_current_user)) -> dict:
    return {"username": current_user.username}

{"username": "olga"}

Смотрите, как аккуратно легли слои: эндпоинт /me вообще не знает про токены — он получает готовый объект User и занимается своими задачами. Вся проверка живёт в get_current_user, а его можно подставить в любой эндпоинт, который надо закрыть: def create_task(payload: TaskCreate, db: Session = Depends(get_db), user: User = Depends(get_current_user)) — и POST /tasks больше не принимает анонимов. Заодно смотрите на algorithms=[ALGORITHM]: список разрешённых алгоритмов обязательно фиксируется при decode, иначе фреймворк поверит тому, что заявлено в header подделки.

Кнопка Authorize в Swagger

OAuth2PasswordBearer с tokenUrl="login" окупается сразу: в документации по адресу /docs появляется кнопка Authorize. Вводите логин и пароль — Swagger сам отправляет POST /login, запоминает токен и подставляет заголовок Authorization во все последующие запросы из интерфейса. За сорок строк аутентификации вы бесплатно получаете интерактивный вход прямо в документации API — тестировщики скажут спасибо.

SECRET_KEY: единственный секрет всего замка

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

# В терминале: случайные 32 байта в hex-кодировке
export SECRET_KEY="$(openssl rand -hex 32)"
uvicorn main:app --reload

# В main.py секрет читается из окружения, а не вшит в код
import os
SECRET_KEY = os.environ["SECRET_KEY"]

Прогоняем сценарий целиком и что дальше

Соберём весь путь пользователя в четыре команды: регистрация, логин, запрос с токеном и попытка пролезть без него. Обратите внимание на разницу в формате: /register принимает JSON, /login — form-data:

# 1. Регистрируем пользователя
curl -X POST localhost:8000/register \
  -H "Content-Type: application/json" \
  -d '{"username": "olga", "password": "s3cret-pass"}'

# 2. Логинимся и получаем токен
curl -X POST localhost:8000/login \
  -d "username=olga" -d "password=s3cret-pass"

# 3. Ходим с токеном в заголовке
curl localhost:8000/me \
  -H "Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJvbGdhIiwiZXhwIjoxNzUyNjAzNjAwfQ.YN2mB884Fyxz7P_tkwzP0OxUOLJ_4wXfpPX5LnKBo8c"

# 4. Тот же запрос без заголовка
curl localhost:8000/me

{"username": "olga"}
{"access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJvbGdhIiwiZXhwIjoxNzUyNjAzNjAwfQ.YN2mB884Fyxz7P_tkwzP0OxUOLJ_4wXfpPX5LnKBo8c", "token_type": "bearer"}
{"username": "olga"}
{"detail": "Not authenticated"}

1. Пароль хранится только в виде хеша с солью; для продакшена — bcrypt или argon2, для песочницы — hashlib.
2. Токен — это header.payload.signature, подпись HMAC-SHA256 секретным ключом сервера.
3. Payload читается кем угодно, но меняется ни кем: подпись ловит подмену, exp ограничивает срок жизни.
4. SECRET_KEY — только из переменных окружения, никогда из кода.
5. Защита эндпоинта — один параметр: user: User = Depends(get_current_user).
6. 401 — не представился, 403 — представился, но нельзя.

Чек-лист короткий, но за ним стоит целый замок: аноним получает 401 до того, как его код вообще выполнится. Задание на сейчас: закройте зависимостью get_current_user эндпоинт DELETE /tasks/{task_id} из седьмого урока и решите вопрос по-честному: задачи удаляет только тот, кто их создал. Подсказка: в таблицу tasks понадобится колонка user_id, а в get_current_user она уже почти есть.

В девятом уроке наш сервис впервые проверит сам себя: TestClient и pytest прогонят регистрацию, логин, запрос без токена и CRUD с базой — и 401 станет не страшной ошибкой, а зелёным assert'ом. А в финальном проекте десятого урока всё это соберётся в один сервис заметок: SQLite, JWT, теги и Swagger-контракт в придачу. Замок готов — ступени, которые остались, собраны в самоучителе FastAPI: тесты и финальный проект.