mirror of
https://github.com/jumpserver/jumpserver.git
synced 2026-01-29 21:51:31 +00:00
83ff8dbf26
* perf: 修复一些错误权限位 * Pr@fix rbac@fix rbac permissions (#7648) * fix: 确保每次 migrate 执行更新 role permissions * perf: 修改 choices * feat: 兼容apple m1 * perf: 修改 migrations role permissions * perf: pymysql 导入 * perf: admin 判断 * fix: 修复消息订阅权限 Co-authored-by: ibuler <ibuler@qq.com> Co-authored-by: Aaron3S <chenyang@fit2cloud.com> Co-authored-by: feng626 <1304903146@qq.com>
136 lines
4.1 KiB
Python
136 lines
4.1 KiB
Python
from django.utils.translation import ugettext_lazy as _, gettext
|
||
from django.db import models
|
||
|
||
from common.db.models import JMSModel
|
||
from common.utils import lazyproperty
|
||
from .permission import Permission
|
||
from ..builtin import BuiltinRole
|
||
from .. import const
|
||
|
||
__all__ = ['Role', 'SystemRole', 'OrgRole']
|
||
|
||
|
||
class SystemRoleManager(models.Manager):
|
||
def get_queryset(self):
|
||
queryset = super().get_queryset()
|
||
return queryset.filter(scope=const.Scope.system)
|
||
|
||
|
||
class OrgRoleManager(models.Manager):
|
||
def get_queryset(self):
|
||
queryset = super().get_queryset()
|
||
return queryset.filter(scope=const.Scope.org)
|
||
|
||
|
||
class Role(JMSModel):
|
||
""" 定义 角色 ｜ 角色-权限 关系 """
|
||
Scope = const.Scope
|
||
|
||
name = models.CharField(max_length=128, verbose_name=_('Name'))
|
||
scope = models.CharField(
|
||
max_length=128, choices=Scope.choices, default=Scope.system, verbose_name=_('Scope')
|
||
)
|
||
permissions = models.ManyToManyField(
|
||
'rbac.Permission', related_name='roles', blank=True, verbose_name=_('Permissions')
|
||
)
|
||
builtin = models.BooleanField(default=False, verbose_name=_('Built-in'))
|
||
comment = models.TextField(max_length=128, default='', blank=True, verbose_name=_('Comment'))
|
||
|
||
BuiltinRole = BuiltinRole
|
||
objects = models.Manager()
|
||
org_roles = OrgRoleManager()
|
||
system_roles = SystemRoleManager()
|
||
|
||
class Meta:
|
||
unique_together = [('name', 'scope')]
|
||
verbose_name = _('Role')
|
||
|
||
def __str__(self):
|
||
return '%s(%s)' % (self.name, self.get_scope_display())
|
||
|
||
def is_system_admin(self):
|
||
return str(self.id) == self.BuiltinRole.system_admin.id and self.builtin
|
||
|
||
def is_org_admin(self):
|
||
return str(self.id) == self.BuiltinRole.org_admin.id and self.builtin
|
||
|
||
def is_admin(self):
|
||
yes = self.is_system_admin() or self.is_org_admin()
|
||
return yes
|
||
|
||
@staticmethod
|
||
def get_scope_roles_perms(roles, scope):
|
||
has_admin = any([r.is_admin() for r in roles])
|
||
if has_admin:
|
||
perms = Permission.objects.all()
|
||
else:
|
||
perms = Permission.objects.filter(roles__in=roles).distinct()
|
||
perms = Permission.clean_permissions(perms, scope=scope)
|
||
return perms
|
||
|
||
@classmethod
|
||
def get_roles_permissions(cls, roles):
|
||
org_roles = [role for role in roles if role.scope == cls.Scope.org]
|
||
org_perms_id = cls.get_scope_roles_perms(org_roles, cls.Scope.org)\
|
||
.values_list('id', flat=True)
|
||
|
||
system_roles = [role for role in roles if role.scope == cls.Scope.system]
|
||
system_perms_id = cls.get_scope_roles_perms(system_roles, cls.Scope.system)\
|
||
.values_list('id', flat=True)
|
||
perms_id = set(org_perms_id) | set(system_perms_id)
|
||
permissions = Permission.objects.filter(id__in=perms_id)\
|
||
.prefetch_related('content_type')
|
||
return permissions
|
||
|
||
@classmethod
|
||
def get_roles_perms(cls, roles):
|
||
permissions = cls.get_roles_permissions(roles)
|
||
return Permission.to_perms(permissions)
|
||
|
||
def get_permissions(self):
|
||
if self.is_admin():
|
||
permissions = Permission.objects.all()
|
||
else:
|
||
permissions = self.permissions.all()
|
||
permissions = Permission.clean_permissions(permissions, self.scope)
|
||
return permissions
|
||
|
||
@lazyproperty
|
||
def users(self):
|
||
from .rolebinding import RoleBinding
|
||
return RoleBinding.get_role_users(self)
|
||
|
||
@lazyproperty
|
||
def users_amount(self):
|
||
return self.users.count()
|
||
|
||
@lazyproperty
|
||
def permissions_amount(self):
|
||
return self.permissions.count()
|
||
|
||
@classmethod
|
||
def create_builtin_roles(cls):
|
||
BuiltinRole.sync_to_db()
|
||
|
||
@property
|
||
def display_name(self):
|
||
if not self.builtin:
|
||
return self.name
|
||
return gettext(self.name)
|
||
|
||
|
||
class SystemRole(Role):
|
||
objects = SystemRoleManager()
|
||
|
||
class Meta:
|
||
proxy = True
|
||
verbose_name = _('System role')
|
||
|
||
|
||
class OrgRole(Role):
|
||
objects = OrgRoleManager()
|
||
|
||
class Meta:
|
||
proxy = True
|
||
verbose_name = _('Organization role')
|