FastAPI+Authlib+Keycloak多K8s副本部署的认证state参数处理

1. 移除硬编码的state参数 #

你当前代码里手动指定固定state="toto"的做法既无法防范CSRF攻击，也不符合多副本部署的需求。Authlib的authorize_redirect方法默认会自动生成随机state值，并将其存储到请求的session中，后续回调时会自动完成state验证。

修改代码如下：

• 删除oauth.register中的state="toto"参数
• 调整login路由，移除手动传入的state参数：

@router.get("/login", tags=["user"])
async def login(request: Request):
    """将用户重定向到Keycloak登录页面"""
    redirect_uri = request.url_for("auth")
    # 交由Authlib自动生成并管理state
    return await oauth_client.authorize_redirect(request, redirect_uri)

2. 配置共享会话存储（核心步骤） #

Starlette/FastAPI默认的会话存储是内存级别的，多副本部署下不同Pod的会话数据相互隔离，导致/auth回调时无法读取/login阶段存储的state，进而验证失败。因此必须配置分布式会话存储（比如Redis），让所有K8s副本共享同一份会话数据源。

安装依赖 #

pip install redis itsdangerous

替换SessionMiddleware为Redis后端 #

在FastAPI应用初始化时，配置基于Redis的会话存储：

from fastapi import FastAPI
from starlette.middleware.sessions import SessionMiddleware
from redis import asyncio as aioredis
from starlette.sessions import SessionBackend, SessionData
from itsdangerous import URLSafeSerializer
from typing import Optional

class RedisSessionBackend(SessionBackend):
    def __init__(self, redis: aioredis.Redis, secret_key: str):
        self.redis = redis
        self.serializer = URLSafeSerializer(secret_key)

    async def read_session(self, session_id: str) -> Optional[SessionData]:
        data = await self.redis.get(f"session:{session_id}")
        if not data:
            return {}
        try:
            return self.serializer.loads(data)
        except Exception:
            return {}

    async def write_session(self, session_id: str, data: SessionData) -> str:
        serialized = self.serializer.dumps(data)
        await self.redis.setex(f"session:{session_id}", 86400, serialized)  # 设置1天过期
        return session_id

    async def delete_session(self, session_id: str) -> None:
        await self.redis.delete(f"session:{session_id}")

async def create_redis_client():
    # 生产环境通过环境变量注入Redis地址
    return aioredis.from_url("redis://your-redis-service:6379")

app = FastAPI()

# 初始化Redis客户端
@app.on_event("startup")
async def startup():
    app.state.redis = await create_redis_client()

@app.on_event("shutdown")
async def shutdown():
    await app.state.redis.close()

# 添加SessionMiddleware，使用Redis作为会话后端
app.add_middleware(
    SessionMiddleware,
    secret_key="your-strong-production-secret",  # 生产环境用K8s Secret管理
    session_backend=RedisSessionBackend(app.state.redis, "your-strong-production-secret")
)

注意：生产环境中，secret_key和Redis连接地址必须通过K8s ConfigMap/Secret注入，禁止硬编码。

3. 确认Keycloak重定向URL配置 #

确保Keycloak客户端的Valid Redirect URIs配置为服务的外部统一入口（比如K8s Ingress的域名：https://your-service-domain/auth），而非单个Pod的IP。这样无论请求转发到哪个副本，Keycloak的回调都会回到统一入口，再由K8s负载均衡到任意副本，而副本通过共享Redis会话能读取到正确的state。

4. 验证流程逻辑 #

1. 用户访问/login，任意Pod处理请求，Authlib生成随机state并存储到Redis会话，随后重定向到Keycloak。
2. 用户在Keycloak完成登录后，回调请求打到服务统一入口，K8s将其转发到任意Pod。
3. 处理/auth的Pod从Redis会话中读取之前存储的state，与Keycloak返回的state对比验证，通过后获取token。
4. 验证通过后，将用户信息写入Redis会话，完成登录流程。

额外安全建议 #

• 为Redis实例启用密码认证和网络隔离，避免会话数据泄露。
• 根据业务需求配置合理的会话过期时间，降低会话劫持风险。
• 生产环境强制使用HTTPS，防止state和会话Cookie被明文传输。

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