1. 为什么需要双Token机制在传统的Web应用中我们通常使用单一的JWT Token进行身份认证。这种方式虽然简单但存在几个明显的安全隐患和用户体验问题安全性问题Token一旦泄露攻击者可以在有效期内无限使用用户体验问题Token过期后用户必须重新登录性能问题频繁签发新Token增加服务器负担双Token机制通过将认证过程拆分为两个部分来解决这些问题Access Token短期有效的令牌通常30分钟用于实际API访问Refresh Token长期有效的令牌通常7天专门用于获取新的Access Token重要提示Refresh Token必须存储在安全的地方如HttpOnly Cookie而Access Token则可以通过前端JavaScript访问。2. JWT双Token的核心实现原理2.1 Token的生成与验证在Python的FastAPI框架中我们可以使用python-jose库来实现JWT的生成和验证from jose import jwt from datetime import datetime, timedelta SECRET_KEY your-256-bit-secret # 建议使用环境变量存储 ALGORITHM HS256 def create_access_token(user_id: int, username: str) - str: 生成30分钟有效的Access Token expire datetime.utcnow() timedelta(minutes30) payload { sub: str(user_id), # JWT标准建议使用sub作为用户标识 username: username, exp: expire, type: access, iat: datetime.utcnow() # 签发时间 } return jwt.encode(payload, SECRET_KEY, algorithmALGORITHM) def create_refresh_token(user_id: int) - str: 生成7天有效的Refresh Token expire datetime.utcnow() timedelta(days7) payload { sub: str(user_id), exp: expire, type: refresh, iat: datetime.utcnow() } return jwt.encode(payload, SECRET_KEY, algorithmALGORITHM)2.2 Token验证的注意事项验证Token时需要特别注意以下几点def verify_token(token: str, token_type: str) - dict: try: payload jwt.decode(token, SECRET_KEY, algorithms[ALGORITHM]) # 检查Token类型是否匹配 if payload.get(type) ! token_type: raise ValueError(f无效的Token类型期望{token_type}) # 检查是否过期jwt.decode会自动检查exp return payload except jwt.ExpiredSignatureError: raise ValueError(Token已过期) except jwt.JWTError as e: raise ValueError(fToken验证失败: {str(e)})3. 完整的认证流程实现3.1 用户登录接口from fastapi import APIRouter, Depends, HTTPException from fastapi.security import OAuth2PasswordRequestForm router APIRouter(prefix/auth, tags[认证]) router.post(/login) async def login(form_data: OAuth2PasswordRequestForm Depends()): # 1. 验证用户凭证 user authenticate_user(form_data.username, form_data.password) if not user: raise HTTPException( status_code401, detail用户名或密码错误, headers{WWW-Authenticate: Bearer}, ) # 2. 生成双Token access_token create_access_token(user.id, user.username) refresh_token create_refresh_token(user.id) # 3. 返回响应 return { access_token: access_token, refresh_token: refresh_token, token_type: bearer }3.2 Token刷新接口router.post(/refresh) async def refresh_token(refresh_token: str Body(..., embedTrue)): try: payload verify_token(refresh_token, refresh) user_id payload[sub] # 检查用户是否存在 user get_user_by_id(user_id) if not user: raise HTTPException(status_code401, detail用户不存在) # 生成新的Access Token new_access_token create_access_token(user.id, user.username) return { access_token: new_access_token, token_type: bearer } except ValueError as e: raise HTTPException(status_code401, detailstr(e))4. 安全防护与最佳实践4.1 Token存储策略Access Token存储在内存或localStorage中适用于SPA应用Refresh Token必须存储在HttpOnly、Secure、SameSiteStrict的Cookie中from fastapi import Response router.post(/login) async def login(..., response: Response): # ...登录逻辑... # 设置Refresh Token Cookie response.set_cookie( keyrefresh_token, valuerefresh_token, httponlyTrue, secureTrue, # 仅HTTPS samesitestrict, max_age60*60*24*7 # 7天 ) return {access_token: access_token, token_type: bearer}4.2 增强安全性的额外措施Token黑名单实现登出功能时将未过期的Token加入黑名单指纹验证在Token中包含客户端指纹防止跨设备使用使用率监控异常频繁的Refresh请求可能是攻击迹象# 增强版的Token生成 def create_access_token(user_id: int, username: str, fingerprint: str) - str: payload { sub: str(user_id), username: username, exp: datetime.utcnow() timedelta(minutes30), type: access, fingerprint: fingerprint # 客户端指纹 } return jwt.encode(payload, SECRET_KEY, algorithmALGORITHM)5. 实际应用中的常见问题与解决方案5.1 并发请求导致的Token刷新问题当Access Token过期时多个并发请求可能同时触发Refresh流程。解决方案import asyncio from fastapi import Request refresh_lock asyncio.Lock() async def refresh_access_token(request: Request, user_id: str): async with refresh_lock: # 检查是否已经有其他请求刷新了Token if new_access_token in request.state: return request.state.new_access_token # 执行刷新逻辑 new_token await do_refresh_token(user_id) request.state.new_access_token new_token return new_token5.2 服务端渲染应用的特殊处理对于SSR应用需要在服务端和客户端都处理Token# 中间件处理 app.middleware(http) async def auth_middleware(request: Request, call_next): # 从Cookie获取Refresh Token refresh_token request.cookies.get(refresh_token) if refresh_token and should_refresh(request): try: new_access_token await refresh_access_token(request, refresh_token) request.state.access_token new_access_token except: pass response await call_next(request) return response5.3 分布式系统的Token验证在微服务架构中所有服务需要共享Token验证逻辑# 使用Redis存储Token黑名单 import redis redis_client redis.Redis(hostlocalhost, port6379, db0) def is_token_revoked(token: str) - bool: return redis_client.exists(frevoked:{token}) 1 router.post(/logout) async def logout(request: Request): access_token get_token_from_request(request) refresh_token request.cookies.get(refresh_token) # 将未过期的Token加入黑名单 if access_token: payload verify_token(access_token, access) remaining payload[exp] - datetime.utcnow().timestamp() if remaining 0: redis_client.setex(frevoked:{access_token}, int(remaining), 1) # 清除Refresh Token Cookie response JSONResponse(content{message: 登出成功}) response.delete_cookie(refresh_token) return response6. 性能优化与扩展思考6.1 无感刷新的前端实现前端可以在Token即将过期时自动刷新let refreshPromise null; async function refreshTokenIfNeeded() { const token getAccessToken(); const payload parseJwt(token); const now Math.floor(Date.now() / 1000); // 提前5分钟刷新 if (payload.exp - now 300) { if (!refreshPromise) { refreshPromise fetch(/auth/refresh, { method: POST, credentials: include }).then(res res.json()) .finally(() refreshPromise null); } return refreshPromise; } return Promise.resolve({ access_token: token }); }6.2 多设备登录管理可以在Refresh Token中包含设备信息def create_refresh_token(user_id: int, device_info: dict) - str: payload { sub: str(user_id), exp: datetime.utcnow() timedelta(days7), type: refresh, device: { id: device_info.get(id), name: device_info.get(name), ip: device_info.get(ip) } } return jwt.encode(payload, SECRET_KEY, algorithmALGORITHM)6.3 监控与告警系统实现Token使用情况的监控from prometheus_client import Counter token_refresh_counter Counter( token_refresh_total, Total number of token refresh requests, [user_id, client_ip] ) router.post(/refresh) async def refresh_token(request: Request, ...): client_ip request.client.host token_refresh_counter.labels(user_iduser_id, client_ipclient_ip).inc() # ...原有逻辑...在实际项目中我遇到过Refresh Token被恶意刷新的情况。通过实现上述的监控系统我们成功识别并阻止了这种攻击。关键是要在Token机制中建立完整的生命周期管理和监控体系而不仅仅是实现基本功能。