Usage¶

The validator contract¶

Every source wraps a callable you provide. It can be sync or async, receives the extracted credential as its first argument, and returns the authenticated identity (any object: a dict, a User model, …). To reject a credential, raise UnauthorizedError (or any HTTPException):

from fastapi_multiauth import UnauthorizedError


async def validate_token(token: str) -> User:
    user = await db.get_user_by_token(token)
    if user is None:
        raise UnauthorizedError()
    return user

Extra keyword arguments passed at instantiation are forwarded to the validator on every call:

admin_bearer = HTTPBearerAuth(validate_token, role=Role.ADMIN)

The same applies per route: .require(**kwargs) returns a copy of the source with extra (or overriding) kwargs, without mutating the original. Configure the source once, then tighten individual endpoints where needed:

bearer = HTTPBearerAuth(validate_token)


@app.get("/admin")
async def admin(user=Security(bearer.require(role=Role.ADMIN))):
    return user

Use the right status code: UnauthorizedError (401) when the credential is absent or invalid, ForbiddenError (403) when the identity is valid but lacks permission. 401 responses from bearer sources automatically carry the WWW-Authenticate: Bearer challenge required by RFC 7235; MultiAuth advertises the union of its sources' challenges.

Three keyword names are reserved, because the library sets them itself: scopes, session_id and request. Passing one at instantiation or through require() raises a ValueError on the spot.

Reaching the request¶

A validator that declares a request parameter is handed the request being authenticated. request.state is then a request-scoped place to leave whatever the validator learned and the route needs later:

from fastapi import Request


async def validate_token(token: str, request: Request | None = None) -> User:
    user = await db.get_user_by_token(token)
    if user is None:
        raise UnauthorizedError()
    if request is not None:
        request.state.token_scopes = user.token_scopes
    return user


bearer = HTTPBearerAuth(validate_token)


@app.get("/me")
async def me(request: Request, user=Security(bearer)):
    return {"user": user, "scopes": request.state.token_scopes}

It is opt-in by signature, exactly like scopes: a validator that does not declare request is called as it was before.

The parameter is Request | None and the None is real, not a formality. authenticate() and authenticate_scoped() are public and reachable from a CLI, a background task or a hand-rolled dependency, none of which has a request to pass. Handle the absent case rather than assuming a route is always upstream.

Custom sources¶

An AuthSource subclass already has the request in extract(), which is where it pulls its credential out of the request to begin with. Reaching it at validation time means overriding authenticate_scoped(), the seam that carries the route's context:

from fastapi_multiauth import AuthSource


class HeaderAuth(AuthSource):
    async def extract(self, request: Request) -> str | None:
        return request.headers.get("X-Token") or None

    async def authenticate(self, credential: str) -> User:
        return await lookup(credential)

    async def authenticate_scoped(
        self, credential: str, scopes: list[str], *, request: Request | None = None
    ) -> User:
        user = await self.authenticate(credential)
        if request is not None:
            request.state.token_scopes = user.token_scopes
        return user

authenticate() is the credential primitive; authenticate_scoped() is where the request and the route's scopes arrive together. An override written before request existed keeps its two-argument shape and goes on being called that way, so a custom source predating this feature needs no edit.

Optional authentication¶

optional() returns a copy of a source (or a MultiAuth) that yields None instead of a 401 when the request carries no credential, for endpoints that serve anonymous callers but render more for signed-in ones:

bearer = HTTPBearerAuth(validate_token)
MaybeUser = Annotated[User | None, Security(bearer.optional())]


@app.get("/articles/{slug}")
async def article(slug: str, user: MaybeUser):
    return render(slug, drafts=user is not None)

Only an absent credential yields None. One that is present and rejected by the validator still raises 401, so optional() can never turn a bad token into an anonymous request. Whether a credential counts as present is extract()'s decision, which means a bearer token that does not match a configured prefix reads as absent rather than invalid.

Copies are independent: the source it came from stays strict, so one configured source serves both kinds of route. The route still advertises its security scheme in OpenAPI, matching fastapi.security's own auto_error=False behaviour.

