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

FastAPI中OAuth2+JWT请求时自动延长Token有效期方案问询

FastAPI JWT自动续期问题解答

问题背景

基于FastAPI官方OAuth2+JWT登录示例,当前Token有效期为30分钟,需求是:用户每次请求任意接口时,自动将Token有效期延长至当前时间+10分钟,避免活跃用户被意外登出。

咨询问题

  1. 是否可以不依赖数据库存储用户信息,仅通过客户端浏览器存储JWT实现该功能?
  2. 此类需求的最佳实践是什么?是每次请求更新Authorization头,还是设置Cookie?

未成功的尝试代码

尝试修改get_current_user函数实现续期,但未生效:

async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception

    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception

    # Check if the token needs to be refreshed
    expiration_time = datetime.fromtimestamp(payload["exp"], tz=timezone.utc)
    if expiration_time - datetime.now(tz=timezone.utc) < timedelta(minutes=REFRESH_INTERVAL_MINUTES):
        # Refresh the token with a new expiration time
        access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
        new_access_token = create_access_token(
            data={"sub": user.username}, expires_delta=access_token_expires
        )
        return user, new_access_token

    return user , token


async def get_current_active_user(
    current_user_and_token: Annotated[User, Depends(get_current_user)]
):
    current_user,_ = current_user_and_token
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

问题解答

1. 能否仅靠客户端存储JWT实现续期?

不行。JWT是无状态的,一旦签发就无法修改其内置的过期时间,客户端存储的Token本身无法被修改。要实现续期必须由服务端重新签发新Token,而签发新Token时需要验证原Token的合法性——这一步要么依赖数据库校验用户状态(比如账号是否被禁用),要么完全信任原Token的payload,但后者存在安全隐患:如果用户账号被禁用,原Token在过期前仍能正常使用。

如果完全不依赖数据库,只能在原Token未过期时基于其payload重新签发新Token,但这种方式无法校验用户当前状态,安全性极低,不推荐使用。

2. 续期需求的最佳实践

推荐采用Refresh Token + 短时效Access Token的组合方案:

  • Access Token:时效设为10-15分钟,放在Authorization头中用于接口鉴权
  • Refresh Token:时效设为较长周期(比如7天),放在HttpOnly、Secure属性的Cookie中,专门用于换取新的Access Token

当Access Token过期时,客户端自动调用/refresh-token接口,用Refresh Token获取新的Access Token,无需用户重新登录。这种方案无需每次请求都续期,仅在Access Token快过期时触发刷新,逻辑更清晰,安全性更高。

如果一定要实现每次请求自动续期,两种方案对比:

  • 更新Authorization头:服务端在响应中返回新Token,客户端收到后更新本地存储,下次请求携带新Token。缺点是客户端需要额外处理逻辑,且并发请求时可能出现Token覆盖问题。
  • 设置Cookie:将Access Token放在HttpOnly Cookie中,服务端每次验证后自动更新Cookie的过期时间。优点是客户端无需额外处理,且HttpOnly Cookie能避免XSS攻击窃取Token,缺点是跨域场景下需要配置Cookie跨域规则。

代码修改方案(实现每次请求自动续期)

你的尝试代码问题在于:get_current_user返回了用户和新Token,但后续依赖只处理了用户,且接口响应未将新Token返回给客户端。以下是修正后的完整方案:

步骤1:新增返回模型,统一封装用户和新Token

from pydantic import BaseModel

class UserWithToken(BaseModel):
    user: User
    new_access_token: str | None = None

步骤2:修改依赖函数,返回封装后的对象

# 新增配置常量
REFRESH_INTERVAL_MINUTES = 5  # Token剩余有效期小于5分钟时触发续期
ACCESS_TOKEN_EXPIRE_MINUTES = 10  # 续期后新Token有效期为10分钟

async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception

    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception

    # 检查是否需要续期
    expiration_time = datetime.fromtimestamp(payload["exp"], tz=timezone.utc)
    time_remaining = expiration_time - datetime.now(tz=timezone.utc)
    new_access_token = None
    if time_remaining < timedelta(minutes=REFRESH_INTERVAL_MINUTES):
        access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
        new_access_token = create_access_token(
            data={"sub": user.username}, expires_delta=access_token_expires
        )

    return UserWithToken(user=User(**user.dict()), new_access_token=new_access_token)

async def get_current_active_user(
    user_with_token: Annotated[UserWithToken, Depends(get_current_user)]
):
    if user_with_token.user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return user_with_token

步骤3:修改接口,在响应头返回新Token

以/users/me/接口为例:

from fastapi.responses import JSONResponse

@app.get("/users/me/", response_model=User)
async def read_users_me(
    user_with_token: Annotated[UserWithToken, Depends(get_current_active_user)]
):
    headers = {}
    # 如果生成了新Token,在响应头中返回
    if user_with_token.new_access_token:
        headers["X-New-Access-Token"] = user_with_token.new_access_token
    return JSONResponse(content=user_with_token.user.dict(), headers=headers)

客户端处理逻辑

客户端需要在每次请求后检查响应头中的X-New-Access-Token字段,若存在则更新本地存储的Token,下次请求时使用新Token发起请求。


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

相关产品推荐
方舟 Agent Plan

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

最近更新时间:2026.06.27 17:34:55