【业务总结】认证方案——JWT

淩亂°似流年 2022-01-28 09:15 302阅读 0赞

JSON WEB Token——在两个组织之间传递安全可靠的消息。  
JWT的两种实现：  
![JWT,JWS,JWE][JWT_JWS_JWE]  
**JSON Web Signature(JWS)**——一个有着简单统一表达方式的字符串。(签名)  
![jws表达方式][jws]

1.  base64enc\{“alg”:指定签名加密算法，  
    “typ”:“JWT”\}——>base64加密后得到的字符串Header
2.  base64enc \{“iss”: 该JWT的签发者，  
    “sub”: 该JWT所面向的用户  
    “aud”:该JWT的接收方  
    “exp”:过期时间，是一个Unix时间戳（expires）  
    “iat”：什么时候签发  
    …  
    \}——base64加密后的字符串PayLoad
3.  header.payload.secret——进行header中申明的加密方式。——signature。
4.  jwt = Header.payload.signature  
    签名：jwt = Header.payload.signature  
    验签：1. jwt2 = Header.payload.signature是否等于jwt. 2. 再检验exp是否过期。

--------------------

**JSON Web Encryption(JWE)**——相对于jws，jwe能同时保证安全性与完整性。（加密）  
![在这里插入图片描述][20190526102133539.png]  
**生成步骤：**

1.  JOSE含义与JWS头部相同。
2.  生成一个随机的Content Encryption Key （CEK）。
3.  使用RSAES-OAEP 加密算法，用公钥加密CEK，生成JWE Encrypted Key。
4.  生成JWE初始化向量。
5.  使用AES GCM加密算法对明文部分进行加密生成密文Ciphertext,算法会随之生成一个128位的认证标记Authentication Tag。 6.对五个部分分别进行base64编码。  
    JWE的计算过程相对繁琐，不够轻量级，因此适合与数据传输而非token认证，但该协议也足够安全可靠，用简短字符串描述了传输内容，兼顾数据的安全性与完整性。

**对比：**

jws:使用时对照签名是否一致。对内容进行base64转码——反转码便可获取内容。  
载荷部分一定不能包含敏感数据。  
jwe:使用时需要解密不知道密钥根本就解不出来——它就是一个密文。payload数据采用加密方式进行加密。收到方不是进行验签而是解密。

以上内容参考来源：[https://www.jianshu.com/p/50ade6f2e4fd][https_www.jianshu.com_p_50ade6f2e4fd]

--------------------

JWT的Python库

*  itsdangerous

JSONWebSignatureSerializer（JWS）  
TimedJSONWebSignatureSerializer （可设置有效期）

*  pyjwt——允许完全定制payload内容和头  
    **使用**

>>> import jwt
    
     >>> encoded_jwt = jwt.encode({
        'some': 'payload'}, 'secret', algorithm='HS256')
      >>> encoded_jwt
      'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzb21lIjoicGF5bG9hZCJ9.4twFt5NiznN84AWoo1d7KO1T_yoc0Z6XOpOVswacPZg'
    
     >>> jwt.decode(encoded_jwt, 'secret', algorithms=['HS256'])
      {
        'some': 'payload'}

编码：jwt.encode(载荷,‘secret’,algorithm=‘HS256’)  
解码：jwt.decode(编码字符串，‘密钥’，algorithms=\[‘加密方式’\])  
\{\}  
**项目中的封装：**

import jwt
    from flask import current_app
    
    
    payload = {
        ...}
    # 封装一个生成JWT的函数
    def generate_iwt(payload,expiry,secret=None)
    """
        生成jwt
        :param payload: dict 载荷
        :param expiry: datetime 有效期
        :param secret: 密钥
        :return: jwt
        """
        _payload = {
        'exp': expiry}
        _payload.update(payload)
    
    	if not secret:
    		secrete = current_app.config['JWT_SECRET']
    	
    	token = jwt.encode(_payload,secret.algorithm='HS256')
    	return token.decode()
    
    
    # 校验jwt
    def verify_jwt(token, secret=None):
        """
        检验jwt
        :param token: jwt
        :param secret: 密钥
        :return: dict: payload
        """
        if not secret:
            secret = current_app.config['JWT_SECRET']
    	
    	# 验签时exp可能过期，程序执行到次直接抛出异常
        try:
            payload = jwt.decode(token, secret, algorithm=['HS256'])
        except jwt.PyJWTError:
            payload = None
    
        return payload

**JWT刷新机制：**

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h5eF94_size_16_color_FFFFFF_t_70]  
token若不设置有效期，则服务器会一直处理请求，问题就是token会长期暴露再网络中——不安全。  
一般token有效期设置在几个小时左右（2）——采用jwt刷新机制，使token过期后用户无感知获取新的token。  
如果客户端携带了token，服务器还返回401，那就客户端就知道时token过期了。  
接下来客户端发送refresh\_token：请求刷新接口。  
服务端验证refresh\_token生成token返回新token。  
refresh\_token ，可以做成很长时间。

