使用JWT生成Token，并实现Token刷新API

Posted 2022-01-26 胖虎是只mao

一、背景

传统的网站用户认证方式严重依赖于 Cookie 。但 很多 项目是一个前后端分离的项目，我们希望前端界面的实现不受 项目API 的影响，这就要求 用户认证没有 Cookie 时也能进行。其实仔细想想，用户的认证鉴权，无非就是在访问 API 时发送一个能够识别出用户的字符串给服务端，然后服务端基于字符串查找出用户，再判断用户是否有权限使用 API 。这串字符串，我们可以将其称之为 Token。由于 rmon 的所有 API 都是基于 HTTP 协议的，而在 HTTP 协议中有一个 Authorization 的头部用于携带用户认证信息，所以我们可以将 Token 放在 HTTP 头部的 Authorization 字段中发送给服务端。需要注意的是，具体使用什么 HTTP 头部携带 Token 其实无关要紧，只需要服务端和客户端使用同样的头部进行处理就可以了。我们选择 Authorization 头部，是因为在相关标准中 Authorization 头部被指定用于实现 HTTP 基本认证。

Token 一般是由服务端下发的，用户在成功登录后，服务端将发送一个 Token，后续的所有的访问中，客户端都将携带这个 Token，而服务端都将基于该 Token 进行用户认证鉴权。由于任何人拿到这个 Token 后就相当于盗取了用户的所有操作权限，所以 Token 的安全性非常重要。因此 Token 有一些要求，例如 Token 需要有一个过期时间，过期后 Token 就作废，服务端收到 Token 时需要验证其是否已经过期。此外需要有一种办法确保服务端发出的 Token 没有被篡改，如果不能验证 Token 是否被篡改，就无法保证 Token 代表的用户是其未被篡改前代表的用户。从零开始实现这些需求非常麻烦，好在我们可以使用 JWT。

二、使用JWT生成Token

JWT 全称为 Json Web Token，是一种轻量的 Token 实现标准， 它满足了我们对 Token 的所有需求，而且使用简单。一个 JWT 由 . 符号分隔成三部分，如下所示：

eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ1aWQiOiIxMjMiLCJleHAiOjE1MDk0OTc2NDZ9.1tCu1bjfveMfMhj5C6nR5P9iRm-O_Z7rSFnmj_BmMno

每一部分都是经过 Base64 编码的，且前两部分都是 json 字符串。其中第一部分被称为头部，包含了 JWT 的签名方式，也就是前文中防止被篡改所使用的算法，例如下面的头部：

  "typ": "JWT",
  "alg": "HS256"

其中 type 字段说明是 JWT，alg 字段值指明了签名所使用的算法。

第二部分被称为 payload 也就是载荷，包含了用户自定义的字段信息和 JWT 标准的字段信息。例如前文所述要通过 Token 传递能够识别出用户身份的标示符，那么标示符就应该放在 JWT 的 payload 中。标准的字段主要有下面几种：

• iss: JWT 签发者
• sub: JWT 所面向的用户
• exp: JWT 的过期时间，这个过期时间必须要大于签发时间
• nbf: 定义在什么时间之前，该 JWT 都是不可用的
• iat: JWT 的签发时间

一个有效的 payload 如下：

  "uid": "123",
  "exp": 1509497646

包含了一个自定义的信息字段 uid 代表用户 ID，和一个标准的 exp 字段。对于一个 JWT 来说，我们可以直接通过对第二部分进行 base64 解码得到其具体内容，这就要求我们不能将机密信息放入 JWT 的第二部分中，比如用户密码。第三部分则是基于前两部分内容生成的签名字符串。

在 Python 中可以使用 pyjwt 软件包用于生成 JWT 和验证 JWT。基本的使用方法如下图所示：

我们主要使用 jwt.encode 和 jwt.decode 方法实现生成和读取 JWT 的操作。调用 jwt.decode 的时候可以通过参数指定是否对 JWT 进行验证，比如验证过期时间，验证签名信息是否正确等等。

总结下，用户通过登录认证 API 进行登录，服务端验证用户发送的用户名和密码是否正确，如果正确则返回一个 JWT 给用户；此后用户访问其它 API 时，例如管理 Redis 服务器时，则需要携带 JWT ；服务端收到请求后验证 JWT，查看相应的用户是否有访问权限。

项目实际使用

