FastAPI JWT 刷新令牌轮换

示例说明

访问令牌短时效把泄漏窗口压到分钟级，长会话交给刷新令牌维持。轮换的要点是每次刷新都发新对并把旧刷新令牌作废——如果已作废的令牌再次出现，说明令牌被复制或泄漏，立即拒绝并可吊销整个会话族。typ 声明区分令牌类型，防止拿刷新令牌直接访问接口。

技术要点

• 访问与刷新令牌双轨独立时效
• 撤销集合检测刷新令牌复用
• typ 声明阻止令牌类型混用

代码解读

issue_tokens 一次性产出对：访问十五分钟、刷新十四天并带唯一 jti；refresh 端点先验签再查撤销集——已撤销的刷新令牌再次出现直接 401 报复用，正常路径把旧令牌加入撤销集后发新对。me 端点校验 typ 必须是 access。

使用建议

撤销集合要持久化并带 TTL 自动清理。复用检测应触发告警与全族吊销。密钥用强随机值并支持轮换。

代码示例

from datetime import datetime, timedelta, timezone

import jwt
from fastapi import Depends, FastAPI, HTTPException
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel

SECRET = 'rotate-me-in-production'
REFRESH_TTL = timedelta(days=14)
ACCESS_TTL = timedelta(minutes=15)

app = FastAPI()
oauth2 = OAuth2PasswordBearer(tokenUrl='/token')
revoked: set[str] = set()  # 生产环境应持久化并设置过期


class TokenPair(BaseModel):
    access: str
    refresh: str


def issue_tokens(user_id: str) -> TokenPair:
    now = datetime.now(timezone.utc)
    return TokenPair(
        access=jwt.encode(
            {'sub': user_id, 'exp': now + ACCESS_TTL, 'typ': 'access'},
            SECRET, algorithm='HS256'),
        refresh=jwt.encode(
            {'sub': user_id, 'exp': now + REFRESH_TTL, 'typ': 'refresh',
             'jti': f'{user_id}-{int(now.timestamp())}'},
            SECRET, algorithm='HS256'),
    )


@app.post('/token/refresh', response_model=TokenPair)
def refresh(pair: TokenPair):
    try:
        claims = jwt.decode(pair.refresh, SECRET, algorithms=['HS256'])
    except jwt.InvalidTokenError:
        raise HTTPException(401, 'invalid refresh token')

    if pair.refresh in revoked:
        raise HTTPException(401, 'token reuse detected')

    revoked.add(pair.refresh)
    return issue_tokens(claims['sub'])


@app.get('/me')
def me(token: str = Depends(oauth2)):
    claims = jwt.decode(token, SECRET, algorithms=['HS256'])
    if claims.get('typ') != 'access':
        raise HTTPException(401, 'wrong token type')
    return {'user': claims['sub']}