PyJWT是一个Python库,用来编码/解码JWT(JSON Web Token)的。
1:安装PyJWT
2: 直接上代码了:
import datetime, jwt, time from app.dao.userDao import UserDao from flask import jsonify from .. import common class Auth(): @staticmethod def encode_auth_token(user_id, login_time): """ 生成认证Token :param user_id: int :param login_time: int(timestamp) :return: string """ try: payload = { 'exp': datetime.datetime.utcnow() + datetime.timedelta(days=0, seconds=10), 'iat': datetime.datetime.utcnow(), 'iss': 'ken', 'data': { 'id':user_id, 'login_time': login_time } } return jwt.encode( payload, 'secret', algorithm='HS256' ) except Exception as e: return e @staticmethod def decode_auth_token(auth_token): """ 验证Token :param auth_token: :return: integer|string """ try: payload = jwt.decode(auth_token, 'secret', options= {'verify_exp':False}) if ('data' in payload and 'id' in payload['data']): return payload else: raise jwt.InvalidTokenError except jwt.ExpiredSignatureError: return "Token过期" except jwt.InvalidTokenError: return "无效的Token" def authenticate(self, username, password): """ 用户登录,登录成功返回token,写将登录时间写入数据库;登录失败返回失败原因 :param password: :return: json """ userDao = UserDao() user = userDao.search(username) if (user is None): return jsonify(common.falseReturn('', '找不到用户')) else: if (user.password == password): login_time = int(time.time()) token = self.encode_auth_token(user.username, login_time) return jsonify(common.trueReturn(token.decode(), '登陆成功')) else: return jsonify(common.falseReturn('', '密码不正确')) def identify(self, request): """ 用户鉴权 :return: list """ auth_header = request.headers.get('Authorization') if (auth_header): auth_tokenArr = auth_header.split(" ") if (not auth_tokenArr or auth_tokenArr[0]!= 'jwt' or len(auth_tokenArr) != 2 ): result = common.falseReturn('','请传递正确的验证头信息') else: auth_token = auth_tokenArr[1] payload = self.decode_auth_token(auth_token) if not isinstance(payload, str): userDao = UserDao() user = userDao.search(payload['data']['id']) if (user is None): result = common.falseReturn('', '找不到该用户信息') else: result = common.trueReturn('', '请求成功') else: result = common.falseReturn('', payload) else: result = common.falseReturn('','没有提供认证token') return result
代码说明:
authenticate: 根据用户名/密码,到DB中进行校验,如果是合法的用户名/密码,调用encode_auth_token生成token返回;username加入到token的payload中。
encode_auth_token: 生成token的函数,payload可以存储一些不敏感的信息,比如用户名等,但是不能存密码;还有指定签名算法和秘钥。
identify: 用户的请求需要携带token信息,这个函数对request进行校验,调用decode_auth_token完成的校验。
decode_auth_token: 调用jwt.decode进行token校验(要指定秘钥,秘钥和生成token的秘钥一样)
3: 拦截所有的请求,都进行token校验
@app.before_request
def before_request():
Auth.identify(Auth,request )