Flask如何用PyJWT实现用户登录token验证?

文章导读
用户登录后需要服务端签发一个 token,后续请求携带这个 token 来证明身份。但经常遇到的问题是:token 生成后客户端用不了,或者很快过期,或者服务端解析报错。排查时我会先确认两件事:一是 token 里到底放了什么信息,二是验证时有没有正确处理异常。很多问题出在 payload 选择、密钥硬编码和异常捕获上。
📋 目录
  1. 先确认现象
  2. 容易误判的地方
  3. 建议的实现顺序
  4. 配置与代码示例
  5. 验证方法与边界情况
  6. 回滚与风险
A A

先确认现象

用户登录后需要服务端签发一个 token,后续请求携带这个 token 来证明身份。但经常遇到的问题是:token 生成后客户端用不了,或者很快过期,或者服务端解析报错。排查时我会先确认两件事:一是 token 里到底放了什么信息,二是验证时有没有正确处理异常。很多问题出在 payload 选择、密钥硬编码和异常捕获上。

容易误判的地方

第一个常见误判是以为 token 里放密码也没问题。实际上 PyJWT 只签名不加密,payload 里任何内容都可以被 base64 解码看到。用户登录成功后,服务端使用 PyJWT 生成 token。需要将用户唯一标识(如用户 ID)作为 payload,并设置过期时间(exp)和签发时间(iat)。注意不要将敏感信息如密码放入 payload,因为 token 仅签名未加密。签名时需指定算法(推荐 HS256)和密钥,密钥应从环境变量读取而非硬编码。第二个误判是验证时只检查签名,忽略过期时间。许多新手写死一个 decode 调用,不捕获异常。正确的做法是在每个受保护的路由前,通过 Flask 请求装饰器或中间件验证 token。从请求头获取 Authorization 字段(格式为 Bearer ),用 PyJWT 的 decode 方法解析。需捕获过期签名异常(如 ExpiredSignatureError、InvalidTokenError),并返回 401 状态码。若验证成功,将用户信息存入 g 对象供后续使用。

建议的实现顺序

我会先搭建一个最简单的登录接口,只做生成 token 和验证 token 两个动作,不急着加刷新和吊销。顺序如下:

  1. 设置环境变量 SECRET_KEY,长度至少 32 字节,建议用 os.urandom(64) 生成。
  2. 写一个 create_access_token 函数,payload 里只放 user_id、exp 和 iat。exp 用 datetime.utcnow() + timedelta(hours=2) 设置短有效期。
  3. 写一个装饰器 @require_auth,从请求头取 Authorization,调用 decode 并捕获异常。如果失败返回 401 JSON 响应。
  4. 在受保护的路由上应用装饰器,从 g.user_id 获取当前用户。

完成这四步后,用 curl 测试一下:先调登录接口拿到 token,再调受保护接口带上 Authorization: Bearer ,确认能正常获取数据和返回 401。

配置与代码示例

生成 token 的示例:

Flask如何用PyJWT实现用户登录token验证?
import jwt
from datetime import datetime, timedelta
from flask import current_app

def create_token(user_id):
    payload = {
        'user_id': user_id,
        'iat': datetime.utcnow(),
        'exp': datetime.utcnow() + timedelta(hours=2)
    }
    token = jwt.encode(payload, current_app.config['SECRET_KEY'], algorithm='HS256')
    return token

验证装饰器:

from functools import wraps
from flask import request, jsonify, g
import jwt

def require_auth(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        auth_header = request.headers.get('Authorization', '')
        if not auth_header.startswith('Bearer '):
            return jsonify({'error': 'Missing or invalid token'}), 401
        token = auth_header.split(' ')[1]
        try:
            data = jwt.decode(token, current_app.config['SECRET_KEY'], algorithms=['HS256'])
            g.user_id = data['user_id']
        except jwt.ExpiredSignatureError:
            return jsonify({'error': 'Token expired'}), 401
        except jwt.InvalidTokenError:
            return jsonify({'error': 'Invalid token'}), 401
        return f(*args, **kwargs)
    return decorated

注意 decode 时 algorithms 参数必须写成列表 ['HS256'],这是 PyJWT 在 2.0 之后的强制要求,否则会有警告且可能未来抛异常。

验证方法与边界情况

写完代码后,验证要点:

  • 用试过期的 token 访问,应返回 401 且提示 token 过期。
  • 用错误的密钥生成 token(比如改一下 SECRET_KEY),服务端 decode 应返回 invalid token。
  • 不使用 Bearer 前缀或直接放 token 在请求头其他位置都会触发 401。
  • 如果用了多个服务共享密钥,需要在 payload 里加 aud(受众)和 iss(签发者),并在 decode 时校验这两个声明,防止一个服务的 token 被另一个服务冒用。使用 PyJWT 时,务必指定 algorithms 参数为列表形式(如 ['HS256']),否则可能收到警告。密钥长度应足够(HS256 推荐 32 字节以上),避免弱密钥被暴力破解。不要忽略 token 的 aud(受众)和 iss(签发者)声明,尤其在多服务间共享密钥时,防止跨服务冒用。另外,不要将 JWT 放在 URL 参数中,推荐放在 Header 或 Cookie(需设置 HttpOnly)。

回滚与风险

如果上线后发现验证逻辑有问题,比如密钥变更导致所有在线用户 token 失效,需要紧急回滚。最简单的回滚是直接把 SECRET_KEY 改回原来的值,所有 token 重新有效。如果是因为算法写错导致 decode 失败,可以立刻恢复代码。另一个风险是忘记捕获异常导致 500 错误,新手常犯。如果上线前测试不充分,可以先在少量路由上应用装饰器,观察日志中是否有未捕获的 jwt 异常。后续维护建议:定期更换密钥(但要考虑旧 token 如何处理,可以保留上一个密钥一段时间用于验证过期 token),以及为注销功能维护 token 黑名单(如 Redis存储撤销的 jti),但不要一开始就上黑名单,先确保基本流程稳定。