令牌与 JWKS：验证 Didit 身份认证 JWT

// npm install jose
import { createRemoteJWKSet, jwtVerify } from "jose";

const ISSUER = process.env.DIDIT_AUTH_ISSUER; // https://<name>.auth.didit.me
const AUDIENCE = process.env.DIDIT_APPLICATION_ID;
// Cache keys for 5 minutes; refetch an unknown kid at most once a minute.
const jwks = createRemoteJWKSet(new URL(`${ISSUER}/.well-known/jwks.json`), {
  cacheMaxAge: 5 * 60 * 1000,
  cooldownDuration: 60 * 1000,
});

export async function verifyAccessToken(token) {
  const { payload } = await jwtVerify(token, jwks, {
    issuer: ISSUER,
    audience: AUDIENCE,
    algorithms: ["ES256"],
    typ: "at+jwt",
    clockTolerance: 60,
    maxTokenAge: "5m", // also rejects an iat in the future
    requiredClaims: ["sub", "sid", "jti", "exp", "iat"],
  });
  if (payload.act !== null) throw new Error("agent tokens are not accepted");
  return payload; // payload.vendor_data is the person across Didit products
}

# pip install "pyjwt[crypto]"
import os
import threading
import time

import jwt

ISSUER = os.environ["DIDIT_AUTH_ISSUER"]  # https://<name>.auth.didit.me
AUDIENCE = os.environ["DIDIT_APPLICATION_ID"]
jwks = jwt.PyJWKClient(f"{ISSUER}/.well-known/jwks.json", lifespan=300)
refresh_lock = threading.Lock()
last_refresh = 0.0


def signing_key(kid: str | None) -> jwt.PyJWK:
    # PyJWKClient refetches for every unknown kid: limit it to once a minute.
    global last_refresh
    keys = {key.key_id: key for key in jwks.get_signing_keys()}
    if kid not in keys:
        with refresh_lock:
            if time.monotonic() - last_refresh >= 60:
                last_refresh = time.monotonic()
                fresh = jwks.get_signing_keys(refresh=True)
                keys = {key.key_id: key for key in fresh}
    if kid not in keys:
        raise jwt.InvalidTokenError("unknown signing key")
    return keys[kid]


def verify_access_token(token: str) -> dict:
    header = jwt.get_unverified_header(token)
    if header.get("typ") != "at+jwt":
        raise jwt.InvalidTokenError("not an access token")
    claims = jwt.decode(
        token,
        signing_key(header.get("kid")).key,
        algorithms=["ES256"],
        audience=AUDIENCE,
        issuer=ISSUER,
        leeway=60,
        options={"require": ["iss", "aud", "sub", "sid", "jti", "exp", "iat"]},
    )
    if "act" not in claims or claims["act"] is not None:
        raise jwt.InvalidTokenError("agent tokens are not accepted")
    return claims