Proteger una API de Django

Este inicio rápido protege una API de Django con tokens de acceso de KeenAuth. Añada el decorador de vista @protect del SDK de Python, exija un permiso en una vista y pruebe la API sin conexión con el kit de pruebas del SDK. El ejemplo usa Django 5.2 y no utiliza base de datos.

La API tiene tres endpoints:

Endpoint Requiere
GET /api/public nada
GET /api/private un token de acceso válido para su API
GET /api/messages un token de acceso válido con el permiso messages:read

Antes de empezar

1. Instalar el SDK

pip install "keenauth-python[django]"

El SDK todavía no está publicado. Hasta entonces, 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

project/settings.py lee todos los valores de variables de entorno:

Variable Obligatoria Significado
DJANGO_SECRET_KEY sí La clave secreta de Django; genere una y nunca la incluya en el control de versiones
DJANGO_ALLOWED_HOSTS no Nombres de host separados por comas, por defecto localhost,127.0.0.1
KEENAUTH_ISSUER sí El emisor de su entorno
KEENAUTH_AUDIENCE sí El identificador de su API
PUBLIC_BASE_URL sí La URL que llaman los clientes (consulte Detrás de un proxy)
KEENAUTH_WEBHOOK_SECRET recomendada El secreto de firma idp_whsec_ del endpoint de webhook que entrega token.revoked (dos, separados por comas, mientras lo rota). Sin él, la API nunca se entera de los tokens revocados

3. Proteger sus vistas

@protect(protector) exige un token válido y coloca las reclamaciones verificadas en request.auth; scopes=[...] además exige permisos. Una comprobación fallida devuelve una respuesta application/problem+json antes de ejecutar la vista.

@require_GET
def public(request: HttpRequest) -> JsonResponse:
    # No token needed.
    return JsonResponse({"message": "Hello from a public endpoint."})


@require_GET
@protect(protector)
def private(request: HttpRequest) -> JsonResponse:
    # Any valid access token for this API; the verified claims are in request.auth.
    return JsonResponse({"message": "Hello from a private endpoint.", "sub": request.auth.subject})


@require_GET
@protect(protector, scopes=["messages:read"])
def messages(request: HttpRequest) -> JsonResponse:
    # A valid access token that also carries the messages:read scope.
    return JsonResponse({"messages": ["Hello from a scoped endpoint."]})

api/auth.py construye el protector a partir de la configuración al arrancar (ApiConfig.ready en api/apps.py), de modo que una configuración ausente detiene el servidor antes de la primera solicitud. El verificador obtiene las claves de firma de su entorno la primera vez que recibe un token.

@cache
def get_protector() -> Protector:
    if not settings.KEENAUTH_ISSUER or not settings.KEENAUTH_AUDIENCE or not settings.KEENAUTH_PUBLIC_BASE_URL:
        raise ImproperlyConfigured("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=settings.KEENAUTH_ISSUER, audience=settings.KEENAUTH_AUDIENCE))
    return Protector(
        verifier,
        public_base_url=settings.KEENAUTH_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),
    )


@cache
def get_webhook() -> Optional[RevocationWebhook]:
    """The token.revoked webhook, or None when KEENAUTH_WEBHOOK_SECRET is not set. A malformed secret raises
    WebhookSecretError (without echoing it), so ApiConfig.ready stops the server at startup."""
    if not settings.KEENAUTH_WEBHOOK_SECRETS:
        return None
    return RevocationWebhook(settings.KEENAUTH_WEBHOOK_SECRETS, revocation)


class ProtectorProxy:
    """Hands each request to `get_protector()`, so the `@protect` decorators can be applied at import time."""

    def authenticate(self, req: RequestInfo, requirement: Optional[Requirement] = None) -> AccessClaims:
        return get_protector().authenticate(req, requirement)


protector = ProtectorProxy()

4. Ejecutar y llamar a la API

export DJANGO_SECRET_KEY="$(python3 -c 'import secrets; print(secrets.token_urlsafe(50))')"
export KEENAUTH_ISSUER=https://your-environment.eu.keenauth.app
export KEENAUTH_AUDIENCE=https://api.example.com
export PUBLIC_BASE_URL=https://api.example.com
python manage.py runserver 8080

runserver es solo para desarrollo; en producción ejecute project.wsgi bajo un servidor WSGI.

Para una primera llamada, obtenga un token con la concesión de credenciales de cliente de una aplicación máquina a máquina que tenga el permiso messages:read en su 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 reemplazan get_protector por un protector que usa ese verificador, así que no necesitan red, ninguna cuenta de KeenAuth ni ningún interruptor de prueba en la configuración. Consulte Kit de pruebas sin conexión.

class ProtectedAPITest(SimpleTestCase):
    def setUp(self) -> None:
        self.cache = MemoryRevocationCache()
        test_protector = Protector(testing.verifier(audience=AUDIENCE), public_base_url=BASE_URL, revocation=self.cache)
        patcher = mock.patch.object(auth, "get_protector", return_value=test_protector)
        patcher.start()
        self.addCleanup(patcher.stop)
    def test_scope(self) -> None:
        self.assert_problem(self.get("/api/messages", testing.mint_access_token(AUDIENCE)), 403, "insufficient_scope",
                            'scope="messages:read"')
        tok = testing.mint_access_token(AUDIENCE, claims={"scope": "profile messages:read"})
        self.assertEqual(self.get("/api/messages", tok).status_code, 200)
set -a; . ./.env.example; set +a
DJANGO_SECRET_KEY=insecure-test-only python manage.py test

Las pruebas del ejemplo comprueban que se acepta un token válido y que se rechazan, con los estados indicados abajo, un token ausente, caducado o con audiencia incorrecta, un permiso ausente, un token en la consulta 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 el protector de producción rechaza el emisor de prueba. Además, envían entregas firmadas de token.revoked: una válida revoca el token, una falsificada, alterada o caducada recibe 400 y no revoca nada, y una repetida tiene un único efecto.

Respuestas

Solicitud Estado code en el cuerpo
Sin encabezado Authorization 401 missing_token
Token caducado, token de otra API, firma incorrecta, token revocado 401 invalid_token
Token sin el permiso requerido 403 insufficient_scope
Token en la consulta de la URL (?access_token=) 400 invalid_request

El cuerpo es application/problem+json y el encabezado WWW-Authenticate contiene el desafío estándar; Django une varios desafíos en un solo encabezado. La tabla completa está en Proteger su 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. Defina PUBLIC_BASE_URL con la URL externa 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 se niega a arrancar sin ella. El middleware nunca lee los encabezados X-Forwarded-* por sí mismo; Django solo los aplica si define SECURE_PROXY_SSL_HEADER o USE_X_FORWARDED_HOST, y con PUBLIC_BASE_URL definida la URL comprobada no depende de ellos.

Tokens revocados

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

  1. En el panel, cree un endpoint de webhook con la URL https://<your host>/webhooks/keenauth que se suscriba a token.revoked, y exporte 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 de tiempo difiere más de cinco minutos. Asocia el selector del evento (access_token, session, user, client, org, org_user, client_user) con un marcador de revocación y responde 204; después, el middleware rechaza cada token coincidente emitido antes de not_before.
  3. KeenAuth entrega al menos una vez, así que el manejador aplica cada webhook-id una sola vez. Si no puede almacenar 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 identificador del evento, el tipo de selector y el motivo, nunca el cuerpo, una firma ni el secreto. Cada proceso trabajador tiene su propia caché; con varios trabajadores o instancias, use en su lugar un almacén compartido. Consulte Caché de revocación.

Próximos pasos