**实现**

*  手机号+验证码（或帐号+密码）验证后颁发接口调用token与refresh\_token（刷新token）
 *  Token 有效期为2小时，在调用接口时携带，每2小时刷新一次
 *  提供refresh\_token，refresh\_token 有效期14天
 *  在接口调用token过期后凭借refresh\_token 获取新token
 *  未携带token 、错误的token或接口调用token过期，返回401状态码
 *  refresh\_token 过期返回403状态码，前端在使用refresh\_token请求新token时遇到403状态码则进入用户登录界面从新认证。
 *  token的携带方式是在请求头中使用如下格式：
    
    Authorization: Bearer （此处有个空格）eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzb21lIjoicGF5bG9hZCJ9.4twFt5NiznN84AWoo1d7KO1T\_yoc0Z6XOpOVswacPZg

登陆：  
接口设计：  
POST：/authentications  
请求参数：JSON\{“mobile” : “” ,“code” : “” \}  
响应：  
短信验证码错误：400  
正常：200

{
    	“message” : "OK",
    	"data" :
    			 {
    			 token" : "",
    			"refresh_token" : ""
    				}
    }

登陆和注册——一个接口完成。

**1. 注册或登陆获取token**

class AuthorizationResource(Resource)：
    	"""
    	认证
    	"""
    	def _generate_tokens(self,user_id,with_refresh_token=True):
    	"""
    	 生成token 和refresh_token
         :param user_id: 用户id
         :return: token, refresh_token
    	"""
    		# 生成调用token
            # 生成刷新token
            # utcnow东八区当前时间，timedelta()设置时间差,hours=中的值放在配置信息里，加载到flask_app
            secret = current_app.config['JWT_SECRET']  # 免得执行两次
            exipry = datetime.utcnow() + timedelta(hours=current_app.config['JWT_exipry_HOURS'])
            token = generate_jwt({
        'user_id':user_id},exipry,secret)
            # 'is_refresh':True 主要是为了区分两个token的身份
            exipry = datetime.utcnow()+timedelta(days=current_app.config['JWT_REFRESH_DAYS'])
            refresh_token = generate_jwt({
        'user_id':user_id,'is_refresh':True},exipry,secret)
            return token,refresh_token
    
    
     def post(self):
            """
            登录创建token
            """
            json_parser = RequestParser()
            json_parser.add_argument('mobile', type=parser.mobile, required=True, location='json')
            json_parser.add_argument('code', type=parser.regex(r'^\d{6}$'), required=True, location='json')
            args = json_parser.parse_args()
            mobile = args.mobile
            code = args.code
    
            # 从redis中获取验证码
            key = 'app:code:{}'.format(mobile)
            try:
                real_code = current_app.redis_master.get(key)
            except ConnectionError as e:
                current_app.logger.error(e)
                real_code = current_app.redis_slave.get(key)
    
            try:
                current_app.redis_master.delete(key)
            except ConnectionError as e:
                current_app.logger.error(e)
    
            if not real_code or real_code.decode() != code:
                return {
        'message': 'Invalid code.'}, 400
    
            # 查询或保存用户
            user = User.query.filter_by(mobile=mobile).first()
    
            if user is None:
                # 用户不存在，注册用户
                # 采用雪花算法生成分布式id
                # 其它会用到雪花算法生成id的地方：文章id,评论id
                # 这三个id在代码中直接操作数据库使用（通过id获取数据），需要全局唯一，使用雪花算法生成。
                user_id = current_app.id_worker.get_id()
                user = User(id=user_id, mobile=mobile, name=mobile, last_login=datetime.now())
                db.session.add(user)
                profile = UserProfile(id=user.id)
                db.session.add(profile)
                db.session.commit()
            else:
                if user.status == User.STATUS.DISABLE:
                    return {
        'message': 'Invalid user.'}, 403
    
            token, refresh_token = self._generate_tokens(user.id)
    
            return {
        'token': token, 'refresh_token': refresh_token}, 201

**2. 请求钩子**  
common/utils/middlewares.py

