mirror of
https://github.com/netbox-community/netbox.git
synced 2024-05-10 07:54:54 +00:00
Remove ViewExemptBackend; use same for model- and object-level permissions
This commit is contained in:
@ -13,19 +13,24 @@ class ObjectPermissionRequiredMixin(AccessMixin):
|
||||
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):
|
||||
user = self.request.user
|
||||
|
||||
# First, check that the user is granted the required permission at either the model or object level.
|
||||
if not user.has_perm(self.permission_required):
|
||||
return False
|
||||
|
||||
# Next, determine whether the permission is model-level or object-level. Model-level permissions grant the
|
||||
# Superusers implicitly have all permissions
|
||||
if user.is_superuser:
|
||||
return True
|
||||
|
||||
# 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.request.user.is_superuser or self.permission_required in self.request.user._perm_cache:
|
||||
if self.permission_required in {*user._user_perm_cache, *user._group_perm_cache}:
|
||||
return True
|
||||
|
||||
# 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)
|
||||
attrs = ObjectPermission.objects.get_attr_constraints(user, self.permission_required)
|
||||
if attrs:
|
||||
# Update the view's QuerySet to filter only the permitted objects
|
||||
self.queryset = self.queryset.filter(attrs)
|
||||
|
@ -333,8 +333,7 @@ TEMPLATES = [
|
||||
|
||||
# Set up authentication backends
|
||||
AUTHENTICATION_BACKENDS = [
|
||||
REMOTE_AUTH_BACKEND,
|
||||
'utilities.auth_backends.ViewExemptModelBackend',
|
||||
# REMOTE_AUTH_BACKEND,
|
||||
'utilities.auth_backends.ObjectPermissionBackend',
|
||||
]
|
||||
|
||||
|
@ -5,11 +5,12 @@ from django.test import Client
|
||||
from django.test.utils import override_settings
|
||||
from django.urls import reverse
|
||||
from netaddr import IPNetwork
|
||||
from rest_framework.test import APIClient
|
||||
|
||||
from dcim.models import Site
|
||||
from ipam.choices import PrefixStatusChoices
|
||||
from ipam.models import Prefix
|
||||
from users.models import ObjectPermission
|
||||
from users.models import ObjectPermission, Token
|
||||
from utilities.testing.testcases import TestCase
|
||||
|
||||
|
||||
@ -167,7 +168,7 @@ class ExternalAuthenticationTestCase(TestCase):
|
||||
self.assertTrue(new_user.has_perms(['dcim.add_site', 'dcim.change_site']))
|
||||
|
||||
|
||||
class ObjectPermissionTestCase(TestCase):
|
||||
class ObjectPermissionViewTestCase(TestCase):
|
||||
|
||||
@classmethod
|
||||
def setUpTestData(cls):
|
||||
@ -193,14 +194,16 @@ class ObjectPermissionTestCase(TestCase):
|
||||
Prefix.objects.bulk_create(cls.prefixes)
|
||||
|
||||
@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
|
||||
def test_ui_get_object(self):
|
||||
def test_get_object(self):
|
||||
|
||||
# Attempt to retrieve object without permission
|
||||
response = self.client.get(self.prefixes[0].get_absolute_url())
|
||||
self.assertHttpStatus(response, 403)
|
||||
|
||||
# Assign object permission
|
||||
obj_perm = ObjectPermission(
|
||||
model=ContentType.objects.get_for_model(Prefix),
|
||||
attrs={
|
||||
'site__name': 'Site 1',
|
||||
},
|
||||
attrs={'site__name': 'Site 1'},
|
||||
can_view=True
|
||||
)
|
||||
obj_perm.save()
|
||||
@ -215,7 +218,7 @@ class ObjectPermissionTestCase(TestCase):
|
||||
self.assertHttpStatus(response, 404)
|
||||
|
||||
@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
|
||||
def test_ui_list_objects(self):
|
||||
def test_list_objects(self):
|
||||
|
||||
# Attempt to list objects without permission
|
||||
response = self.client.get(reverse('ipam:prefix_list'))
|
||||
@ -224,9 +227,7 @@ class ObjectPermissionTestCase(TestCase):
|
||||
# Assign object permission
|
||||
obj_perm = ObjectPermission(
|
||||
model=ContentType.objects.get_for_model(Prefix),
|
||||
attrs={
|
||||
'site__name': 'Site 1',
|
||||
},
|
||||
attrs={'site__name': 'Site 1'},
|
||||
can_view=True
|
||||
)
|
||||
obj_perm.save()
|
||||
@ -239,7 +240,7 @@ class ObjectPermissionTestCase(TestCase):
|
||||
self.assertNotIn(str(self.prefixes[3].prefix), str(response.content))
|
||||
|
||||
@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
|
||||
def test_ui_create_object(self):
|
||||
def test_create_object(self):
|
||||
initial_count = Prefix.objects.count()
|
||||
form_data = {
|
||||
'prefix': '10.0.9.0/24',
|
||||
@ -260,9 +261,7 @@ class ObjectPermissionTestCase(TestCase):
|
||||
# Assign object permission
|
||||
obj_perm = ObjectPermission(
|
||||
model=ContentType.objects.get_for_model(Prefix),
|
||||
attrs={
|
||||
'site__name': 'Site 1',
|
||||
},
|
||||
attrs={'site__name': 'Site 1'},
|
||||
can_view=True,
|
||||
can_add=True
|
||||
)
|
||||
@ -277,7 +276,7 @@ class ObjectPermissionTestCase(TestCase):
|
||||
}
|
||||
response = self.client.post(**request)
|
||||
self.assertHttpStatus(response, 200)
|
||||
self.assertEqual(initial_count, Prefix.objects.count())
|
||||
self.assertEqual(Prefix.objects.count(), initial_count)
|
||||
|
||||
# Create a permitted object
|
||||
form_data['site'] = self.sites[0].pk
|
||||
@ -288,10 +287,10 @@ class ObjectPermissionTestCase(TestCase):
|
||||
}
|
||||
response = self.client.post(**request)
|
||||
self.assertHttpStatus(response, 200)
|
||||
self.assertEqual(initial_count + 1, Prefix.objects.count())
|
||||
self.assertEqual(Prefix.objects.count(), initial_count + 1)
|
||||
|
||||
@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
|
||||
def test_ui_edit_object(self):
|
||||
def test_edit_object(self):
|
||||
form_data = {
|
||||
'prefix': '10.0.9.0/24',
|
||||
'site': self.sites[0].pk,
|
||||
@ -310,9 +309,7 @@ class ObjectPermissionTestCase(TestCase):
|
||||
# Assign object permission
|
||||
obj_perm = ObjectPermission(
|
||||
model=ContentType.objects.get_for_model(Prefix),
|
||||
attrs={
|
||||
'site__name': 'Site 1',
|
||||
},
|
||||
attrs={'site__name': 'Site 1'},
|
||||
can_view=True,
|
||||
can_change=True
|
||||
)
|
||||
@ -340,7 +337,7 @@ class ObjectPermissionTestCase(TestCase):
|
||||
self.assertEqual(prefix.status, PrefixStatusChoices.STATUS_RESERVED)
|
||||
|
||||
@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
|
||||
def test_ui_delete_object(self):
|
||||
def test_delete_object(self):
|
||||
form_data = {
|
||||
'confirm': True
|
||||
}
|
||||
@ -348,9 +345,7 @@ class ObjectPermissionTestCase(TestCase):
|
||||
# Assign object permission
|
||||
obj_perm = ObjectPermission(
|
||||
model=ContentType.objects.get_for_model(Prefix),
|
||||
attrs={
|
||||
'site__name': 'Site 1',
|
||||
},
|
||||
attrs={'site__name': 'Site 1'},
|
||||
can_view=True,
|
||||
can_delete=True
|
||||
)
|
||||
|
@ -8,115 +8,89 @@ from django.db.models import Q
|
||||
from users.models import ObjectPermission
|
||||
|
||||
|
||||
class ViewExemptModelBackend(ModelBackend):
|
||||
class ObjectPermissionBackend(ModelBackend):
|
||||
|
||||
def get_object_permissions(self, user_obj):
|
||||
"""
|
||||
Custom implementation of Django's stock ModelBackend which allows for the exemption of arbitrary models from view
|
||||
permission enforcement.
|
||||
Return all model-level permissions granted to the user by an ObjectPermission.
|
||||
"""
|
||||
def _get_user_permissions(self, user_obj):
|
||||
if not hasattr(user_obj, '_object_perm_cache'):
|
||||
|
||||
if not settings.EXEMPT_VIEW_PERMISSIONS:
|
||||
# No view permissions have been exempted from enforcement, so fall back to the built-in logic.
|
||||
return super()._get_user_permissions(user_obj)
|
||||
# Cache all assigned ObjectPermissions on the User instance
|
||||
perms = 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}"):
|
||||
perms.add(f"{obj_perm.model.app_label}.{action}_{obj_perm.model.model}")
|
||||
setattr(user_obj, '_object_perm_cache', perms)
|
||||
|
||||
if '*' in settings.EXEMPT_VIEW_PERMISSIONS:
|
||||
# All view permissions have been exempted from enforcement, so include all view permissions when fetching
|
||||
# User permissions.
|
||||
return Permission.objects.filter(
|
||||
Q(user=user_obj) | Q(codename__startswith='view_')
|
||||
)
|
||||
return user_obj._object_perm_cache
|
||||
|
||||
# Return all Permissions that are either assigned to the user or that are view permissions listed in
|
||||
# EXEMPT_VIEW_PERMISSIONS.
|
||||
qs_filter = Q(user=user_obj)
|
||||
for model in settings.EXEMPT_VIEW_PERMISSIONS:
|
||||
app, name = model.split('.')
|
||||
qs_filter |= Q(content_type__app_label=app, codename=f'view_{name}')
|
||||
return Permission.objects.filter(qs_filter)
|
||||
def get_all_permissions(self, user_obj, obj=None):
|
||||
|
||||
# Handle inactive/anonymous users
|
||||
if not user_obj.is_active or user_obj.is_anonymous:
|
||||
return set()
|
||||
|
||||
# Cache model-level permissions on the User instance
|
||||
if not hasattr(user_obj, '_perm_cache'):
|
||||
user_obj._perm_cache = {
|
||||
*self.get_user_permissions(user_obj, obj=obj),
|
||||
*self.get_group_permissions(user_obj, obj=obj),
|
||||
*self.get_object_permissions(user_obj)
|
||||
}
|
||||
|
||||
return user_obj._perm_cache
|
||||
|
||||
def has_perm(self, user_obj, perm, obj=None):
|
||||
|
||||
# Authenticated users need to have the view permissions cached for assessment
|
||||
if user_obj.is_authenticated:
|
||||
return super().has_perm(user_obj, perm, obj)
|
||||
app_label, codename = perm.split('.')
|
||||
action, model_name = codename.split('_')
|
||||
|
||||
# If this is a view permission, check whether the model has been exempted from enforcement
|
||||
try:
|
||||
app, codename = perm.split('.')
|
||||
action, model = codename.split('_')
|
||||
if action == 'view':
|
||||
if (
|
||||
# All models are exempt from view permission enforcement
|
||||
'*' in settings.EXEMPT_VIEW_PERMISSIONS
|
||||
) or (
|
||||
# This specific model is exempt from view permission enforcement
|
||||
'{}.{}'.format(app, model) in settings.EXEMPT_VIEW_PERMISSIONS
|
||||
'{}.{}'.format(app_label, model_name) in settings.EXEMPT_VIEW_PERMISSIONS
|
||||
):
|
||||
return True
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
|
||||
class ObjectPermissionBackend(ModelBackend):
|
||||
"""
|
||||
Evaluates permission of a user to access or modify a specific object based on the assignment of ObjectPermissions
|
||||
either to the user directly or to a group of which the user is a member. Model-level permissions supersede this
|
||||
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):
|
||||
|
||||
# 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 no object is specified, evaluate model-level permissions. The presence of a permission in this set tells
|
||||
# us that the user has permission for *some* objects, but not necessarily a specific object.
|
||||
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
|
||||
|
||||
# Sanity check: Ensure that the requested permission applies to the specified object
|
||||
model = obj._meta.model
|
||||
|
||||
# Check that the requested permission applies to the specified object
|
||||
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.
|
||||
if model.objects.filter(attrs, pk=obj.pk).exists():
|
||||
# If the user has been granted model-level permission for the object, return True
|
||||
model_perms = {
|
||||
*self.get_user_permissions(user_obj),
|
||||
*self.get_group_permissions(user_obj),
|
||||
}
|
||||
if perm in model_perms:
|
||||
return True
|
||||
|
||||
# Gather all ObjectPermissions pertinent to the requested permission. If none are found, the User has no
|
||||
# applicable permissions.
|
||||
attrs = ObjectPermission.objects.get_attr_constraints(user_obj, perm)
|
||||
if not attrs:
|
||||
return False
|
||||
|
||||
class RemoteUserBackend(ViewExemptModelBackend, RemoteUserBackend_):
|
||||
# Permission to perform the requested action on the object depends on whether the specified object matches
|
||||
# the specified attributes. Note that this check is made against the *database* record representing the object,
|
||||
# not the instance itself.
|
||||
return model.objects.filter(attrs, pk=obj.pk).exists()
|
||||
|
||||
|
||||
class RemoteUserBackend(RemoteUserBackend_):
|
||||
"""
|
||||
Custom implementation of Django's RemoteUserBackend which provides configuration hooks for basic customization.
|
||||
"""
|
||||
|
Reference in New Issue
Block a user