SDK de servidor para Python
keenauth-python es el SDK de servidor para Python 3.10 o posterior. Llama a la API de gestión con métodos tipados y añade el comportamiento que comparten todos los SDK de servidor de KeenAuth: autenticación con una clave de API o con un token máquina a máquina private_key_jwt, reintentos, paginación por cursor, errores tipados, verificación de firmas de webhooks y advertencias de obsolescencia. El paquete todavía no está publicado en PyPI.
Para iniciar sesión en una aplicación web, consulte Django, Flask y FastAPI. Para proteger una API que recibe tokens de portador, consulte Proteger su API.
Instalar
pip install keenauth-python
El firmante private_key_jwt necesita el paquete cryptography. Instálelo con el extra m2m:
pip install "keenauth-python[m2m]"
Llamar a la API de gestión con una clave de API
import os
from keenauth import KeenAuth
client = KeenAuth("https://api.eu.keenauth.com/v1", os.environ["KEENAUTH_API_KEY"])
page = client.users.list_users(limit=20)
for user in page.data:
print(user.id)
Cada clase de API generada está disponible como atributo: client.users, client.organizations, client.api_keys y así sucesivamente. La URL base debe usar https; http simple solo se acepta para localhost. El SDK nunca sigue una redirección, de modo que su credencial nunca se envía a otro host.
Usar un token de gestión private_key_jwt
Una aplicación de gestión se autentica con su clave privada en lugar de una clave de API. El SDK firma una aserción de cliente (RFC 7523), obtiene un token de la API de gestión del endpoint de tokens de su entorno y lo renueva 60 segundos antes de que caduque:
import os
from keenauth import KeenAuth
from keenauth.m2m import ManagementTokenSource
source = ManagementTokenSource(
issuer="https://acme.eu.keenauth.app", # your environment's default host
client_id="app_01j9zzzzzzzzzzzzzzzzzzzz", # a management application
private_key=os.environ["KEENAUTH_PRIVATE_KEY"], # unencrypted PEM
resource="https://api.eu.keenauth.com", # the management API of your region
scopes=["users:read", "users:write"], # optional; default: the grant's default scopes
)
client = KeenAuth("https://api.eu.keenauth.com/v1", token_source=source)
- La aserción tiene
issysubcon el valor declient_id,audcon el valor del emisor, unjtialeatorio y una duración de 60 segundos. Sukides la huella RFC 7638 de la clave pública. - Claves admitidas: EC P-256 (
ES256), EC P-384 (ES384), Ed25519 (EdDSA) y RSA de al menos 2048 bits (PS256por defecto, oalgorithm="RS256"). - Registre la clave pública en la aplicación de gestión.
keenauth.m2m.load_private_key(pem).public_jwk()la devuelve como JWK. - El token se almacena en caché. Cuando varios hilos necesitan un token nuevo al mismo tiempo, solo una solicitud llega al endpoint de tokens.
- Una solicitud rechazada genera
keenauth.m2m.TokenErrorconstatus,erroryerror_description. El mensaje nunca contiene la aserción ni el token.
Reintentos
El SDK reintenta una solicitud que falló con 429, 502, 503 o 504, o con un error de conexión o un tiempo de espera agotado. Solo reintenta solicitudes GET, PUT, DELETE y POST que incluyan un Idempotency-Key. Un PATCH que incluya If-Match se reintenta solo con 429: el limitador de frecuencia rechaza una solicitud antes de ejecutarla, y If-Match evita que un cambio se aplique dos veces. Cualquier otro PATCH nunca se reintenta. Realiza como máximo 3 intentos en total.
- Si la respuesta tiene un encabezado
Retry-After(segundos o una fecha HTTP), el SDK espera ese tiempo. SiRetry-Aftersupera los 60 segundos, el SDK no espera y le devuelve la respuesta. - Sin
Retry-After, espera 0,5 segundos y luego 1 segundo, cada uno con un jitter aleatorio de ±25 %. - El SDK añade un
Idempotency-KeyUUIDv4 aleatorio a cada operaciónPOSTque lo acepte, a menos que usted pase su propioidempotency_key=. Un reintento envía la misma clave, de modo que el servidor ejecuta la operación una sola vez.
Paginación
Los métodos de listado devuelven una página: data y list_metadata con los cursores after y before. paginate recorre todas las páginas y solicita la página siguiente solo cuando usted llega a ella:
from keenauth import paginate
for user in paginate(client.users.list_users, limit=100, status="active"):
if user.email == "ada@example.com":
break # no further page is requested
- Otros argumentos, como los filtros y
order, se envían con cada solicitud de página. - Pase
before=para recorrer hacia atrás desde un cursor.afterybeforeno pueden usarse juntos. pages(...)acepta los mismos argumentos y devuelve páginas completas en lugar de elementos.
Errores
Una respuesta de error en application/problem+json (RFC 9457) genera keenauth.ProblemError. Use sus campos code, status, detail, request_id y errors. Un code que esta versión del SDK no conoce se sigue devolviendo como cadena, de modo que los nuevos códigos de error del servidor no rompen su código.
from keenauth import ProblemError
try:
client.users.get_user("usr_01j9zzzzzzzzzzzzzzzzzzzz")
except ProblemError as e:
if e.code == "not_found":
...
print(e.request_id)
Verificar webhooks
KeenAuth firma los webhooks con el esquema Standard Webhooks. WebhookVerifier comprueba los encabezados webhook-id, webhook-timestamp y webhook-signature con el secreto idp_whsec_ de su endpoint:
import os
from flask import Flask, request
from keenauth.webhooks import WebhookVerificationError, WebhookVerifier
app = Flask(__name__)
verifier = WebhookVerifier(os.environ["KEENAUTH_WEBHOOK_SECRET"])
@app.post("/webhooks/keenauth")
def keenauth_webhook():
try:
verifier.verify(request.get_data(), request.headers)
except WebhookVerificationError as e:
return {"error": e.reason}, 400
event = request.get_json()
...
return "", 204
- Pase el cuerpo sin procesar de la solicitud, antes de cualquier análisis JSON; un byte modificado hace fallar la comprobación.
- La marca de tiempo debe estar dentro de 5 minutos del reloj de su servidor, en cualquier dirección.
- Mientras rota el secreto, pase ambos:
WebhookVerifier([new_secret, old_secret]). La solicitud se acepta cuando cualquier firma del encabezado coincide con cualquier secreto. - Los errores son tipados:
WebhookHeadersError,WebhookTimestampError,WebhookSignatureErroryWebhookSecretError(un secreto mal formado, que se genera al crear el verificador). Cada uno tiene un camporeason. - La comparación se ejecuta en tiempo constante, y el mensaje de error nunca contiene el secreto ni la firma.
Eventos y claims tipados
keenauth.events tiene un dataclass inmutable para cada tipo de evento de cliente, generado a partir de los esquemas públicos de eventos. parse_event comprueba un cuerpo verificado y devuelve la clase de su tipo. keenauth.webhooks.verify_event verifica y analiza en una sola llamada. Refine el tipo con match o isinstance:
from keenauth.events import EmailSendingSuspendedEvent, UnknownEvent, UserCreatedEvent
from keenauth.webhooks import verify_event
event = verify_event(verifier, request.get_data(), request.headers)
match event:
case UserCreatedEvent():
print(event.data.object.id)
case EmailSendingSuspendedEvent():
print(event.data.object.reason)
case UnknownEvent():
pass # a type this SDK version does not know yet; event.type and event.raw are still available
parse_eventacepta bytes, una cadena o un mapeo ya decodificado. Comprueba el sobre y los miembros que declara el esquema del tipo. Un miembro ausente o con un tipo incorrecto generaEventParseErrorcon su ruta, comodata.object.reason.- Un tipo desconocido, o un tipo conocido con un
schema_versionque este SDK no modela, es unUnknownEvent. No es un error. - Los miembros que el esquema no declara se conservan en
raw(en el evento y endata.object). - Los arrays son tuplas. Los valores de los campos enumerados no se comprueban frente a los valores conocidos.
- Los modelos superan
mypy --strict.
keenauth.claims tipa los claims de un token verificado: parse_id_token_claims (OpenID Connect Core) y parse_access_token_claims (RFC 9068). Comprueban los tipos de los claims, pero no verifican nada; llámelos sobre los claims que devuelve el verificador:
from keenauth.claims import parse_access_token_claims
at = verifier.verify_access_token(token)
claims = parse_access_token_claims(at.claims)
claims.client_id # str
claims.org_roles # tuple[str, ...] | None
claims.custom # your own claims
aud es siempre una tupla. Un claim que no encaja genera ClaimsParseError con su nombre.
Advertencias de obsolescencia
Cuando una respuesta tiene un encabezado Deprecation (RFC 9745) o Sunset (RFC 8594), el SDK escribe una advertencia por proceso en el logger keenauth con el módulo estándar logging. La advertencia indica el método, el recurso y los valores de los encabezados; nunca contiene IDs, cadenas de consulta ni credenciales. La solicitud en sí no se ve afectada.
import logging
logging.getLogger("keenauth").setLevel(logging.WARNING)
Encabezado de telemetría
Cada solicitud incluye Idp-Client: keenauth-python/<version> (python/<version>). No contiene datos personales y no puede desactivarse; KeenAuth lo usa para soporte y avisos de obsolescencia.