如何基于FastAPI文档的python-jose实现标准JWT刷新令牌？

核心问题解答 #

1. 登录返回格式规范 #

登录接口应同时返回access_token、refresh_token和token_type，这是OAuth2协议的常见实践，无需拆分单独路由。前端可一次性获取两个令牌，分别存储（例如将refresh_token存入HttpOnly Cookie提升安全性，access_token存入内存或localStorage），减少冗余请求。

2. 刷新令牌的依赖项实现 #

刷新令牌本质仍是JWT，可复用官方文档的OAuth2认证逻辑，但需额外验证令牌类型（避免用访问令牌冒充刷新令牌），以下是完整代码实现：

完整代码示例（基于python-jose） #

from datetime import datetime, timedelta
from typing import Optional

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel

# JWT核心配置（生产环境请用随机生成的密钥）
SECRET_KEY = "your-32-character-secret-key-here"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
REFRESH_TOKEN_EXPIRE_DAYS = 7

# 密码加密上下文
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# 模拟用户数据库
fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "$2b$12$EixZaY3sTHK08pSakH5L6uS7v1X9c30E20c8Q9c30E20c8Q9c30E2",
        "disabled": False,
    }
}

# Pydantic模型定义
class Token(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str

class TokenData(BaseModel):
    username: Optional[str] = None

class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[bool] = None

class UserInDB(User):
    hashed_password: str

# 工具函数
def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    return pwd_context.hash(password)

def get_user(db, username: str):
    if username in db:
        return UserInDB(**db[username])

def authenticate_user(db, username: str, password: str):
    user = get_user(db, username)
    if not user or not verify_password(password, user.hashed_password):
        return False
    return user

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire, "type": "access"})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def create_refresh_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(days=7))
    to_encode.update({"exp": expire, "type": "refresh"})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

# 依赖项定义
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user_from_refresh(token: str = Depends(OAuth2PasswordBearer(tokenUrl="refresh-token"))):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无效的凭证",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        token_type = payload.get("type")
        if not username or token_type != "refresh":
            raise credentials_exception
        user = get_user(fake_users_db, username)
        if not user:
            raise credentials_exception
        return user
    except JWTError:
        raise credentials_exception

# FastAPI路由
app = FastAPI()

@app.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(
        data={"sub": user.username},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    refresh_token = create_refresh_token(
        data={"sub": user.username},
        expires_delta=timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
    )
    return {"access_token": access_token, "refresh_token": refresh_token, "token_type": "bearer"}

@app.post("/refresh-token")
async def refresh_access_token(current_user: User = Depends(get_current_user_from_refresh)):
    new_access_token = create_access_token(
        data={"sub": current_user.username},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    return {"access_token": new_access_token, "token_type": "bearer"}

@app.get("/users/me", response_model=User)
async def get_current_user(current_user: User = Depends(oauth2_scheme)):
    return current_user

关键细节说明 #

• 刷新令牌的JWT中添加了type: "refresh"声明，验证时强制检查该字段，防止访问令牌被滥用。
• 刷新令牌的过期时间远长于访问令牌（示例中为7天），符合无轮换刷新令牌的设计逻辑。
• 生产环境中SECRET_KEY需用openssl rand -hex 32生成随机安全密钥，绝对不能硬编码在代码中。

内容的提问来源于stack exchange，提问作者Harry Myburgh