将Keycloak与FastAPI集成：替换自定义JWT认证方案

一、Keycloak基础配置 #

先在Keycloak后台完成核心配置：

• 创建专属Realm（例如命名为my-app-realm）
• 为SvelteKit创建公开客户端：Access Type选public，Valid Redirect URIs设为SvelteKit的回调地址（本地开发用http://localhost:5173/auth/callback，生产替换为实际域名），Web Origins设为SvelteKit的域名（如http://localhost:5173）
• 为FastAPI创建保密客户端：Access Type选confidential，记录客户端ID和秘钥，后续FastAPI会用它验证Keycloak令牌

二、SvelteKit端：对接Keycloak获取令牌 #

用OAuth2授权码流程实现跳转登录，推荐用@auth/sveltekit简化开发：

1. 安装依赖：

npm install @auth/sveltekit @auth/core

2. 在src/hooks.server.ts配置认证逻辑：

import { SvelteKitAuth } from "@auth/sveltekit";
import Keycloak from "@auth/core/providers/keycloak";
import { env } from "$env/dynamic/private";

export const handle = SvelteKitAuth({
  providers: [
    Keycloak({
      clientId: env.KEYCLOAK_SVELTE_CLIENT_ID,
      clientSecret: "", // 公开客户端无需秘钥
      issuer: env.KEYCLOAK_ISSUER, // 格式：http://keycloak地址/realms/你的Realm名
    }),
  ],
  callbacks: {
    async jwt({ token, account }) {
      // 缓存Keycloak的access_token，用于后续传给FastAPI
      if (account) token.keycloakToken = account.access_token;
      return token;
    },
    async session({ session, token }) {
      session.keycloakToken = token.keycloakToken as string;
      return session;
    },
  },
});

3. 在页面中添加登录触发逻辑：调用signIn("keycloak")即可跳转到Keycloak登录页，登录成功后会自动回调并拿到Keycloak令牌。

三、FastAPI端：验证Keycloak令牌并生成自有JWT #

1. 安装依赖：

pip install python-jose[cryptography] fastapi python-multipart

2. 核心代码实现：

from fastapi import FastAPI, Depends, HTTPException
from jose import JWTError, jwt
from datetime import datetime, timedelta
from pydantic import BaseModel
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()

# 配置项，根据实际环境替换
KEYCLOAK_ISSUER = "http://keycloak:8080/realms/my-app-realm"
# 从Keycloak的Realm设置页面获取公钥，地址：/realms/你的Realm名/protocol/openid-connect/certs
KEYCLOAK_PUBLIC_KEY = "-----BEGIN PUBLIC KEY-----\n你的公钥内容\n-----END PUBLIC KEY-----"
FASTAPI_JWT_SECRET = "your-fastapi-jwt-secret-key"
FASTAPI_JWT_ALGORITHM = "HS256"
FASTAPI_JWT_EXPIRE_MINUTES = 30

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

class KeycloakTokenRequest(BaseModel):
    keycloak_token: str

# 验证Keycloak令牌的依赖函数
def verify_keycloak_token(token: str):
    try:
        payload = jwt.decode(
            token,
            KEYCLOAK_PUBLIC_KEY,
            algorithms=["RS256"],
            issuer=KEYCLOAK_ISSUER,
            options={"verify_aud": False}  # 无需验证aud时开启，按需调整
        )
        # 可在此处添加角色、权限校验逻辑
        return payload
    except JWTError:
        raise HTTPException(status_code=401, detail="无效的Keycloak令牌")

# 生成FastAPI自有JWT
def create_fastapi_token(user_data: dict):
    to_encode = user_data.copy()
    expire = datetime.utcnow() + timedelta(minutes=FASTAPI_JWT_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, FASTAPI_JWT_SECRET, algorithm=FASTAPI_JWT_ALGORITHM)

# 令牌交换接口：接收Keycloak令牌，返回FastAPI自有令牌
@app.post("/api/exchange-token")
async def exchange_token(request: KeycloakTokenRequest, keycloak_payload: dict = Depends(verify_keycloak_token)):
    # 从Keycloak的令牌载荷中提取用户核心信息
    user_info = {
        "sub": keycloak_payload.get("sub"),
        "email": keycloak_payload.get("email"),
        "roles": keycloak_payload.get("realm_access", {}).get("roles", [])
    }
    fastapi_token = create_fastapi_token(user_info)
    return {"access_token": fastapi_token, "token_type": "bearer"}

# 验证FastAPI令牌的依赖函数
def verify_fastapi_token(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, FASTAPI_JWT_SECRET, algorithms=[FASTAPI_JWT_ALGORITHM])
        return payload
    except JWTError:
        raise HTTPException(status_code=401, detail="无效的FastAPI令牌")

# 示例受保护接口
@app.get("/api/protected")
async def protected_route(user_payload: dict = Depends(verify_fastapi_token)):
    return {"message": "这是受保护接口", "用户信息": user_payload}

四、SvelteKit端：存储FastAPI令牌并发起请求 #

1. 调用FastAPI的/api/exchange-token接口，传入Keycloak令牌，拿到FastAPI自有令牌后，推荐存储在HttpOnly Cookie中（安全性更高）。
2. 后续发起请求时，自动在请求头携带Authorization: Bearer {fastapi_token}，可通过SvelteKit的handleFetch全局拦截实现：

import type { HandleFetch } from "@sveltejs/kit";
import { getSession } from "@auth/sveltekit";

export const handleFetch: HandleFetch = async ({ request, fetch, event }) => {
    if (request.url.startsWith("http://fastapi:8000/api/")) {
        const session = await getSession(event);
        if (session?.fastapiToken) {
            request.headers.set("Authorization", `Bearer ${session.fastapiToken}`);
        }
    }
    return fetch(request);
};

五、容器部署注意事项 #

• 确保Keycloak、FastAPI、SvelteKit容器在同一网络下，FastAPI和SvelteKit能通过容器服务名访问Keycloak（例如http://keycloak:8080）
• 生产环境中，SvelteKit的回调地址和Web Origins要设置为实际域名，避免跨域问题
• Keycloak的客户端配置要严格匹配生产环境的域名和端口

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