← Back to the section

When a client sends a request with a token, the service must make sure the token is genuine, the signature is valid, and it has not expired. It seems simple — "decode the Base64 and read it". But this is exactly where people most often make mistakes that open the door to attackers.

Let's look at how to do it correctly in FastAPI.

Why you cannot just "decode" the token

A JWT is three parts separated by dots: the header, the payload, and the signature. The signature is created by the authentication server (IdP) using a private key.

A common mistake is to read only the middle part and not verify the signature:

# Never do this
import base64, json

def bad_decode(token: str) -> dict:
    payload_b64 = token.split(".")[1]
    padding = "=" * (4 - len(payload_b64) % 4)
    return json.loads(base64.urlsafe_b64decode(payload_b64 + padding))

What goes wrong:

  • The signature is not verified — an attacker inserts any data, including admin: true.
  • The expiration is not checked — an expired token is accepted as fresh.
  • The issuer and audience are not checked — a token from another service or a foreign IdP passes too.
  • The none algorithm may pass — a classic vulnerability: an unsigned token declares "alg": "none" and is accepted without verification.

The correct way is to use PyJWT together with PyJWKClient. The library performs all the necessary checks itself.

How correct validation works

The authentication server publishes public keys at a known address — this is called JWKS (JSON Web Key Set). Your service downloads these keys and uses them to verify the signature of every token.

All of this code lives in one place: adapters/in/http/security.py. The rest of the application receives an already-verified Principal object through FastAPI's dependency mechanism — Depends.

# adapters/in/http/security.py
import jwt
from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from jwt import PyJWKClient, PyJWKClientConnectionError
from pydantic import BaseModel

from app.config import settings

_jwks = PyJWKClient(settings.jwks_uri, cache_keys=True, lifespan=300)

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token", auto_error=True)


class Principal(BaseModel):
    sub: str
    roles: list[str]


def _extract_roles(claims: dict) -> list[str]:
    realm = claims.get("realm_access", {})
    return realm.get("roles", [])


async def principal(token: str = Depends(oauth2_scheme)) -> Principal:
    try:
        signing_key = _jwks.get_signing_key_from_jwt(token).key
    except PyJWKClientConnectionError as exc:
        raise HTTPException(status_code=503, detail="identity provider unavailable") from exc
    try:
        claims = jwt.decode(
            token,
            signing_key,
            algorithms=["RS256"],
            audience=settings.audience,
            issuer=settings.issuer,
        )
    except jwt.PyJWTError as exc:
        raise HTTPException(status_code=401, detail="invalid token") from exc
    return Principal(sub=claims["sub"], roles=_extract_roles(claims))


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

What happens here, step by step:

  1. _jwks.get_signing_key_from_jwt(token) — finds the right key by the kid field in the token header.
  2. jwt.decode(...) — verifies the signature, expiration (exp), issuer (iss), and audience (aud). The RS256 algorithm is specified explicitly — this closes the alg: none attack.
  3. If something is wrong — an HTTPException(status_code=401) is raised automatically.
  4. If everything is fine — a Principal with sub and a list of roles is returned.

Settings are read from environment variables via pydantic-settings:

# config.py
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    jwks_uri: str
    audience: str
    issuer: str

    model_config = {"env_file": ".env"}


settings = Settings()

An example .env for a local run (not committed to git):

JWKS_URI=https://idp.example.com/realms/main/protocol/openid-connect/certs
AUDIENCE=order-service
ISSUER=https://idp.example.com/realms/main

How to use it in routers

Each endpoint declares which roles are required for access. This is done via Depends(require_roles(...)):

# adapters/in/http/orders_router.py
from fastapi import APIRouter, Depends

from adapters.in.http.security import Principal, require_roles
from application.use_cases.create_order import CreateOrderUseCase

router = APIRouter(prefix="/orders")


@router.post("/", status_code=201)
async def create_order(
    body: CreateOrderRequest,
    principal: Principal = Depends(require_roles("customer", "admin")),
    use_case: CreateOrderUseCase = Depends(),
) -> OrderResponse:
    return await use_case.execute(body, principal)


@router.get("/{order_id}")
async def get_order(
    order_id: str,
    principal: Principal = Depends(require_roles("customer", "seller", "admin")),
    use_case: GetOrderUseCase = Depends(),
) -> OrderResponse:
    return await use_case.execute(order_id, principal)

The Principal object is passed into the UseCase — checks like "this order belongs to the current user" (order.customer_id == principal.sub) are done there, not in the router.

JWKS cache and key rotation

Fetching the keys from the IdP on every request is a bad idea: it is slow and creates unnecessary load. PyJWKClient with the parameter lifespan=300 caches the keys for 5 minutes:

_jwks = PyJWKClient(
    settings.jwks_uri,
    cache_keys=True,
    lifespan=300,
)

What happens when the IdP rotates its keys: if the kid from a new token is not found in the cache, PyJWKClient automatically refreshes the cache and immediately finds the new key. Nothing extra needs to be written.

Common mistakes:

  • Storing the public key directly in the configuration — when the keys rotate, the service stops validating tokens.
  • Manually calling _jwks.fetch_data() on a timer — PyJWKClient does this itself.
  • Parsing the JWK manually via cryptography without PyJWKClient.

If the IdP is unavailable at service startup — the service fails its readiness check. This is correct behavior: you must not accept requests without the ability to validate a token.

The difference between 401 and 403

These are different situations with different consequences for the client:

CodeWhenWhat the client does
401 UnauthorizedThe token is invalid: bad signature, expired, missing headerStart a token refresh or redirect to login
403 ForbiddenThe token is valid, but the role does not fit or access to the resource is closedShow "access denied"; a token refresh will not help

If you return 403 for an expired token, the client will not try to refresh it and will "hang". If you return 401 for insufficient rights, the client will endlessly try to log in again.

In the code above this is separated by dependencies: principal throws 401, require_roles throws 403.

An example of handling this in client code, when one service calls another:

# adapters/out/http/order_client.py
async def get_order(order_id: str, token: str) -> OrderDTO:
    resp = await client.get(f"/orders/{order_id}", headers={"Authorization": f"Bearer {token}"})
    if resp.status_code == 401:
        raise TokenExpiredError()
    if resp.status_code == 403:
        raise ForbiddenError()
    resp.raise_for_status()
    return OrderDTO.model_validate(resp.json())

In short

  • A JWT contains a signature — you must verify it; simply "decoding the Base64" is not enough.
  • Use PyJWT + PyJWKClient: they verify the signature, expiration, issuer, and audience.
  • Explicitly specify algorithms=["RS256"] — this closes the attack via an unsigned token (alg: none).
  • All token validation logic lives in one place (security.py) and is handed to the endpoint via Depends(principal).
  • PyJWKClient caches the keys and refreshes the cache itself when the IdP rotates its keys.
  • 401 — the token is invalid (the client refreshes the token); 403 — the token is valid but access is closed (a refresh will not help).
  • Read the configuration (jwks_uri, audience, issuer) from environment variables; do not hardcode it.

Further reading