RBAC权限管理

橘生淮南 · 发布于 2023-7-2 22:23:19

目录
用户权限
权限框架
1.RBAC介绍
RBAC概述
RBAC生命周期
2. RBAC表
模型类定义
3. RBAC中间件管理
自定义路由中间件
用户权限

一个商城，后台维护人员不只一个，还有其他后台人员，如：信息录入、财务、运营、采购等，他们对应的角色不同，所对应的权限也不相同，需要后台管理员来分配权限。
权限框架

RBAC模型5大属性，分别是： 1 用户属性（张三、李四、王五） 2 角色属性（销售经理、销售、前台） 3 用户与角色的关系（张三是销售经理、李四王五是销售） 4 权限（添加客户、编辑客户、删除客户，查看客户） 5 权限与角色的关系（销售拥有查看客户的权限、销售经理可以查看/添加/删除/编辑客户的）
一个RBAC权限模块，必然要实现三个功能用户管理用户列表添加用户编辑用户设置用户角色角色管理角色列表添加角色编辑角色设置角色权限权限管理权限列表新增权限编辑权限
1.RBAC介绍

RBAC概述

在网站中，用户通过URL地址，进入网站的后端逻辑，从而对网站的数据库停止操作管理。可以让拥有权限的用户来完成操作，而没有权限的用户无法操作.
RBAC是基于角色的访问控制（Role-Based Access Control ），在RBAC中，权限与角色相关联，用户通过成为适当角色的成员而得到这些角色的权限，这就极大地简化了权限的管理。将权限赋予角色，而把角色又赋予用户，这样的权限设计很清楚，管理起来很方便
RBAC认为受权实际上是Who 、What 、How 三元组之间的关系，也就是Who 对What 停止How 的操作，也就是“主体”对“客体”的操作。
Who：是权限的拥有者或主体（如：User，Role）。
What：是操作或对象（operation，object）。
How：详细的权限（Privilege,正向受权与负向受权）。
RBAC生命周期

1、用户登陆验证
2、根据用户身份验证信息，获取用户的角色---所有
3、通过用户所绑定的角色，获取这个角色绑定的所有权限，并去重----存储session/ redis
4、查询用户所==访问的URL==是否在角色的权限内，假设在，则继续访问，假设不在，回绝访问
2. RBAC表

模型类定义

from django.db import models
from django.contrib.auth.models import AbstractUser
# 权限类型（角色）
# name varchar(30) '权限类型的名称'
class PermissionType(models.Model):
"""权限的类型"""
name = models.CharField(max_length=30, verbose_name='权限类型的名称')
class Meta:
db_table = 'permission_type'
def __str__(self):
return self.name
# 用户模型类，继承AbstractUser
# phone, varchar(11) 允许空唯一 '用户手机'
# last_login
# role fk--->PermissionType
class User(AbstractUser):
"""
用户表
"""
phone = models.CharField(max_length=11, null=True, unique=True, blank=True, verbose_name='用户手机')
last_login = models.DateTimeField()
# 已定义好的模型类，可添加字段
role = models.ManyToManyField(PermissionType, verbose_name='所属角色')
class Meta:
db_table = 'user'
def __str__(self):
return self.username
# 权限表
# name varchar(30) '权限名称'
# code_name varchar(300) '权限识别名称'
# type_name fk --->PermissionType
class Permission(models.Model):
"""
权限模型类
"""
name = models.CharField(max_length=30, verbose_name='权限名称')
code_name = models.CharField(max_length=300, verbose_name='权限识别名称')
type_name = models.ForeignKey(PermissionType, on_delete=models.CASCADE)
def __str__(self):
return self.name
class Meta:
db_table = 'permission'
# 用户组
# name varchar(30) '组名称'
# user m:n
# permission m:n
class UserGroup(models.Model):
"""用户组的模型类"""
name = models.CharField(max_length=30, verbose_name='组名称')
user = models.ManyToManyField(User)
# 更改模型类: 改为多对多
permission = models.ManyToManyField(Permission)
class Meta:
db_table = 'user_group'
def __str__(self):
return self.name

复制代码

3. RBAC中间件管理

自定义路由中间件

from django.utils.deprecation import MiddlewareMixin
from django.http import JsonResponse, HttpResponse
from rest_framework_jwt.utils import jwt_decode_handler
from jwt.exceptions import ExpiredSignatureError, DecodeError
import re
import redis
import json
# 我的权限控制中间件
class MyRBAC(MiddlewareMixin):
def __init__(self, handler):
super(MyRBAC, self).__init__(handler)
self.redis_cli = redis.Redis(host="localhost", port=6379, password="laufing", db=3)
# 路由中间件
def process_request(self, request):
# 获取恳求的地址，不含查询参数，以/开头
path = request.path_info[1:]
# 恳求的方法
method = request.method.lower()
# 判断是否登录/管理员的恳求
if re.findall(r"login|^admin", path, re.I):
# 继续向后走
return None
# 不是登录，则判断是否有会话坚持 cookie or session or jwt
# 这里简单一些，就对应前端项目的jwt token
payload = self.validate_token(request)
if payload:
# 有会话坚持
print("解析的用户payload：", payload)
# 检查用户是否有权限访问
# 获取当前用户的所有权限
permissions = self.redis_cli.get("user_permissions_%s"%payload.get("user_id"))
permissions_list = json.loads(permissions.decode())
print("当前用户的所有权限：", permissions_list)
# 判断当前恳求的地址是否在权限列表
for permisson in permissions_list:
# 判断恳求方法
if permisson.startswith(method):
# 获取路由的正则形式
re_path_info = permisson.split()[1]
if re.findall(re_path_info, path, re.I):
# 具有权限，继续向后走
return None
#print(path, method)
return JsonResponse({"msg": "无权限访问", "code": 403}, status=403)
#return HttpResponse(json.dumps({"msg": "正在检查...", "payload": payload}, ensure_ascii=False))
else:
return JsonResponse({"code": 401, "msg": "未认证"}, status=401)
@staticmethod
def validate_token(request):
# 获取jwt token
token = request.headers.get("Authorization")
if not token:
return False
# 解析token
token_ = token.split()[1]
try:
payload = jwt_decode_handler(token_)
except DecodeError:
return False
except ExpiredSignatureError:
return False
return payload

复制代码

注意：此时用户完成登录认证时，需要存储用户的所有权限

# 获取当前用户的角色集合
roles = user.role.all()
# 根据角色获取所有的权限 [(v1,), (v2,),.....]
permissions_list = []
for role in roles:
permissions = role.permission_set.all().values_list("code_name")
permissions_list += [i[0] for i in permissions]
# 权限去重
permissions_list = list(set(permissions_list))
# 权限存入redis
user_permissions = "user_permissions_%s"%user.id
redis_cli.set(user_permissions, json.dumps(permissions_list))
redis_cli.close()
return Response({
"code": "200",
"msg": "登录胜利，进入首页",
"token": token,
"user": user.username
})

复制代码