Django-xadmin+rule对象级权限的实现方式

1. 需求vs现状

1.1 需求

要求做一个ERP后台辅助管理的程序，有以下几项基本要求：

1. 基本的增删改查功能

2. 基于对象的权限控制（如：系统用户分为平台运营人员和商家用户，商家用户小A只能查看编辑所属商家记录，而管理员可以纵览全局）

3. 数据库记录导入导出（xsl, json等），并且拥有对象级的权限控制（如：小A不能导出小B公司的信息，更不能导入小B公司信息进行更新和新增）

1.2 现状

实现需求1：Django-admin让我们能够很方便的实现一个管理后台程序。django-xadmin则在拥有admin基本功能的基础上增加了更为丰富的功能、界面也更加漂亮。类似还有django-suit等，本文使用xadmin（功能更丰富）；

实现需求2：django-admin，以及xadmin都只有基于model级的权限控制机制，需要自己扩展或者使用开源解决方案，如django-guardian，django-rules，本文结合django-rules实现了该功能；

实现需求3：xadmin虽然自带导出功能，但是导入功能没有实现，django自带后台结合django-import-export可以很容易实现，但是xadmin并不直接兼容，只有通过xadmin的插件机制实现。

2. 功能实现

本节主要展示对象级权限功能实现。django工程、xadmin替换原生admin的设置，请参照官方文档。

2.1 安装并配置rules

pip安装:pip install django-rules

配置settings.py

# settings.py
INSTALLED_APPS = (
 # ...
 'rules',
)
AUTHENTICATION_BACKENDS = (
 'rules.permissions.ObjectPermissionBackend',
 'django.contrib.auth.backends.ModelBackend',
)

2.2 建立model

新增CompanyUser模型表示商家账户（即对django自带user模块进行扩展，使每个账号绑定自己的公司码），新增Customer模型表示商家的客户信息并包含公司码字段，商家账号只能查看、编辑、导入、导出公司码一致的商家客户信息

# model.py
class CompanyUser(models.Model):
 user = models.OneToOneField(User, verbose_name='用户名')
 is_taixiang_admin = models.BooleanField('是否运营人员', default=False)
 company_code = models.CharField('公司码', max_length=20, blank=True, default='')

def __unicode__(self):
   return '％s' ％ self.user

class Meta:
   verbose_name = '导入账号'
   verbose_name_plural = verbose_name

class Customer(models.Model):
 name = models.CharField('客户姓名', max_length=50)
 phone = models.CharField('客户电话', max_length=12)
 type_choice = ((1, '普通'), (2, '批发'), (3, 'VIP'))
 creator = models.ForeignKey(settings.AUTH_USER_MODEL, verbose_name='创建人', blank=True, null=True)
 company_code = models.CharField('公司码', max_length=20, blank=True, null=True)

def __unicode__(self):
   return '％s-％s-％s' ％ (self.company_code, self.name, self.phone1)

class Meta:
   permissions = (
     ("simulate_import_customer", "允许模拟导入客户"),
     ("import_customer", "允许导入客户至商家系统"),
           )
   verbose_name = "客户"
   verbose_name_plural = verbose_name

2.2 使用rule

在model统计目录新增rules.py，配置该app相关的对象权限

引用rules

# rules.py
# On Python 2, you must also add the following to the top of your rules.py file, or you'll get import errors trying to import django-rules itself
from __future__ import absolute_import

import rules

# 使用修饰符@rules.predicate自定义predicates（判断），返回True表示有权限，False表示无权限

# Predicates

@rules.predicate
def is_colleague(user, entry):
 if not entry or not hasattr(user, 'companyuser'):
   return False
 return entry.company_code == user.companyuser.company_code

@rules.predicate
def is_taixiang_admin(user):
 if not hasattr(user, 'companyuser'):
   return False
 return user.companyuser.is_taixiang_admin

# predicates间可以进行运算
is_colleague_or_taixiang_admin = is_colleague | is_taixiang_admin | rules.is_superuser

# 设置Rules

rules.add_rule('can_view_customer', is_colleague_or_taixiang_admin)
rules.add_rule('can_delete_customer', is_colleague_or_taixiang_admin)
rules.add_perm('can_change_customer', is_colleague_or_taixiang_admin)

# 设置Permissions

rules.add_perm('data_import.view_customer', is_colleague_or_taixiang_admin)
rules.add_perm('data_import.delete_customer', is_colleague_or_taixiang_admin)
rules.add_perm('data_import.add_customer', is_colleague_or_taixiang_admin)
rules.add_perm('data_import.change_customer', is_colleague_or_taixiang_admin)

2.3 admin.py以及adminx.py设置

如果使用原生的django-admin，admin.py做如下设置：

# admin.py
from __future__ import absolute_import

from django.contrib import admin
from rules.contrib.admin import ObjectPermissionsModelAdmin
from .models import Customer

# ModelAdmin class继承ObjectPermissionsModelAdmin即可
class CustomerAdmin(ObjectPermissionsModelAdmin):
 pass

admin.site.register(Customer, CustomerAdmin)

使用xadmin，由于ObjectPermissionsModelAdmin无法直接使用，故参照源码重写has_change_permission和has_delete_permission方法即可。

注意：必须引用rules文件，权限规则才会生效，对于xadmin，添加

from .rules import *即可

# adminx.py
class CustomerAdmin(object):
 def has_change_permission(self, obj=None):
   codename = get_permission_codename('change', self.opts)
   return self.user.has_perm('％s.％s' ％ (self.app_label, codename), obj)

def has_delete_permission(self, obj=None):
   codename = get_permission_codename('delete', self.opts)
   return self.user.has_perm('％s.％s' ％ (self.app_label, codename), obj)

# 重写queryset()或者get_list_display()，list view的权限也做到了对象级隔离
 def queryset(self):
   qs = super(CustomerAdmin, self).queryset()
   if self.request.user.is_superuser or is_taixiang_admin(self.request.user):
     return qs
   try:
     return qs.filter(company_code=self.request.user.companyuser.company_code)
   except AttributeError:
     return None

class CompanyUserAdmin(object):
 pass

xadmin.sites.site.register(Customer, CustomerAdmin)
xadmin.sites.site.register(CompanyUser, CompanyUserAdmin)

2.4 效果展示

CompanyUser设置：

商家账号只有所属公司信息权限

运营人员拥有所有记录权限

补充知识：django 扩展自带权限,使其支持对象权限

扩展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

来源：https://blog.csdn.net/bbwangj/article/details/89380925