python flask 框架实现jwt用户登录 接口权限认证 案例

环境：python3.6+

模块：flask、jwt

目的：实现用于登录并返回token令牌，用于后续的接口权限验证。

前言介绍：

jwt（JSON Web Tokens），在用户认证当中常用的方式，在如今的前后端分离项目当中应用广泛，提高了后端代码的简洁和效能。

传统token和jwt区别

• 传统token：服务端会对登录成功的用户生成一个随机token返回，同时也会在本地保留对应的token（如在数据库中存入：token、用户名、过期时间等），当用户再次访问时，会携带之前的token给服务端进行校验，服务端则通过与本地保留的token进行对比，若寻找到符合条件的token数据，则校验成功
• jwt验证：服务端会对登录成功的用户生成一个随机token返回，但并不会在服务端本地保留（这是jwt和传统token最大的区别），而当用户再次访问时，服务端会基于jwt对token进行校验和解码（由于jwt是基于base64url编码，因此是可以反向解码的，建议不要在token中存放敏感数据）

JWT 的官方文档: JSON Web Token Introduction - jwt.io

代码如下：

from flask import g, request, Flask, current_app, jsonify
import jwt
from jwt import exceptions
import functools
import datetime

app = Flask(__name__)

# 处理中文编码
app.config['JSON_AS_ASCII'] = False

# 跨域支持
def after_request(resp):
    resp.headers['Access-Control-Allow-Origin'] = '*'
    return resp

app.after_request(after_request)

# 构造header
headers = {
    'typ': 'jwt',
    'alg': 'HS256'
}

# 密钥
SALT = 'iv%i6xo7l8_t9bf_u!8#g#m*)*+ej@bek6)(@u3kh*42+unjv='

def create_token(username, password):
    # 构造payload
    payload = {
        'username': username,
        'password': password,  # 自定义用户ID
        'exp': datetime.datetime.utcnow() + datetime.timedelta(days=7)  # 超时时间
    }
    result = jwt.encode(payload=payload, key=SALT, algorithm="HS256", headers=headers).decode('utf-8')
    return result

def verify_jwt(token, secret=None):
    """
    检验jwt
    :param token: jwt
    :param secret: 密钥
    :return: dict: payload
    """
    if not secret:
        secret = current_app.config['JWT_SECRET']

    try:
        payload = jwt.decode(token, secret, algorithms=['HS256'])
        return payload
    except exceptions.ExpiredSignatureError:  # 'token已失效'
        return 1
    except jwt.DecodeError:  # 'token认证失败'
        return 2
    except jwt.InvalidTokenError:  # '非法的token'
        return 3

def login_required(f):
    '让装饰器装饰的函数属性不会变 -- name属性'
    '第1种方法,使用functools模块的wraps装饰内部函数'

    @functools.wraps(f)
    def wrapper(*args, **kwargs):
        try:
            if g.username == 1:
                return {'code': 4001, 'message': 'token已失效'}, 401
            elif g.username == 2:
                return {'code': 4001, 'message': 'token认证失败'}, 401
            elif g.username == 2:
                return {'code': 4001, 'message': '非法的token'}, 401
            else:
                return f(*args, **kwargs)
        except BaseException as e:
            return {'code': 4001, 'message': '请先登录认证.'}, 401

    '第2种方法,在返回内部函数之前,先修改wrapper的name属性'
    # wrapper.__name__ = f.__name__
    return wrapper

@app.before_request
def jwt_authentication():
    """
    1.获取请求头Authorization中的token
    2.判断是否以 Bearer开头
    3.使用jwt模块进行校验
    4.判断校验结果,成功就提取token中的载荷信息,赋值给g对象保存
    """
    auth = request.headers.get('Authorization')
    if auth and auth.startswith('Bearer '):
        "提取token 0-6 被Bearer和空格占用 取下标7以后的所有字符"
        token = auth[7:]
        "校验token"
        g.username = None
        try:
            "判断token的校验结果"
            payload = jwt.decode(token, SALT, algorithms=['HS256'])
            "获取载荷中的信息赋值给g对象"
            g.username = payload.get('username')
        except exceptions.ExpiredSignatureError:  # 'token已失效'
            g.username = 1
        except jwt.DecodeError:  # 'token认证失败'
            g.username = 2
        except jwt.InvalidTokenError:  # '非法的token'
            g.username = 3

@app.route('/')
def hello_world():
    return "ok"

# 登录
@app.route('/api/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        data = request.form
        type_ = data.get("type")
        if type_ == 'login':
            username = data.get("username")
            password = data.get("password")
            # 验证账号密码，正确则返回token，用于后续接口权限验证
            if username == "root" and password == "123456":
                token = create_token(username, password)
                return {"code": 200, "message": "success", "data": {"token": token}}
            else:
                return {"code": 501, "message": "登陆失败"}
        else:
            return {"code": 201, "message": "type is false"}

    elif request.method == 'GET':
        return {"code": 202, "message": "get is nothing"}
    else:
        return {"code": 203, "message": "'not support other method'"}

# 测试接口
@app.route('/api/test', methods=['GET', 'POST'])
@login_required
def submit_test_info_():
    username = g.username
    return username

if __name__ == "__main__":
    app.run(host='0.0.0.0', port=8090, debug=True)

运行项目后，首先测下登录接口

效果如下：

这个时候返回token令牌。

我们先测下没有令牌的情况下访问/api/test测试接口，效果如下：

这个时候是无法访问接口的，因为身份信息验证并为通过，

接下来，我们把token令牌添加到请求头中，添加参数：Authorization

效果如下：

这个时候可以看出，只要请求头中携带了正确的token口令，才可以访问这个接口。