RBAC 权限管理系统

梦里梦外; 2023-09-30 23:53 136阅读 0赞 
  1. # -- coding: utf-8 --
  2. # @Time : 2023/2/28 22:16
  3. # @File : models.py
  4. from werkzeug.security import check_password_hash, generate_password_hash
  6. from app import db
  8. """
  9. user - dept：一对一
  10. user - rose：多对多
  11. permission - rose：多对多
  12. """
  15. # 用户模型
  16. class User(db.Model):
  17.     __tablename__ = 'admin_user'
  19.     id = db.Column(db.Integer, primary_key=True)
  20.     username = db.Column(db.String(32), unique=True)
  21.     password_hash = db.Column(db.String(128))
  22.     enable = db.Column(db.Integer, default=0, comment='启用')
  23.     dept_id = db.Column(db.Integer, comment='部门id')  # 一对一
  25.     role = db.relationship('Role', secondary="admin_user_role", backref=db.backref('user'), lazy='dynamic')
  27.     @staticmethod
  28.     def generate_hash(password):
  29.         return generate_password_hash(password)
  31.     @staticmethod
  32.     def verify_hash(password, hash):
  33.         return check_password_hash(hash, password)
  35.     def get_id(self):
  36.         return str(self.id)
  38.     @classmethod
  39.     def find_by_username(cls, username):
  40.         return cls.query.filter_by(username=username).first()
  42.     def create(self):
  43.         db.session.add(self)
  44.         db.session.commit()
  45.         return self
  48. # 创建中间表：用户和角色多对多
  49. user_role = db.Table(
  50.     "admin_user_role",  # 中间表名称
  51.     db.Column("id", db.Integer, primary_key=True, autoincrement=True, comment='标识'),  # 主键
  52.     db.Column("user_id", db.Integer, db.ForeignKey("admin_user.id"), comment='用户编号'),  # 属性 外键
  53.     db.Column("role_id", db.Integer, db.ForeignKey("admin_role.id"), comment='角色编号'),  # 属性 外键
  54. )
  57. # 角色模型
  58. class Role(db.Model):
  59.     __tablename__ = 'admin_role'
  60.     id = db.Column(db.Integer, primary_key=True, comment='角色ID')
  61.     name = db.Column(db.String(255), comment='角色名称')
  62.     code = db.Column(db.String(255), comment='角色标识')
  63.     enable = db.Column(db.Integer, comment='是否启用')
  64.     remark = db.Column(db.String(255), comment='备注')
  65.     details = db.Column(db.String(255), comment='详情')
  66.     sort = db.Column(db.Integer, comment='排序')
  68.     permissions = db.relationship('Permission', secondary="admin_role_permission", backref=db.backref('role'))
  70.     def create(self):
  71.         db.session.add(self)
  72.         db.session.commit()
  73.         return self
  76. # 创建中间表：角色和权限多对多
  77. role_power = db.Table(
  78.     "admin_role_permission",  # 中间表名称
  79.     db.Column("id", db.Integer, primary_key=True, autoincrement=True, comment='标识'),  # 主键
  80.     db.Column("permission_id", db.Integer, db.ForeignKey("admin_permission.id"), comment='用户编号'),  # 属性 外键
  81.     db.Column("role_id", db.Integer, db.ForeignKey("admin_role.id"), comment='角色编号'),  # 属性 外键
  82. )
  85. # 权限表
  86. class Permission(db.Model):
  87.     __tablename__ = 'admin_permission'
  88.     id = db.Column(db.Integer, primary_key=True, comment='权限编号')
  89.     name = db.Column(db.String(255), comment='权限名称')
  90.     type = db.Column(db.String(1), comment='权限类型')
  91.     code = db.Column(db.String(30), comment='权限标识')
  92.     url = db.Column(db.String(255), comment='权限路径')
  93.     open_type = db.Column(db.String(10), comment='打开方式')
  94.     parent_id = db.Column(db.Integer, comment='父类编号')
  95.     icon = db.Column(db.String(128), comment='图标')
  96.     sort = db.Column(db.Integer, comment='排序')
  97.     enable = db.Column(db.Integer, comment='是否开启')
  99.     def create(self):
  100.         db.session.add(self)
  101.         db.session.commit()
  102.         return self
  105. # 部门表：部门和用户：一对一
  106. class Dept(db.Model):
  107.     __tablename__ = 'admin_dept'
  108.     id = db.Column(db.Integer, primary_key=True, comment="部门ID")
  109.     parent_id = db.Column(db.Integer, comment="父级编号")
  110.     dept_name = db.Column(db.String(50), comment="部门名称")
  111.     sort = db.Column(db.Integer, comment="排序")
  112.     leader = db.Column(db.String(50), comment="负责人")
  113.     phone = db.Column(db.String(20), comment="联系方式")
  114.     email = db.Column(db.String(50), comment="邮箱")
  115.     status = db.Column(db.Integer, comment='状态(1开启,0关闭)')
  116.     remark = db.Column(db.Text, comment="备注")
  118.     def create(self):
  119.         db.session.add(self)
  120.         db.session.commit()
  121.         return self
  123. # -- coding: utf-8 --
  124. # @Time : 2023/2/28 22:22
  125. # @File : routes.py
  126. import json
  127. import traceback
  129. from flask import jsonify, make_response, request
  130. from flask_jwt_extended import create_access_token, jwt_required, get_jwt_identity
  131. from flask_restful import Api, Resource, abort
  132. from app import db
  133. from app.rbac import rbac_bp
  134. from app.rbac.models import User, Role, Dept, Permission
  135. from app.rbac.schema import UserSchema, DeptSchema, RoleSchema, PermissionSchema
  136. from app.utils.rights import authorize
  137. from app.utils.responses import response_with
  138. from app.utils import responses as resp
  140. api = Api(rbac_bp)
  143. @rbac_bp.route("/init_db/")
  144. def init_db():
  145.     # 创建数据库
  146.     db.create_all()
  147.     return response_with(resp.SUCCESS_200, message="init table success.")
  150. @rbac_bp.route('/login/', methods=["POST"])
  151. def login():
  152.     """登录"""
  153.     try:
  154.         data = json.loads(request.get_data())
  156.         user_obj = User.find_by_username(data['username'])
  157.         if not user_obj:
  158.             return response_with(resp.NOT_FOUND_HANDLER_404, message="user not found.")
  160.         # 登录成功，需要 生成 jwt字符串
  161.         if User.verify_hash(data['password'], user_obj.password_hash):
  163.             # 用户权限
  164.             user_permissions = []
  165.             for role in user_obj.role:
  166.                 # 未启用
  167.                 if role.enable == 0:
  168.                     continue
  170.                 for p in role.permissions:
  171.                     if p.enable == 0:
  172.                         continue
  174.                     user_permissions.append(p.code)
  176.             payload = {
  178.                 'user_id': user_obj.id,
  179.                 'username': user_obj.username,
  180.                 'roles': [r.name for r in user_obj.role],
  181.                 'permissions': user_permissions
  182.             }
  183.             print(f"payload: {
  184.       payload}")
  186.             """
  187.             {'user_id': 3, 'username': 'admin', 'roles': ['管理员'], 
  188.             'permissions': ['admin:user:main', 'admin:permission:main', 'admin:permission:add', 'admin:permission:edit', 
  189.             'admin:permission:remove', 'admin:role:main', 'admin:role:add', 'admin:role:edit', 'admin:role:remove', 
  190.             'admin:user:add', 'admin:user:edit', 'admin:user:remove', 'admin:dept:main', 'admin:dept:add', 
  191.             'admin:dept:edit', 'admin:dept:remove']}
  192.             """
  193.             access_token = create_access_token(identity=payload)
  195.             return response_with(resp.SUCCESS_201, message=f"{
  196.       user_obj.username} login success.",
  197.                                  value={
  198.     "access_token": access_token})
  200.         else:
  201.             return response_with((resp.UNAUTHORIZED_401))
  202.     except Exception as e:
  203.         print(traceback.format_exc())
  204.         return response_with(resp.INVALID_INPUT_422)
  207. class UserListResource(Resource):
  208.     """
  209.     用户
  210.     """
  212.     @authorize("admin:user:main")
  213.     def get(self):
  214.         users = User.query.all()
  216.         return UserSchema(many=True, only=("id", "username", "dept_id", "role")).dump(users)
  218.     @authorize("admin:user:add")
  219.     def post(self):
  220.         json_data = json.loads(request.get_data())
  221.         if not json_data:
  222.             return response_with(resp.BAD_REQUEST_400, message="No input data provided.")
  224.         try:
  225.             json_data['password_hash'] = User.generate_hash(json_data.pop('password', None))
  227.             print(json_data)
  228.             # user = user_schema.load(json_data)
  229.             # user.create()
  231.             user = User(**json_data)
  232.             db.session.add(user)
  233.             db.session.commit()
  234.         except Exception as e:
  235.             db.session.rollback()
  236.             print(traceback.format_exc())
  237.             return response_with(resp.BAD_REQUEST_400, message="Could not create user. Error: {}".format(str(e)))
  239.         return response_with(resp.SUCCESS_201, message="User create success.")
  242. class DeptListResource(Resource):
  243.     """
  244.     部门
  245.     """
  247.     @authorize("admin:dept:main")
  248.     def get(self):
  249.         depts = Dept.query.all()
  250.         return DeptSchema(many=True, only=("id", "dept_name")).dump(depts)
  252.     @authorize("admin:dept:add")
  253.     def post(self):
  254.         json_data = json.loads(request.get_data())
  255.         if not json_data:
  256.             return response_with(resp.BAD_REQUEST_400, message="No input data provided.")
  258.         try:
  259.             dept = DeptSchema().load(json_data)
  260.             print(json_data)
  261.             print(dept)
  263.             dept.data.create()
  264.         except Exception as e:
  265.             db.session.rollback()
  266.             print(traceback.format_exc())
  267.             return response_with(resp.BAD_REQUEST_400, message="Could not create dept. Error: {}".format(str(e)))
  269.         return response_with(resp.SUCCESS_201, message="Dept create success.")
  272. class PermissionListResource(Resource):
  273.     """
  274.     权限
  275.     """
  277.     @authorize("admin:permission:main")
  278.     def get(self):
  279.         permissions = Permission.query.all()
  280.         return PermissionSchema(many=True,
  281.                                 only=("id", "name", "code", "type", "url", "enable", "open_type", "parent_id")).dump(
  282.             permissions)
  284.     @authorize("admin:permission:add")
  285.     def post(self):
  286.         json_data = json.loads(request.get_data())
  287.         if not json_data:
  288.             return response_with(resp.BAD_REQUEST_400, message="No input data provided.")
  289.         try:
  290.             permission = PermissionSchema().load(json_data)
  291.             print(json_data)
  292.             print(permission)
  294.             permission.data.create()
  295.         except Exception as e:
  296.             db.session.rollback()
  297.             print(traceback.format_exc())
  298.             return response_with(resp.BAD_REQUEST_400, message="Could not create permission. Error: {}".format(str(e)))
  300.         return response_with(resp.SUCCESS_201, message="Permission create success.")
  303. class RoleListResource(Resource):
  304.     """
  305.     角色视图：获取所有角色、创建角色
  306.     """
  308.     # @authorize("admin:role:main")
  309.     def get(self):
  310.         roles = Role.query.all()
  311.         print(f"roles: {
  312.       roles}")
  313.         return RoleSchema(many=True).dump(roles)
  315.     @authorize("admin:role:add")
  316.     def post(self):
  317.         args = request.get_data()
  318.         if not args:
  319.             return response_with(resp.BAD_REQUEST_400, message="No input data provided.")
  321.         try:
  322.             json_data = json.loads(args)
  324.             role = RoleSchema().load(json_data)
  325.             print(json_data)
  326.             print(role)
  328.             role.data.create()
  329.         except Exception as e:
  330.             db.session.rollback()
  331.             print(traceback.format_exc())
  332.             abort(400, message="Could not create permission. Error: {}".format(str(e)))
  334.         return response_with(resp.SUCCESS_201, message="Role create success.")
  337. class RoleResource(Resource):
  338.     """
  339.     角色视图：获取角色详情、修改角色、删除角色
  340.     特征：单独操作某一个角色
  341.     """
  343.     @authorize("admin:role:main")
  344.     def get(self, role_id):
  345.         """单独获取某个角色"""
  346.         role = Role.query.get(role_id)
  347.         if not role:
  348.             return response_with(resp.NOT_FOUND_HANDLER_404, message="Role not found.")
  350.         return RoleSchema().dump(role)
  352.     @authorize("admin:role:edit")
  353.     def put(self, role_id):
  354.         """修改角色"""
  355.         role = Role.query.get(role_id)
  356.         if not role:
  357.             return response_with(resp.NOT_FOUND_HANDLER_404, message="Role not found.")
  359.         args = request.get_data()
  360.         if not args:
  361.             return response_with(resp.BAD_REQUEST_400, message="No input data provided.")
  363.         try:
  364.             json_data = json.loads(args)
  366.             role.name = json_data.get("name", "")
  367.             role.details = json_data.get("details", "")
  368.             role.enable = json_data.get("enable", 1)
  370.             db.session.add(role)
  371.             db.session.commit()
  372.         except Exception as e:
  373.             print(traceback.format_exc())
  374.             db.session.rollback()
  375.             return response_with(resp.BAD_REQUEST_400, message="Could not update role. Error: {}".format(str(e)))
  377.         return RoleSchema().dump(role)
  379.     @authorize("admin:role:remove")
  380.     def delete(self, role_id):
  381.         """删除角色"""
  382.         role = Role.query.get(role_id)
  383.         if not role:
  384.             return response_with(resp.NOT_FOUND_HANDLER_404, message="Role not found.")
  386.         try:
  387.             db.session.delete(role)
  388.             db.session.commit()
  389.         except Exception as e:
  390.             db.session.rollback()
  391.             abort(400, message="Could not delete role. Error: {}".format(str(e)))
  393.         return response_with(resp.SUCCESS_204, message="Role deleted successfully.")
  396. api.add_resource(UserListResource, '/users')
  397. api.add_resource(DeptListResource, '/depts')
  398. api.add_resource(PermissionListResource, '/permissions')
  399. api.add_resource(RoleListResource, '/roles')
  400. api.add_resource(RoleResource, '/roles/<int:role_id>')
  402. # -- coding: utf-8 --
  403. # @Time : 2023/3/11 23:10
  404. # @Author : hj
  405. # @File : rights.py
  407. import traceback
  408. from functools import wraps
  409. from flask_jwt_extended import get_jwt_identity, get_jwt, verify_jwt_in_request
  411. from flask import request, jsonify
  412. from app.utils.responses import response_with
  413. from app.utils import responses as resp
  416. def authorize(permission_name: str):
  417.     """
  418.     权限校验装饰器
  419.     :param permission_name:
  420.     :return:
  421.     """
  423.     def decorator(func):
  424.         @wraps(func)
  425.         def wrapper(*args, **kwargs):
  426.             # 获取 Authorization 头
  427.             auth_header = request.headers.get('Authorization', "")
  428.             print(auth_header)
  429.             if not auth_header:
  430.                 return response_with(resp.UNAUTHORIZED_401)
  432.             # 解析 jwt
  433.             try:
  434.                 # 验证请求中存在有效的JWT，除非' ' optional=True ' '
  435.                 verify_jwt_in_request()
  436.                 token = auth_header.split()[1]
  438.                 # 解析 jwt
  439.                 user_info = get_jwt_identity()
  440.                 print(f"user_info: {
  441.       user_info}")
  443.                 roles = user_info.get("roles", [])
  444.                 permissions = user_info.get("permissions", [])
  446.                 print(f"roles: {
  447.       roles}")
  448.                 print(f"permissions: {
  449.       permissions}")
  451.                 # 管理员
  452.                 if "管理员" in roles:
  453.                     return func(*args, **kwargs)
  455.                 # 检查权限
  456.                 if permission_name not in permissions:
  457.                     return response_with(resp.FORBIDDEN_403)
  458.             except:
  459.                 print(traceback.format_exc())
  460.                 return {
  461.     "msg": "Invalid token"}, 401
  463.             return func(*args, **kwargs)
  465.         return wrapper
  467.     return decorator

