flask框架：根据角色权限调用接口

àì夳堔傛蜴生んèń 2022-12-22 09:27 33阅读 0赞

# 前言 #

根据角色进行调用接口

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

# 代码实现 #

*  删除功能，首选贴出token
 *  生成token

# -*- coding: utf-8 -*-
    # @Author  : ZhaoCunWei
    # @Time    : 2020/11/4 4:46 下午
    # @Function:
    from flask import current_app, jsonify
    from itsdangerous import Serializer
    
    from app.api.validators.forms import ClientForm
    from app.libs.enums import ClientTypeEnum
    from app.libs.redprint import Redprint
    from app.models.user import User
    
    api = Redprint('token')
    
    
    @api.route('', methods=['POST'])
    def get_token():
        form = ClientForm().validate_for_api()
        promise = { 
            ClientTypeEnum.USER_EMAIL: User.verify,
        }
        identity = promise[ClientTypeEnum(form.type.data)](
            form.account.data,
            form.secret.data
        )
        # Token
        expiration = current_app.config['TOKEN_EXPIRATION']
        token = generate_auth_token(identity['uid'],
                                    form.type.data,
                                    identity['is_admin'],
                                    expiration)
    
        t = { 
            'token': token.decode('ascii')
        }
        return jsonify(t), 201
    
    
    def generate_auth_token(uid, ac_type, is_admin=None, expiration=7200):
        """生成令牌"""
        s = Serializer(current_app.config['SECRET_KEY'],
                       expires_in=expiration)
        return s.dumps({ 
            'uid': uid,
            'type': ac_type.value,
            'is_admin': is_admin
        })

*  读取token

# -*- coding: utf-8 -*-
    # @Author  : ZhaoCunWei
    # @Time    : 2020/11/4 5:49 下午
    # @Function:
    from collections import namedtuple
    
    from flask import current_app, g
    from flask_httpauth import HTTPBasicAuth
    from itsdangerous import Serializer, BadSignature, SignatureExpired
    
    from app.libs.error_code import AuthFailed
    
    auth = HTTPBasicAuth()
    User = namedtuple('User', ['uid', 'ac_type', 'is_admin'])
    
    
    @auth.verify_password
    def verify_password(token, password):
        user_info = verify_auth_token(token)
        if not user_info:
            return False
        else:
            g.user = user_info
            return True
    
    
    # 读取令牌
    def verify_auth_token(token):
        s = Serializer(current_app.config['SECRET_KEY'])
    
        try:
            data = s.loads(token)
        except BadSignature:
            raise AuthFailed(msg='token is invalid', error_code=1002)
        except SignatureExpired:
            raise AuthFailed(msg='token is expired', error_code=1003)
    
        uid = data['uid']
        ac_type = data['type']
        is_admin = data['is_admin']
        return User(uid, ac_type, is_admin)

*  用户对象

# -*- coding: utf-8 -*-
    # @Author  : ZhaoCunWei
    # @Time    : 2020/11/3 6:26 下午
    # @Function:
    from datetime import datetime
    
    from sqlalchemy import Column, Integer, String, SmallInteger
    from werkzeug.security import generate_password_hash, check_password_hash
    
    from app.libs.error_code import NotFound, AuthFailed
    from app.models.base import Base, db
    
    
    class User(Base):
        id = Column(Integer, primary_key=True)
        email = Column(String(24), unique=True, nullable=False)
        nickname = Column(String(24), unique=True)
        auth = Column(SmallInteger, default=1)
        time = datetime.date(2018, 5, 20)
        _password = Column('password', String(100))
    
        # 需要序列化的参数
        def keys(self):
            return ['id', 'email', 'nickname', 'auth', 'time']
    
        def __getitem__(self, item):
            return getattr(self, item)
    
        # =================
        @property
        def password(self):
            return self._password
    
        @password.setter
        def password(self, raw):
            self._password = generate_password_hash(raw)
    
        @staticmethod
        def register_by_email(nickname, account, secret):
            with db.auto_commit():
                user = User()
                user.nickname = nickname
                user.email = account
                user.password = secret
                db.session.add(user)
    
        # 生成token时，进行区分用户的权限
        @staticmethod
        def verify(email, password):
            user = User.query.filter_by(email=email).first_or_404()
            if not user.check_password(password):
                raise AuthFailed()
            is_admin = True if user.auth == 2 else False
            return { 'uid': user.id, 'is_admin': is_admin}
    
        def check_password(self, raw):
            if not self._password:
                return False
            return check_password_hash(self._password, raw)

