Protéger une API FastAPI

Ce guide de démarrage protège une API FastAPI avec des jetons d'accès KeenAuth. Vous ajoutez la dépendance RequireAuth du SDK Python, exigez une autorisation sur une route et testez l'API hors ligne avec le kit de test du SDK.

L'API comporte trois points de terminaison :

Avant de commencer

• Un environnement KeenAuth. Son émetteur ressemble à https://your-environment.eu.keenauth.app.
• Une API enregistrée dans l'environnement (Créer une API). Son identifier, par exemple https://api.example.com, est l'audience des jetons acceptés par votre API.
• L'autorisation messages:read sur cette API, accordée à l'application qui l'appelle.
• Python 3.10 ou version ultérieure.

1. Installer le SDK

pip install "keenauth-python[fastapi]" uvicorn

Le SDK n'est pas encore publié. Tant qu'il ne l'est pas, l'exemple l'installe à partir de son code source. L'exemple épingle chaque paquet à une version exacte dans requirements.txt.

2. Configurer

L'API lit sa configuration uniquement depuis des variables d'environnement :

3. Protéger vos routes

create_app crée le protecteur et les routes. Depends(RequireAuth(protector)) exige un jeton valide et renvoie ses revendications vérifiées ; scopes=[...] exige en plus des autorisations. install(app) fait que les erreurs renvoient des réponses application/problem+json, avec un en-tête WWW-Authenticate par défi.

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 construit le vérificateur réel à partir de la configuration. Le vérificateur récupère les clés de signature auprès de votre environnement la première fois qu'il voit un jeton. La dépendance est synchrone, donc FastAPI l'exécute dans son pool de threads.

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. Lancer et appeler l'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

Pour un premier appel, obtenez un jeton avec le flux d'identifiants client d'une application machine à machine qui possède l'autorisation messages:read sur votre 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. Tester sans réseau

keenauth.testing émet des jetons signés avec une clé de test publique, et testing.verifier n'accepte que ces jetons. Les tests transmettent ce vérificateur à create_app et appellent l'application avec le client de test de Starlette ; ils n'ont donc besoin ni de réseau ni de compte KeenAuth. Voir Kit de test hors ligne.

@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

Les tests de l'exemple vérifient qu'un jeton valide est accepté, et qu'un jeton absent, expiré ou destiné à une autre audience, un jeton sans l'autorisation requise, un jeton placé dans la chaîne de requête de l'URL et un jeton révoqué sont rejetés avec les statuts ci-dessous. Ils envoient aussi un jeton lié à DPoP avec une preuve pour PUBLIC_BASE_URL, et vérifient que app_from_env refuse l'émetteur de test. Les tests envoient également des livraisons token.revoked signées : une livraison valide révoque le jeton, une livraison falsifiée, altérée ou périmée reçoit 400 et ne révoque rien, et une livraison répétée n'a qu'un seul effet.

Réponses

Le corps est application/problem+json et l'en-tête WWW-Authenticate transporte le défi standard. Le tableau complet figure dans Protéger votre API avec un middleware.

Derrière un proxy

Si les clients utilisent DPoP, le middleware vérifie que chaque preuve a été émise pour l'URL appelée par le client. Définissez PUBLIC_BASE_URL sur l'URL externe appelée par les clients, par exemple https://api.example.com, y compris derrière un proxy ou un équilibreur de charge. Cette valeur est obligatoire et doit utiliser https (http uniquement pour localhost, 127.0.0.1 ou [::1]) ; l'application refuse de démarrer sans elle. Le middleware ne lit jamais lui-même les en-têtes X-Forwarded-*. Uvicorn, en revanche, applique X-Forwarded-Proto à l'URL de la requête lorsque la connexion provient d'une adresse figurant dans --forwarded-allow-ips (par défaut 127.0.0.1,::1) ; avec PUBLIC_BASE_URL défini, l'URL vérifiée ne dépend pas de ce comportement.

Jetons révoqués

Un jeton d'accès reste valide jusqu'à son expiration ; KeenAuth informe donc votre API lorsque des jetons sont révoqués. L'exemple transmet un cache de révocation en mémoire au protecteur et expose un webhook token.revoked sur POST /webhooks/keenauth (webhook.py) qui l'alimente.

1. Dans le tableau de bord, créez un point de terminaison webhook avec l'URL https://<your host>/webhooks/keenauth qui s'abonne à token.revoked, puis exportez son secret de signature sous KEENAUTH_WEBHOOK_SECRET.
2. Le gestionnaire vérifie la signature avec le SDK sur le corps brut et répond 400 lorsqu'elle est invalide ou lorsque l'horodatage s'écarte de plus de cinq minutes. Il associe le sélecteur de l'événement (access_token, session, user, client, org, org_user, client_user) à un marqueur de révocation et répond 204 ; le middleware rejette ensuite chaque jeton correspondant émis avant not_before.
3. KeenAuth livre au moins une fois, donc le gestionnaire applique chaque webhook-id une seule fois. Si le marqueur ne peut pas être enregistré, il répond 503 et KeenAuth renvoie l'événement. Il ignore les autres types d'événements et les sélecteurs qu'aucun jeton d'accès ne peut correspondre (refresh_family, api).

Le gestionnaire journalise l'identifiant de l'événement, le type de sélecteur et la raison, jamais le corps, une signature ni le secret. Chaque processus de travail a son propre cache ; avec plusieurs workers ou instances, utilisez plutôt un stockage partagé. Voir Cache de révocation.

Étapes suivantes

• Protéger votre API avec un middleware : Flask, exigences de step-up, DPoP et toutes les options.
• Vérifier les jetons : le vérificateur sous le middleware.
• Kit de test hors ligne.