FastAPI-API schützen

Diese Schnellstartanleitung schützt eine FastAPI-API mit KeenAuth-Zugriffstokens. Sie fügen die RequireAuth-Abhängigkeit des Python-SDK hinzu, verlangen einen Scope auf einer Route und testen die API offline mit dem SDK-Testkit.

Die API hat drei Endpunkte:

Bevor Sie beginnen

• Eine KeenAuth-Umgebung. Ihr Issuer sieht so aus: https://your-environment.eu.keenauth.app.
• Eine in der Umgebung registrierte API (Eine API erstellen). Ihre identifier, zum Beispiel https://api.example.com, ist die Audience der Tokens, die Ihre API akzeptiert.
• Der Scope messages:read für diese API, gewährt an die Anwendung, die sie aufruft.
• Python 3.10 oder neuer.

1. SDK installieren

pip install "keenauth-python[fastapi]" uvicorn

Das SDK ist noch nicht veröffentlicht. Bis dahin installiert das Beispiel es aus dem SDK-Quellcode. Das Beispiel fixiert jedes Paket in requirements.txt auf eine exakte Version.

2. Konfigurieren

Die API liest ihre Konfiguration ausschließlich aus Umgebungsvariablen:

3. Ihre Routen schützen

create_app erstellt den Protector und die Routen. Depends(RequireAuth(protector)) verlangt ein gültiges Token und gibt dessen geprüfte Claims zurück; scopes=[...] verlangt außerdem Scopes. install(app) macht die Fehler zu application/problem+json-Antworten mit je einem WWW-Authenticate-Header pro 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 erstellt den echten Verifier aus der Konfiguration. Der Verifier lädt die Signierschlüssel beim ersten Token aus Ihrer Umgebung. Die Abhängigkeit ist synchron, daher führt FastAPI sie in seinem Thread-Pool aus.

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. API starten und aufrufen

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

Für einen ersten Aufruf holen Sie ein Token mit dem Client-Credentials-Grant einer Maschine-zu-Maschine-Anwendung, die den Scope messages:read für Ihre API hat:

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. Ohne Netzwerk testen

keenauth.testing erzeugt Tokens, die mit einem öffentlichen Testschlüssel signiert sind, und testing.verifier akzeptiert nur diese Tokens. Die Tests übergeben diesen Verifier an create_app und rufen die App mit dem Test-Client von Starlette auf; sie benötigen daher weder Netzwerk noch KeenAuth-Konto. Siehe Offline-Testkit.

@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

Die Tests des Beispiels prüfen, dass ein gültiges Token akzeptiert wird und dass ein fehlendes, abgelaufenes oder für eine andere Audience ausgestelltes Token, ein fehlender Scope, ein Token in der URL-Query und ein widerrufenes Token mit den unten genannten Statuscodes abgelehnt werden. Außerdem senden sie ein DPoP-gebundenes Token mit einem Nachweis für PUBLIC_BASE_URL und prüfen, dass app_from_env den Test-Issuer ablehnt. Die Tests senden zudem signierte token.revoked-Zustellungen: Eine gültige widerruft das Token, eine gefälschte, manipulierte oder veraltete erhält 400 und widerruft nichts, und eine wiederholte hat genau eine Wirkung.

Antworten

Der Body ist application/problem+json, und der WWW-Authenticate-Header trägt die Standard-Challenge. Die vollständige Tabelle steht unter Ihre API mit Middleware schützen.

Hinter einem Proxy

Wenn Clients DPoP verwenden, prüft die Middleware, dass jeder Nachweis für die URL erstellt wurde, die der Client aufgerufen hat. Setzen Sie PUBLIC_BASE_URL auf die externe URL, die Clients aufrufen, zum Beispiel https://api.example.com, auch hinter einem Proxy oder Load Balancer. Die Variable ist erforderlich und muss https verwenden (http nur für localhost, 127.0.0.1 oder [::1]); ohne sie startet die App nicht. Die Middleware liest X-Forwarded-*-Header selbst nie aus. Uvicorn wendet X-Forwarded-Proto jedoch auf die Request-URL an, wenn die Verbindung von einer Adresse in --forwarded-allow-ips kommt (standardmäßig 127.0.0.1,::1); mit gesetztem PUBLIC_BASE_URL hängt die geprüfte URL davon nicht ab.

Widerrufene Tokens

Ein Zugriffstoken bleibt bis zu seinem Ablauf gültig, daher teilt KeenAuth Ihrer API mit, wann Tokens widerrufen werden. Das Beispiel übergibt dem Protector einen In-Memory-Widerrufscache und stellt unter POST /webhooks/keenauth (webhook.py) einen token.revoked-Webhook bereit, der diesen Cache befüllt.

1. Erstellen Sie im Dashboard einen Webhook-Endpunkt mit der URL https://<your host>/webhooks/keenauth, der token.revoked abonniert, und exportieren Sie sein Signiergeheimnis als KEENAUTH_WEBHOOK_SECRET.
2. Der Handler prüft die Signatur mit dem SDK auf dem rohen Body und antwortet mit 400, wenn sie ungültig ist oder der Zeitstempel mehr als fünf Minuten abweicht. Er ordnet den Selektor des Ereignisses (access_token, session, user, client, org, org_user, client_user) einer Widerrufsmarkierung zu und antwortet mit 204; die Middleware lehnt danach jedes passende Token ab, das vor not_before ausgestellt wurde.
3. KeenAuth liefert mindestens einmal aus, daher wendet der Handler jede webhook-id genau einmal an. Kann die Markierung nicht gespeichert werden, antwortet er mit 503, und KeenAuth liefert das Ereignis erneut. Andere Ereignistypen und Selektoren, auf die kein Zugriffstoken passen kann (refresh_family, api), ignoriert er.

Der Handler protokolliert die Ereignis-ID, den Selektortyp und den Grund, niemals den Body, eine Signatur oder das Geheimnis. Jeder Worker-Prozess hat seinen eigenen Cache; bei mehreren Workern oder Instanzen verwenden Sie stattdessen einen gemeinsamen Speicher. Siehe Widerrufscache.

Nächste Schritte

• Ihre API mit Middleware schützen: Flask, Step-up-Anforderungen, DPoP und alle Optionen.
• Tokens prüfen: der Verifier unter der Middleware.
• Offline-Testkit.