*  对外接口

# -*- coding: utf-8 -*-
    # @Author  : ZhaoCunWei
    # @Time    : 2020/11/3 2:33 下午
    # @Function:
    from flask import jsonify, g
    
    from app.libs.error_code import DeleteSuccess, AuthFailed
    from app.libs.redprint import Redprint
    from app.libs.token_auth import auth
    from app.models.base import db
    from app.models.user import User
    
    api = Redprint('user')
    
    
    class ZCW:
        name = 'zcw'
        age = 18
    
        def __init__(self):
            self.gender = 'male'
    
        def keys(self):
            return ['name', 'age']
    
        def __getitem__(self, item):
            return getattr(self, item)
    
    
    @api.route('/<int:uid>', methods=['GET'])
    @auth.login_required
    def get_user(uid):
        user = User.query.get_or_404(uid)
        return jsonify(user)
    
    
    @api.route('/<int:uid>', methods=['DELETE'])
    @auth.login_required
    def delete_user(uid):
        with db.auto_commit():
            user = User.query.filter_by(id=uid).first_or_404()
            user.delete()
        return DeleteSuccess()
    
    
    # 管理员
    @api.route('/<int:uid>', methods=['DELETE'])
    def super_delete_user(uid):
        pass
    
    
    # 管理员调用接口
    @api.route('/<int:uid>', methods=['GET'])
    @auth.login_required
    def super_get_user(uid):
        is_admin = g.user.is_admin
        if not is_admin:
            raise AuthFailed()
        user = User.query.filter_by(id=uid).first_or_404()
        return jsonify(user)
    
    
    # 普通用户
    @api.route('', methods=['GET'])
    @auth.login_required
    def get_user():
        uid = g.user.id
        user = User.query.filter_by(id=uid).first_or_404()
        return jsonify(user)

*  优化  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMzcwOTEz_size_16_color_FFFFFF_t_70_pic_center]
 *  创建表结构  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMzcwOTEz_size_16_color_FFFFFF_t_70_pic_center 1]  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMzcwOTEz_size_16_color_FFFFFF_t_70_pic_center 2]  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMzcwOTEz_size_16_color_FFFFFF_t_70_pic_center 3]  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMzcwOTEz_size_16_color_FFFFFF_t_70_pic_center 4]

# -*- coding: utf-8 -*-
    # @Author  : ZhaoCunWei
    # @Time    : 2020/11/4 5:49 下午
    # @Function:
    from collections import namedtuple
    
    from flask import current_app, g, request
    from flask_httpauth import HTTPBasicAuth
    from itsdangerous import Serializer, BadSignature, SignatureExpired
    from app.libs.error_code import AuthFailed, Forbidden
    from app.libs.scope import is_in_scope
    
    auth = HTTPBasicAuth()
    User = namedtuple('User', ['uid', 'ac_type', 'scope'])
    
    
    @auth.verify_password
    def verify_password(token, password):
        user_info = verify_auth_token(token)
        if not user_info:
            return False
        else:
            g.user = user_info
            return True
    
    
    # 读取令牌
    def verify_auth_token(token):
        s = Serializer(current_app.config['SECRET_KEY'])
    
        try:
            data = s.loads(token)
        except BadSignature:
            raise AuthFailed(msg='token is invalid', error_code=1002)
        except SignatureExpired:
            raise AuthFailed(msg='token is expired', error_code=1003)
    
        uid = data['uid']
        ac_type = data['type']
        scope = data['scope']
        # request 视图函数
        allow = is_in_scope(scope, request.endpoint)
        if not allow:
            raise Forbidden()
        return User(uid, ac_type, scope)

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMzcwOTEz_size_16_color_FFFFFF_t_70_pic_center 5]