Combining with require()¶

require() and optional() compose in either order, and mean "anonymous is welcome, but a credential that is presented must still qualify":

service_only = bearer.require(kind="service").optional()


@app.get("/status")
async def status(caller: Annotated[Service | None, Security(service_only)]):
    return {"detail": caller is not None}

Sources¶

The library ships one request-time source per standard fastapi.security scheme. Each extracts a credential from the request and hands it to your validator (the contract above); returning None when the credential is absent lets MultiAuth fall through to the next source.

Bearer tokens¶

from fastapi import FastAPI, Security
from fastapi_multiauth import HTTPBearerAuth

bearer = HTTPBearerAuth(validate_token)

app = FastAPI()


@app.get("/me")
async def me(user=Security(bearer)):
    return user

Token prefixes¶

Use prefixes to run several token types side by side (Stripe-style user_ / org_). Only tokens starting with the prefix are matched, and the prefix is kept in the value passed to the validator:

user_bearer = HTTPBearerAuth(validate_user_token, prefix="user_")
org_bearer = HTTPBearerAuth(validate_org_token, prefix="org_")

Generating and storing tokens¶

generate_token() returns 256 bits of CSPRNG entropy (43 url-safe chars by default), with the source's prefix prepended; a recognizable prefix also lets secret-scanning tools flag leaked tokens. Store the hash, never the token:

from fastapi_multiauth import hash_token, verify_token_hash

token = user_bearer.generate_token()  # "user_Xk3...": show it to the user once
await db.save_api_token(user_id, token_hash=hash_token(token))


async def validate_user_token(token: str) -> User:
    row = await db.get_api_token(token_hash=hash_token(token))
    if row is None:
        raise UnauthorizedError()
    return row.user

Look tokens up by their SHA-256 hash (no salt needed: the token itself is high-entropy, unlike a password), or compare explicitly with verify_token_hash(token, stored_hash), a constant-time comparison.

nbytes changes the length of the token (generate_token(64)). To control its format as well, pass a token_generator: it receives that nbytes and returns the body, and the source still prepends its prefix. The default is secrets.token_urlsafe.

import secrets

user_bearer = HTTPBearerAuth(
    validate_user_token,
    prefix="user_",
    token_generator=secrets.token_hex,  # "user_9f3b..."
)

A generator is free to ignore nbytes (lambda _n: ...), in which case generate_token(64) has no effect. The callable owns the entropy: use secrets, never random. Leading or trailing whitespace on the finished token is rejected, since Authorization header parsing strips it and the token would no longer match the hash you stored.

Cookie sessions¶

APIKeyCookieAuth reads a cookie and hands its value to your validator. With a secret_key, the cookie is signed (HMAC-SHA256 via itsdangerous, salted with the cookie name) with an embedded timestamp checked against ttl: a stateless, tamper-proof session without any database entry:

from fastapi_multiauth import APIKeyCookieAuth

session = APIKeyCookieAuth(
    "session",
    validate_session,
    secret_key=settings.SECRET_KEY,  # ≥ 32 bytes, enforced at startup
    ttl=86400,
    samesite="lax",  # default; also: domain=..., path=...
)


@app.post("/login")
async def login(response: Response, credentials: LoginForm):
    user = await check_password(credentials)
    session.set_cookie(response, str(user.id))
    return {"ok": True}


@app.post("/logout")
async def logout(response: Response):
    session.delete_cookie(response)
    return {"ok": True}

Key rotation¶

Pass a sequence of keys to rotate a secret without logging everyone out. The first key signs new cookies; every key verifies:

session = APIKeyCookieAuth(
    "session",
    validate_session,
    secret_key=[settings.NEW_KEY, settings.OLD_KEY],
)

Deploy with both keys, wait until ttl has passed, then drop the old key.

Cookies are bound to their name: two APIKeyCookieAuth instances sharing a secret_key (e.g. session and admin_session) can never accept each other's cookies.

Per-session identity¶

