← назад к разделу

Когда добавляешь авторизацию в FastAPI, первый порыв — сделать одну зависимость и поставить её везде. Но авторизация — это три разные проверки, и каждая делается в своём месте. Если их смешать, появляются дыры: кто-то видит чужие данные, или Gateway начинает ходить в базу за каждым запросом.

Разберём, где что проверяется и почему именно там.

Обязательно

Три разных вопроса — три разных места

Каждый уровень отвечает на свой вопрос:

УровеньВопросFastAPI-инструмент
Gateway / API edgeКто этот клиент? (JWT валиден?)PyJWKClient + зависимость principal
BFF / Application LayerМожет ли этот клиент вообще обратиться к endpoint?Depends(require_roles(...)) на роутере
Domain HandlerМожет ли этот клиент работать с конкретным ресурсом?сравнение aggregate.owner_id == principal.sub

Путаница начинается, когда эти вопросы смешивают в одном месте.

Gateway — аутентификация

Gateway отвечает только на вопрос «кто этот клиент». Он проверяет подпись JWT, срок действия (exp), издателя (iss) и аудиторию (aud). Ничего больше.

В FastAPI это оформляется одной общей зависимостью:

# adapters/in/http/security.py
from jwt import PyJWKClient
import jwt as pyjwt
from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
_jwks = PyJWKClient(settings.jwks_uri, cache_keys=True, lifespan=300)


async def principal(token: str = Depends(oauth2_scheme)) -> Principal:
    try:
        key = _jwks.get_signing_key_from_jwt(token).key
        claims = pyjwt.decode(
            token, key,
            algorithms=["RS256"],
            audience=settings.audience,
            issuer=settings.issuer,
        )
    except pyjwt.PyJWTError as e:
        raise HTTPException(status_code=401, detail="invalid token") from e
    return Principal(sub=claims["sub"], roles=_roles(claims))

PyJWKClient кеширует публичные ключи ~5 минут — не надо ходить за ними на каждый запрос. Невалидная подпись или просроченный токен → 401 Unauthorized. До бизнес-логики запрос не доходит.

Что Gateway не делает:

  • Не проверяет, какие роли нужны для конкретного endpoint.
  • Не знает ничего про доменные объекты — не ходит в базу за заказами или продуктами.

BFF — грубая проверка по роли

BFF отвечает на вопрос «может ли этот клиент вообще обратиться к этому endpoint». Это не «чей ресурс», а просто «есть ли у пользователя нужная роль».

В FastAPI удобно сделать фабрику зависимостей:

def require_roles(*roles: str):
    async def dep(p: Principal = Depends(principal)) -> Principal:
        if not set(roles) & set(p.roles):
            raise HTTPException(status_code=403, detail="forbidden")
        return p
    return dep

И декларировать роли прямо на каждом endpoint:

router = APIRouter(prefix="/orders")

@router.post("/")
async def create_order(
    body: CreateOrderRequest,
    p: Principal = Depends(require_roles("customer")),
    handler: CreateOrderHandler = Depends(),
) -> OrderResponse:
    return handler.handle(CreateOrderCommand(customer_id=p.sub, **body.model_dump()))

@router.get("/{order_id}")
async def get_order(
    order_id: UUID,
    p: Principal = Depends(require_roles("customer", "admin")),
    handler: GetOrderByIdHandler = Depends(),
) -> OrderResponse:
    return handler.handle(GetOrderByIdQuery(order_id=order_id, principal=p))

Нет нужной роли — 403 Forbidden до вызова Handler. Handler вообще не запускается.

Частая ошибка — оставить endpoint без require_roles. Тогда любой пользователь с валидным токеном сможет к нему обратиться.

Domain Handler — проверка по ресурсу

Handler отвечает на вопрос «может ли этот конкретный пользователь работать с этим конкретным ресурсом». Здесь уже нужна доменная модель.

Роль customer проверена на уровне BFF, но это не значит, что customer может читать чужой заказ:

# application/use_cases/get_order_by_id.py
@dataclass(frozen=True)
class GetOrderByIdQuery:
    order_id: UUID
    principal: Principal


class GetOrderByIdHandler:
    def __init__(self, repo: OrderRepository) -> None:
        self._repo = repo

    def handle(self, query: GetOrderByIdQuery) -> Order:
        order = self._repo.find_by_id(query.order_id)
        if order is None:
            raise NotFoundError(f"Order {query.order_id} not found")

        if str(order.customer_id) != query.principal.sub and "admin" not in query.principal.roles:
            raise ForbiddenError("Order does not belong to current user")

        return order

order.customer_id != principal.sub — именно здесь решается вопрос владения. И только здесь, потому что только Handler знает, что такое Order и кому он принадлежит.

Тот же принцип для других агрегатов:

class UpdateProductHandler:
    def handle(self, cmd: UpdateProductCommand) -> Product:
        product = self._repo.find_by_id(cmd.product_id)
        if product is None:
            raise NotFoundError(f"Product {cmd.product_id} not found")

        if str(product.seller_id) != cmd.principal.sub and "admin" not in cmd.principal.roles:
            raise ForbiddenError("Product does not belong to current seller")

        return self._repo.save(product.update(cmd))

Роль admin обходит проверку владения — но это осознанное исключение, а не дыра.

Почему не делать проверку владения на Gateway

Кажется удобным: токен валиден, проверим сразу кому принадлежит ресурс. Но тогда Gateway должен:

  1. Получить запрос GET /orders/abc123.
  2. Пойти в базу или в order-service, чтобы узнать, кому принадлежит этот заказ.
  3. Сравнить результат с sub из токена.

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

Маппинг ролей из токена

Разные провайдеры хранят роли в разных полях:

def _roles(claims: dict) -> list[str]:
    realm = claims.get("realm_access", {})
    keycloak_roles = realm.get("roles", [])
    scope_roles = claims.get("scope", "").split()
    return keycloak_roles or scope_roles

Keycloak кладёт роли в realm_access.roles, стандартный OAuth2 — в scope. Функция _roles скрывает это различие. Principal всегда получает простой список строк — без знания о провайдере.

Полная цепочка на примере

POST /orders oauth2_scheme: достать Bearer principal: разобрать JWT require_roles("customer") CreateOrderHandler GET /orders/{id} principal: sub и роли require_roles: customer или admin GetOrderByIdHandler сверка владельца: 403

Роль проверяется на входе, а владение — внутри обработчика: на входе ещё неизвестно, чей это заказ. Поэтому «customer создаёт свой заказ» проверять нечего, а «customer читает чужой» ловится только после загрузки заказа.

Частые ошибки

RBAC на Gateway. Gateway только аутентифицирует; проверку ролей делает require_roles на роутере.

Проверка владения в роутере. Роутер не знает, что лежит в базе. Сравнение owner_id == sub — только внутри Handler.

jwt.decode(token, options={"verify_signature": False}). Без проверки подписи любой токен считается валидным. Используйте PyJWKClient.

Endpoint без Depends(require_roles(...)). Любой пользователь с токеном получит доступ. Каждый endpoint должен явно декларировать роли.

403 при невалидном JWT. Невалидный токен — это 401 Unauthorized. Нехватка роли — это 403 Forbidden. Коды различаются.

Только проверка роли без проверки владения для own-resource endpoint. Роль customer не означает, что customer может читать заказы других customers.

Дополнительно: при первом чтении можно пропустить

Глубже: граница: TLS, заголовки и что шлюз срезаетрасширенное

Три уровня выше отвечают на вопрос «кто ты и что тебе можно». У границы есть ещё обязанности, которые к личности не относятся, и без них проверка прав защищает только от вежливых.