# -*- coding: utf-8 -*-
    # @Author  : ZhaoCunWei
    # @Time    : 2020/11/19 2:33 下午
    # @Function:
    
    
    class AdminScope:
        allow_api = ['super_get_user']
    
    
    class UserScope:
        allow_api = []
    
    
    def is_in_scope(scope, endpoint):
        # scope()
        # 反射
        # globals
        # v1.view_func v1.module_name+view_func
        # v1.red_name+view_func
        scope = globals()[scope]()
        splits = endpoint.split('+')
        red_name = splits[0]
        if endpoint in scope.forbidden:
            return False
        if endpoint in scope.allow_api:
            return True
        if red_name in scope.allow_module:
            return True
        else:
            return False

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMzcwOTEz_size_16_color_FFFFFF_t_70_pic_center 6]

# 封装权限校验类 #

# -*- coding: utf-8 -*-
    # @Author  : ZhaoCunWei
    # @Time    : 2020/11/19 2:33 下午
    # @Function:
    
    
    class Scope:
        allow_api = []
        allow_module = []
        forbidden = []
    
        def __add__(self, other):
            self.allow_api = self.allow_api + other.allow_api
            self.allow_api = list(set(self.allow_api))
            # 运算符重载
    
            self.allow_module = self.allow_module + \
                                other.allow_module
            self.allow_module = list(set(self.allow_module))
    
            self.forbidden = self.forbidden + other.forbidden
            self.forbidden = list(set(self.forbidden))
    
            return self
    
    
    class AdminScope(Scope):
        # allow_api = ['v1.user+super_get_user',
        #              'v1.user+super_delete_user']
        allow_module = ['v1.user']
    
        def __init__(self):
            # 排除
            pass
            # self + UserScope()
    
    
    class UserScope(Scope):
        allow_module = ['v1.gift']
        forbidden = ['v1.user+super_get_user',
                     'v1.user+super_delete_user']
    
        def __init__(self):
            self + AdminScope()
        # allow_api = ['v1.user+get_user', 'v1.user+delete_user']
    
    
    def is_in_scope(scope, endpoint):
        # scope()
        # 反射
        # globals
        # v1.view_func v1.module_name+view_func
        # v1.red_name+view_func
        scope = globals()[scope]()
        splits = endpoint.split('+')
        red_name = splits[0]
        if endpoint in scope.forbidden:
            return False
        if endpoint in scope.allow_api:
            return True
        if red_name in scope.allow_module:
            return True
        else:
            return False

# -*- coding: utf-8 -*-
    # @Author  : ZhaoCunWei
    # @Time    : 2020/11/3 3:03 下午
    # @Function:
    
    
    class Redprint:
        def __init__(self, name):
            self.name = name
            self.mound = []
    
        def route(self, rule, **options):
            def decorator(f):
                self.mound.append((f, rule, options))
            return decorator
    
        def register(self, bp, url_prefix=None):
            if url_prefix is None:
                url_prefix = '/' + self.name
            for f, rule, options in self.mound:
                endpoint = self.name + '+' + \
                           options.pop("endpoint", f.__name__)
                bp.add_url_rule(url_prefix + rule, endpoint, f, **options)

*  验证token是否过期 – body里面传递参数–不是head里面传递

