django-guardian
创建每个对象的权限。但是我想在这些权限周围添加一层逻辑。例如,如果某人对
Book
有edit_book
权限,则他们对该书中的 Pages
进行编辑的权限应该是隐含的。rules
包似乎非常理想。django-guardian
创建每个对象的权限。Book
有edit_book
权限,则他们对该书中的 Pages
进行编辑的权限应该是隐含的。rules
包似乎非常理想。简而言之:是的,他们可以这样做,我们可以解决一些规则可扩展性问题,但你无法避免每个对象运行一个查询,因此经过权限过滤的查询很昂贵。在底部建议了一种混合但更复杂的解决方案,通过将惰性规则集编译成 SQL 在运行时使用惰性Q
-like 对象来绕过此限制。
以下看起来是有效的:
import rules
import guardian
@rules.predicate
def is_page_book_editor(user, page):
return user.has_perm('books.edit_book', page.book)
@rules.predicate
def is_page_editor(user, page):
return user.has_perm('pages.edit_page', page)
rules.add_perm('pages.can_edit_page', is_page_book_editor | is_page_editor)
joe.has_perm('pages.can_edit_page', page34)
或者:
@permission_required('pages.can_edit_page', fn=objectgetter(Page, 'page_id'))
def post_update(request, page_id):
# ...
在定义了认证后端之后:
AUTHENTICATION_BACKENDS = (
'rules.permissions.ObjectPermissionBackend',
'django.contrib.auth.backends.ModelBackend',
'guardian.backends.ObjectPermissionBackend',
)
导入:
from django.contrib.auth.models import User
import rules
import guardian
from guardian.shortcuts import assign_perm
from myapp.models import Book, Page
测试内容:
joe = User.objects.create(username='joe', email='joe@example.com')
page23 = Page.objects.filter(id=123)
assign_perm('edit_page', joe, page23)
joe.has_perm('edit_page', page23)
is_page_editor(joe, page23) # returns True
joe.has_perm('can_edit_page', i) # returns True
rules.remove_perm('can_edit_page')
rules.add_perm('can_edit_page', is_page_book_editor & is_page_editor)
joe.has_perm('can_edit_page', i) # returns False
@rules.predicate
def is_page_book_viewer(user, instance):
if is_page_book_viewer.context.get('user_perms') is None:
is_page_book_viewer.context['user_perms'] = guardian.shortcuts.get_perms(user, page.book)
return 'view_book' in is_page_book_viewer.context.get('user_perms')
@rules.predicate(bind=True)
def is_page_viewer(self, user, instance):
if self.context.get('user_perms') is None:
self.context['user_perms'] = guardian.shortcuts.get_perms(user, instance)
return 'view_page' in self.context.get('user_perms')
(在第二个例子中,我绑定并使用了self
,但这与使用谓词名称是相同的。)
当您执行复杂的组合权限时,最好将django-guardian的通用外键替换为真实外键,从而使数据库能够进行优化和索引,具体操作如下:(参见链接)
class PageUserObjectPermission(UserObjectPermissionBase):
content_object = models.ForeignKey(Page)
class PageGroupObjectPermission(GroupObjectPermissionBase):
content_object = models.ForeignKey(Page)
class BookUserObjectPermission(UserObjectPermissionBase):
content_object = models.ForeignKey(Book)
class BookGroupObjectPermission(GroupObjectPermissionBase):
content_object = models.ForeignKey(Book)
有一个错误。我们正在同一位置缓存Page
和Book
的权限 - 我们需要区分并单独缓存它们。此外,让我们将重复代码封装到自己的方法中。最后,让我们给get()
一个默认值,以确保当用户没有任何权限时不再重新查询他们的权限。
def cache_permissions(predicate, user, instance):
"""
Cache all permissions this user has on this instance, for potential reuse by other predicates in this rule check.
"""
key = 'user_%s_perms_%s_%s' % (user.pk, type(instance).__name__, instance.pk)
if predicate.context.get(key, -1) == -1:
predicate.context[key] = guardian.shortcuts.get_perms(user, instance)
return predicate.context[key]
这样对象权限将被单独缓存。(在键中包含用户id是不必要的,因为任何规则只会检查一个用户,但这样做更有未来性.)
然后我们可以按如下方式定义谓词:
@rules.predicate(bind=True)
def is_page_book_viewer(self, user, instance: Page):
return 'view_book' in cache_permissions(self, user, instance.book)
rules
的一个限制是必须根据用户逐个进行权限检查,但通常我们需要获取用户拥有给定权限的所有对象。例如,要获取用户具有编辑页面权限的所有页面列表,我需要反复调用[p for p in Pages.objects.all() if usr.has_perm('can_edit_page', p)]
,而不是usr.has_perm('can_edit_page')
以一次查询返回所有允许的对象。
我们无法完全解决这个限制,但在我们不需要检查列表中的每个对象时,我们可以使用next
和基于惰性生成器协程的查询集来减少查询数量。在上面的示例中,如果我们可能不会到达列表的末尾,我们可以使用(...)
而不是[...]
,如果我们只需要检查列表中是否有任何一个对象具有该权限,则可以使用next(...)
。在正常循环代码中,break
或return
将是等效的,如下所示。
我有一个情况,其中一个模型具有自连接层次结构,我只需要知道模型的任何后代是否具有权限。该代码必须使用连续节点的后代递归查询表。但是,一旦我们找到具有权限的对象,就不需要进一步查询。我已经按如下方式完成了此操作。(请注意,我对是否有人在对象上具有权限感兴趣,并且已指定非通用键。如果您正在为特定用户检查权限,则可以使用user.has_perm('perm_name', obj)
来使用您的规则。)
class Foo(models.Model):
parent = models.ForeignKey('Foo', blank=True, null=True)
def descendants(self):
"""
When callers don't need the complete list (eg, checking if any dependent is
viewable by any user), we run fewer queries by only going into the dependent
hierarchy as much as necessary.
"""
immediate_descendants = Foo.objects.filter(parent=self)
for x in immediate_descendants:
yield x
for x in immediate_descendants:
for y in x.descendants():
yield y
def obj_or_descendant_has_perm(self, perm_code):
perm_id = Permission.objects.get(codename=perm_code).id
if FooUserObjectPermission.objects.filter(permission_id=perm_id,
content_object=self).exists()
return True
if FooGroupObjectPermission.objects.filter(permission_id=perm_id,
content_object=self).exists()
return True
for o in self.descendants():
if FooUserObjectPermission.objects.filter(permission_id=perm_id,
content_object=self).exists()
return True
if FooGroupObjectPermission.objects.filter(permission_id=perm_id,
content_object=self).exists()
return True
return False
treebeard
,以获得建模层次结构(材料化路径、嵌套集或邻接列表)更有效的方法。在我的情况下,由于这是通过其他表进行的自连接,所以这不可能。class Foo(models.Model):
parent = models.ForeignKey('Foo', blank=True, null=True)
def descendants(self):
"""
When callers don't need the complete list (eg, checking if any dependent is
viewable by any user), we run fewer queries by only going into the dependent
hierarchy as much as necessary. Returns a generator of querysets of Foo objects.
"""
immediate_descendants = Foo.objects.filter(parent=self)
yield immediate_descendants
for x in immediate_descendants:
for y in x.descendants():
yield y
def obj_or_descendant_has_perm(self, perm_code):
perm_id = Permission.objects.get(codename=perm_code).id
if FooUserObjectPermission.objects.filter(permission_id=perm_id,
content_object=self).exists()
return True
if FooGroupObjectPermission.objects.filter(permission_id=perm_id,
content_object=self).exists()
return True
for gen in self.descendants():
if FooUserObjectPermission.objects.filter(permission_id=perm_id,
content_object__in=gen).exists()
return True
if FooGroupObjectPermission.objects.filter(permission_id=perm_id,
content_object__in=gen).exists()
return True
return False
不幸的是,您只能预取和缓存每个对象,而不能跨查询所有对象进行缓存,因为django-guardian
在检查每个对象时会生成一个Rules
层次结构检查的SQL子查询,这是无法预先缓存的。此外,在Django中注册两个权限后端意味着每次都会检查两个权限后端,如果名称重复可能会导致意外批准。
因此,我选择了类似于Rules
的东西,但它会在执行时惰性地编译为Queryset过滤器。以下是一个人为制造的示例,以演示Guardian权限、Django Q对象和惰性评估Q对象的布尔组合成规则:
add_rule('kimsim_app.model_run.view',
LazyGuardianPermission('kimsim_app.view_model') &
(
LazyGuardianPermission('kimsim_app.saved_model') |
LazyGuardianPermission('kimsim_app.saved_model')
)
& ~LazyQ('modelgroupobjectpermission__group__user', 'request.user')
& ~Q(number_of_failures__lte=42)
)
lazy = LazyPermission('kimsim_app.model_run.view') & ~LazyGuardianPermission('kimsim_app.view_model')
Model.objects.filter(lazy.convert_to_q(user=u))
要判断用户是否有访问模型m
的权限(如果有,返回m
,否则返回None
):
Model.objects.filter(lazy.convert_to_q(user=u, obj=m))
这可以构建成DRF、Django admin、模板标签等的权限后端。代码如下:
class BaseLazyQ(Q):
"""
Is the type instantiated by the Q library when it parses the operators linking Q, LazyQ, LazyPermission and other
BaseLazyQ subclasses, and generates the 'lazy' Q node tree for a rule.
"""
def __init__(self, *args, **kwargs):
# If no args, this is a connector node joining two sub-clauses
# Or if args[0] is not a string, this is a standard, fully declared Q object
super(BaseLazyQ, self).__init__(*args, **kwargs)
logging.info('Instantiated fully declared BaseLazyQ %s, children %s', self.connector, self.children)
# These two overrides force the connector nodes to be of type `BaseLazyQ`, ie, `LazyQ() & LazyGuardian()`
# creates a BaseLazyQ connector node. These are then simply cloned on calling `convert_to_q()`.
# Sub-classes do different conversion processing in `convert_to_q`.
# They also disable `squash`, as child Q objects have not been instantiated yet: 'lazy' Q or
# guardian permissions are still awaiting values request & obj.
def _combine(self, other, conn):
if not isinstance(other, Q):
raise TypeError(other)
obj = BaseLazyQ()
obj.connector = conn
obj.add(self, conn, squash=False)
obj.add(other, conn, squash=False)
return obj
def __invert__(self):
obj = BaseLazyQ()
obj.add(self, self.AND, squash=False)
obj.negate()
return obj
def convert_to_q(self, *args, **kwargs):
"""
Generates a tree of fully specified Q() objects at run time from our tree of lazy Q, Guardian and LazyPermission
objects, by passing them the ``request`` and ``obj`` objects of the current request.
Note that only kwargs ``request`` or ``obj`` can be used if you will integrate with Django ModelAdmin and
django-restframework permissions classes.
:param request: From the current request
:param obj: Optional - the object permissions are being tested for, if this is object-specific.
:return: A tree of Q() objects that can be applied to a queryset of type ``obj``
"""
logging.info('Converting fully declared BaseLazyQ conn %s children %s', self.connector, self.children)
q = self.__class__._new_instance(children=[], connector=self.connector, negated=self.negated)
for predicate in self.children:
if isinstance(predicate, BaseLazyQ):
# Including subclasses
q.children.append(predicate.convert_to_q(*args, **kwargs))
else:
# Q or Node
q.children.append(predicate.clone())
logging.info('Cloning child Q %s', predicate)
return q
class AlwaysQ(BaseLazyQ):
"""
This class is used for permissions that are always granted or denied regardless of user, request, object, etc.
"""
def __init__(self, always_allow, *args, **kwargs):
"""
Initializes a class which always permits or denies a particular permission. Still subject to boolean operators,
ie, `AlwaysQ('allow') & [some failing test/s]` will refuse permission.
Likewise `AlwaysQ('deny') | [some passing test/s]` will grant permission.
:param always_allow: Must be set to `'allow'` to always allow, or `'deny'` to always deny.
"""
super(AlwaysQ, self).__init__(*args, **kwargs)
if not always_allow in ['allow', 'deny']:
raise LazyPermDeclarationError('AlwaysQ must be declared as either \'allow\' or \'deny\'.')
self.always_allow = always_allow
def convert_to_q(self, *args, **kwargs):
return Q(pk__isnull=not self.always_allow)
class LazyQ(BaseLazyQ):
def __init__(self, *args, **kwargs):
super(LazyQ, self).__init__(*args, **kwargs)
if args and len(args) == 2 and isinstance(args[0], str) and isinstance(args[1], str):
logging.info('Instantiating LazyQ %s %s', args[0], args[1])
self.field = args[0]
attrs = args[1].split('.')
self.parameter = attrs[0]
self.attributes = attrs[1:]
else:
raise LazyPermDeclarationError('LazyQ must be declared with a Q query string and the naming of the '
'parameter attributes to assign it.')
def convert_to_q(self, *args, **kwargs):
"""
Generates a tree of fully specified Q() objects at run time, from our tree of lazy LazyPermission() and LazyQ()
objects, by passing them the ``request`` and ``obj`` objects of the current request.
Note that only kwargs ``request`` or ``obj`` can be used if we are to integrate with Django ModelAdmin and
django-restframework permissions classes.
:param request: From the current request
:param obj: Optional - the object permissions are being tested for, if this is object-specific.
:return: A tree of Q() objects that can be applied to a queryset of type ``obj``
"""
logging.info('Converting LazyQ conn %s children %s args %s kwargs %s', self.connector, self.children, args, kwargs)
value = kwargs[self.parameter]
for attr in self.attributes:
value = getattr(value, attr)
logging.info('attr %s = %s', attr, value)
return Q((self.field, value))
class LazyGuardianPermission(BaseLazyQ):
"""
This class supports lazy guardian permissions, whose request and obj are to be passed at runtime.
"""
def __init__(self, permission, globals_override=False, use_groups=True, related_object=None, *args, **kwargs):
"""
Instantiates a lazy guardian permission that can later be converted to fully defined Q objects when passed
request and (optionally) obj at request time.
:param permission: The fully qualified guardian permission name, including the app label, eg, app.action_model
:param globals_override: If 'allow', if the user has the permission on the model, then they have the
permission on every object. It is not possible to disable global permissions inherited through group ownership.
If 'deny', the user must have *both* the global permission, and the permission on the object. Removing the
global permission for a user effectively removes their permission on all that model's objects.
It is not possible to disable global permissions inherited through group ownership, and so only use those
allocated to a user, ie, use_groups has no effect and is always True for global permission checks, as they are
provided by the Django auth ModelBackend.
Default False, which means global permissions are ignored.
:param use_groups: If False, permissions a user has by group membership will not be considered.
Default True, which means this check will check the permissions of groups the user is in. Note that this does
not affect the `allow_groups` option, or checks that are not object-specific. These will always include group
permissions, as determined by the Django auth ModelBackend.
:param related_object: If the guardian permission is on a model related to the current one, this is the
query string path from the current model to that model.
"""
logging.info('Instantiating LazyGuardianPermission %s', permission)
super(LazyGuardianPermission, self).__init__(*args, **kwargs)
if isinstance(permission, str):
perm_elems = permission.split('.')
if len(perm_elems) == 2:
# This specifies a guardian permission
self.app_label = perm_elems[0]
try:
self.permission = Permission.objects.select_related('content_type')\
.get(content_type__app_label=perm_elems[0],
codename=perm_elems[1])
except Permission.DoesNotExist:
raise LazyPermDeclarationError('Guardian permission %s not found. LazyGuardianPermission must be '
'passed a fully qualified guardian permission, eg, '
'app.action_model. Q, LazyQ or LazyPermission objects can also be '
'used.' % permission)
self.related_object = related_object
self.use_groups = use_groups
self.globals_override = globals_override
self.model_cls = self.permission.content_type.model_class()
else:
raise LazyPermDeclarationError('Guardian permission %s not found. LazyGuardianPermission must contain '
'a fully qualified guardian permission, eg, app_action_model. Q, LazyQ '
'or LazyPermission objects can also be used.' % permission)
else:
raise LazyPermDeclarationError('LazyGuardianPermission must be declared with a fully qualified guardian '
'permission name, eg, app.action_model. <%s> not a valid parameter.' %
str(permission))
def convert_to_q(self, user, obj=None):
"""
Generates a tree of fully specified Q() objects at run time to test this Guardian permission, by passing them
the `request` and `obj` objects of the current request.
:param user: From the current request
:param obj: Optional - the object permissions are being tested for, if this is object-specific.
:return: A tree of Q() objects that can be applied to a queryset of type ``obj``
"""
logging.info('Converting LazyGuardianPermission %s%s', '~' if self.negated else '', self.permission.codename)
if self.globals_override:
has_global = user.has_perm('%s.%s' % (self.app_label, self.permission.codename))
if has_global and self.globals_override == 'allow':
return Q(pk__isnull=False)
elif not has_global and self.globals_override == 'deny':
return Q(pk__isnull=True)
related_object_prefix = '%s__' % self.related_object if self.related_object else ''
user_obj_perms_model = get_user_obj_perms_model(self.model_cls)
group_obj_perms_model = get_group_obj_perms_model(self.model_cls)
# logging.info('%s %s %s', self.model_cls, user_obj_perms_model, user_obj_perms_model.objects)
if user_obj_perms_model.objects.is_generic():
raise LazyPermDeclarationError('%s appears to be using generic foreign keys. LazyPermissions '
'does not support Guardian permissions maintained via generic '
'foreign keys, and insists you specify a custom table joining '
'object, permission and user, for example `class '
'DatasetUserObjectPermission(UserObjectPermissionBase): '
'content_object = models.ForeignKey(Dataset)` and likewise '
'for Groups. This is also more performant and maintains '
'referential integrity.' % self.permission)
user_obj_perms_model_ref = '%s%s' % (related_object_prefix,
user_obj_perms_model.content_object.field.related_query_name())
if obj:
filters = (
Q(('%s__user' % user_obj_perms_model_ref, user)) &
Q(('%s__permission' % user_obj_perms_model_ref, self.permission)) &
Q(('%s__content_object' % user_obj_perms_model_ref, obj.pk))
)
else:
filters = (
Q(('%s__user' % user_obj_perms_model_ref, user)) &
Q(('%s__permission' % user_obj_perms_model_ref, self.permission))
)
if self.use_groups:
if user_obj_perms_model.objects.is_generic():
raise LazyPermDeclarationError('%s appears to be using generic foreign keys. LazyPermissions '
'does not support Guardian permissions maintained via generic '
'foreign keys, and insists you specify a custom table joining '
'object, permission and user, for example `class '
'DatasetGroupObjectPermission(GroupObjectPermissionBase): '
'content_object = models.ForeignKey(Dataset)` and likewise '
'for Users. This is also more performant and maintains '
'referential integrity.' % self.permission)
group_obj_perms_model_ref = '%s%s' % (related_object_prefix,
group_obj_perms_model.content_object.field.related_query_name())
if obj:
filters |= (
Q(('%s__group__user' % group_obj_perms_model_ref, user)) &
Q(('%s__permission' % group_obj_perms_model_ref, self.permission)) &
Q(('%s__content_object' % group_obj_perms_model_ref, obj.pk))
)
else:
filters |= (
Q(('%s__group__user' % group_obj_perms_model_ref, user)) &
Q(('%s__permission' % group_obj_perms_model_ref, self.permission))
)
logging.info('Converted non-declared LazyGuardianPermission %s%s filters %s',
'~' if self.negated else '', self.permission.codename, filters)
return Q(filters)
class LazyPermission(BaseLazyQ):
"""
This class supports recursive LazyPermission references, converted to lazy q or guardian checks on
declaration then treated identically on calling.
"""
def __init__(self, permission=None, *args, **kwargs):
logging.info('instantiating gorm permission=%s', permission)
super(LazyPermission, self).__init__(*args, **kwargs)
if isinstance(permission, str):
try:
# This is a recursive LazyPermission reference, so add it as a sub-tree
self.children.append(default_rules[permission])
except KeyError:
raise LazyPermDeclarationError('%s not found in rule_set. LazyPermission must contain a fully '
'qualified guardian permission, eg, app.action_model, or another '
'LazyPermission\'s key.' % permission)
logging.info('Instantiated LazyPermission %s as LazyGuardianPermission sub-tree %s.',
permission, self.children)
else:
raise LazyPermDeclarationError('LazyPermission must be declared with either a fully qualified guardian '
'permission, eg, app.action_model, or another LazyPermission\' key.')
class RuleSet(dict):
def test_rule(self, name, *args, **kwargs):
return name in self and self[name].convert_to_q(*args, **kwargs)
def rule_exists(self, name):
return name in self
def add_rule(self, name, pred):
if name in self:
raise KeyError('A rule with name `%s` already exists' % name)
self[name] = pred
def remove_rule(self, name):
del self[name]