Cache object-level permissions on the User instance for evaluation

netbox/netbox/authentication.py

@@ -1,32 +1,50 @@
 from django.contrib.auth.mixins import AccessMixin
-from django.contrib.contenttypes.models import ContentType
-from django.db.models import Q
+from django.core.exceptions import ImproperlyConfigured
 
 from users.models import ObjectPermission
 
 
 class ObjectPermissionRequiredMixin(AccessMixin):
+    """
+    Similar to Django's built-in PermissionRequiredMixin, but extended to check for both model-level and object-level
+    permission assignments. If the user has only object-level permissions assigned, the view's queryset is filtered
+    to return only those objects on which the user is permitted to perform the specified action.
+    """
     permission_required = None
 
     def has_permission(self):
+        # First, check whether the user is granted the requested permissions from any backend.
+        if not self.request.user.has_perm(self.permission_required):
+            return False
 
-        # First, check whether the user has a model-level permission assigned
-        if self.request.user.has_perm(self.permission_required):
+        # Next, determine whether the permission is model-level or object-level. Model-level permissions grant the
+        # specified action to *all* objects, so no further action is needed.
+        if self.permission_required in self.request.user._perm_cache:
             return True
 
-        # If not, check for object-level permissions
-        app, codename = self.permission_required.split('.')
-        action, model_name = codename.split('_')
-        model = self.queryset.model
-        attrs = ObjectPermission.objects.get_attr_constraints(self.request.user, model, action)
-        if attrs:
-            # Update the view's QuerySet to filter only the permitted objects
-            self.queryset = self.queryset.filter(**attrs)
-            return True
-
-        return False
+        # If the permission is granted only at the object level, filter the view's queryset to return only objects
+        # on which the user is permitted to perform the specified action.
+        if self.permission_required in self.request.user._obj_perm_cache:
+            attrs = ObjectPermission.objects.get_attr_constraints(self.request.user, self.permission_required)
+            if attrs:
+                # Update the view's QuerySet to filter only the permitted objects
+                self.queryset = self.queryset.filter(**attrs)
+                return True
 
     def dispatch(self, request, *args, **kwargs):
+        if self.permission_required is None:
+            raise ImproperlyConfigured(
+                '{0} is missing the permission_required attribute. Define {0}.permission_required, or override '
+                '{0}.get_permission_required().'.format(self.__class__.__name__)
+            )
+
+        if not hasattr(self, 'queryset'):
+            raise ImproperlyConfigured(
+                '{} has no queryset defined. ObjectPermissionRequiredMixin may only be used on views which define '
+                'a base queryset'.format(self.__class__.__name__)
+            )
+
         if not self.has_permission():
             return self.handle_no_permission()
+
         return super().dispatch(request, *args, **kwargs)

netbox/users/models.py

@@ -197,16 +197,19 @@ class Token(models.Model):
 
 class ObjectPermissionManager(models.Manager):
 
-    def get_attr_constraints(self, user, model, action):
+    def get_attr_constraints(self, user, perm):
         """
         Compile all ObjectPermission attributes applicable to a specific combination of user, model, and action. Returns
         a dictionary that can be passed directly to .filter() on a QuerySet.
         """
+        app_label, codename = perm.split('.')
+        action, model_name = codename.split('_')
         assert action in ['view', 'add', 'change', 'delete'], f"Invalid action: {action}"
 
+        content_type = ContentType.objects.get(app_label=app_label, model=model_name)
         qs = self.get_queryset().filter(
             Q(users=user) | Q(groups__user=user),
-            model=ContentType.objects.get_for_model(model),
+            model=content_type,
             **{f'can_{action}': True}
         )
 
@@ -216,16 +219,6 @@ class ObjectPermissionManager(models.Manager):
 
         return attrs
 
-    def validate_queryset(self, queryset, user, action):
-        """
-        Check that the specified user has permission to perform the specified action on all objects in the QuerySet.
-        """
-        assert action in ['view', 'add', 'change', 'delete'], f"Invalid action: {action}"
-
-        model = queryset.model
-        attrs = self.get_attr_constraints(user, model, action)
-        return queryset.count() == model.objects.filter(**attrs).count()
-
 
 class ObjectPermission(models.Model):
     """

netbox/utilities/auth_backends.py

@@ -3,6 +3,8 @@ import logging
 from django.conf import settings
 from django.contrib.auth.backends import ModelBackend, RemoteUserBackend as RemoteUserBackend_
 from django.contrib.auth.models import Group, Permission
+from django.contrib.contenttypes.models import ContentType
+from django.db.models import Q
 
 from users.models import ObjectPermission
 
@@ -43,23 +45,54 @@ class ObjectPermissionBackend(ModelBackend):
     check: For example, if a user has the dcim.view_site model-level permission assigned, the ViewExemptModelBackend
     will grant permission before this backend is evaluated for permission to view a specific site.
     """
+    def _get_all_permissions(self, user_obj):
+        """
+        Retrieve all ObjectPermissions assigned to this User (either directly or through a Group) and return the model-
+        level equivalent codenames.
+        """
+        perm_names = set()
+        for obj_perm in ObjectPermission.objects.filter(
+            Q(users=user_obj) | Q(groups__user=user_obj)
+        ).prefetch_related('model'):
+            for action in ['view', 'add', 'change', 'delete']:
+                if getattr(obj_perm, f"can_{action}"):
+                    perm_names.add(f"{obj_perm.model.app_label}.{action}_{obj_perm.model.model}")
+        return perm_names
+
+    def get_all_permissions(self, user_obj, obj=None):
+        """
+        Get all model-level permissions assigned by this backend. Permissions are cached on the User instance.
+        """
+        if not user_obj.is_active or user_obj.is_anonymous:
+            return set()
+        if not hasattr(user_obj, '_obj_perm_cache'):
+            user_obj._obj_perm_cache = self._get_all_permissions(user_obj)
+        return user_obj._obj_perm_cache
+
     def has_perm(self, user_obj, perm, obj=None):
 
-        # This backend only checks for permissions on specific objects
+        # If no object is specified, look for any matching ObjectPermissions. If one or more are found, this indicates
+        # that the user has permission to perform the requested action on at least *some* objects, but not necessarily
+        # on all of them.
         if obj is None:
+            return perm in self.get_all_permissions(user_obj)
+
+        attrs = ObjectPermission.objects.get_attr_constraints(user_obj, perm)
+
+        # No ObjectPermissions found for this combination of user, model, and action
+        if not attrs:
             return
 
-        app, codename = perm.split('.')
-        action, model_name = codename.split('_')
         model = obj._meta.model
 
         # Check that the requested permission applies to the specified object
-        if model._meta.model_name != model_name:
+        app_label, codename = perm.split('.')
+        action, model_name = codename.split('_')
+        if model._meta.label_lower != '.'.join((app_label, model_name)):
             raise ValueError(f"Invalid permission {perm} for model {model}")
 
         # Attempt to retrieve the model from the database using the attributes defined in the
         # ObjectPermission. If we have a match, assert that the user has permission.
-        attrs = ObjectPermission.objects.get_attr_constraints(user_obj, obj, action)
         if model.objects.filter(pk=obj.pk, **attrs).exists():
             return True