FastAPI中OAuth2+JWT请求时自动延长Token有效期方案问询
问题背景
基于FastAPI官方OAuth2+JWT登录示例,当前Token有效期为30分钟,需求是:用户每次请求任意接口时,自动将Token有效期延长至当前时间+10分钟,避免活跃用户被意外登出。
咨询问题
- 是否可以不依赖数据库存储用户信息,仅通过客户端浏览器存储JWT实现该功能?
- 此类需求的最佳实践是什么?是每次请求更新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放在
HttpOnlyCookie中,服务端每次验证后自动更新Cookie的过期时间。优点是客户端无需额外处理,且HttpOnlyCookie能避免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

