JWT 认证实现方法详解(AI 视频下载总结器的无状态鉴权)

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 里别放敏感信息,它只保证防篡改,不保证防偷看。验签流程其实就是三步:

  1. 读 header 里声明的算法;
  2. 用服务端 secret 对 header 和 payload 重新算一遍签名;
  3. 比对是否与 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。

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌抄袭侵权/违法违规的内容, 请发送邮件至 qiqicto@qq.com 举报,一经查实,本站将立刻删除。
赞 (0)
其AI的头像其AI普通用户

相关推荐

返回顶部