generate_token 实例方法，用于产生用户实例对应的 JWT，代码如下：

    def generate_token(self):
        """生成 json web token

        生成 token，有效期为 1 天，过期后十分钟内可以使用老 token 刷新获取新的
        token
        """
        # 设置 token 的过期时间为 1 天
        exp = datetime.utcnow() + timedelta(days=1)
        # token 过期后 10 分钟内，可以刷新旧的 token 获取新 token
        # datetime.datetime 对象的 utctimetuple 方法的返回值
        # 是 time.struct_time 对象，该对象为类 tuple 对象
        struct_time = (exp + timedelta(minutes=10)).utctimetuple()
        # calendar 模块中的 timegm 函数接收元组对象作为参数
        # 返回值是 GMT 计算时间戳的数值
        # 即从 1970 年 1 月 1 日到 struct_time 所指时间的秒数
        refresh_exp = timegm(struct_time)
        # 定义载荷，四个字段：ID 、管理员标识、过期时间、刷新时间
        payload = 
            'uid': self.id,
            'is_admin': self.is_admin,
            'exp': exp,
            'refresh_exp': refresh_exp
        
        # 创建 token ，三个参数：载荷、秘钥、算法
        token = jwt.encode(payload, current_app.secret_key, algorithm='HS512')
        return token.decode()

生成 JWT 的过程，主要使用 pyjwt 软件包完成。需要注意的是，我们设置了 Token 的过期时间为一天以后，这样就保证了该 Token 只有一天的有效期。同时我们还设置了一个 refresh_exp 字段，这个字段用于设置 token 过期后仍可刷新的时间。生成 JWT 时，还携带了自定义信息，比如代表用户 ID 的 uid 字段，和标示用户是否是管理员的 is_admin 字段，这些信息都来自于当前的用户实例。还需要注意的是，在 Python3 中，jwt.encode 方法返回的是 bytes 字节类型，需要通过 decode() 操作使其转换为字符串。

verify_token 用于基于 JWT 认证用户，代码如下：

    @classmethod
    def verify_token(cls, token, verify_exp=True):
        '''客户端向服务器发送请求时，验证 token

        Args:
            token (str): JSON WEB TOKEN
            verify_exp (bool): 是否验证 token 的过期时间

        Return:
            object: 返回用户对象（User 类实例）
        '''
        # verify_exp 的值如果为 False ，则不验证 token 的过期时间
        # 也就是说即使 token 过期也没关系，还是可以得到 payload
        # 如果 verify_exp 等于 True ，则验证 token 的过期时间
        # token 过期则抛出 ExpiredSignatureError 异常
        options = None if verify_exp else 'verify_exp': False
        # 获取载荷，它是字典对象
        try:
            # 前两个位置参数：token 、秘钥
            # verify ：是否验证秘钥；algorithms ：算法列表
            # options ：其它选项；require_exp ：载荷必须有 exp 字段
            payload = jwt.decode(token, current_app.secret_key, verify=True,
                    algorithms=['HS512'], options=options, require_exp=True)
        except jwt.InvalidTokenError as e:
            raise InvalidTokenError(403, str(e))
        # 验证载荷中的关键字段是否存在
        conditions = ('uid' in payload, 'is_admin' in payload,
                'refresh_exp' in payload)
        if not all(conditions):
            raise InvalidTokenError(403, 'invalid token')
        # 检查是否过了允许刷新的时间，刷新时间是 token 过期后 10 分钟内
        if payload['refresh_exp'] < timegm(datetime.utcnow().utctimetuple()):
            raise InvalidTokenError(403, 'invalid token')
        # 经过上面的层层检查后
        user = cls.query.get(payload['uid'])
        if not user:
            raise InvalidTokenError(403, 'User not exist')
        return user

代码中，主要通过 jwt.decode 对 Token 进行解码，在解码的过程中需要验证 Token 是否正确，包括 Token 是否过期，签名是否正确等。验证成功后，就返回通过 uid 字段代表的用户，否则抛出 InvalidTokenError 异常，代表 Token 验证失败。需要注意的是，代码中同时也对 refresh_exp 字段进行了处理，判断了是否 refresh_exp 也已经过期。

我们知道有了用户系统就要求 Redis 服务器管理 API 必须对用户进行鉴权，只有被认证的用户才能访问这些 API。如果在所有的 API 中都实现一遍用户认证代码，会有大量重复工作。方便的是，在前一周我们实现的 RestView 视图控制器基类中可以使用装饰器。所以我们只需要定义一个视图装饰器类，在装饰器中认证用户，并把装饰器应用到所有 API 对应的视图控制器方法上即可，而现在我们将要实现用户认证装饰器，这些装饰器都应用于视图控制器方法上