在这里插入图片描述

在这里插入图片描述

在这里插入图片描述

在这里插入图片描述

在这里插入图片描述

发表评论

表情：
评论列表 （有 0 条评论，136人围观）

还没有评论，来说两句吧...

相关阅读

    相关 RBAC权限管理

    RBAC（Role-Based Access Control，基于角色的访问控制），就是用户通过角色与权限进行关联。简单地说，一个用户拥有若干角色，每一个角色拥有若干权限。这样

    电玩女神电玩女神/ 0 赞/ 326 阅读

    相关 RBAC权限管理系统

    RBAC权限管理系统 RBAC权限控制机制 权限所要控制的资源类别是根据应用系统的需要而定义的，具有的语义和控制规则也是应用系统提供的，对于权限管理系统来说是透明的，权限将

    分手后的思念是犯贱分手后的思念是犯贱/ 0 赞/ 387 阅读

    相关 RBAC权限管理

    1. 简介 RBAC是Role-Based Access Control的首字母，即基于角色的访问控制，是最简单的权限管理解决方案。它对权限的控制精度一般为节点。 基本

    桃扇骨桃扇骨/ 0 赞/ 392 阅读

    相关 RBAC权限管理

    RBAC(Role-Based Access Controller,基于角色的访问控制), 概念: 就是用户通过角色与权限进行相关联; 模型: “用户—>角色—>权限”;

    小灰灰小灰灰/ 0 赞/ 401 阅读