Protect a FastAPI API

This quickstart protects a FastAPI API with KeenAuth access tokens. You add the RequireAuth dependency of the Python SDK, require a scope on one route and test the API offline with the SDK test kit.

The API has three endpoints:

Before you begin

• A KeenAuth environment. Its issuer looks like https://your-environment.eu.keenauth.app.
• An API registered in the environment (Create an API). Its identifier, for example https://api.example.com, is the audience of the tokens your API accepts.
• The messages:read scope on that API, granted to the application that calls it.
• Python 3.10 or later.

1. Install the SDK

pip install "keenauth-python[fastapi]" uvicorn

The SDK is not published yet. Until it is, the example installs it from the SDK source. The example pins every package to an exact version in requirements.txt.

2. Configure

The API reads its configuration from environment variables only:

3. Protect your routes

create_app creates the protector and the routes. Depends(RequireAuth(protector)) requires a valid token and returns its verified claims; scopes=[...] also requires scopes. install(app) makes the errors application/problem+json responses with one WWW-Authenticate header per challenge.

def create_app(
    verifier: AccessTokenVerifier,
    *,
    public_base_url: str,
    revocation: Optional[RevocationChecker] = None,
    webhook_secrets: Sequence[str] = (),
) -> FastAPI:
    """Builds the API. The verifier is a parameter so tests can pass the offline test kit's verifier. With
    webhook_secrets (the idp_whsec_ secrets of your webhook endpoint) the token.revoked webhook is served and feeds
    `revocation`, which must then be a MemoryRevocationCache."""
    protector = Protector(
        verifier,
        public_base_url=public_base_url,
        revocation=revocation,
        # The error never contains the token or claim values; never log the Authorization or DPoP header.
        on_error=lambda err: log.warning("rejected: %s", err),
    )
    app = FastAPI()
    install(app)  # problem+json responses and one WWW-Authenticate header per challenge

    @app.get("/api/public")
    def public() -> dict[str, Any]:
        # No token needed.
        return {"message": "Hello from a public endpoint."}

    @app.get("/api/private")
    def private(claims: AccessClaims = Depends(RequireAuth(protector))) -> dict[str, Any]:
        # Any valid access token for this API.
        return {"message": "Hello from a private endpoint.", "sub": claims.subject}

    @app.get("/api/messages")
    def messages(claims: AccessClaims = Depends(RequireAuth(protector, scopes=["messages:read"]))) -> dict[str, Any]:
        # A valid access token that also carries the messages:read scope.
        return {"messages": ["Hello from a scoped endpoint."]}

    # token.revoked deliveries from KeenAuth.
    if webhook_secrets:
        if not isinstance(revocation, MemoryRevocationCache):
            raise ValueError("the token.revoked webhook needs a MemoryRevocationCache to feed")
        app.include_router(webhook_router(RevocationWebhook(webhook_secrets, revocation)))
    else:
        log.warning("KEENAUTH_WEBHOOK_SECRET is not set: token.revoked events are not received, so revoked tokens stay valid until they expire")

    return app

app_from_env builds the real verifier from the configuration. The verifier fetches the signing keys from your environment the first time it sees a token. The dependency is synchronous, so FastAPI runs it in its thread pool.

def app_from_env() -> FastAPI:
    """The production app: KEENAUTH_ISSUER, KEENAUTH_AUDIENCE and PUBLIC_BASE_URL are required."""
    issuer = os.environ.get("KEENAUTH_ISSUER", "")
    audience = os.environ.get("KEENAUTH_AUDIENCE", "")
    # The URL clients call: DPoP proofs are checked against it, never against the Host header.
    public_base_url = os.environ.get("PUBLIC_BASE_URL", "")
    if not issuer or not audience or not public_base_url:
        raise SystemExit("KEENAUTH_ISSUER, KEENAUTH_AUDIENCE and PUBLIC_BASE_URL are required")
    # The verifier refuses an issuer under the reserved .invalid domain, so a test token can never pass here.
    verifier = Verifier(VerifierConfig(issuer=issuer, audience=audience))
    # Fed by the token.revoked webhook.
    revocation = MemoryRevocationCache()
    # The signing secret (idp_whsec_...) of the webhook endpoint that delivers token.revoked; comma separated while rotating.
    webhook_secrets = [x.strip() for x in os.environ.get("KEENAUTH_WEBHOOK_SECRET", "").split(",") if x.strip()]
    return create_app(verifier, public_base_url=public_base_url, revocation=revocation, webhook_secrets=webhook_secrets)

