You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

FastAPI集成Authlib OAuth生成JWT后无法重定向至根路径的求助

问题描述

我有如下main.py文件,在/auth端点生成JWT token,但生成后无法重定向到根路径("/")。当尝试访问根路径时,会被重定向到/auth端点并显示Bearer token,希望获取实现正确重定向的方法或指引。

现有代码
from authlib.integrations.starlette_client import OAuth
oauth = OAuth()

CONF_URL = "https://localhost:8080/.well-known/openid-configuration"
oauth.register(
    name="cad",
    server_metadata_url=CONF_URL,
    client_id=settings.CLIENT_ID,
    client_secret=settings.CLIENT_SECRET,
    client_kwargs={"scope": "openid email profile authorization_group"},
)
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='/auth')

def create_access_token(*, data: dict, expires_delta: datetime.timedelta = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.datetime.utcnow() + expires_delta
    else:
        expire = datetime.datetime.utcnow() + datetime.timedelta(minutes=15)
    to_encode.update({'exp': expire})
    encoded_jwt = jwt.encode(to_encode, "abcd", algorithm="HS256")
    return encoded_jwt


def create_token(id):
    access_token_expires = datetime.timedelta(minutes=120)
    access_token = create_access_token(data={'sub': id}, expires_delta=access_token_expires)
    return access_token


@app.middleware("http")
async def authorize(request: Request, call_next):
    if not (request.scope["path"].startswith("/login") or request.scope["path"].startswith("/auth")):
        if not is_session_okay(request.session):
            return RedirectResponse(url="/login")
    return await call_next(request)


@app.get("/login")
async def login(request: Request):
    redirect_uri = request.url_for("auth")
    return await oauth.cad.authorize_redirect(request, redirect_uri)


@app.get("/auth")
async def auth(request: Request):
    try:
        token = await oauth.cad.authorize_access_token(request)
    except OAuthError as error:
        return HTMLResponse(f"<h1>{error.error}</h1>")
    user = await oauth.cad.parse_id_token(request, token)
    access_token = create_token(user['sub'])
    return {"access_token": access_token, "token_type": "bearer"}


@app.get("/", tags=["Web-UI"])
def index():
    frontend_root = "./ui"
    return FileResponse(str(frontend_root) + "/index.html", media_type="text/html")
解决方案

问题核心在于/auth直接返回token但未标记用户登录状态,中间件一直判定会话无效,导致循环重定向;同时is_session_okay逻辑缺失或未正确校验会话状态。以下是具体修正步骤:

1. 修改/auth端点,存储token并重定向

生成JWT后,将其存入会话,再重定向到根路径,让中间件识别用户已登录:

@app.get("/auth")
async def auth(request: Request):
    try:
        token = await oauth.cad.authorize_access_token(request)
    except OAuthError as error:
        return HTMLResponse(f"<h1>{error.error}</h1>")
    user = await oauth.cad.parse_id_token(request, token)
    access_token = create_token(user['sub'])
    # 将token存入会话,标记登录状态
    request.session['access_token'] = access_token
    # 重定向到根路径
    return RedirectResponse(url="/")

2. 完善is_session_okay校验逻辑

确保该函数能正确验证会话中的token有效性:

def is_session_okay(session):
    access_token = session.get('access_token')
    if not access_token:
        return False
    try:
        # 解码token,验证签名和过期时间
        jwt.decode(access_token, "abcd", algorithms=["HS256"])
        return True
    except jwt.PyJWTError:
        # token无效则清除会话,避免无效状态残留
        session.pop('access_token', None)
        return False

3. 可选:改用HttpOnly Cookie存储token(更安全)

若不想依赖服务端会话,可将token存入HttpOnly Cookie,防止XSS攻击,同时适配前后端分离场景:

from starlette.responses import RedirectResponse

@app.get("/auth")
async def auth(request: Request):
    # ... 其他逻辑不变
    access_token = create_token(user['sub'])
    response = RedirectResponse(url="/")
    # 设置Cookie,有效期与token保持一致
    response.set_cookie(
        key="access_token",
        value=access_token,
        httponly=True,
        max_age=120*60,  # 120分钟
        secure=True,  # 生产环境开启,仅HTTPS传输
        samesite="lax"
    )
    return response

同时修改中间件和is_session_okay,改为从request.cookies中获取token校验:

async def authorize(request: Request, call_next):
    if not (request.scope["path"].startswith("/login") or request.scope["path"].startswith("/auth")):
        if not is_session_okay(request.cookies):
            return RedirectResponse(url="/login")
    return await call_next(request)

def is_session_okay(cookies):
    access_token = cookies.get('access_token')
    # 后续校验逻辑同前
关键说明
  • 必须先完成登录状态的存储(会话或Cookie),再重定向,否则中间件会一直判定用户未登录,触发循环重定向。
  • is_session_okay是登录校验的核心,需确保能准确识别有效/无效的登录状态。
  • 会话存储适合服务端渲染应用,Cookie存储更适合前后端分离架构,按需选择即可。

内容的提问来源于stack exchange,提问作者Deb

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.24 21:24:20