session_id=True mints a random id on every set_cookie, carries it inside the signed payload, and injects it into your validator as session_id. The library stays stateless: it mints and forwards the id, you decide what to store and what revoked means.

session = APIKeyCookieAuth(
    "session",
    validate_session,
    secret_key=settings.SECRET_KEY,  # required with session_id
    session_id=True,
)


async def validate_session(user_id: str, *, session_id: str) -> User:
    if await revoked(session_id):  # your store
        raise UnauthorizedError()
    return await db.get_user(user_id)


@app.post("/login")
async def login(response: Response, credentials: LoginForm):
    user = await check_password(credentials)
    sid = session.set_cookie(response, str(user.id))
    await db.create_session(sid, user.id)
    return {"ok": True}


@app.post("/logout")
async def logout(request: Request, response: Response):
    sid = session.delete_cookie(
        response, request
    )  # None if the cookie was absent/invalid
    if sid:
        await db.revoke_session(sid)  # this device only
    return {"ok": True}

session_id_of(request) reads the id off the request's cookie without running the validator, so a route can name the session it is serving: mark "this device" in a session list, or spare it from a bulk revoke. It returns None when session_id is off, or the cookie is absent, expired or forged.

@app.post("/sessions/revoke-others")
async def revoke_others(request: Request, user=Security(session)):
    await db.revoke_sessions(user.id, except_id=session.session_id_of(request))
    return {"ok": True}

Points to watch:

• The validator must declare a session_id parameter and secret_key must be set; both are checked at construction.
• Turning session_id on or off changes the signing salt, so cookies minted the other way stop verifying: existing sessions are logged out on deploy.
• Rotate after a privilege change (password change, elevation): mint the new id with set_cookie before revoking the old one, so a failure in between leaves the user signed in rather than the old id valid.
• Your revoked() check is one lookup per authenticated request; verifying the cookie itself does zero I/O.
• The validator receives the id but not the Request, so anything request-derived (IP, user-agent) has to be captured where you call set_cookie.
• name and ttl are readable on the source: derive your session row's expiry from session.ttl instead of duplicating the constant.

API keys¶

APIKeyHeaderAuth reads the named header and hands its value to your validator. There is no WWW-Authenticate challenge (the apiKey scheme defines none), and an absent or empty header yields None so MultiAuth can fall through:

from fastapi_multiauth import APIKeyHeaderAuth

api_key = APIKeyHeaderAuth("X-API-Key", validate_api_key)

APIKeyQueryAuth is the same source for the query string (?api_key=...), for legacy clients that cannot set a header. Prefer the header where you can: query strings leak into access logs, browser history, and Referer headers.

from fastapi_multiauth import APIKeyQueryAuth

api_key = APIKeyQueryAuth("api_key", validate_api_key)

Basic auth¶

HTTPBasicAuth decodes the RFC 7617 Authorization: Basic header (UTF-8 charset) and calls validator(username, password). Compare secrets in constant time, never with ==:

import secrets
from fastapi_multiauth import HTTPBasicAuth, UnauthorizedError


async def validate_basic(username: str, password: str) -> dict:
    user = await db.get_user(username)
    if user is None or not secrets.compare_digest(
        hash_password(password), user.password_hash
    ):
        raise UnauthorizedError()
    return user


basic = HTTPBasicAuth(validate_basic, realm="api")

The realm shows up in the WWW-Authenticate: Basic realm="api" challenge on 401 responses, which is what makes browsers prompt for credentials.

OAuth 2.0 and OpenID Connect¶

Three sources validate a bearer token an OAuth 2.0 or OIDC provider already issued. They declare no flow of their own: the URLs and scope catalogue are OpenAPI metadata, which is what makes /docs offer a real Authorize dialog with per-scope checkboxes instead of a bare token box.

from fastapi_multiauth import OAuth2AuthorizationCodeBearerAuth

auth = OAuth2AuthorizationCodeBearerAuth(
    validate_access_token,
    authorization_url="https://idp.example/authorize",
    token_url="https://idp.example/oauth/token",
    scopes={"admin": "Administer everything", "billing": "Read invoices"},
)