Где заканчивается TLS. Шифрование обычно терминируется на балансировщике или ingress, и до uvicorn запрос доходит обычным HTTP. Приложение узнаёт о схеме и адресе клиента только из заголовков X-Forwarded-Proto и X-Forwarded-For, а верить им можно лишь от своего прокси: uvicorn по умолчанию читает их только с 127.0.0.1, и в кластере это означает «не читает». Флаг --forwarded-allow-ips с адресами или подсетью ingress включает доверие; значение * открывает подмену адреса любому клиенту, который сам пришлёт X-Forwarded-For. Внутренний трафик между подами либо шифрует сетка (mTLS, статья про межсервисную аутентификацию), либо он идёт открытым, и это осознанное решение, а не умолчание.

Что шлюз срезает на входе. Всё, что внутренние сервисы считают доверенным, шлюз обязан перезаписать, а не дополнить: X-User-Id, X-User-Roles, X-Forwarded-*. Пришедший снаружи X-User-Id: 777 должен исчезнуть до того, как запрос уйдёт дальше; как это делается через MutableHeaders, показано в статье про структурные паттерны.

Заголовки безопасности ставят один раз на границе, а не в каждом сервисе: Strict-Transport-Security, X-Content-Type-Options: nosniff, X-Frame-Options: DENY и Content-Security-Policy для страниц. И лимит частоты по адресу и пути живёт там же, до аутентификации: перебор паролей и токенов должен упираться в 429 раньше, чем в вашу базу.

Глубже: дыры помимо доступа: инъекции, SSRF, обход пути и десериализациярасширенное

Половину инцидентов даёт не ошибка в правах, а код, который правильно проверил пользователя и потом сделал то, что он попросил, буквально.

Инъекции. text(f"... WHERE email = '{email}'") выполняет всё, что пришло в email. SQLAlchemy с параметрами text("... WHERE email = :email").bindparams(email=email) и ORM-запросы передают значение отдельно от текста, и это правило без исключений, включая ORDER BY по имени поля из запроса: его сверяют со списком допустимых колонок, а не подставляют.

SSRF. Сервис скачивает картинку по адресу от пользователя или зовёт вебхук по адресу, который ввёл клиент. Адрес http://169.254.169.254/ отдаст метаданные облака, http://redis:6379/ достанет до внутренних систем. Защита: список разрешённых хостов, запрет частных диапазонов после разрешения DNS, таймаут и отключённые перенаправления (httpx их по умолчанию не следует, и это стоит оставить).

Обход пути. open(f"/srv/files/{name}") с именем ../../etc/passwd читает что угодно. Проверяют после нормализации: path = (base / name).resolve() и path.is_relative_to(base), иначе отказ; лучше вовсе не принимать имена файлов, а выдавать свои идентификаторы.

Десериализация. pickle.loads на данных снаружи это выполнение чужого кода, yaml.load без SafeLoader то же самое; берут yaml.safe_load и Pydantic как границу, через которую внешний JSON становится вашим объектом. Туда же subprocess с shell=True и строкой, собранной из ввода.

Эти четыре проверки ловят в сборке bandit и ревью; какие ещё проверки ставят в CI, разобрано в статье про персональные данные и секреты.

Коротко

  • Три уровня — три разных вопроса: кто, может ли вообще, может ли работать с этим ресурсом.
  • Gateway проверяет JWT: подпись, exp, iss, aud. Возвращает Principal. Не знает про доменную модель.
  • BFF проверяет роль через Depends(require_roles(...)) на каждом endpoint. Нет роли — 403.
  • Handler проверяет владение: aggregate.owner_id == principal.sub. Только Handler знает, кому принадлежит ресурс.
  • Невалидный токен — 401, нехватка роли — 403. Не путать.
  • PyJWKClient кеширует публичные ключи — не надо ходить за ними на каждый запрос.
  • У границы свои обязанности: TLS до ingress и доверие X-Forwarded-* только своему прокси, перезапись X-User-Id, заголовки безопасности; а инъекции, SSRF, обход пути и pickle обходят права вовсе.

Что почитать дальше