django 扩展自带权限,使其支持对象权限

Posted 2021-01-13

扩展django 自带权限

说明

在不重写 自带权限的基础上，完成支持对象权限，适用于小型项目。
欢迎提出修改意见

软件支持

jsonfield

数据库

新建3个表

from django.db import models
from django.contrib.auth.models import AbstractUser, Group ,User

from jsonfield import JSONField

class Request(models.Model):
    request = models.CharField(max_length=16, verbose_name=‘请求类型(大写)‘)

    class Meta:
        db_table = "request"
        verbose_name = "请求类型"
        verbose_name_plural = verbose_name

    def __str__(self):
        return self.request

class RolePermission(models.Model):
    role = models.CharField(max_length=32, verbose_name=‘角色组‘)
    table = models.CharField(max_length=32, verbose_name=‘表名字‘)
    request = models.ManyToManyField(Request, verbose_name=‘请求‘, related_name=‘re‘, )
    permission = JSONField(max_length=1024, verbose_name=‘权限条件‘)

    class Meta:
        db_table = "role_permission"
        verbose_name = "角色组权限"
        verbose_name_plural = verbose_name

    def __str__(self):
        return self.role

class Role(models.Model):
    group = models.ForeignKey(Group, verbose_name=‘用户组‘, on_delete=models.CASCADE)
    roles = models.ManyToManyField(RolePermission, verbose_name=‘角色组权限‘, blank=True,related_name=‘roles‘ )

    class Meta:
        db_table = "role"
        verbose_name = "角色组关系"
        verbose_name_plural = verbose_name

    def __str__(self):
        return self.group.name

system/models
Role                 角色组关系    : 系统用户组  <-->  角色组权限
Request             请求类型      : GET ,POST
RolePermission      角色组权限    : 角色  表名字  请求  权限条件(JSON类型)

重点为 RolePermission 表。

例子

• 以常见的资产 asset 为例

  表名字  asset     字段 groups     (分组 为 dev,ops)

• 权限划分
• 新建用户 hequan

• 新建组 dev

• 在Request 表 添加

  GET （代表只读）
  POST （代表更新 删除）

• 在RolePermission 添加

  角色 asset-dev只读
  表名字assset
  请求 GET
  权限条件 {"groups":‘dev‘}

• 在Role 表中 添加

  系统用户组 dev
  角色组权限 asset-dev只读

权限验证代码

import json
from system.models import Role
from functools import wraps
from django.shortcuts import HttpResponse

def role_permission_get_list(function):
    """
    列表页面 控制权限
    :param function:
    :return:
    """

    @wraps(function)
    def wrapped(self):
        user = self.request.user
        groups = [x[‘name‘] for x in self.request.user.groups.values()]
        request_type = self.request.method
        model = str(self.model._meta).split(".")[1]

        filter_dict = {}
        not_list = [‘page‘, ‘order_by‘, ‘csrfmiddlewaretoken‘]
        for k, v in dict(self.request.GET).items():
            if [i for i in v if i != ‘‘] and (k not in not_list):
                if ‘__in‘ in k:
                    filter_dict[k] = v
                else:
                    filter_dict[k] = v[0]

        if not user.is_superuser:
            role_groups = Role.objects.filter(group__name__in=groups).values_list(‘roles__table‘,
                                                                                  ‘roles__request__request‘,
                                                                                  ‘roles__permission‘)

            permission_dict = {}
            for i in role_groups:
                if i[0] == model and i[1] == request_type:
                    permission_dict = json.loads(i[2])

            if permission_dict:
                if filter_dict:
                    for k, v in permission_dict.items():
                        if ‘__in‘ in k:
                            k1 = k.replace(‘__in‘, ‘‘)
                        if ‘__gt‘ in k:
                            k1 = k.replace(‘__gt‘, ‘‘)
                        if ‘__lt‘ in k:
                            k1 = k.replace(‘__lt‘, ‘‘)
                        else:
                            k1 = k
                        if k1 in list(filter_dict.keys()):
                            del filter_dict[k1]

                    if filter_dict:
                        filter_dict.update(**permission_dict)
                    else:
                        print(‘查询条件处理后为空,默认权限‘)
                        filter_dict = permission_dict
                else:
                    print(‘查询条件为空,默认权限‘)
                    filter_dict = permission_dict
            else:
                print(‘没有权限‘)
                filter_dict = {‘id‘: -1}

        self.filter_dict = filter_dict
        result = function(self)
        return result

    return wrapped

def role_permission_detail(function):
    """
    详情页面 控制权限
    :param function:
    :return:
    """

    @wraps(function)
    def wrapped(self, request, *args, **kwargs):
        user = self.request.user

        if not user.is_superuser:
            groups = [x[‘name‘] for x in self.request.user.groups.values()]
            request_type = self.request.method
            model = str(self.model._meta).split(".")[1]
            pk = self.kwargs.get(self.pk_url_kwarg, None)

            role_groups = Role.objects.filter(group__name__in=groups).values_list(‘roles__table‘,
                                                                                  ‘roles__request__request‘,
                                                                                  ‘roles__permission‘)

            permission_dict = {}
            for i in role_groups:
                if i[0] == model and i[1] == request_type:
                    permission_dict = json.loads(i[2])

            permission_dict[‘id‘] = pk
            obj = self.model.objects.filter(**permission_dict).count()
            if not obj:
                return HttpResponse(status=403)

        result = function(self, request, *args, **kwargs)
        return result

    return wrapped

def role_permission_update_delete(function):
    """
    详情页面 控制权限
    :param function:
    :return:
    """

    @wraps(function)
    def wrapped(self, request):
        user = self.request.user
        if not user.is_superuser:

            groups = [x[‘name‘] for x in self.request.user.groups.values()]
            request_type = self.request.method
            model = str(self.model._meta).split(".")[1]
            pk = self.request.POST.get(‘nid‘, None)

            role_groups = Role.objects.filter(group__name__in=groups).values_list(‘roles__table‘,
                                                                                  ‘roles__request__request‘,
                                                                                  ‘roles__permission‘)

            permission_dict = {}
            for i in role_groups:
                if i[0] == model and i[1] == request_type:
                    permission_dict = json.loads(i[2])

            permission_dict[‘id‘] = pk
            obj = self.model.objects.filter(**permission_dict).count()
            if not obj:
                ret = {‘status‘: None, ‘error‘: "没有权限,拒绝", ‘msg‘: ‘Without permission, rejected‘}
                return HttpResponse(json.dumps(ret))

        result = function(self, request)
        return result

    return wrapped

CBV 例子

• 省略部分代码

  class AssetListAll(LoginRequiredMixin, ListView):
  model = Ecs
  
  @role_permission_get_list
  def get_queryset(self):
      filter_dict = self.filter_dict
      self.queryset = self.model.objects.filter(**filter_dict)
      return self.queryset

class AssetChange(LoginRequiredMixin, UpdateView):
    model = Ecs

    @role_permission_detail
    def dispatch(self, request, *args, **kwargs):
        return super().dispatch(request, *args, **kwargs)

    @role_permission_update_delete
    def form_valid(self, form):
        self.object = form.save()
        return super().form_valid(form)

class AssetDetail(LoginRequiredMixin, DetailView):
    model = Ecs

    @role_permission_detail
    def dispatch(self, request, *args, **kwargs):
        return super().dispatch(request, *args, **kwargs)

class AssetDel(LoginRequiredMixin, View):
    model = Ecs

    @role_permission_update_delete
    def post(self, request):
        pass