sdk/python

SDK de servidor Python asíncrono para verificación JWT sin llamadas de red, autenticación de solicitudes y validación de firma de webhook.

Estado

Implementado y verificado localmente. La verificación de ida y vuelta contra un IdP real (obtención de JWKS, firma/verificación de tokens contra una instancia XID en producción) aún no se ha realizado y debe completarse antes del uso en producción.

Instalación

pip install xid

Inicio rápido

Construye un único XidClient al arrancar y reutilízalo. El cliente almacena el JWKS en caché internamente.

from xid import XidClient

client = XidClient(
    issuer="https://xid.dev",
    audience="https://api.yourapp.com",  # optional
)

# Verify a token
claims = await client.verify_token("eyJ...")
print(claims.sub, claims.email, claims.scope)

# Authenticate a request
status = await client.authenticate_request(
    headers=dict(request.headers),
    cookies=dict(request.cookies),
)
if not status.authenticated:
    raise Unauthorized()
user_id = status.claims.sub

Verificar webhook

from xid import WebhookVerificationError

try:
    webhook = client.verify_webhook(
        payload=request.body,
        headers=dict(request.headers),
        secret="whsec_xxx",
    )
    import json
    event = json.loads(webhook.body)
except WebhookVerificationError as exc:
    raise BadRequest(str(exc))

Integración con FastAPI

from fastapi import FastAPI, Depends, HTTPException, Request
from xid import XidClient, TokenClaims

app = FastAPI()
xid = XidClient(issuer="https://xid.dev")

@app.on_event("shutdown")
async def shutdown():
    await xid.aclose()

async def require_auth(request: Request) -> TokenClaims:
    status = await xid.authenticate_request(dict(request.headers))
    if not status.authenticated:
        raise HTTPException(status_code=401)
    return status.claims

@app.get("/me")
async def me(claims: TokenClaims = Depends(require_auth)):
    return {"sub": claims.sub, "email": claims.email}

Opciones de XidClient

Parámetro	Por defecto	Descripción
`issuer`	obligatorio	URL del emisor XID
`audience`	`None`	Claim aud esperado; None omite la validación
`jwks_ttl`	`3600`	TTL del caché en memoria de JWKS en segundos
`http_timeout`	`10.0`	Tiempo de espera de obtención de JWKS en segundos
`cookie_name`	`__session`	Clave de cookie para extraer el token de sesión
`leeway`	`0`	Tolerancia de desfase de reloj en segundos

API principal

Método	Descripción
`await client.verify_token(token)`	Verifica una cadena JWT; lanza `TokenVerificationError` al fallar.
`await client.authenticate_request(headers, cookies)`	Extrae y verifica el token de encabezados/cookies. Devuelve `AuthStatus`; no lanza excepciones.
`client.verify_webhook(payload, headers, secret)`	Síncrono. Valida el HMAC-SHA256 de svix + ventana de reproducción de 5 minutos. Lanza `WebhookVerificationError` al fallar.
`await client.aclose()`	Libera los recursos del cliente HTTP subyacente.

Notas de plataforma

Async-first. Las llamadas síncronas (Django/Flask) pueden envolverse con asyncio.run().
Requiere pyjwt[crypto] >=2.8 y httpx >=0.27. Se necesita Python 3.10+.
Los despliegues multi-worker no comparten el caché de JWKS entre procesos. Un caché compartido (Redis) es una mejora planificada.