Zustand
Implementiert und lokal verifiziert. Die echteIdP-Round-Trip-Verifizierung (JWKS-Abruf, Token-Signierung/Prüfung gegeneine Live-XID-Instanz) wurde noch nicht durchgeführt und muss vor demProduktionseinsatz abgeschlossen werden.
Installieren
pip install xidSchnellstart
Erstellen Sie beim Start einen XidClient und verwenden Sie ihnwieder. Der Client speichert JWKS intern zwischen.
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.subWebhook prüfen
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))FastAPI-Integration
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}XidClient-Optionen
| Parameter | Standard | Beschreibung |
|---|---|---|
issuer |
erforderlich | XID-Issuer-URL |
audience |
None |
Erwarteter aud-Claim; None überspringt die Validierung |
jwks_ttl |
3600 |
JWKS-In-Memory-Cache-TTL in Sekunden |
http_timeout |
10.0 |
JWKS-Abruf-Timeout in Sekunden |
cookie_name |
__session |
Cookie-Schlüssel für die Sitzungs-Token-Extraktion |
leeway |
0 |
Taktversatztoleranz in Sekunden |
Kern-API
| Methode | Beschreibung |
|---|---|
await client.verify_token(token) |
JWT-String prüfen; löst bei Fehler TokenVerificationError aus. |
await client.authenticate_request(headers, cookies) |
Token aus Headern/Cookies extrahieren und prüfen. Gibt AuthStatuszurück; löst keine Exception aus. |
client.verify_webhook(payload, headers, secret) |
Synchron. Validiert svix-HMAC-SHA256 + 5-Minuten-Replay-Fenster. Löst beiFehler WebhookVerificationError aus. |
await client.aclose() |
Zugrunde liegende HTTP-Client-Ressourcen freigeben. |
Plattformhinweise
- Async-first. Synchrone Aufrufer (Django/Flask) können mit
asyncio.run()wrappen. - Setzt
pyjwt[crypto] >=2.8undhttpx >=0.27voraus. Python3.10+ erforderlich. - Multi-Worker-Deployments teilen keinen JWKS-Cache über Prozesse hinweg.Ein gemeinsamer Cache (Redis) ist eine geplante Verbesserung.