实现 TokenAuthenticate 装饰器类，基于 JWT 对用户进行认证，其代码如下：

class TokenAuthenticate:
    """通过 jwt 认证用户

    验证 HTTP Authorization 头所包含的 token
    """

    def __init__(self, admin=True):
        """
        Args:
            admin(bool): 是否需要验证管理员权限
        """
        self.admin = admin

    def __call__(self, func):
        """装饰器实现
        """
        @wraps(func)
        def wrapper(*args, **kwargs):

            pack = request.headers.get('Authorization', None)
            if pack is None:
                raise AuthenticationError(401, 'token not found')
            parts = pack.split()
            # Authorization 头部值必须为 'jwt <token_value>' 这种形式
            if parts[0].lower() != 'jwt':
                raise AuthenticationError(401, 'invalid token header')
            elif len(parts) == 1:
                raise AuthenticationError(401, 'token missing')
            elif len(parts) > 2:
                raise AuthenticationError(401, 'invalid token')
            token = parts[1]
            user = User.verify_token(token)

            # 如果需要验证是否是管理员
            if self.admin and not user.is_admin:
                raise AuthenticationError(403, 'no permission')

            # 将当前用户存入到 g 对象中
            g.user = user
            return func(*args, **kwargs)
        return wrapper

在前文中，我们知道服务端可以基于 HTTP 请求头部 Authorization 所包含的 Token 对用户进行认证，而 TokenAuthenticate 实例正是实现了这样的功能。 TokenAuthenticate 实例装饰器在执行真正的视图控制器代码前，会首先通过 flask.request 对象获取 Authorization 头部对应的值，并要该值必须是 jwt <token> 这样的格式，其中 jwt 表示这是一个 Json Web Token，而 <token> 则代表具体的 Token，两者间通过空格连接，当正确获取到 Token 后，会通过 User.verify_token 类方法对 Token 进行验证，如果验证成功则接着判断 Token 代表的用户是否有是管理员，通过层层验证，最后将用户存储在 flask.g 对象中，这样后续的控制器代码就可以直接通过 g.user 获取到当前请求关联的具体用户。 TokenAuthenticate 认证的 HTTP 请求如下图所示：

上图的 HTTP 请求中，就通过 Authorization 头部发送了 Token，而服务端则会使用 TokenAuthenticate 实例装饰器对 Token 进行验证并识别出对应的用户。

三、实现token刷新API

在前面的实验中，、实现了基于 Json Web Token 的认证方式。并且 返回的 token 具有过期时间，所以我们需要有一种方式在 token 过期后可以刷新获取新的 token。

背景
项目中通过 User.generate_token 方法生成 token，且生成的 token 中已经有一个用于刷新获取 token 的字段 refresh_exp 字段。在 token 过期且 refresh_exp 字段还未过期的情况下，可以使用老的 token 获取新的 token。但是通过什么 API 刷新 token，目前项目中还没有实现。

本次需要实现一个支持使用已过期 token 且 refresh_exp 字段还未过期时获取新 token 的 API。在实现的 API 中，用户可以通过带有 Authorization JWT <token_value> 头部的 HTTP GET 请求访问 /token/refresh 路径获取新的 token。

API 执行成功后会返回格式如 'ok': True, 'token': 'token_value' json 数据，其中 token_value 代表具体的 token 值

提示

• 充分理解 rmon 用户系统认证实现方式
• views.decorators.TokenAuthenticate 装饰器会强制验证 token 是否过期，所以你可能需要修改 TokenAuthenticate 代码才能应用到你的 API 实现中

参考代码

修改以下三个文件中的相关内容：

# File Name: /home/shiyanlou/Code/rmon/rmon/views/auth.py

# 增加以下代码

from flask import g


class RefreshTokenView(RestView):

    method_decorators = (TokenAuthenticate(admin=False, verify_exp=False), )

    def get(self):
        return 'ok': True, 'token': g.user.generate_token()

# File Name: /home/shiyanlou/Code/rmon/rmon/views/decorators.py

# 修改 TokenAuthenticate 类的 __init__ 方法


class TokenAuthenticate:

    def __init__(self, admin=True, verify_exp=True):
        self.admin = admin

# File Name: /home/shiyanlou/Code/rmon/rmon/views/urls.py

# 增加以下代码

from rmon.views.auth import RefreshTokenView

api.add_url_rule('/token/refresh',
        view_func=RefreshTokenView.as_view('refresh_token'))