Add web UI view tests for object-level permissions

netbox/ipam/views.py

@@ -8,6 +8,7 @@ from django.views.generic import View
 from django_tables2 import RequestConfig
 
 from dcim.models import Device, Interface
+from netbox.authentication import ObjectPermissionRequiredMixin
 from utilities.paginator import EnhancedPaginator
 from utilities.views import (
     BulkCreateView, BulkDeleteView, BulkEditView, BulkImportView, ObjectDeleteView, ObjectEditView, ObjectListView,
@@ -440,7 +441,7 @@ class RoleBulkDeleteView(PermissionRequiredMixin, BulkDeleteView):
 # Prefixes
 #
 
-class PrefixListView(PermissionRequiredMixin, ObjectListView):
+class PrefixListView(ObjectPermissionRequiredMixin, ObjectListView):
     permission_required = 'ipam.view_prefix'
     queryset = Prefix.objects.prefetch_related('site', 'vrf__tenant', 'tenant', 'vlan', 'role')
     filterset = filters.PrefixFilterSet
@@ -454,14 +455,13 @@ class PrefixListView(PermissionRequiredMixin, ObjectListView):
         return self.queryset.annotate_depth(limit=limit)
 
 
-class PrefixView(PermissionRequiredMixin, View):
+class PrefixView(ObjectPermissionRequiredMixin, View):
     permission_required = 'ipam.view_prefix'
+    queryset = Prefix.objects.prefetch_related('vrf', 'site__region', 'tenant__group', 'vlan__group', 'role')
 
     def get(self, request, pk):
 
-        prefix = get_object_or_404(Prefix.objects.prefetch_related(
-            'vrf', 'site__region', 'tenant__group', 'vlan__group', 'role'
-        ), pk=pk)
+        prefix = get_object_or_404(self.queryset, pk=pk)
 
         try:
             aggregate = Aggregate.objects.get(prefix__net_contains_or_equals=str(prefix.prefix))
@@ -586,7 +586,7 @@ class PrefixIPAddressesView(PermissionRequiredMixin, View):
         })
 
 
-class PrefixCreateView(PermissionRequiredMixin, ObjectEditView):
+class PrefixCreateView(ObjectPermissionRequiredMixin, ObjectEditView):
     permission_required = 'ipam.add_prefix'
     queryset = Prefix.objects.all()
     model_form = forms.PrefixForm
@@ -598,7 +598,7 @@ class PrefixEditView(PrefixCreateView):
     permission_required = 'ipam.change_prefix'
 
 
-class PrefixDeleteView(PermissionRequiredMixin, ObjectDeleteView):
+class PrefixDeleteView(ObjectPermissionRequiredMixin, ObjectDeleteView):
     permission_required = 'ipam.delete_prefix'
     queryset = Prefix.objects.all()
     template_name = 'ipam/prefix_delete.html'

netbox/netbox/tests/test_authentication.py

@@ -1,8 +1,16 @@
 from django.conf import settings
 from django.contrib.auth.models import Group, User
-from django.test import Client, TestCase
+from django.contrib.contenttypes.models import ContentType
+from django.test import Client
 from django.test.utils import override_settings
 from django.urls import reverse
+from netaddr import IPNetwork
+
+from dcim.models import Site
+from ipam.choices import PrefixStatusChoices
+from ipam.models import Prefix
+from users.models import ObjectPermission
+from utilities.testing.testcases import TestCase
 
 
 class ExternalAuthenticationTestCase(TestCase):
@@ -157,3 +165,214 @@ class ExternalAuthenticationTestCase(TestCase):
         new_user = User.objects.get(username='remoteuser2')
         self.assertEqual(int(self.client.session.get('_auth_user_id')), new_user.pk, msg='Authentication failed')
         self.assertTrue(new_user.has_perms(['dcim.add_site', 'dcim.change_site']))
