← Back to the section

When you add authorization to FastAPI, the first impulse is to make one dependency and put it everywhere. But authorization is three different checks, and each is done in its own place. If you mix them, gaps appear: someone sees other people's data, or the Gateway starts going to the database on every request.

Let's look at where each thing is checked and why exactly there.

Three different questions — three different places

Each level answers its own question:

LevelQuestionFastAPI tool
Gateway / API edgeWho is this client? (Is the JWT valid?)PyJWKClient + the principal dependency
BFF / Application LayerCan this client access the endpoint at all?Depends(require_roles(...)) on the router
Domain HandlerCan this client work with this specific resource?comparing aggregate.owner_id == principal.sub

Confusion begins when these questions are mixed in one place.

Gateway — authentication

The Gateway answers only the question "who is this client". It verifies the JWT's signature, expiration (exp), issuer (iss), and audience (aud). Nothing more.

In FastAPI this is expressed as one shared dependency:

# 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 caches the public keys for ~5 minutes — no need to fetch them on every request. An invalid signature or an expired token → 401 Unauthorized. The request never reaches the business logic.

What the Gateway does not do:

  • It does not check which roles are needed for a specific endpoint.
  • It knows nothing about domain objects — it does not go to the database for orders or products.

BFF — a coarse role check

The BFF answers the question "can this client access this endpoint at all". This is not "whose resource" but simply "does the user have the required role".

In FastAPI it is convenient to make a dependency factory:

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

And declare the roles right on each 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))

No required role — 403 Forbidden before the Handler is called. The Handler does not run at all.

A common mistake is leaving an endpoint without require_roles. Then any user with a valid token will be able to access it.

Domain Handler — a resource check

The Handler answers the question "can this specific user work with this specific resource". Here the domain model is already needed.

The customer role was checked at the BFF level, but that does not mean a customer can read someone else's order:

# 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 — this is exactly where the ownership question is decided. And only here, because only the Handler knows what an Order is and to whom it belongs.

The same principle for other aggregates:

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))

The admin role bypasses the ownership check — but this is a deliberate exception, not a gap.

Why not do the ownership check at the Gateway

It seems convenient: the token is valid, let's immediately check who the resource belongs to. But then the Gateway has to:

  1. Receive the request GET /orders/abc123.
  2. Go to the database or to order-service to find out who this order belongs to.
  3. Compare the result with the sub from the token.

This turns the Gateway into a service that knows the domain model. When co-owners, delegation, or an organizational hierarchy appear — you will have to change both the domain model and the Gateway in parallel. This does not scale.

Mapping roles from the token

Different providers store roles in different fields:

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 puts roles in realm_access.roles, standard OAuth2 — in scope. The _roles function hides this difference. Principal always gets a plain list of strings — without knowing about the provider.

The full chain by example

POST /orders
  │
  ├─ oauth2_scheme          → extract the Bearer token
  ├─ principal              → PyJWKClient, JWT decode → Principal(sub, roles)
  ├─ require_roles("customer") → check roles ∋ "customer" → 403 if not
  │
  └─ CreateOrderHandler.handle(cmd)
       └─ (no ownership check: the customer creates their own order, sub goes into 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

Common mistakes

RBAC at the Gateway. The Gateway only authenticates; the role check is done by require_roles on the router.

An ownership check in the router. The router does not know what is in the database. The owner_id == sub comparison — only inside the Handler.

jwt.decode(token, options={"verify_signature": False}). Without signature verification any token is considered valid. Use PyJWKClient.

An endpoint without Depends(require_roles(...)). Any user with a token gets access. Every endpoint must declare its roles explicitly.

403 for an invalid JWT. An invalid token is 401 Unauthorized. A missing role is 403 Forbidden. The codes are different.

Only a role check without an ownership check for an own-resource endpoint. The customer role does not mean the customer can read other customers' orders.

In short

  • Three levels — three different questions: who, may they at all, may they work with this resource.
  • Gateway verifies the JWT: signature, exp, iss, aud. Returns a Principal. Knows nothing about the domain model.
  • BFF checks the role via Depends(require_roles(...)) on every endpoint. No role — 403.
  • Handler checks ownership: aggregate.owner_id == principal.sub. Only the Handler knows to whom the resource belongs.
  • An invalid token — 401, a missing role — 403. Do not confuse them.
  • PyJWKClient caches the public keys — no need to fetch them on every request.

Further reading