@app.get("/invoices")
async def invoices(user=Security(auth, scopes=["billing"])):
    return user

• OAuth2AuthorizationCodeBearerAuth: the flow to prefer for anything user-facing; pairs with the login helpers that mint the tokens.
• OAuth2PasswordBearerAuth: token_url only, for an API that issues its own tokens from a password endpoint. OAuth 2.1 drops this grant.
• OpenIdConnectAuth: one openid_connect_url, and /docs reads the provider's discovery document for the rest. Unlike fastapi.security.OpenIdConnect, which hands the validator the raw Authorization header, this source extracts the bearer token.

Publishing a catalogue is a claim, so a route that declares a scope outside it is refused outright: a typo reaches neither the validator nor the client as an ordinary 403. Omit scopes= to make no claim and accept whatever routes declare, as every other source does. Checking a credential against its own scopes is still the validator's job, and it must accept a scopes parameter for routes that declare any. Pair these sources with JWTValidator when the provider issues JWTs.

Combining sources with MultiAuth¶

MultiAuth tries each source in order and authenticates with the first one that finds a credential in the request. All underlying schemes are documented in OpenAPI:

from fastapi_multiauth import MultiAuth

auth = MultiAuth(bearer, session)


@app.get("/me")
async def me(user=Security(auth)):
    return user

Security scopes¶

Scopes declared on the route are forwarded to validators that declare a scopes parameter. If the validator does not support scopes and the route declares some, the request fails closed instead of silently skipping the check:

async def validate_token(token: str, scopes: list[str]) -> User:
    user = await db.get_user_by_token(token)
    if user is None or not set(scopes) <= set(user.scopes):
        raise UnauthorizedError()
    return user


bearer = HTTPBearerAuth(validate_token)


@app.post("/challenges")
async def create(user=Security(bearer, scopes=["challenges:write"])): ...

JWT validation¶

JWTValidator plugs into HTTPBearerAuth to validate JWTs issued by an identity provider. It requires the jwt extra:

uv add "fastapi-multiauth[jwt]"

For Keycloak / Auth0 / Entra ID / Authentik, point it at the provider's JWKS (the URL is also available from oauth.oauth_resolve_provider_urls(...).jwks_uri):

from fastapi_multiauth import HTTPBearerAuth, JWTValidator

bearer = HTTPBearerAuth(
    JWTValidator(
        jwks_url="https://idp.example.com/realms/main/protocol/openid-connect/certs",
        audience="my-api",
        issuer="https://idp.example.com/realms/main",
    )
)


@app.get("/me")
async def me(claims=Security(bearer)):
    return claims

Signing keys are fetched over HTTPS with a timeout, cached for an hour, and refreshed once when a token carries an unknown kid: provider key rotation needs no restart. A provider that breaks does not take your API down with it: whether the endpoint is unreachable, errors, or answers with a document carrying no usable key, the cached keys keep serving and the refresh is retried under a cooldown. Only a cold start with nothing cached fails, and it fails as a 503 rather than a 401, so clients retry instead of discarding valid tokens. Symmetric mode is JWTValidator(secret=...) (HS256, secret ≥ 32 bytes); HS* algorithms are rejected in JWKS mode to close the classic key-confusion attack.

Every token is checked for signature, exp/nbf/iat (with configurable leeway), aud/iss when configured, and exp is required by default; pass required_claims=() if your provider really issues non-expiring tokens.

Scopes integrate with Security(..., scopes=[...]): the claim is configurable (scope space-separated string by default; scp or roles lists via scopes_claim), and a valid token missing a required scope gets a 403 (RFC 6750 insufficient_scope), not a 401.

By default the source returns the validated claims dict. To return your own identity object instead (exactly like every other source's validator), pass claims_to_identity; the endpoint then receives whatever it returns:

validator = JWTValidator(
    jwks_url=...,
    audience="my-api",
    scopes_claim="scp",
    claims_to_identity=lambda claims: User(id=claims["sub"], email=claims["email"]),
)