Опирается на правила: AUTH-1, AUTH-2, AUTH-3 из Auth Patterns Style Guide → раздел 1. Где какая проверка делается.

Важно знать

  • Gateway / API edge — аутентификация: валидация JWT (подпись, exp, iss, aud) через PyJWKClient + JWKS-кеш. Прокидывает identity вниз.
  • BFF / Application Layer — грубая авторизация по роли (RBAC) через Depends(require_roles(...)) на каждом endpoint.
  • Domain Service / Handler — авторизация по ресурсу (ABAC): order.customer_id == principal.sub.
  • ABAC никогда на Gateway — Gateway не знает доменную модель.
  • Endpoint без Depends(require_roles(...)) — критическое нарушение AUTH-9.
  • Невалидная подпись / просроченный exp → 401, не 403 — коды различаются (AUTH-6).
  • Размытие ответственности по слоям — главный источник дыр в авторизации.

Auth — это не один монолитный шаг, а три разных проверки на трёх разных уровнях. В FastAPI каждый уровень оформляется через зависимости (Depends): одна общая зависимость principal валидирует токен, фабрика require_roles защищает endpoint, Handler сравнивает агрегат с текущим пользователем.

Три уровня и три ответственности

УровеньЧто проверяетFastAPI-механизм
Gateway / API edgeподпись JWT, exp, iss, aud, rate limitPyJWKClient + JWKS-кеш, зависимость principal
BFF / Application LayerRBAC: есть ли роль для этого endpointDepends(require_roles("admin")) на роутере
Domain HandlerABAC: владеет ли этот user этим ресурсомсравнение aggregate.owner_id == principal.sub в Handler

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

AUTH-1: edge отвечает на вопрос «кто этот клиент».

В FastAPI роль gateway может играть внешний Istio/Kong, либо сам сервис, если внешнего gateway нет. В обоих случаях JWT валидируется одной общей зависимостью:

# adapters/in/http/security.py
from jose import JWTError
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))

_jwks кеширует ключи ~5 минут — JWK Set не запрашивается на каждый запрос (AUTH-5). Невалидная подпись или просроченный exp → 401 Unauthorized, до бизнес-логики запрос не доходит.

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

  • Не проверяет, какие роли нужны для endpoint.
  • Не знает доменную модель — не сравнивает order.customer_id ни с чем.

BFF — грубая авторизация по роли

AUTH-2: BFF отвечает на вопрос «может ли этот клиент вообще обратиться к этому endpoint».

Фабрика require_roles возвращает зависимость, которая проверяет пересечение ролей и бросает 403 при отсутствии:

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

Роутеры используют её декларативно:

# adapters/in/http/order_router.py
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))

Разделение endpoint-ов по ролям:

  • POST /orders — только customer (продавец не создаёт заказы клиента).
  • GET /orders/{id} — customer или admin.
  • POST /admin/orders/{id}/refund — только admin.

Если роль не подходит — 403 до вызова Handler.

Domain Handler — авторизация по ресурсу

AUTH-3: Handler отвечает на вопрос «может ли этот клиент работать с этим конкретным ресурсом».

# application/use_cases/get_order_by_id.py
from dataclasses import dataclass
from uuid import UUID
from domain.order import Order
from ports.order_repository import OrderRepository
from adapters.in.http.security import Principal
from shared.errors import ForbiddenError, NotFoundError


@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

ABAC в Handler — это не RBAC. Роль customer валидна, endpoint доступен, но конкретный customer не может читать чужой заказ. auth-3 запрещает проверять это на Gateway: Gateway не знает, кому принадлежит order_id=abc123.

Для домена Product / Seller структура аналогична:

# application/use_cases/update_product.py
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 обходит ABAC — полный доступ — но каждое действие должно попасть в audit log (AUTH-12, AUTH-15).

Почему ABAC не на Gateway

AUTH-3 запрещает ABAC на Gateway. Причина прямолинейна:

  1. Gateway получает GET /orders/abc123, JWT валиден, sub=user-99.
  2. Чтобы ответить «чей этот заказ» — нужно сходить в БД или в order-service.
  3. Gateway становится сервисом, дублирует доменную логику.
  4. При добавлении co-owners, делегирования, организационной структуры — Gateway обновляется параллельно с domain-моделью.

Это разрушает единый источник правды о владении. ABAC живёт там, где живёт Order aggregate.

То же относится к BFF: RBAC на роутере, ABAC в Handler — не наоборот.

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

AUTH-7: роли берутся из claim, не захардкожены в сервисе.

# adapters/in/http/security.py
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. Функция скрывает это различие за Principal.roles.

Полная цепочка для заказа

POST /orders
  │
  ├─ oauth2_scheme         → извлечь Bearer-токен
  ├─ principal             → PyJWKClient, JWT decode → Principal(sub, roles)
  ├─ require_roles("customer") → проверить roles ∋ "customer" → 403 если нет
  │
  └─ CreateOrderHandler.handle(cmd)
       └─ (нет ABAC: 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 на GatewayAUTH-2Gateway только аутентификация; require_roles на роутере
ABAC на Gateway или в роутереAUTH-3ABAC в Handler, где живёт агрегат
jwt.decode(token, options={"verify_signature": False})AUTH-4PyJWKClient + проверка подписи обязательна
Endpoint без Depends(require_roles(...))AUTH-9каждый endpoint декларирует роли явно
status_code=403 при невалидном JWTAUTH-6невалидный токен → 401, нехватка роли → 403
Только RBAC без ABAC для own-resource endpointsAUTH-3RBAC на endpoint + ABAC в Handler
Ручной парсинг Authorization headerAUTH-4OAuth2PasswordBearer + PyJWKClient

Куда дальше

  • JWT validation — PyJWKClient, PyJWT, алгоритмы, claims.
  • RBAC: маппинг ролей — require_roles, _roles(claims), разрешённые роли.
  • ABAC: владение ресурсом — AccessPolicy, owner_id == sub, admin-bypass.
  • Service-to-service — mTLS, Client Credentials Flow, outbound-клиенты.
  • Audit admin-команд — *_audit_log, декоратор аудита в Handler.
  • PII и секреты — что не попадает в логи, pydantic-settings, Vault.
  • Идемпотентность — Idempotency-Key, повтор команды без дубля.
  • Хранение токенов на клиенте — HttpOnly cookie, RT rotation.