如何高效处理422无效token：从原理到实战的解决方案

1次阅读

没有评论

共计 2462 个字符，预计需要花费 7 分钟才能阅读完成。

422 无效 token 错误是 API 开发中常见的身份验证问题，尤其在分布式系统中更为棘手。当客户端携带的 token 无效或过期时，服务器会返回 422 状态码，表示请求无法处理。这种情况通常发生在以下几种场景：

Token 过期：JWT 等 token 通常有有效期，过期后需要重新获取
Token 被撤销：用户主动注销或管理员禁用账户
Token 被篡改：客户端传递的 token 被恶意修改

这些情况会导致用户体验下降，增加服务器无效请求处理负担，甚至可能引发安全问题。

在处理 token 验证方面，常见的技术方案有 JWT、OAuth 和 Session 等，它们各有优缺点：

JWT：
优点：无状态、轻量级、易于分布式部署
缺点：一旦签发难以撤销，需要额外的刷新机制
OAuth：
优点：标准化流程、支持第三方授权
缺点：实现复杂、性能开销较大
Session：
优点：易于管理、可即时失效
缺点：有状态、不适合分布式系统

对于大多数现代 Web 应用，JWT 结合 Redis 的方案提供了较好的平衡。

下面是一个基于 Python 的 JWT 和 Redis 实现方案，包含 token 刷新机制：

import jwt
import redis
from datetime import datetime, timedelta

# 配置
SECRET_KEY = 'your-secret-key'
REDIS_HOST = 'localhost'
REDIS_PORT = 6379
ACCESS_TOKEN_EXPIRE = timedelta(minutes=15)
REFRESH_TOKEN_EXPIRE = timedelta(days=7)

# 连接 Redis
redis_client = redis.StrictRedis(host=REDIS_HOST, port=REDIS_PORT)

def generate_tokens(user_id):
    """生成 access token 和 refresh token"""
    access_payload = {
        'user_id': user_id,
        'exp': datetime.utcnow() + ACCESS_TOKEN_EXPIRE,
        'type': 'access'
    }

    refresh_payload = {
        'user_id': user_id,
        'exp': datetime.utcnow() + REFRESH_TOKEN_EXPIRE,
        'type': 'refresh'
    }

    access_token = jwt.encode(access_payload, SECRET_KEY, algorithm='HS256')
    refresh_token = jwt.encode(refresh_payload, SECRET_KEY, algorithm='HS256')

    # 将 refresh token 存入 Redis
    redis_client.set(f'refresh_token:{user_id}', refresh_token, ex=int(REFRESH_TOKEN_EXPIRE.total_seconds()))

    return {
        'access_token': access_token,
        'refresh_token': refresh_token
    }

def refresh_access_token(refresh_token):
    """使用 refresh token 获取新的 access token"""
    try:
        payload = jwt.decode(refresh_token, SECRET_KEY, algorithms=['HS256'])

        if payload['type'] != 'refresh':
            raise jwt.InvalidTokenError('Invalid token type')

        # 检查 Redis 中的 refresh token 是否匹配
        stored_refresh = redis_client.get(f'refresh_token:{payload["user_id"]}')
        if not stored_refresh or stored_refresh.decode() != refresh_token:
            raise jwt.InvalidTokenError('Refresh token mismatch')

        # 生成新的 access token
        return generate_tokens(payload['user_id'])

    except jwt.ExpiredSignatureError:
        raise jwt.InvalidTokenError('Refresh token expired')
    except jwt.InvalidTokenError:
        raise

在实现 token 刷新机制时，需要考虑以下关键点：