如何在FastAPI中实现HTTPBasic取Token与Bearer认证双模式

以下是满足你需求的具体代码实现，可直接复用并根据实际环境调整：

核心代码实现 #

from fastapi import FastAPI, Security, HTTPException, status
from fastapi.security import HTTPBearer, HTTPBasic, HTTPBasicCredentials
from jose import JWTError, jwt
from passlib.context import CryptContext
import httpx
import os

# 从环境变量加载配置（生产环境必须用这种方式）
AZURE_TENANT_ID = os.getenv("AZURE_TENANT_ID", "your-tenant-id")
AZURE_CLIENT_ID = os.getenv("AZURE_CLIENT_ID", "your-client-id")
AZURE_CLIENT_SECRET = os.getenv("AZURE_CLIENT_SECRET", "your-client-secret")
AZURE_TOKEN_URL = f"https://login.microsoftonline.com/{AZURE_TENANT_ID}/oauth2/v2.0/token"
ALGORITHM = "RS256"
JWKS_URL = f"https://login.microsoftonline.com/{AZURE_TENANT_ID}/discovery/v2.0/keys"

# 初始化安全组件
bearer_scheme = HTTPBearer()
basic_scheme = HTTPBasic()
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# 模拟用户存储（实际替换为Azure AD用户查询或企业用户数据库）
fake_users_db = {
    "user@example.com": {
        "username": "user@example.com",
        "hashed_password": "$2b$12$EixZaY3sTHK48LbsmX0M/.t0C8xq86e5RcQ6V2XpOvY8eX8eX8eX8",
        "email": "user@example.com"
    }
}

app = FastAPI()

async def verify_azure_id_token(token: str) -> dict:
    """验证Azure ID Token的合法性"""
    try:
        # 获取Azure JWKS公钥
        async with httpx.AsyncClient() as client:
            jwks_resp = await client.get(JWKS_URL)
            jwks = jwks_resp.json()
        
        # 匹配Token头部对应的公钥
        unverified_header = jwt.get_unverified_header(token)
        rsa_key = next(
            (key for key in jwks["keys"] if key["kid"] == unverified_header["kid"]),
            None
        )
        
        if not rsa_key:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="无法找到Token验证密钥",
                headers={"WWW-Authenticate": "Bearer"}
            )
        
        # 解码并验证Token核心字段
        payload = jwt.decode(
            token,
            rsa_key,
            algorithms=[ALGORITHM],
            audience=AZURE_CLIENT_ID,
            issuer=f"https://login.microsoftonline.com/{AZURE_TENANT_ID}/v2.0"
        )
        return payload
    
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="无效的Azure ID Token",
            headers={"WWW-Authenticate": "Bearer"}
        )

async def authenticate_via_basic(credentials: HTTPBasicCredentials) -> dict:
    """通过Basic Auth验证用户，并获取/验证Azure ID Token"""
    # 验证用户名密码
    user = fake_users_db.get(credentials.username)
    if not user or not pwd_context.verify(credentials.password, user["hashed_password"]):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
            headers={"WWW-Authenticate": "Basic"}
        )
    
    # 调用Azure密码授权流获取ID Token
    async with httpx.AsyncClient() as client:
        token_payload = {
            "grant_type": "password",
            "client_id": AZURE_CLIENT_ID,
            "client_secret": AZURE_CLIENT_SECRET,
            "username": credentials.username,
            "password": credentials.password,
            "scope": f"openid profile email {AZURE_CLIENT_ID}/.default"
        }
        token_resp = await client.post(AZURE_TOKEN_URL, data=token_payload)
        
        if token_resp.status_code != 200:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Azure Token获取失败",
                headers={"WWW-Authenticate": "Basic"}
            )
    
    # 验证获取到的ID Token
    return await verify_azure_id_token(token_resp.json()["id_token"])

async def get_authenticated_user(
    bearer_creds: HTTPBearer = Security(bearer_scheme),
    basic_creds: HTTPBasicCredentials = Security(basic_scheme)
) -> dict:
    """自定义认证依赖：优先用Bearer Token，失败则降级到Basic Auth"""
    # 先尝试Bearer Token认证
    try:
        return await verify_azure_id_token(bearer_creds.credentials)
    except HTTPException:
        # Bearer认证失败，尝试Basic Auth
        try:
            return await authenticate_via_basic(basic_creds)
        except HTTPException:
            # 两种认证都失败
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="认证失败，请提供有效的Bearer Token或用户名密码",
                headers={"WWW-Authenticate": "Bearer, Basic"}
            )

# 受保护的API示例
@app.get("/api/protected")
async def protected_resource(current_user: dict = Security(get_authenticated_user)):
    return {"message": "资源访问成功", "user_info": current_user}

# 单独的登录接口（可选：用于用户获取Token后用curl调用）
@app.post("/api/login")
async def get_token(credentials: HTTPBasicCredentials = Security(basic_scheme)):
    user = fake_users_db.get(credentials.username)
    if not user or not pwd_context.verify(credentials.password, user["hashed_password"]):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
            headers={"WWW-Authenticate": "Basic"}
        )
    
    async with httpx.AsyncClient() as client:
        token_payload = {
            "grant_type": "password",
            "client_id": AZURE_CLIENT_ID,
            "client_secret": AZURE_CLIENT_SECRET,
            "username": credentials.username,
            "password": credentials.password,
            "scope": f"openid profile email {AZURE_CLIENT_ID}/.default"
        }
        token_resp = await client.post(AZURE_TOKEN_URL, data=token_payload)
        
        if token_resp.status_code != 200:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Azure Token获取失败"
            )
    
    return token_resp.json()

关键逻辑说明 #

1. 双认证共存实现：

   • 自定义依赖get_authenticated_user同时接收Bearer和Basic两种安全凭证
   • 优先验证外部传入的Azure ID Token，失败后自动尝试Basic Auth流程
   • Swagger UI会自动识别两种认证方式，用户可在界面上选择输入Bearer Token或用户名密码

2. Azure ID Token验证：

   • 通过Azure官方JWKS端点动态获取公钥，避免硬编码公钥导致的维护问题
   • 严格验证Token的签名、受众（aud）、发行方（iss）等核心字段，确保Token合法有效

3. Basic Auth流程：

   • 先验证用户名密码（示例用模拟数据库，实际可对接Azure AD用户验证接口）
   • 调用Azure密码授权流获取ID Token，再对Token进行二次验证，确保符合认证要求

注意事项 #

• 密码授权流需要在Azure AD应用中提前启用，且应用需配置允许该授权方式
• 敏感配置（租户ID、客户端ID等）必须用环境变量管理，禁止硬编码到代码中
• 模拟用户数据库仅作示例，实际应对接企业用户系统或Azure AD用户查询接口
• 生产环境建议添加Token缓存机制，减少重复调用Azure接口的次数

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