@api.route('/secret', methods=['POST'])
    def get_token_info():
        """获取令牌信息"""
        form = TokenForm().validate_for_api()
        s = Serializer(current_app.config['SECRET_KEY'])
        try:
            data = s.loads(form.token.data, return_header=True)
        except SignatureExpired:
            raise AuthFailed(msg='token is expired', error_code=1003)
        except BadSignature:
            raise AuthFailed(msg='token is invalid', error_code=1002)
    
        r = { 
            'scope': data[0]['scope'],
            'create_at': data[1]['iat'],
            'expire_in': data[1]['exp'],
            'uid': data[0]['uid']
        }
        return jsonify(r)

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMzcwOTEz_size_16_color_FFFFFF_t_70_pic_center 7]

# -*- coding: utf-8 -*-
    # @Author  : ZhaoCunWei
    # @Time    : 2020/11/4 4:46 下午
    # @Function:
    from flask import current_app, jsonify
    from itsdangerous import Serializer, SignatureExpired, BadSignature
    
    from app.api.validators.forms import ClientForm, TokenForm
    from app.libs.enums import ClientTypeEnum
    from app.libs.error_code import AuthFailed
    from app.libs.redprint import Redprint
    from app.models.user import User
    
    api = Redprint('token')
    
    
    @api.route('', methods=['POST'])
    def get_token():
        form = ClientForm().validate_for_api()
        promise = { 
            ClientTypeEnum.USER_EMAIL: User.verify,
        }
        identity = promise[ClientTypeEnum(form.type.data)](
            form.account.data,
            form.secret.data
        )
        # Token
        expiration = current_app.config['TOKEN_EXPIRATION']
        token = generate_auth_token(identity['uid'],
                                    form.type.data,
                                    identity['scope'],
                                    expiration)
    
        t = { 
            'token': token.decode('ascii')
        }
        return jsonify(t), 201
    
    
    @api.route('/secret', methods=['POST'])
    def get_token_info():
        """获取令牌信息"""
        form = TokenForm().validate_for_api()
        s = Serializer(current_app.config['SECRET_KEY'])
        try:
            data = s.loads(form.token.data, return_header=True)
        except SignatureExpired:
            raise AuthFailed(msg='token is expired', error_code=1003)
        except BadSignature:
            raise AuthFailed(msg='token is invalid', error_code=1002)
    
        r = { 
            'scope': data[0]['scope'],
            'create_at': data[1]['iat'],
            'expire_in': data[1]['exp'],
            'uid': data[0]['uid']
        }
        return jsonify(r)
    
    
    def generate_auth_token(uid, ac_type, scope=None, expiration=7200):
        """生成令牌"""
        s = Serializer(current_app.config['SECRET_KEY'],
                       expires_in=expiration)
        return s.dumps({ 
            'uid': uid,
            'type': ac_type.value,
            'scope': scope
        })

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMzcwOTEz_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/2df138776373400f9a5247c6849eefd5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMzcwOTEz_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/37329dcc00ab4add8d222323c964c5a1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMzcwOTEz_size_16_color_FFFFFF_t_70_pic_center 2]: https://img-blog.csdnimg.cn/20201119143041361.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMzcwOTEz,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMzcwOTEz_size_16_color_FFFFFF_t_70_pic_center 3]: https://img-blog.csdnimg.cn/20201119143224684.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMzcwOTEz,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMzcwOTEz_size_16_color_FFFFFF_t_70_pic_center 4]: https://img-blog.csdnimg.cn/20201119144202866.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMzcwOTEz,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMzcwOTEz_size_16_color_FFFFFF_t_70_pic_center 5]: https://img-blog.csdnimg.cn/20201119144547256.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMzcwOTEz,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMzcwOTEz_size_16_color_FFFFFF_t_70_pic_center 6]: https://img-blog.csdnimg.cn/20201119152352861.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMzcwOTEz,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMzcwOTEz_size_16_color_FFFFFF_t_70_pic_center 7]: https://img-blog.csdnimg.cn/20201119191342766.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMzcwOTEz,size_16,color_FFFFFF,t_70#pic_center