sdk/python

异步 Python 服务端 SDK，提供 networkless JWT 验证、请求认证和 webhook 签名校验。

状态

已在本地实现并验证。针对真实 XID 实例的 IdP 往返验证（JWKS 获取、token 签名/验证）尚未执行，生产使用前必须完成。

安装

pip install xid

快速开始

在启动时创建一个 XidClient 并复用。客户端在内部缓存 JWKS。

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

验证 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))

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}

XidClient 选项

参数	默认	描述
`issuer`	必填	XID 签发方 URL
`audience`	`None`	期望的 aud claim；None 跳过验证
`jwks_ttl`	`3600`	JWKS 内存缓存 TTL（秒）
`http_timeout`	`10.0`	JWKS 获取超时（秒）
`cookie_name`	`__session`	提取会话 token 的 cookie key
`leeway`	`0`	时钟偏差容忍度（秒）

核心 API

方式	描述
`await client.verify_token(token)`	验证 JWT 字符串；失败时抛出 `TokenVerificationError`。
`await client.authenticate_request(headers, cookies)`	从请求头/cookie 中提取并验证 token。返回 `AuthStatus`；不抛出异常。
`client.verify_webhook(payload, headers, secret)`	同步执行。验证 svix HMAC-SHA256 + 5 分钟重放窗口。失败时抛出 `WebhookVerificationError`。
`await client.aclose()`	释放底层 HTTP 客户端资源。

平台注意事项

异步优先。同步调用方（Django/Flask）可用 asyncio.run() 包裹。
依赖 pyjwt[crypto] >=2.8 和 httpx >=0.27。需要 Python 3.10+。
多 Worker 部署不在进程间共享 JWKS 缓存。共享缓存（Redis）是计划中的改进。