FastAPI集成Keycloak OpenID：实现认证授权与受保护路由

要实现无需重复提交表单的受保护路由和登出功能，核心是从请求头的Authorization字段获取Access Token并验证有效性，而非每次依赖用户名密码表单。以下是针对你现有代码的修改方案：

关键修改说明 #

1. 替换原有依赖，实现从请求头提取并验证Token的逻辑
2. 优化登出流程，直接使用客户端保存的Refresh Token完成操作
3. 添加受保护路由示例，通过依赖自动验证用户身份

修改后的完整代码 #

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import (
    OAuth2PasswordBearer,
    OAuth2PasswordRequestForm,
    OAuth2AuthorizationCodeBearer,
)
from keycloak import KeycloakOpenID, KeycloakAuthenticationError

auth_router = APIRouter(prefix="/auth")

keycloak_url = "http://localhost:8080"
realm_name = "dive-dev"
client_id = "quarkus-be"
client_secret = "74vIzXHvrI5Mnsb15LYYQo4nmMutD3g3"

keycloak_openid = KeycloakOpenID(
    server_url=keycloak_url,
    realm_name=realm_name,
    client_id=client_id,
    client_secret_key=client_secret,
)

# 用于从请求头的Authorization: Bearer <token>中提取Access Token
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

# 获取Keycloak公钥，用于验证Token签名
public_key = f"-----BEGIN PUBLIC KEY-----\n{keycloak_openid.public_key()}\n-----END PUBLIC KEY-----"

async def get_current_user(token: str = Depends(oauth2_scheme)):
    """验证Access Token有效性，返回当前用户信息"""
    try:
        # 解析并验证Token
        token_info = keycloak_openid.decode_token(
            token,
            key=public_key,
            options={
                "verify_signature": True,
                "verify_aud": False,  # 根据业务需求决定是否验证受众
                "exp": True  # 验证Token是否过期
            }
        )
        # 获取用户详细信息
        user_info = keycloak_openid.userinfo(token)
        return user_info
    except KeycloakAuthenticationError as e:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="无效或过期的Token",
            headers={"WWW-Authenticate": "Bearer"},
        ) from e

@auth_router.post("/token", response_model=dict)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    """密码流登录，返回Access Token、Refresh Token及用户信息"""
    try:
        token_response = keycloak_openid.token(
            username=form_data.username,
            password=form_data.password,
            grant_type="password",
        )
        token_response["user_info"] = keycloak_openid.userinfo(token_response["access_token"])
        return token_response
    except KeycloakAuthenticationError as e:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
        ) from e

@auth_router.post("/logout")
async def logout(refresh_token: str):
    """使用Refresh Token登出，无需重复输入用户名密码"""
    try:
        keycloak_openid.logout(refresh_token=refresh_token)
        return {"message": "登出成功"}
    except KeycloakAuthenticationError as e:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="无效的Refresh Token",
        ) from e

@auth_router.get("/protected")
async def protected_route(current_user: dict = Depends(get_current_user)):
    """受保护路由示例，仅持有有效Token的用户可访问"""
    return {
        "message": "这是受保护的路由",
        "current_user": current_user
    }

授权码流（Authorization Code Flow）适配 #

如果需要使用你提到的OAuth2AuthorizationCodeBearer实现标准授权码流（适合前后端分离场景），可添加以下代码：

# 授权码流的Scheme配置
oauth2_auth_code_scheme = OAuth2AuthorizationCodeBearer(
    authorizationUrl=f"{keycloak_url}/realms/{realm_name}/protocol/openid-connect/auth",
    tokenUrl=f"{keycloak_url}/realms/{realm_name}/protocol/openid-connect/token",
)

# 授权码回调端点（前端跳转Keycloak授权页面后，后端接收授权码并交换Token）
@auth_router.get("/callback")
async def auth_callback(code: str):
    token_response = keycloak_openid.token(
        grant_type="authorization_code",
        code=code,
        redirect_uri="http://your-frontend-callback-url"  # 需与Keycloak客户端配置的回调地址一致
    )
    return token_response

内容的提问来源于stack exchange，提问作者Pierre-Alexandre