FastAPI集成Azure AD OAuth2认证：常见问题与解决方案

import os
from authlib.integrations.starlette_client import OAuth

# 从环境变量加载配置
CLIENT_ID = os.getenv("ASPEN_APP_AUTH_CLIENT_ID")
TENANT_ID = os.getenv("ASPEN_APP_AUTH_TENANT_ID")
CLIENT_SECRET = os.getenv("ASPEN_APP_AUTH_SECRET")

# Azure AD 端点
AZURE_AD_AUTHORIZE_URL = f'https://login.microsoftonline.com/{TENANT_ID}/oauth2/v2.0/authorize'
AZURE_AD_TOKEN_ENDPOINT = f'https://login.microsoftonline.com/{TENANT_ID}/oauth2/v2.0/token'
JWKS_URI = f"https://login.microsoftonline.com/{TENANT_ID}/discovery/v2.0/keys"

# 初始化 OAuth
oauth = OAuth()
oauth.register(
    name='azure',
    client_id=CLIENT_ID,
    client_secret=CLIENT_SECRET,
    authorize_url=AZURE_AD_AUTHORIZE_URL,
    token_endpoint=AZURE_AD_TOKEN_ENDPOINT, # 使用 token_endpoint 替代 token_url
    client_kwargs={'scope': 'openid email profile'}, # 确保包含 openid 范围以获取 ID Token
    jwks_uri=JWKS_URI # 必须提供 JWKS URI 以便验证 ID Token
)

from fastapi import FastAPI, Request, HTTPException, status
from fastapi.responses import JSONResponse
from starlette.middleware.sessions import SessionMiddleware

# 假设 app 和 oauth 已在别处初始化

app = FastAPI()
app.add_middleware(SessionMiddleware, secret_key="YOUR_SESSION_SECRET_KEY") # 确保使用一个强随机密钥

# ... (oauth.register 配置如上所示) ...

@app.get("/login")
async def login(request: Request):
    # 在这里生成并存储 nonce 到 session，以便在 /auth 中验证
    # Authlib 通常会自动处理 nonce 的生成和验证，但手动处理可以更灵活
    # 对于 Azure AD，Authlib 可能会从 token 响应中提取 nonce
    redirect_uri = request.url_for('auth')
    return await oauth.azure.authorize_redirect(request, redirect_uri)

@app.get("/auth")
async def auth(request: Request):
    try:
        # 1. 获取访问令牌
        # Authlib 的 authorize_access_token 方法会处理大部分 OAuth2 流程
        token = await oauth.azure.authorize_access_token(request)

        # 2. 从 token 响应中尝试获取 nonce
        # 注意：nonce 通常在认证请求时生成并存储在会话中，然后在此处进行验证。
        # 这里的 userinfo.nonce 是一种特定场景下的获取方式，更标准做法是从 session 获取。
        # 如果 token 字典中直接包含 'nonce'，也可以直接用 token.get('nonce')
        nonce = token.get('userinfo', {}).get('nonce') # 尝试从 userinfo 中获取 nonce

        # 3. 解析 ID Token
        # Authlib 的 parse_id_token 期望 token 参数是一个包含 id_token 字段的字典
        # 并且会使用 nonce 进行验证
        user_info = await oauth.azure.parse_id_token(token=token, nonce=nonce)

        # 认证成功，返回用户信息
        return JSONResponse(content={"user_info": user_info})

    except HTTPException as e:
        # 捕获并重新抛出 Authlib 内部可能抛出的 HTTP 异常
        raise e
    except Exception as e:
        # 捕获其他未知错误，并返回通用认证失败信息
        print(f"Error during authentication: {e}")
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Authentication failed due to an unexpected error.")

# 示例：健康检查路由
@app.get("/")
async def health():
    return JSONResponse(content={"status": "healthy"}, status_code=200)

# 示例：受保护的资源（需要认证）
from fastapi.security import OAuth2AuthorizationCodeBearer, OAuth2PasswordBearer, Security
from authlib.integrations.httpx_client import AsyncOAuth2Client

# 假设 oauth2_scheme 已经定义，用于依赖注入
# oauth2_scheme = OAuth2AuthorizationCodeBearer(...)
# 实际使用时，get_current_user 会通过解析 Access Token 来验证用户
async def get_current_user(request: Request, token: str = Security(oauth2_scheme)):
    try:
        # 注意：这里 oauth.azure.parse_id_token 应该是用于验证 ID Token，
        # 而不是 Access Token。对于 Access Token 验证，通常需要调用 UserInfo 端点
        # 或验证 JWT Access Token 的签名。
        # 如果您的 Access Token 是 JWT 格式且包含用户信息，可以尝试解析。
        # 更严谨的做法是调用 Azure AD 的 /userinfo 端点或验证 JWT 签名。
        # 这里为了演示，我们假设 parse_id_token 能够处理返回的 token
        # 实际生产中，您需要根据 Access Token 的类型进行验证。
        # 例如，如果 Access Token 是 JWT，可以使用 Authlib 的 decode 方法。
        # user_info = await oauth.azure.parse_id_token(request, token) # 此处 token 应为 id_token 字符串
        # 简化示例，假设 token 包含了可解析的用户信息
        return {"authenticated_token": token, "message": "User authenticated via Access Token"}
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=f"Invalid authentication credentials: {e}"
        )

@app.get("/protected")
async def protected_route(user: dict = Depends(get_current_user)):
    return JSONResponse(content={"message": "You accessed a protected route!", "user": user})