tfse-admin-api-v0.2/user/user_auth.py

import functools

from flask import request
from itsdangerous import Serializer
from itsdangerous import TimedJSONWebSignatureSerializer as Serializer, SignatureExpired, BadSignature

from DBHelper.MongoHelper import MongoHelper

TOKEN_KEY = "P0eHym@&CbaLzWkq"
TOKEN_EXPIRE = 60*60*8


def create_token(param):
    """
    创建token
    Parameters:
        param: 传入参数，用于创建token
    Returns:
        token: 用户访问令牌
    """
    s = Serializer(TOKEN_KEY, expires_in=TOKEN_EXPIRE)
    token = '' + s.dumps(param).decode('ascii')
    return token


def verify_token(func):
    """
    校验token
    return:
        type:str
        desc: token被解析后的值
    """
    @functools.wraps(func)
    def internal(*args, **kwargs):
        try:
            # step 1.1
            # 解析请求头传送的token
            s = Serializer(TOKEN_KEY)
            session_id = s.loads(request.headers.get('token'))

            # step 1.2
            # 请求头中没有token参数 返回错误提示
            if session_id is None:
                return {"info": "缺少token"}, 401

            # step 2.1
            # token解析成功 从token记录中查询session_id对应的uid
            db = MongoHelper("tfse_v0.21")
            uid = db.find_single_column(
                "管理端",
                "token记录",
                {"session_id": session_id},
                "UID"
            )

            # step 2.2
            # 根据session_id没有找到对应的uid 返回错误提示
            if not uid:
                return {"info": "提示: 账号已在别处登录"}, 401

        except TypeError:
            return {"info": "异常token"}, 401
        except KeyError:
            return {"info": "异常token"}, 401
        except BadSignature:
            return {"info": "错误token"}, 401
        except SignatureExpired:
            return {"info": "过期token"}, 401
        return func(*args, **kwargs, uid=uid)
    return internal


def authority_scope(scope):
    def decorate(func):
        @functools.wraps(func)
        def internal(*args, ** kwargs):

            db = MongoHelper("tfse_v0.21")
            records = db.find_single_data(
                "管理端",
                "用户",
                {"UID": kwargs['uid']},
                ['status', 'role']
            )

            if not records:
                return {"info": "提示: 账户不存在"}, 401

            if records['status'] != "normal":
                return {"info": "提示: 账户已被禁用"}, 401

            if records['role'] not in scope:
                return {"info": "提示: 没有此项操作权限"}, 401

            return func(*args, ** kwargs)
        return internal
    return decorate