+
+
+class ObjectPermissionTestCase(TestCase):
+
+    @classmethod
+    def setUpTestData(cls):
+
+        cls.sites = (
+            Site(name='Site 1', slug='site-1'),
+            Site(name='Site 2', slug='site-2'),
+            Site(name='Site 3', slug='site-3'),
+        )
+        Site.objects.bulk_create(cls.sites)
+
+        cls.prefixes = (
+            Prefix(prefix=IPNetwork('10.0.0.0/24'), site=cls.sites[0]),
+            Prefix(prefix=IPNetwork('10.0.1.0/24'), site=cls.sites[0]),
+            Prefix(prefix=IPNetwork('10.0.2.0/24'), site=cls.sites[0]),
+            Prefix(prefix=IPNetwork('10.0.3.0/24'), site=cls.sites[1]),
+            Prefix(prefix=IPNetwork('10.0.4.0/24'), site=cls.sites[1]),
+            Prefix(prefix=IPNetwork('10.0.5.0/24'), site=cls.sites[1]),
+            Prefix(prefix=IPNetwork('10.0.6.0/24'), site=cls.sites[2]),
+            Prefix(prefix=IPNetwork('10.0.7.0/24'), site=cls.sites[2]),
+            Prefix(prefix=IPNetwork('10.0.8.0/24'), site=cls.sites[2]),
+        )
+        Prefix.objects.bulk_create(cls.prefixes)
+
+    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
+    def test_ui_get_object(self):
+
+        # Assign object permission
+        obj_perm = ObjectPermission(
+            model=ContentType.objects.get_for_model(Prefix),
+            attrs={
+                'site__name': 'Site 1',
+            },
+            can_view=True
+        )
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+
+        # Retrieve permitted object
+        response = self.client.get(self.prefixes[0].get_absolute_url())
+        self.assertHttpStatus(response, 200)
+
+        # Attempt to retrieve non-permitted object
+        response = self.client.get(self.prefixes[3].get_absolute_url())
+        self.assertHttpStatus(response, 404)
+
+    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
+    def test_ui_list_objects(self):
+
+        # Attempt to list objects without permission
+        response = self.client.get(reverse('ipam:prefix_list'))
+        self.assertHttpStatus(response, 403)
+
+        # Assign object permission
+        obj_perm = ObjectPermission(
+            model=ContentType.objects.get_for_model(Prefix),
+            attrs={
+                'site__name': 'Site 1',
+            },
+            can_view=True
+        )
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+
+        # Retrieve all objects. Only permitted objects should be returned.
+        response = self.client.get(reverse('ipam:prefix_list'))
+        self.assertHttpStatus(response, 200)
+        self.assertIn(str(self.prefixes[0].prefix), str(response.content))
+        self.assertNotIn(str(self.prefixes[3].prefix), str(response.content))
+
+    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
+    def test_ui_create_object(self):
+        initial_count = Prefix.objects.count()
+        form_data = {
+            'prefix': '10.0.9.0/24',
+            'site': self.sites[1].pk,
+            'status': PrefixStatusChoices.STATUS_ACTIVE,
+        }
+
+        # Attempt to create an object without permission
+        request = {
+            'path': reverse('ipam:prefix_add'),
+            'data': form_data,
+            'follow': False,  # Do not follow 302 redirects
+        }
+        response = self.client.post(**request)
+        self.assertHttpStatus(response, 403)
+        self.assertEqual(initial_count, Prefix.objects.count())
+
+        # Assign object permission
+        obj_perm = ObjectPermission(
+            model=ContentType.objects.get_for_model(Prefix),
+            attrs={
+                'site__name': 'Site 1',
+            },
+            can_view=True,
+            can_add=True
+        )
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+
+        # Attempt to create a non-permitted object
+        request = {
+            'path': reverse('ipam:prefix_add'),
+            'data': form_data,
+            'follow': True,  # Follow 302 redirects
+        }
+        response = self.client.post(**request)
+        self.assertHttpStatus(response, 200)
+        self.assertEqual(initial_count, Prefix.objects.count())
+
+        # Create a permitted object
+        form_data['site'] = self.sites[0].pk
+        request = {
+            'path': reverse('ipam:prefix_add'),
+            'data': form_data,
+            'follow': True,  # Follow 302 redirects
+        }
+        response = self.client.post(**request)
+        self.assertHttpStatus(response, 200)
+        self.assertEqual(initial_count + 1, Prefix.objects.count())
+
+    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
+    def test_ui_edit_object(self):
+        form_data = {
+            'prefix': '10.0.9.0/24',
+            'site': self.sites[0].pk,
+            'status': PrefixStatusChoices.STATUS_RESERVED,
+        }
+
+        # Attempt to edit an object without permission
+        request = {
+            'path': reverse('ipam:prefix_edit', kwargs={'pk': self.prefixes[0].pk}),
+            'data': form_data,
+            'follow': False,  # Do not follow 302 redirects
+        }
+        response = self.client.post(**request)
+        self.assertHttpStatus(response, 403)
+
+        # Assign object permission
+        obj_perm = ObjectPermission(
+            model=ContentType.objects.get_for_model(Prefix),
+            attrs={
+                'site__name': 'Site 1',
+            },
+            can_view=True,
+            can_change=True
+        )
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+
+        # Attempt to edit a non-permitted object
+        request = {
+            'path': reverse('ipam:prefix_edit', kwargs={'pk': self.prefixes[3].pk}),
+            'data': form_data,
+            'follow': True,  # Follow 302 redirects
+        }
+        response = self.client.post(**request)
+        self.assertHttpStatus(response, 404)
+
+        # Edit a permitted object
+        request = {
+            'path': reverse('ipam:prefix_edit', kwargs={'pk': self.prefixes[0].pk}),
+            'data': form_data,
+            'follow': True,  # Follow 302 redirects
+        }
+        response = self.client.post(**request)
+        self.assertHttpStatus(response, 200)
+        prefix = Prefix.objects.get(pk=self.prefixes[0].pk)
+        self.assertEqual(prefix.status, PrefixStatusChoices.STATUS_RESERVED)
+
+    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
+    def test_ui_delete_object(self):
+        form_data = {
+            'confirm': True
+        }
+
+        # Assign object permission
+        obj_perm = ObjectPermission(
+            model=ContentType.objects.get_for_model(Prefix),
+            attrs={
+                'site__name': 'Site 1',
+            },
+            can_view=True,
+            can_delete=True
+        )
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+
+        # Delete permitted object
+        request = {
+            'path': reverse('ipam:prefix_delete', kwargs={'pk': self.prefixes[0].pk}),
+            'data': form_data,
+            'follow': True,  # Follow 302 redirects
+        }
+        response = self.client.post(**request)
+        self.assertHttpStatus(response, 200)
+        self.assertFalse(Prefix.objects.filter(pk=self.prefixes[0].pk).exists())
+
+        # Attempt to delete non-permitted object
+        request = {
+            'path': reverse('ipam:prefix_delete', kwargs={'pk': self.prefixes[3].pk}),
+            'data': form_data,
+            'follow': True,  # Follow 302 redirects
+        }
+        response = self.client.post(**request)
+        self.assertHttpStatus(response, 404)
+        self.assertTrue(Prefix.objects.filter(pk=self.prefixes[3].pk).exists())