Proteger una API de FastAPI

Este inicio rápido protege una API de FastAPI con tokens de acceso de KeenAuth. Añades la dependencia RequireAuth del SDK de Python, exiges un scope en una ruta y pruebas la API sin conexión con el kit de pruebas del SDK.

La API tiene tres endpoints:

Antes de empezar

• Un entorno de KeenAuth. Su emisor tiene el formato https://your-environment.eu.keenauth.app.
• Una API registrada en el entorno (Crear una API). Su identifier, por ejemplo https://api.example.com, es la audiencia de los tokens que acepta tu API.
• El scope messages:read en esa API, concedido a la aplicación que la llama.
• Python 3.10 o posterior.

1. Instalar el SDK

pip install "keenauth-python[fastapi]" uvicorn

El SDK todavía no está publicado. Hasta que lo esté, el ejemplo lo instala desde el código fuente del SDK. El ejemplo fija cada paquete a una versión exacta en requirements.txt.

2. Configurar

La API lee su configuración solo desde variables de entorno:

3. Proteger tus rutas

create_app crea el protector y las rutas. Depends(RequireAuth(protector)) exige un token válido y devuelve sus claims verificados; scopes=[...] además exige scopes. install(app) hace que los errores devuelvan respuestas application/problem+json, con un encabezado WWW-Authenticate por cada desafío.

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 crea el verificador real a partir de la configuración. El verificador obtiene las claves de firma de tu entorno la primera vez que ve un token. La dependencia es síncrona, así que FastAPI la ejecuta en su pool de hilos.

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. Ejecutar y llamar a la 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

Para una primera llamada, obtén un token con el grant client credentials de una aplicación máquina a máquina que tenga el scope messages:read en tu 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. Probar sin red

keenauth.testing emite tokens firmados con una clave de prueba pública, y testing.verifier acepta solo esos tokens. Las pruebas pasan ese verificador a create_app y llaman a la aplicación con el cliente de pruebas de Starlette, así que no necesitan red ni una cuenta de KeenAuth. Consulta Kit de pruebas sin conexión.

@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

Las pruebas del ejemplo comprueban que se acepta un token válido y que se rechazan con los estados indicados a continuación un token ausente, caducado o con audiencia incorrecta, un scope ausente, un token en la query de la URL y un token revocado. También envían un token vinculado a DPoP con una prueba para PUBLIC_BASE_URL, y comprueban que app_from_env rechaza el emisor de prueba. Las pruebas también envían entregas firmadas de token.revoked: una válida revoca el token, una falsificada, alterada o obsoleta obtiene 400 y no revoca nada, y una repetida tiene un único efecto.

Respuestas

El cuerpo es application/problem+json y el encabezado WWW-Authenticate lleva el desafío estándar. La tabla completa está en Proteger tu API con middleware.

Detrás de un proxy

Si los clientes usan DPoP, el middleware comprueba que cada prueba se hizo para la URL que llamó el cliente. Define PUBLIC_BASE_URL con la URL externa a la que llaman los clientes, por ejemplo https://api.example.com, también detrás de un proxy o balanceador de carga. Es obligatoria y debe usar https (http solo para localhost, 127.0.0.1 o [::1]); la aplicación no arranca sin ella. El middleware nunca lee encabezados X-Forwarded-* por sí mismo. Sin embargo, Uvicorn aplica X-Forwarded-Proto a la URL de la petición cuando la conexión procede de una dirección en --forwarded-allow-ips (por defecto 127.0.0.1,::1); con PUBLIC_BASE_URL definido, la URL comprobada no depende de eso.

Tokens revocados

Un token de acceso sigue siendo válido hasta que caduca, por eso KeenAuth le indica a tu API cuándo se revocan tokens. El ejemplo pasa una caché de revocación en memoria al protector y sirve un webhook de token.revoked en POST /webhooks/keenauth (webhook.py) que la alimenta.

1. En el panel, crea un endpoint de webhook con la URL https://<your host>/webhooks/keenauth que se suscribe a token.revoked, y exporta su secreto de firma como KEENAUTH_WEBHOOK_SECRET.
2. El manejador verifica la firma con el SDK sobre el cuerpo sin procesar y responde 400 cuando no es válida o cuando la marca temporal se desvía más de cinco minutos. Asigna el selector del evento (access_token, session, user, client, org, org_user, client_user) a un marcador de revocación y responde 204; el middleware entonces rechaza todos los tokens coincidentes emitidos antes de not_before.
3. KeenAuth entrega al menos una vez, así que el manejador aplica cada evento webhook-id una sola vez. Si no puede guardar el marcador, responde 503 y KeenAuth vuelve a entregar el evento. Ignora otros tipos de evento y selectores que ningún token de acceso puede coincidir (refresh_family, api).

El manejador registra el id del evento, el tipo de selector y el motivo, nunca el cuerpo, una firma ni el secreto. Cada proceso worker tiene su propia caché; con varios workers o instancias, usa un almacén compartido. Consulta Caché de revocación.

Próximos pasos

• Proteger tu API con middleware: Flask, requisitos de step-up, DPoP y todas las opciones.
• Verificar tokens: el verificador bajo el middleware.
• Kit de pruebas sin conexión.