from flask import request, g
    from .jwt_util import verify_jwt
    
    def jwt_authentication():
        """
        根据jwt验证用户身份
        """
        # 设置初始值
        g.user_id = None
        g.is_refresh_token = False
        # 获取请求头中的token
        token = request.headers.get('Authorization')
        if token and token.startswith('Bearer '):
            token = token.strip()[7:]
            # 验证token
            payload = verify_jwt(token)
            if payload:
            	# 保存到g对象中
                g.user_id = payload.get('user_id')
                g.is_refresh_token = payload.get('refresh')

notes：

1.  请求钩子使用：@app.before\_request装饰器。问题——app加载不进来该装饰器拿不到
2.  使用current\_app——前提：current\_app属于应用上下文，此处是在一个独立的环境中。？？？到底什么是处于应用上下文，什么是处于独立的环境？？？
3.  进入app创建的工厂函数中，

# 添加请求钩子
        from utils.middlewares import jwt_authentication
        app.before_request(jwt_authentication)

**3. 强制登录装饰器**  
common/utils/decorators.py

def login_required(func):
        """
        用户必须登录装饰器
        使用方法：放在method_decorators中
        """
        @wraps(func)
        def wrapper(*args, **kwargs):
            if not g.user_id:
                return {
        'message': 'User must be authorized.'}, 401
            elif g.is_refresh_token:
                return {
        'message': 'Do not use refresh token.'}, 403
            else:
                return func(*args, **kwargs)
    
        return wrapper

**4. 更新token接口**  
**接口**  
PUT /authentications  
请求头 Authorization: Bearer refresh\_token  
返回  
错误：403  
正常：  
\{  
“message”:“OK”,  
“data”:\{  
“token”:  
\}  
\}  
toutiao/resources/user/passport.py

# 补充put方式 更新token接口
        def put(self):
            if g.user_id is not None and g.is_refresh is True:
                token, refresh_token = self.generate_tokens(g.user_id,refresh = False)
                return {
        'token':token}   # 在不需要refresh_token的情况下返回refresh_token会增大开销
            else:
                return {
        'meaasge':'Invalid refresh token'},403
    
    # 对生成token的修改,新增参数：refresh = True d对生成refresh_token进行if逻辑判断。
     def _generate_tokens(self, user_id, with_refresh_token=True,refresh = True):
            ....
            ....
            if refresh:
                exipry = datetime.utcnow()+timedelta(days=current_app.config['JWT_REFRESH_DAYS'])
                refresh_token = generate_jwt({
        'user_id':user_id,'is_refresh':True},exipry,secret)
            else:
                refresh_token = None
            return token,refresh_token

## JWT禁用问题 ##

**需求:**  
token颁发给用户后，在有效期内服务端都会认可，但是如果在token的有效期内需要让token失效，该怎么办？

此问题的应用场景：

*  用户修改密码，需要颁发新的token，禁用还在有效期内的老token
 *  后台封禁用户

**解决方案**  
在redis中使用set类型保存新生成的token

key = 'user:{}:token'.format(user_id)
    pl = redis_client.pipeline()
    pl.sadd(key, new_token)
    pl.expire(key, token有效期)
    pl.execute()

键：user:\{user\_id\}:token  
值：新token  
类型：set

客户端使用token进行请求时，如果验证token通过，则从redis中判断是否存在该用户的user:\{\}:token记录：

*  若不存在记录，放行，进入视图进行业务处理
 *  若存在，则对比本次请求的token是否在redis保存的set中：
    
     *  若存在，则放行
     *  若不在set的数值中，则返回403状态码，不再处理业务逻辑

key = 'user:{}:token'.format(user_id)
    valid_tokens = redis_client.smembers(key, token)
    if valid_tokens and token not in valid_tokens:
      return {
        'message': 'Invalid token'.}, 403

说明：

1.  redis记录设置有效期的时长是一个token的有效期，保证旧token过期后，redis的记录也能自动清除，不占用空间。
2.  使用set保存新token的原因是，考虑到用户可能在旧token的有效期内，在其他多个设备进行了登录，需要生成多个新token，这些新token都要保存下来，既保证新token都能正常登录，又能保证旧token被禁用

[JWT_JWS_JWE]: /images/20220128/d2b3386cf6ea4813916d1c36e24a8889.png
[jws]: /images/20220128/0f2a531465f143018e3b22917bef80eb.png
[20190526102133539.png]: /images/20220128/30dc2391427a4b058c9fbfa30d87f0d3.png
[https_www.jianshu.com_p_50ade6f2e4fd]: https://www.jianshu.com/p/50ade6f2e4fd
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h5eF94_size_16_color_FFFFFF_t_70]: /images/20220128/6a88b4b2de684d21ab9ff58130790000.png