# 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:

| Endpoint | Requires |
|---|---|
| `GET /api/public` | nothing |
| `GET /api/private` | a valid access token for your API |
| `GET /api/messages` | a valid access token with the `messages:read` scope |

## 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](/docs/api/management/apis/create)). 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

```sh
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:

| Variable | Required | Meaning |
|---|---|---|
| `KEENAUTH_ISSUER` | yes | Your environment's issuer |
| `KEENAUTH_AUDIENCE` | yes | Your API's identifier |
| `PUBLIC_BASE_URL` | yes | The URL clients call (see [Behind a proxy](#behind-a-proxy)) |
| `KEENAUTH_WEBHOOK_SECRET` | recommended | The `idp_whsec_` signing secret of the webhook endpoint that delivers `token.revoked` (two, comma separated, while you rotate it). Without it the API never learns of revoked tokens |

## 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.

```python
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.

```python
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

```sh
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:

```sh
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](/docs/sdk/testing).

```python
@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
```
```python
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
```

```sh
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

| Request | Status | `code` in the body |
|---|---|---|
| No `Authorization` header | `401` | `missing_token` |
| Expired token, token for another API, bad signature, revoked token | `401` | `invalid_token` |
| Token without the required scope | `403` | `insufficient_scope` |
| Token in the URL query (`?access_token=`) | `400` | `invalid_request` |

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](/docs/sdk/protect-api#responses).

## 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](/docs/sdk/protect-api#revocation-cache-tokenrevoked).

## Next steps

- [Protecting your API with middleware](/docs/sdk/protect-api): Flask, step-up requirements, DPoP and every option.
- [Verifying tokens](/docs/sdk/token-verification): the verifier under the middleware.
- [Offline test kit](/docs/sdk/testing).