JSON WEB Token——在两个组织之间传递安全可靠的消息。 JWT的两种实现： JSON Web Signature(JWS)——一个有着简单统一表达方式的字符串。(签名)
base64enc{“alg”:指定签名加密算法， “typ”:“JWT”}——>base64加密后得到的字符串Header
base64enc {“iss”: 该JWT的签发者， “sub”: 该JWT所面向的用户 “aud”:该JWT的接收方 “exp”:过期时间，是一个Unix时间戳（expires） “iat”：什么时候签发 … }——base64加密后的字符串PayLoad
header.payload.secret——进行header中申明的加密方式。——signature。
jwt = Header.payload.signature 签名：jwt = Header.payload.signature 验签：1. jwt2 = Header.payload.signature是否等于jwt. 2. 再检验exp是否过期。
JSON Web Encryption(JWE)——相对于jws，jwe能同时保证安全性与完整性。（加密） 生成步骤：
JOSE含义与JWS头部相同。生成一个随机的Content Encryption Key （CEK）。使用RSAES-OAEP 加密算法，用公钥加密CEK，生成JWE Encrypted Key。生成JWE初始化向量。使用AES GCM加密算法对明文部分进行加密生成密文Ciphertext,算法会随之生成一个128位的认证标记Authentication Tag。 6.对五个部分分别进行base64编码。 JWE的计算过程相对繁琐，不够轻量级，因此适合与数据传输而非token认证，但该协议也足够安全可靠，用简短字符串描述了传输内容，兼顾数据的安全性与完整性。对比：
jws:使用时对照签名是否一致。对内容进行base64转码——反转码便可获取内容。 载荷部分一定不能包含敏感数据。 jwe:使用时需要解密不知道密钥根本就解不出来——它就是一个密文。payload数据采用加密方式进行加密。收到方不是进行验签而是解密。
以上内容参考来源：https://www.jianshu.com/p/50ade6f2e4fd
JWT的Python库
itsdangerousJSONWebSignatureSerializer（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 payloadJWT刷新机制：
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}, 2012. 请求钩子 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：
请求钩子使用：@app.before_request装饰器。问题——app加载不进来该装饰器拿不到使用current_app——前提：current_app属于应用上下文，此处是在一个独立的环境中。？？？到底什么是处于应用上下文，什么是处于独立的环境？？？进入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 wrapper4. 更新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需求: 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说明：
redis记录设置有效期的时长是一个token的有效期，保证旧token过期后，redis的记录也能自动清除，不占用空间。使用set保存新token的原因是，考虑到用户可能在旧token的有效期内，在其他多个设备进行了登录，需要生成多个新token，这些新token都要保存下来，既保证新token都能正常登录，又能保证旧token被禁用