Custom sources¶

When no built-in source reads your credential, subclass AuthSource and write two methods:

from fastapi import FastAPI, Security
from fastapi.requests import HTTPConnection
from fastapi.security import APIKeyHeader

from fastapi_multiauth import AuthSource, UnauthorizedError

CLIENTS = {"ab:cd:ef": "billing-service"}
TRUSTED_PROXIES = {"10.0.0.1"}


class ClientCertAuth(AuthSource):
    def __init__(self) -> None:
        super().__init__(
            scheme=APIKeyHeader(name="X-Client-Cert", scheme_name="ClientCert")
        )

    async def extract(self, request: HTTPConnection) -> str | None:
        if request.client is None or request.client.host not in TRUSTED_PROXIES:
            return None
        return request.headers.get("X-Client-Cert") or None

    async def authenticate(self, credential: str) -> str:
        client = CLIENTS.get(credential)
        if client is None:
            raise UnauthorizedError()
        return client


client_cert = ClientCertAuth()

app = FastAPI()


@app.get("/invoices")
async def invoices(client: str = Security(client_cert)):
    return {"client": client}

extract(request): Find the credential in the request. Return None when it is absent, so MultiAuth and optional() work.
authenticate(credential): Return the user or raise, like a validator.

X-Client-Cert comes from the proxy that terminates TLS, so extract() ignores it on connections from anywhere else. The proxy must overwrite any value the client sent. If uvicorn trusts the proxy through --forwarded-allow-ips, request.client is the original client's address instead, and this check refuses every request.

The scheme passed to super().__init__() is only used for OpenAPI. Your source works with MultiAuth, optional(), WebSockets, logging and auditing like the built-in ones.

To check scopes, override authenticate_scoped(). To use fresh() or stepped_up() without a reader, override authenticated_at() or authenticated_acr(). See the API reference.