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.

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.

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.

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"]),
)