JWT and OAuth 2.0¶

When an identity provider issues the tokens (Keycloak, Auth0, Entra ID, Authentik...), you do not write the validator: JWTValidator checks them for you.

uv add "fastapi-multiauth[jwt]"

Validate provider tokens¶

from fastapi import FastAPI, Security

from fastapi_multiauth import HTTPBearerAuth, JWTValidator

ISSUER = "https://idp.example.com/realms/main"

bearer = HTTPBearerAuth(
    validator=JWTValidator(
        jwks_url=f"{ISSUER}/protocol/openid-connect/certs",
        audience="my-api",
        issuer=ISSUER,
    ),
    bearer_format="JWT",
)

app = FastAPI()


@app.get("/me")
async def me(claims: dict = Security(bearer)):
    return {"sub": claims["sub"]}

The validator fetches the provider's public keys (the JWKS) and checks the signature, expiry, audience and issuer. Keys are cached for an hour and refreshed when the provider rotates them; if the provider is down, the cached keys keep working.

By default the route receives the token's claims. Pass claims_to_identity= to turn them into your own user object. For tokens you sign yourself, use JWTValidator(secret=...) instead of jwks_url.

Sender-constrained tokens (DPoP)¶

Whoever holds a bearer token can use it. A DPoP-bound token only works with the private key the client proved when it got the token. Keycloak, Auth0, Okta and others issue them; HTTPDPoPAuth checks them:

import time

from fastapi import FastAPI, Security

from fastapi_multiauth import HTTPDPoPAuth, JWTValidator

ISSUER = "https://idp.example.com/realms/main"
SEEN_PROOFS: dict[str, float] = {}  # use Redis SET NX across processes


def remember_proof(jti: str, expires_at: float) -> bool:
    now = time.time()
    for old in [j for j, until in SEEN_PROOFS.items() if until < now]:
        del SEEN_PROOFS[old]
    if jti in SEEN_PROOFS:
        return False
    SEEN_PROOFS[jti] = expires_at
    return True


dpop = HTTPDPoPAuth(
    validator=JWTValidator(
        jwks_url=f"{ISSUER}/protocol/openid-connect/certs",
        audience="my-api",
        issuer=ISSUER,
    ),
    replay_check=remember_proof,
)

app = FastAPI()


@app.get("/me")
async def me(claims: dict = Security(dpop)):
    return {"sub": claims["sub"]}

The client sends Authorization: DPoP <token> and a DPoP header: a short JWT signed with its key, naming the method and URL of this request and the hash of the token. The source checks that proof, passes the token to your validator, then checks the token's cnf.jkt claim names the key that signed the proof. A stolen token is useless without the key, and a proof cannot be moved to another URL or token.

A proof stays valid for proof_max_age (60 seconds). replay_check records each proof's jti so it cannot be sent twice. Make it atomic across your workers, e.g. Redis SET NX with the expiry it receives. Pass HTTPDPoPAuth.NO_REPLAY_CHECK to skip it knowingly.

Behind a proxy that changes the scheme or host, pass public_origin="https://api.example.com": the proof names the URL the client called. With claims_to_identity, pass jkt= to read the thumbprint from your identity.

Opaque tokens¶

Some providers issue opaque tokens instead of JWTs, or let you choose them because they can be revoked. IntrospectionValidator asks the provider's introspection endpoint whether each token is still active:

import os

from fastapi import FastAPI, Security

from fastapi_multiauth import HTTPBearerAuth, IntrospectionValidator

ISSUER = "https://idp.example.com/realms/main"

bearer = HTTPBearerAuth(
    validator=IntrospectionValidator(
        url=f"{ISSUER}/protocol/openid-connect/token/introspect",
        client_id="my-api",
        client_secret=os.environ["INTROSPECTION_CLIENT_SECRET"],
        audience="my-api",
    )
)

app = FastAPI()


@app.get("/me")
async def me(claims: dict = Security(bearer)):
    return {"sub": claims["sub"]}

The route receives the provider's answer, which carries the same fields as JWT claims (sub, scope, exp...), so scopes, fresh() and stepped_up() work the same way. No extra is needed. Answers describing a refresh or ID token (Keycloak's typ, Ory Hydra's token_use) are refused.

Active answers are cached for cache_ttl seconds (60 by default), never past the token's exp. A revoked token therefore keeps working for up to a minute; set cache_ttl=0 to ask on every request. Inactive answers are not cached, so a client sending junk tokens costs one request to the provider each: rate-limit the routes. If the provider is down, requests get a 503.

A real Authorize button¶

With HTTPBearerAuth, /docs only shows a token box. The OAuth 2.0 sources also describe the login flow and its scopes, so /docs can log in for real:

from fastapi import FastAPI, Security

from fastapi_multiauth import JWTValidator, OAuth2AuthorizationCodeBearerAuth

ISSUER = "https://idp.example.com/realms/main"

auth = OAuth2AuthorizationCodeBearerAuth(
    validator=JWTValidator(
        jwks_url=f"{ISSUER}/protocol/openid-connect/certs",
        audience="my-api",
        issuer=ISSUER,
    ),
    authorization_url=f"{ISSUER}/protocol/openid-connect/auth",
    token_url=f"{ISSUER}/protocol/openid-connect/token",
    scopes={"invoices:read": "Read invoices"},
)

app = FastAPI()


@app.get("/invoices")
async def invoices(claims: dict = Security(auth, scopes=["invoices:read"])):
    return {"sub": claims["sub"], "invoices": []}

Route scopes are checked against the token's scope claim; a missing scope gets a 403. Also available: OAuth2PasswordBearerAuth and OpenIdConnectAuth.