4. Run and call the API

export KEENAUTH_ISSUER=https://your-environment.eu.keenauth.app
export KEENAUTH_AUDIENCE=https://api.example.com
export PUBLIC_BASE_URL=https://api.example.com
uvicorn main:app_from_env --factory --port 8080

For a first call, get a token with the client credentials grant of a machine-to-machine application that has the messages:read scope on your API:

ACCESS_TOKEN=$(curl -s -u "$CLIENT_ID:$CLIENT_SECRET" \
  -d grant_type=client_credentials \
  -d resource=https://api.example.com \
  -d scope=messages:read \
  https://your-environment.eu.keenauth.app/oauth2/token | jq -r .access_token)

curl -i http://localhost:8080/api/messages -H "Authorization: Bearer $ACCESS_TOKEN"

5. Test without a network

keenauth.testing mints tokens signed with a public test key, and testing.verifier accepts only those tokens. The tests pass that verifier to create_app and call the app with Starlette's test client, so they need no network and no KeenAuth account. See Offline test kit.

@pytest.fixture()
def api() -> Api:
    cache = MemoryRevocationCache()
    app = create_app(
        testing.verifier(audience=AUDIENCE),
        public_base_url=BASE_URL,
        revocation=cache,
        webhook_secrets=[testing.webhook_secret()],
    )
    return TestClient(app, base_url=BASE_URL), cache

def test_scope(api: Api) -> None:
    client, _ = api
    no_scope = testing.mint_access_token(AUDIENCE)
    assert_problem(client.get("/api/messages", headers=bearer(no_scope)), 403, "insufficient_scope", 'scope="messages:read"')
    with_scope = testing.mint_access_token(AUDIENCE, claims={"scope": "profile messages:read"})
    assert client.get("/api/messages", headers=bearer(with_scope)).status_code == 200

pip install --no-deps -r requirements-dev.txt
python -m pytest -q

The example's tests check that a valid token is accepted and that a missing, expired or wrong-audience token, a missing scope, a token in the URL query and a revoked token are rejected with the statuses below. They also send a DPoP-bound token with a proof for PUBLIC_BASE_URL, and check that app_from_env refuses the test issuer. The tests also send signed token.revoked deliveries: a valid one revokes the token, a forged, tampered or stale one gets 400 and revokes nothing, and a repeated one has one effect.

Responses

The body is application/problem+json and the WWW-Authenticate header carries the standard challenge. The full table is in Protecting your API with middleware.

Behind a proxy

If clients use DPoP, the middleware checks that each proof was made for the URL the client called. Set PUBLIC_BASE_URL to the external URL clients call, for example https://api.example.com, also behind a proxy or load balancer. It is required and must use https (http only for localhost, 127.0.0.1 or [::1]); the app refuses to start without it. The middleware never reads X-Forwarded-* headers itself. Uvicorn, however, applies X-Forwarded-Proto to the request URL when the connection comes from an address in --forwarded-allow-ips (by default 127.0.0.1,::1); with PUBLIC_BASE_URL set, the checked URL does not depend on that.

Revoked tokens

An access token stays valid until it expires, so KeenAuth tells your API when tokens are revoked. The example passes an in-memory revocation cache to the protector and serves a token.revoked webhook at POST /webhooks/keenauth (webhook.py) that feeds it.

1. In the dashboard, create a webhook endpoint with the URL https://<your host>/webhooks/keenauth that subscribes to token.revoked, and export its signing secret as KEENAUTH_WEBHOOK_SECRET.
2. The handler verifies the signature with the SDK on the raw body and answers 400 when it is invalid or the timestamp is more than five minutes off. It maps the event's selector (access_token, session, user, client, org, org_user, client_user) to a revocation marker and answers 204; the middleware then rejects every matching token issued before not_before.
3. KeenAuth delivers at least once, so the handler applies each webhook-id once. If the marker cannot be stored it answers 503 and KeenAuth delivers the event again. It ignores other event types and selectors that no access token can match (refresh_family, api).

The handler logs the event id, the selector type and the reason, never the body, a signature or the secret. Each worker process has its own cache; with several workers or instances, use a shared store instead. See Revocation cache.

Next steps

• Protecting your API with middleware: Flask, step-up requirements, DPoP and every option.
• Verifying tokens: the verifier under the middleware.
• Offline test kit.