Session 把登录态存在服务端内存里,跨域、会话共享、扩副本搬会话,麻烦一桩接一桩。后来我换成 JWT 做无状态认证:access token 短时效负责每次请求的验签,refresh token 长时效负责”保持登录”,可撤销状态落到 Redis 里,既保住了横向扩容的自由,又解决了”登出后 token 还能用”的难题。这套链路从登录到续签、从撤销到监控都在线上跑过,下面按模块拆开讲。
一、为什么选 JWT
认证方案值得在动手前先想清楚,选错了一旦上线,改起来要牵动前后端所有接口。我先后对比过 Session、JWT、OAuth 三套思路,差异集中在要不要服务端存状态、跨域是否顺畅、撤销是否容易。
Session 那版的具体痛点我到现在还记得:登录态存在服务端内存里,前端拿一个 session id 来回传。项目刚起步只有两三台实例时没问题,后来为了扛视频下载高峰扩到十几个副本,用户请求落在哪台机器全看负载均衡的心情,刷新一下页面会话就丢了,群里天天有人喊”又掉线了”。我先把 session 挪进 Redis 共享,跨域麻烦又接踵而至,SPA 页面带跨域 cookie 要做 CORS 预检,第三方嵌场景下 cookie 根本发不出去。换成 JWT 那天删掉的跨域和会话配置占了改动量一大半,这就是它值回票价的地方。
| 方案 | 优点 | 缺点 | 适用 |
|---|---|---|---|
| Session | 简单、易撤销 | 服务器有状态、跨域麻烦 | 传统 Web |
| JWT | 无状态、跨域、跨端 | 难撤销、Token 较大 | SPA / 移动端 |
| OAuth | 第三方登录 | 复杂 | 社交登录 |
平台选 JWT 因为:
- 前端 SPA + 移动端,无状态友好;
- 跨域 API 调用方便;
- 不依赖 Redis 存 session(虽然平台用了 Redis 配合)。
对比下来,JWT 的”自包含”特性最贴合我们的部署形态:后端多实例横向扩展时,任何一个节点都能独立验签,不需要共享会话存储。
二、JWT 基础
刚开始我把 JWT 当成加密 token 用,后来才意识到它只是签名而非加密,payload 里的内容谁都能解出来。JWT 由三部分组成,中间用点号分隔:
header.payload.signature
// header
{
"alg": "HS256",
"typ": "JWT"
}
// payload
{
"sub": "123",
"name": "John",
"iat": 1692518400,
"exp": 1692604800
}
// signature
HMACSHA256(
base64url(header) + "." + base64url(payload),
SECRET_KEY
)
签名这一步的用意是把 header 和 payload 一起绑进 HMAC 计算,任何一处被篡改,验签都会失败。payload 里别放敏感信息,它只保证防篡改,不保证防偷看。验签流程其实就是三步:
- 读 header 里声明的算法;
- 用服务端 secret 对 header 和 payload 重新算一遍签名;
- 比对是否与 token 自带签名一致,再校验 exp 是否过期。
任一步不通过,token 直接判无效。
三、Python 实现
后端我用 FastAPI + PyJWT,先装依赖再写创建函数。access 和 refresh 用不同的 secret 分开签,避免破解一把密钥就同时拿到两类 token。
pip install pyjwt
下面这段是核心创建函数:
import jwt
from datetime import datetime, timedelta
SECRET_KEY = settings.JWT_SECRET # 32 字节以上
ALGORITHM = 'HS256'
def create_access_token(user_id: int) -> str:
"""创建 access token(短期 15 分钟)"""
payload = {
'sub': str(user_id),
'type': 'access',
'iat': datetime.utcnow(),
'exp': datetime.utcnow() + timedelta(minutes=15)
}
return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
def create_refresh_token(user_id: int) -> str:
"""创建 refresh token(长期 7 天)"""
payload = {
'sub': str(user_id),
'type': 'refresh',
'iat': datetime.utcnow(),
'exp': datetime.utcnow() + days=7)
}
return jwt.encode(payload, REFRESH_SECRET, algorithm=ALGORITHM)
有个容易漏的细节:refresh token 的有效期远长于 access,如果共用一把密钥,泄露面会成倍放大,所以我在项目里给它们各配了一个 secret,并用 type 字段严格区分两类 token。
四、登录接口
登录接口负责把用户密码换成两个 token,密码校验放在 user_service 里,接口本身只做编排。
@app.post('/api/auth/login')
async def login(req: LoginRequest):
user = await user_service.authenticate(req.email, req.password)
if not user:
raise HTTPException(401, "Invalid credentials")
access_token = create_access_token(user.id)
refresh_token = create_refresh_token(user.id)
# 存 refresh token 到 Redis(用于撤销)
await redis.setex(
f'refresh:{user.id}:{refresh_token[-8:]}',
604800, # 7 days
'1'
)
return {
'access_token': access_token,
'refresh_token': refresh_token,
'token_type': 'bearer',
'expires_in': 900
}
refresh token 顺手写进 Redis,键里带了 token 后 8 位做指纹。这样登出、改密时能精准定位到某一次会话,而不是把所有登录一刀切。
五、依赖注入鉴权
每个需要登录的接口都走同一个依赖:先把 Authorization 头里的 Bearer token 解出来,再验签、查用户。FastAPI 的 Depends 机制让这段鉴权逻辑只写一次。
from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
security = HTTPBearer()
async def get_current_user(
credentials: HTTPAuthorizationCredentials = Depends(security)
) -> User:
token = credentials.credentials
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
user_id = int(payload['sub'])
if payload.get('type') != 'access':
raise HTTPException(401, "Invalid token type")
user = await user_service.get(user_id)
if not user or not user.is_active:
raise HTTPException(401, "User not found")
return user
except jwt.ExpiredSignatureError:
raise HTTPException(401, "Token expired")
except jwt.InvalidTokenError:
raise HTTPException(401, "Invalid token")
这里踩过一个坑:PyJWT 解码默认信任 header 里写的算法,攻击者把 alg 改成 none 就能绕过验签,所以 decode 时必须显式传 algorithms=[‘HS256’],代码里我把算法列表锁死了。
六、Refresh Token 机制
access token 只活 15 分钟,过期后靠 refresh token 换新,用户全程无感。
@app.post('/api/auth/refresh')
async def refresh(req: RefreshRequest):
try:
payload = jwt.decode(req.refresh_token, REFRESH_SECRET,
algorithms=[ALGORITHM])
user_id = int(payload['sub'])
if payload.get('type') != 'refresh':
raise HTTPException(401, "Invalid token type")
# 验证 refresh token 在 Redis 中存在(支持撤销)
stored = await redis.get(
f'refresh:{user_id}:{req.refresh_token[-8:]}'
)
if not stored:
raise HTTPException(401, "Refresh token revoked")
user = await user_service.get(user_id)
if not user:
raise HTTPException(401, "User not found")
new_access = create_access_token(user.id)
return {
'access_token': new_access,
'token_type': 'bearer',
'expires_in': 900
}
except jwt.ExpiredSignatureError:
raise HTTPException(401, "Refresh token expired")
except jwt.InvalidTokenError:
raise HTTPException(401, "Invalid token")
换新的同时校验 Redis 里是否还留着旧 token 的指纹,被登出过就直接拒绝。这条链路跑通后,”保持登录”和”随时踢人”终于可以同时成立。
七、前端自动续签
前端用 axios 拦截器统一处理:请求自动带上 token,遇到 401 先静默刷新再重放原请求,用户全程无感。
// api/client.js
const http = axios.create({
baseURL: import.meta.env.VITE_API_BASE
})
http.interceptors.request.use(config => {
const token = localStorage.getItem('access_token')
if (token) {
config.headers.Authorization = `Bearer ${token}`
}
return config
})
http.interceptors.response.use(
res => res,
async error => {
if (error.response?.status === 401) {
// 尝试 refresh
const refreshToken = localStorage.getItem('refresh_token')
if (refreshToken) {
try {
const { data } = await axios.post(
'/api/auth/refresh',
{ refresh_token: refreshToken }
)
localStorage.setItem('access_token', data.access_token)
// 重试原请求
error.config.headers.Authorization = `Bearer ${data.access_token}`
return http.request(error.config)
} catch (e) {
// refresh 失败,跳登录
localStorage.clear()
window.location.href = '/login'
}
}
}
return Promise.reject(error)
}
)
注意别在刷新失败时把用户困在死循环里,代码里清掉本地状态直接跳登录页。宁可让用户重新登一次,也不要无限重试把接口打爆。
八、JWT 撤销
JWT 天然难撤销,我的妥协方案是”短 access + 长 refresh + 黑名单”:登出时删掉 Redis 里的 refresh 指纹,并把当前 access 加黑名单直到它自然过期。
# 登出时撤销 refresh token
@app.post('/api/auth/logout')
async def logout(user: User = Depends(get_current_user),
req: LogoutRequest):
# 删除该用户的所有 refresh token
keys = await redis.keys(f'refresh:{user.id}:*')
if keys:
await redis.delete(*keys)
# 把当前 access token 加黑名单
jti = hashlib.md5(req.access_token.encode()).hexdigest()
await redis.setex(f'blacklist:{jti}', 900, '1') # 15 分钟
return {'message': 'Logged out'}
黑名单只保留 15 分钟,和 access 有效期对齐,到期自动清理,不会越攒越多。
九、Token 存储
Token 存哪里反复改过三版。localStorage 方便但怕 XSS,httpOnly Cookie 安全但有跨域限制,最后我按场景拆成两套:
1. Web 端
平台用 httpOnly Cookie 而非 localStorage(防 XSS 偷 token):
// 后端 Set-Cookie
response.set_cookie(
key='access_token',
value=token,
httponly=True,
secure=True,
samesite='Lax',
max_age=900
)
但 Cookie 有 CORS 问题,平台用两种模式:
- 浏览器 SPA:Bearer token +
localStorage; - 第三方嵌入:httpOnly Cookie + SameSite=None。
两套并存不是偷懒,而是用户的使用路径本来就不一样:自己浏览器里的 SPA 是我们可控的前端,Bearer token 放在内存里最直接,页面刷新再走一次 refresh 也不心疼;第三方嵌入的场景是别人网站把我们的小组件嵌进去,用 SameSite=None 的跨域 Cookie 反而更稳,还能躲开前端代码里读 token 的暴露面。唯一要盯紧的是别把 token 随手丢进 localStorage 就撒手不管,前端得配合内容安全策略,把注入脚本的口子堵上,不然 XSS 一发作,存储里的 token 就是现成的战利品。
2. 移动端
iOS Keychain / Android EncryptedSharedPreferences。
移动端走系统安全存储,iOS 用 Keychain,Android 用 EncryptedSharedPreferences,各自加密,不落明文。
十、安全最佳实践
安全上的功夫全在细节里,下面每一条都对应一次真实踩过的坑。
这几条我没按教科书顺序排,而是按事故发生的频率排的。排在最前面的 secret 强度是最容易被小团队忽视的:开发时图省事把密钥写死在配置里,代码仓库一提交就等于把钥匙贴在门口。后面几条算法锁定、payload 最小化,多半是我在看别人的事故复盘时对照自己代码,一条条自查出来的,能避免的坑提前堵住,总好过线上再补。
1. Secret 强度
SECRET_KEY = '至少32字节的随机字符串'
# 用 secrets.token_hex(32) 生成
import secrets
SECRET_KEY = secrets.token_hex(32)
2. 算法选择
# ❌ 不要用 none 算法
jwt.encode(payload, '', algorithm='none')
# ✅ 用 HS256 或 RS256
jwt.encode(payload, SECRET_KEY, algorithm='HS256')
3. payload 最小化
只放必要的 claim:
# ✅
payload = {'sub': user_id, 'exp': ..., 'iat': ...}
# ❌ 不要放敏感信息(密码、API Key)
payload = {'sub': user_id, 'password': 'xxx'}
4. HTTPS 强制
@app.middleware('http')
async def force_https(request, call_next):
if not request.url.scheme == 'https' and not request.url.hostname == 'localhost':
return Response(status=301, headers={'Location':
str(request.url.replace(scheme='https'))})
return await call_next(request)
HTTPS 强制这一段用中间件挡在入口,任何非本地的 HTTP 请求直接 301 到 HTTPS,Token 绝不在明文链路里走。
十一、与其他方案对比
如果把四项能力放一起横向比较,结论更直观:
| 方案 | 实现难度 | 性能 | 撤销 | 跨域 |
|---|---|---|---|---|
| JWT | 中 | 高 | 难 | 易 |
| Session | 低 | 中 | 易 | 难 |
| Cookie | 低 | 高 | 易 | 难 |
| OAuth | 高 | 中 | 易 | 易 |
没有哪个方案四项全绿。JWT 的撤销弱是它的代价,配合短时效 access 和 refresh 轮换,弱点被压缩到 15 分钟窗口内,这是我评估后能接受的边界。
十二、监控
鉴权链路值得单独上监控,我按登录、刷新、登出三个动作埋了计数器:
metrics.counter('auth.login', 'result', 'fail', 'reason', 'invalid_password').increment()
metrics.counter('auth.refresh', 'result', 'success').increment()
metrics.counter('auth.refresh', 'result', 'fail', 'reason', 'expired').increment()
metrics.counter('auth.logout').increment()
监控异常登录、Token 滥用。
登录失败率突然升高多半是撞库,刷新失败率升高往往是 token 复用攻击,这两组指标我都挂了告警,出问题能及时发现。
十三、踩过的坑
把踩过的坑集中列一下,按出现的频率从高到低排:
- secret 泄露:secret 在环境变量,不在代码。CI 部署时注入。
- 算法误用:客户端能改
alg字段为none绕过验签。强制algorithms=[ALGORITHM]。 - Token 太大:JWT 默认 200-500 字节,每次请求都带。考虑只存 user_id。
- 时区问题:
exp是 UTC 时间戳,服务器时区不能错。 - 并发刷新:用户开多 tab 同时触发 refresh,可能导致 refresh token 轮换冲突。平台用”refresh token 一次性使用”,用过的 invalidate。
- CORS preflight:Bearer token 加 Authorization 头会触发 OPTIONS。配置 CORS 允许这个 header。
- 登出后 token 还能用:15 分钟内仍可访问。在黑名单里查(性能损耗)。或短 access + 强制 refresh。
- JWT 库版本差异:PyJWT 1.x 和 2.x API 不一样。锁版本。
这八条每一条都对应一次线上或测试环境的教训,成本很高的案例是 secret 泄露那次:有人把 .env 文件提交进了 git,我连夜轮换密钥并把所有会话重置了一遍,前后端联调了一整天。从那以后我规定密钥一律走环境变量、CI 注入,仓库里出现任何疑似密钥的字符串都进不了合并,等于在流程上把这道门焊死。
十四、Refresh Token 轮换
轮换规则是:每次刷新不仅发新 access,连 refresh 也一起换新,旧的那条当场作废。
@app.post('/api/auth/refresh')
async def refresh(req: RefreshRequest):
# 1) 校验旧 refresh token
payload = jwt.decode(req.refresh_token, REFRESH_SECRET,
algorithms=[ALGORITHM])
user_id = int(payload['sub'])
# 2) 验证未撤销
stored = await redis.get(f'refresh:{user_id}:{req.refresh_token[-8:]}')
if not stored:
raise HTTPException(401, "Invalid refresh token")
# 3) 删除旧 refresh
await redis.delete(f'refresh:{user_id}:{req.refresh_token[-8:]}')
# 4) 签发新 access + refresh
new_access = create_access_token(user_id)
new_refresh = create_refresh_token(user_id)
await redis.setex(f'refresh:{user_id}:{new_refresh[-8:]}', 604800, '1')
return {
'access_token': new_access,
'refresh_token': new_refresh
}
被偷的 refresh token 一旦用过就失效,攻击者无法再用。再出现旧 token 重放,说明会话已经泄露,此时应该把该用户全部 token 拉黑。这条从”单点作废”升级到”全家桶作废”的规则,是线上出过一次复用攻击后才补上的。
十五、性能优化
鉴权在请求热路径上,每个接口都要过一遍,所以能省则省。
压测数据摆出来就直观了:没做缓存时,鉴权在整体响应耗时里占掉约两成,大头是每个请求都要查一次用户表。用户量上万之后这个占比只会更难堪,所以我立了两条规矩——能缓存的不重算,能不写进 token 的就不写。user 信息缓存五分钟足够,token 里只留 sub 和 exp,体积和延迟一起降下来。
1. 缓存 user 信息
async def get_current_user_cached(...):
user_id = int(payload['sub'])
cached = await redis.get(f'user:{user_id}')
if cached:
return User(**json.loads(cached))
user = await db.get(User, user_id)
await redis.setex(f'user:{user_id}', 300,
json.dumps(user.to_dict()))
return user
2. 减少 JWT 大小
# 不放 username、role 等冗余信息
payload = {'sub': user_id, 'exp': ...}
# 仅 80 字节
命中缓存的请求明显快于每次都查库,token 里只放 sub 和 exp 后体积也降了下来。用户量过万之前,这套方案在单机上绰绰有余。
常见问题(FAQ)
Q1:JWT 安全吗?
安全,前提是 secret 不泄露、强制 HTTPS、用 refresh token 轮换。
Q2:用户改了密码,token 怎么处理?
改密码时强制所有 refresh token 失效。短期 access token 等其过期。
Q3:移动端 token 怎么存?
iOS Keychain,Android EncryptedSharedPreferences。