1
{
2
  "alg": "RS256", // 签名算法: HS256, RS256, ES256
3
  "typ": "JWT", // 令牌类型
4
  "kid": "key-1" // 密钥 ID (用于密钥轮换)
5
}

1
import jwt
2
from jwt import PyJWKClient
3

4
# 1. 获取公钥 (从 JWKS 端点)
5
jwks_client = PyJWKClient("https://auth.example.com/.well-known/jwks.json")
6
signing_key = jwks_client.get_signing_key_from_jwt(token)
7

8
# 2. 解码验证
9
claims = jwt.decode(
10
    token,
11
    signing_key.key,
12
    algorithms=["RS256"],
13
    audience="my_client_id",
14
    issuer="https://auth.example.com"
15
)
16

17
user_id = claims["sub"]

1
import time
2

3
class TokenManager:
4
    def __init__(self, refresh_threshold: int = 300):
5
        self.refresh_threshold = refresh_threshold  # 提前 5 分钟刷新
6

7
    def get_valid_token(self) -> str:
8
        # 1. 检查 access_token 是否即将过期
9
        if self.access_token and not self._is_expiring_soon():
10
            return self.access_token
11

12
        # 2. 使用 refresh_token 刷新
13
        if self.refresh_token:
14
            return self._refresh_access_token()
15

16
        # 3. 需要重新授权
17
        raise TokenExpiredError("请重新登录")
18

19
    def _is_expiring_soon(self) -> bool:
20
        expires_at = self.token_claims.get("exp", 0)
21
        return time.time() > (expires_at - self.refresh_threshold)
22

23
    def _refresh_access_token(self) -> str:
24
        response = httpx.post(
25
            f"{AUTH_SERVER}/oauth/token",
26
            data={
27
                "grant_type": "refresh_token",
28
                "refresh_token": self.refresh_token,
29
                "client_id": CLIENT_ID,
30
            }
31
        )
32
        self.access_token = response.json()["access_token"]
33
        return self.access_token

1
# Token Introspection - 检查 token 状态
2
def introspect_token(token: str) -> dict:
3
    response = httpx.post(
4
        f"{AUTH_SERVER}/oauth/introspect",
5
        data={"token": token},
6
        auth=(CLIENT_ID, CLIENT_SECRET)
7
    )
8
    return response.json()  # {"active": true/false, ...}
9

10
# Token Revocation - 撤回 token
11
def revoke_token(token: str, token_type: str = "access_token"):
12
    httpx.post(
13
        f"{AUTH_SERVER}/oauth/revoke",
14
        data={"token": token, "token_type_hint": token_type},
15
        auth=(CLIENT_ID, CLIENT_SECRET)
16
    )

1
{
2
  "iss": "https://auth.example.com",
3
  "sub": "user-123",
4
  "aud": "my-client-id",
5
  "exp": 1234567890,
6
  "iat": 1234567800,
7
  "auth_time": 1234567700,
8
  "nonce": "n-0S6_WzA2Mj",
9
  "acr": "urn:mace:incommon:iap:silver",
10
  "name": "张三",
11
  "email": "zhang@example.com",
12
  "email_verified": true,
13
  "picture": "https://example.com/avatar.jpg"
14
}

1
# Session vs Token
2
session_cookie = request.session.get("user_id")
3
# 或 JWT 存 Cookie