Saltar al contenido

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.

Ver como Markdown

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.

Estado del registro: UNPUBLISHED. Instala este SDK únicamente desde el checkout del código fuente del repositorio; no uses un registro de paquetes externo.

La autenticación de solicitudes acepta solo Bearer de forma predeterminada. Una cookie JWT propiedad de la aplicación solo se lee cuando se configura su nombre exacto. La cookie opaca de Core __Host-xid.rt.* nunca se busca ni se verifica localmente; intercámbiala reenviando el header Cookie completo al POST /v1/sessions/token del mismo origen exacto, con las redirecciones desactivadas, y acepta solo una respuesta que contenga únicamente el campo token.

Instalación

pip install "xid @ git+https://github.com/StringKe/xid#subdirectory=sdk/python"

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 (Bearer-only by default)
status = await client.authenticate_request(headers=dict(request.headers))
if not status.authenticated:
    raise Unauthorized()
user_id = status.claims.sub

# Explicit same-origin Core session -> JWT exchange
token = await client.exchange_session_token(
    incoming_request_url="https://app.example.com/account",
    cookie_header=request.headers["cookie"],
)

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 disabled Nombre de cookie JWT propiedad de la aplicación; desactivado salvo configuración explícita
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.
Navegación

Escribe para buscar...

Usa las flechas para navegarPulsa Intro para seleccionarPulsa Escape para cerrar