FastAPI中多JWT认证流程与权限范围的实现问题

核心问题修正 #

你的场景是用户用已有的User JWT换取专属Store JWT，不需要使用OAuth2AuthorizationCodeBearer（这是标准授权码流程的组件，不适合你的自定义令牌交换场景）。核心思路是：

• 用get_current_user保护Store Token生成接口，确保只有已认证用户能请求
• 为Store API单独定义JWT验证逻辑，区分开User JWT的认证流程

完整实现代码 #

1. 定义Store JWT的OAuth2 Scheme #

用OAuth2PasswordBearer处理Store Token的Bearer认证（和User JWT的认证方式一致，只是令牌内容不同）：

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel
import jwt
from typing import Annotated, List

# 全局配置（和User JWT共用或单独定义）
SECRET_KEY = "your-secret-key-here"
ALGORITHM = "HS256"

# Store Token的数据模型
class StoreTokenData(BaseModel):
    store_id: str
    scopes: List[str] = []

# Store JWT的认证Scheme，tokenUrl填占位符即可（因为我们是自定义生成令牌）
oauth2_store_scheme = OAuth2PasswordBearer(tokenUrl="dummy", auto_error=True)

# 假设你已经实现的User相关模型和get_current_user
class User(BaseModel):
    id: str
    username: str

async def get_current_user(token: str = Depends(oauth2_user_scheme)):
    # 你的User JWT验证逻辑，返回当前用户
    pass

2. 实现Store Token生成接口 #

该接口仅对已通过User JWT认证的用户开放，生成带店铺权限的Store JWT：

@app.post("/stores/auth/{store_id}")
def create_store_token(
    store_id: str,
    user: Annotated[User, Depends(get_current_user)]
):
    # 先校验用户是否有权限访问该店铺（比如用户是店铺管理员/店主）
    if not check_user_store_permission(user.id, store_id):
        raise HTTPException(status_code=403, detail="无该店铺访问权限")
    
    # 生成Store JWT，payload包含店铺ID和权限范围
    store_token_data = StoreTokenData(store_id=store_id, scopes=["store:items"])
    store_token = jwt.encode(store_token_data.dict(), SECRET_KEY, algorithm=ALGORITHM)
    
    return {"access_token": store_token, "token_type": "bearer"}

# 示例：校验用户-店铺权限的辅助函数
def check_user_store_permission(user_id: str, store_id: str) -> bool:
    # 从数据库查询用户是否关联该店铺，返回布尔值
    return True

3. 实现Store Token验证依赖 #

get_current_store负责解码Store JWT、校验权限并返回店铺信息：

async def get_current_store(
    token: Annotated[str, Depends(oauth2_store_scheme)],
    required_scopes: Annotated[List[str], Depends()]
):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无法验证凭证",
        headers={"WWW-Authenticate": "Bearer"},
    )
    
    try:
        # 解码Store JWT
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        store_id: str = payload.get("store_id")
        token_scopes = payload.get("scopes", [])
        
        if store_id is None:
            raise credentials_exception
        
        token_data = StoreTokenData(store_id=store_id, scopes=token_scopes)
    except jwt.JWTError:
        raise credentials_exception
    
    # 校验所需权限是否存在
    for scope in required_scopes:
        if scope not in token_data.scopes:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="权限不足"
            )
    
    # 从数据库获取店铺信息
    store = get_store_by_id(store_id)
    if store is None:
        raise credentials_exception
    
    return store

# 示例：从数据库获取店铺的辅助函数
def get_store_by_id(store_id: str):
    # 返回店铺对象，包含items等属性
    return type('Store', (object,), {"items": ["item1", "item2"], "id": store_id})()

4. Store API接口使用验证 #

用Security依赖get_current_store并指定所需权限：

@app.get("/stores/items/{store_id}")
def get_store_items(
    store: Annotated[object, Depends(get_current_store)]
):
    # 这里的store参数已经是验证后的店铺对象，可直接使用
    return {"store_id": store.id, "items": store.items}

关键注意事项 #

• 令牌区分：可以在JWT payload中加入token_type字段（如"user"或"store"），验证时先判断类型，避免两种令牌混淆
• 权限粒度：根据业务需求细化Store Token的scopes（如"store:items"、"store:orders"）
• 错误处理：确保两个认证流程的错误提示清晰，方便前端做对应处理

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