Где делать auth-проверку в FastAPI: Gateway, BFF или Handler

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

Когда добавляешь авторизацию в 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 должен:

Получить запрос GET /orders/abc123.
Пойти в базу или в order-service, чтобы узнать, кому принадлежит этот заказ.
Сравнить результат с 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              → PyJWKClient, JWT decode → Principal(sub, roles)
  ├─ require_roles("customer") → проверить roles ∋ "customer" → 403 если нет
  │
  └─ CreateOrderHandler.handle(cmd)
       └─ (нет проверки владения: customer создаёт свой заказ, sub идёт в customer_id)

GET /orders/{id}
  │
  ├─ principal              → Principal(sub="user-42", roles=["customer"])
  ├─ require_roles("customer","admin") → OK
  │
  └─ GetOrderByIdHandler.handle(query)
       ├─ repo.find_by_id(order_id) → Order(customer_id="user-99")
       └─ "user-42" != "user-99" AND "admin" ∉ roles → ForbiddenError → 403

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

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.

Коротко

Три уровня — три разных вопроса: кто, может ли вообще, может ли работать с этим ресурсом.
Gateway проверяет JWT: подпись, exp, iss, aud. Возвращает Principal. Не знает про доменную модель.
BFF проверяет роль через Depends(require_roles(...)) на каждом endpoint. Нет роли — 403.
Handler проверяет владение: aggregate.owner_id == principal.sub. Только Handler знает, кому принадлежит ресурс.
Невалидный токен — 401, нехватка роли — 403. Не путать.
PyJWKClient кеширует публичные ключи — не надо ходить за ними на каждый запрос.

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

JWT validation в FastAPI — PyJWKClient, PyJWT, алгоритмы, claims.
RBAC: маппинг ролей — require_roles, _roles(claims), разрешённые роли.
ABAC: владение ресурсом — owner_id == sub, admin-bypass.
Service-to-service — mTLS, Client Credentials Flow.