Prechádzať zdrojové kódy

Fixes #23398: Omit dependent objects the user cannot view from protected-delete errors

The error reported for a deletion prevented by dependent objects (in the UI and in the REST API's 409 response) now
names only those dependent objects which the user is permitted to view, and reports the remainder as a count. The
object delete view's GET handler also now names the object being deleted, rather than listing the dependent objects
twice.
Jeremy Stretch 2 dní pred
rodič
commit
b8c070bad6

+ 24 - 21
netbox/netbox/api/viewsets/__init__.py

@@ -16,6 +16,7 @@ from rest_framework.viewsets import GenericViewSet
 from netbox.api.fields import RelatedObjectCountField
 from netbox.api.serializers.features import ChangeLogMessageSerializer
 from utilities.api import get_annotations_for_serializer, get_fields_for_serializer, get_prefetches_for_serializer
+from utilities.error_handlers import get_viewable_protected_objects
 from utilities.exceptions import AbortRequest, PreconditionFailed
 from utilities.query import reapply_model_ordering
 
@@ -266,19 +267,7 @@ class NetBoxModelViewSet(
         try:
             return super().dispatch(request, *args, **kwargs)
         except (ProtectedError, RestrictedError) as e:
-            if type(e) is ProtectedError:
-                protected_objects = list(e.protected_objects)
-            else:
-                protected_objects = list(e.restricted_objects)
-            msg = f'Unable to delete object. {len(protected_objects)} dependent objects were found: '
-            msg += ', '.join([f'{obj} ({obj.pk})' for obj in protected_objects])
-            logger.warning(msg)
-            return self.finalize_response(
-                request,
-                Response({'detail': msg}, status=409),
-                *args,
-                **kwargs
-            )
+            return self.finalize_response(request, self._get_protected_error_response(e), *args, **kwargs)
         except AbortRequest as e:
             logger.debug(e.message)
             return self.finalize_response(
@@ -288,6 +277,27 @@ class NetBoxModelViewSet(
                 **kwargs
             )
 
+    def _get_protected_error_response(self, exc):
+        """
+        Log all dependent objects of a ProtectedError or RestrictedError, and return a 409 response listing only those
+        which the user is permitted to view.
+        """
+        def get_message(objects, hidden_count=0):
+            dependent_objects = [f'{obj} ({obj.pk})' for obj in objects]
+            if hidden_count:
+                noun = 'object' if hidden_count == 1 else 'objects'
+                dependent_objects.append(f'{hidden_count} {noun} you do not have permission to view')
+            msg = f'Unable to delete object. {len(objects) + hidden_count} dependent objects were found: '
+            return msg + ', '.join(dependent_objects)
+
+        protected_objects = exc.protected_objects if isinstance(exc, ProtectedError) else exc.restricted_objects
+        logger = logging.getLogger(f'netbox.api.views.{self.__class__.__name__}')
+        logger.warning(get_message(protected_objects))
+
+        return Response({
+            'detail': get_message(*get_viewable_protected_objects(exc, self.request.user)),
+        }, status=409)
+
     def exception_to_response(self, exc):
         """
         Translate a NetBox/Django exception that is not a DRF APIException into the same
@@ -301,14 +311,7 @@ class NetBoxModelViewSet(
         """
         logger = logging.getLogger(f'netbox.api.views.{self.__class__.__name__}')
         if isinstance(exc, (ProtectedError, RestrictedError)):
-            if type(exc) is ProtectedError:
-                protected_objects = list(exc.protected_objects)
-            else:
-                protected_objects = list(exc.restricted_objects)
-            msg = f'Unable to delete object. {len(protected_objects)} dependent objects were found: '
-            msg += ', '.join([f'{obj} ({obj.pk})' for obj in protected_objects])
-            logger.warning(msg)
-            return Response({'detail': msg}, status=409)
+            return self._get_protected_error_response(exc)
         if isinstance(exc, AbortRequest):
             logger.debug(exc.message)
             return Response({'detail': exc.message}, status=400)

+ 1 - 4
netbox/netbox/api/viewsets/mixins.py

@@ -805,10 +805,7 @@ class BulkDestroyModelMixin:
                         e.protected_objects if isinstance(e, ProtectedError) else e.restricted_objects
                     )
                     # Report only the count, not names or PKs, to keep each per-object error
-                    # entry small in a batch response. Note: the single-object delete endpoint
-                    # (NetBoxModelViewSet.dispatch()) does include names and PKs of dependent
-                    # objects, so this is not a hard security boundary — just a narrower
-                    # response shape for the bulk case.
+                    # entry small in a batch response.
                     errors.append({
                         'id': pk,
                         'errors': {

+ 51 - 2
netbox/netbox/tests/test_api.py

@@ -6,6 +6,7 @@ from django.contrib.contenttypes.models import ContentType
 from django.core.exceptions import NON_FIELD_ERRORS
 from django.db import DEFAULT_DB_ALIAS, connection, connections
 from django.db.backends.postgresql.psycopg_any import NumericRange
+from django.db.models import ProtectedError
 from django.test import RequestFactory, TestCase, TransactionTestCase, override_settings, tag
 from django.test.utils import CaptureQueriesContext
 from django.urls import reverse
@@ -17,7 +18,7 @@ from rest_framework.test import APIClient
 
 from core.models import DataFile, DataSource, ObjectType
 from dcim.api.serializers import InterfaceSerializer, RackSerializer
-from dcim.api.views import SiteViewSet
+from dcim.api.views import DeviceTypeViewSet, SiteViewSet
 from dcim.models import Device, Rack, Site
 from extras.models import ExportTemplate
 from ipam.api.serializers import RouteTargetSerializer, VRFSerializer
@@ -28,7 +29,7 @@ from netbox.api.pagination import NetBoxPagination
 from netbox.api.serializers import ValidatedModelSerializer, WritableNestedSerializer
 from netbox.api.viewsets.mixins import BulkUpdateModelMixin
 from users.constants import TOKEN_PREFIX
-from users.models import Token, User
+from users.models import ObjectPermission, Token, User
 from utilities.api import get_serializer_for_model
 from utilities.testing import APITestCase, create_test_device
 from vpn.api.serializers import L2VPNSerializer
@@ -706,3 +707,51 @@ class ObjectUpdateLockingTestCase(APITestCase):
                 with viewset.lock_object():
                     self.assertEqual(viewset.locked_pks, locked_pks)
                 self.assertEqual(viewset.locked_pks, set())
+
+
+class ProtectedObjectsMessageTestCase(APITestCase):
+    """
+    The error for a deletion prevented by dependent objects must name only those dependent objects which the user is
+    permitted to view. (Ref: #23398)
+    """
+
+    @classmethod
+    def setUpTestData(cls):
+        cls.devices = (
+            create_test_device('Device 1'),
+            create_test_device('Device 2'),
+        )
+        cls.device_type = cls.devices[0].device_type
+        cls.expected_message = (
+            f'Unable to delete object. 2 dependent objects were found: Device 1 ({cls.devices[0].pk}), '
+            '1 object you do not have permission to view'
+        )
+
+    def setUp(self):
+        super().setUp()
+        self.add_permissions('dcim.delete_devicetype')
+
+        # Permit viewing only the first Device
+        obj_perm = ObjectPermission.objects.create(
+            name='View Device 1', actions=['view'], constraints={'pk': self.devices[0].pk}
+        )
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(Device))
+
+    def test_delete_object(self):
+        url = reverse('dcim-api:devicetype-detail', kwargs={'pk': self.device_type.pk})
+        with self.assertLogs('netbox.api.views.DeviceTypeViewSet', 'WARNING') as logs:
+            response = self.client.delete(url, **self.header)
+        self.assertHttpStatus(response, 409)
+        self.assertEqual(response.data['detail'], self.expected_message)
+
+        # The log lists all dependent objects
+        self.assertIn(f'Device 2 ({self.devices[1].pk})', logs.output[0])
+
+    def test_exception_to_response(self):
+        viewset = DeviceTypeViewSet()
+        viewset.request = Request(RequestFactory().delete('/'))
+        viewset.request.user = self.user
+        response = viewset.exception_to_response(ProtectedError('', set(self.devices)))
+        self.assertEqual(response.status_code, 409)
+        self.assertEqual(response.data['detail'], self.expected_message)

+ 54 - 3
netbox/netbox/tests/test_views.py

@@ -3,6 +3,7 @@ from contextlib import contextmanager
 from unittest.mock import patch
 
 from django.contrib.contenttypes.models import ContentType
+from django.contrib.messages import get_messages
 from django.core.files.uploadedfile import SimpleUploadedFile
 from django.db.models.signals import pre_delete
 from django.http import HttpResponse
@@ -10,7 +11,7 @@ from django.test import Client, TransactionTestCase, override_settings, tag
 from django.urls import reverse
 from django.utils import timezone
 
-from core.models import DataFile, DataSource, ObjectChange
+from core.models import DataFile, DataSource, ObjectChange, ObjectType
 from dcim.choices import DeviceStatusChoices, InterfaceTypeChoices, SiteStatusChoices
 from dcim.models import Device, DeviceRole, DeviceType, Interface, Manufacturer, Site, VirtualChassis
 from extras.events import enqueue_event
@@ -22,9 +23,9 @@ from ipam.views import VLANGroupVLANsView
 from netbox.choices import CSVDelimiterChoices, ImportFormatChoices, ImportMethodChoices
 from netbox.constants import EMPTY_TABLE_TEXT
 from netbox.search.backends import search_backend
-from users.models import User
+from users.models import ObjectPermission, User
 from utilities.exceptions import AbortRequest
-from utilities.testing import TestCase
+from utilities.testing import TestCase, create_test_device
 from utilities.views import get_action_url
 
 
@@ -600,3 +601,53 @@ class DeleteAtomicityTestCase(DeleteAtomicityMixin, TransactionTestCase):
         # Without the view's atomic block, the member device's cleared vc_priority commits here,
         # leaving a change which dispatches no events
         self.assertDeleteRolledBack(virtual_chassis, device)
+
+
+class ProtectedObjectsMessageTestCase(TestCase):
+    """
+    The error message for a deletion prevented by dependent objects must name only those dependent objects which the
+    user is permitted to view. (Ref: #23398)
+    """
+
+    @classmethod
+    def setUpTestData(cls):
+        cls.devices = (
+            create_test_device('Device 1'),
+            create_test_device('Device 2'),
+        )
+        cls.device_type = cls.devices[0].device_type
+
+    def setUp(self):
+        super().setUp()
+        self.add_permissions('dcim.view_devicetype', 'dcim.delete_devicetype')
+
+        # Permit viewing only the first Device
+        obj_perm = ObjectPermission.objects.create(
+            name='View Device 1', actions=['view'], constraints={'pk': self.devices[0].pk}
+        )
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(Device))
+
+    def assertProtectedObjectsMessage(self, response):
+        self.assertHttpStatus(response, 302)
+        self.assertEqual([str(m) for m in get_messages(response.wsgi_request)], [
+            f'Unable to delete <strong>Device Type 1</strong>. 2 dependent objects were found: '
+            f'<a href="{self.devices[0].get_absolute_url()}">Device 1</a>, '
+            '1 object you do not have permission to view'
+        ])
+
+    def test_get_object_delete_view(self):
+        url = get_action_url(DeviceType, action='delete', kwargs={'pk': self.device_type.pk})
+        self.assertProtectedObjectsMessage(self.client.get(url))
+
+    def test_post_object_delete_view(self):
+        url = get_action_url(DeviceType, action='delete', kwargs={'pk': self.device_type.pk})
+        self.assertProtectedObjectsMessage(self.client.post(url, data={'confirm': True}))
+
+    def test_post_bulk_delete_view(self):
+        response = self.client.post(reverse('dcim:devicetype_bulk_delete'), data={
+            'pk': [self.device_type.pk],
+            'confirm': True,
+            '_confirm': True,
+        })
+        self.assertProtectedObjectsMessage(response)

+ 4 - 6
netbox/netbox/views/generic/object_views.py

@@ -419,11 +419,11 @@ class ObjectDeleteView(GetReturnURLMixin, BaseObjectView):
 
         return dependent_objects
 
-    def _handle_protected_objects(self, obj, protected_objects, request, exc):
+    def _handle_protected_objects(self, obj, request, exc):
         """
         Handle a ProtectedError or RestrictedError exception raised while attempt to resolve dependent objects.
         """
-        handle_protectederror(protected_objects, request, exc)
+        handle_protectederror([obj], request, exc)
 
         if request.htmx:
             return HttpResponse(headers={
@@ -447,10 +447,8 @@ class ObjectDeleteView(GetReturnURLMixin, BaseObjectView):
 
         try:
             dependent_objects = self._get_dependent_objects(obj)
-        except ProtectedError as e:
-            return self._handle_protected_objects(obj, e.protected_objects, request, e)
-        except RestrictedError as e:
-            return self._handle_protected_objects(obj, e.restricted_objects, request, e)
+        except (ProtectedError, RestrictedError) as e:
+            return self._handle_protected_objects(obj, request, e)
 
         # If this is an HTMX request, return only the rendered deletion form as modal content
         if htmx_partial(request):

+ 48 - 8
netbox/utilities/error_handlers.py

@@ -1,5 +1,6 @@
 import platform
 import sys
+from collections import defaultdict
 
 from django.conf import settings
 from django.contrib import messages
@@ -8,38 +9,77 @@ from django.http import JsonResponse
 from django.utils.html import escape
 from django.utils.safestring import mark_safe
 from django.utils.translation import gettext_lazy as _
+from django.utils.translation import ngettext
 from rest_framework import status
 
 __all__ = (
+    'get_viewable_protected_objects',
     'handle_protectederror',
     'handle_rest_api_exception',
 )
 
 
+def get_viewable_protected_objects(e, user):
+    """
+    Return the dependent objects of a ProtectedError or RestrictedError which the user is permitted to view, and the
+    number of those which they are not.
+    """
+    protected_objects = list(e.protected_objects if isinstance(e, ProtectedError) else e.restricted_objects)
+    if user.is_active and user.is_superuser:
+        return protected_objects, 0
+
+    # Group the objects by the database from which each was loaded and by model
+    pks_by_model = defaultdict(list)
+    for obj in protected_objects:
+        pks_by_model[(obj._state.db, obj._meta.model)].append(obj.pk)
+    viewable_pks = defaultdict(set)
+    for (using, model), pks in pks_by_model.items():
+        queryset = model._default_manager.using(using)
+        # Objects of a model which doesn't support restriction are treated as not viewable
+        if hasattr(queryset, 'restrict'):
+            viewable_pks[(using, model)] = set(
+                queryset.restrict(user, 'view').filter(pk__in=pks).values_list('pk', flat=True)
+            )
+
+    viewable_objects = [
+        obj for obj in protected_objects if obj.pk in viewable_pks[(obj._state.db, obj._meta.model)]
+    ]
+    return viewable_objects, len(protected_objects) - len(viewable_objects)
+
+
 def handle_protectederror(obj_list, request, e):
     """
     Generate a user-friendly error message in response to a ProtectedError or RestrictedError exception.
     """
-    if type(e) is ProtectedError:
-        protected_objects = list(e.protected_objects)
-    elif type(e) is RestrictedError:
-        protected_objects = list(e.restricted_objects)
-    else:
+    if type(e) not in (ProtectedError, RestrictedError):
         raise e
+    viewable_objects, hidden_count = get_viewable_protected_objects(e, request.user)
 
     # Formulate the error message
     err_message = _('Unable to delete <strong>{objects}</strong>. {count} dependent objects were found: ').format(
         objects=', '.join(escape(obj) for obj in obj_list),
-        count=len(protected_objects) if len(protected_objects) <= 50 else _('More than 50'),
+        count=len(viewable_objects) + hidden_count,
     )
 
-    # Append dependent objects to error message
+    # Append dependent objects to error message, naming only those which the user is permitted to view
     dependent_objects = []
-    for dependent in protected_objects[:50]:
+    for dependent in viewable_objects[:50]:
         if hasattr(dependent, 'get_absolute_url'):
             dependent_objects.append(f'<a href="{dependent.get_absolute_url()}">{escape(dependent)}</a>')
         else:
             dependent_objects.append(escape(str(dependent)))
+    if (more_count := len(viewable_objects) - 50) > 0:
+        dependent_objects.append(escape(ngettext(
+            '{count} more object',
+            '{count} more objects',
+            more_count,
+        ).format(count=more_count)))
+    if hidden_count:
+        dependent_objects.append(escape(ngettext(
+            '{count} object you do not have permission to view',
+            '{count} objects you do not have permission to view',
+            hidden_count,
+        ).format(count=hidden_count)))
     err_message += ', '.join(dependent_objects)
 
     messages.error(request, mark_safe(err_message))

+ 24 - 1
netbox/utilities/tests/test_error_handlers.py

@@ -1,8 +1,31 @@
 import json
+from unittest.mock import patch
 
+from django.db.models import ProtectedError
 from django.test import RequestFactory, TestCase
 
-from utilities.error_handlers import handle_rest_api_exception
+from dcim.models import Site
+from users.models import User
+from utilities.error_handlers import handle_protectederror, handle_rest_api_exception
+
+
+class HandleProtectedErrorTestCase(TestCase):
+    """
+    Test handle_protectederror() message formatting.
+    """
+
+    def test_dependent_objects_truncated(self):
+        request = RequestFactory().get('/')
+        request.user = User(username='superuser', is_superuser=True)
+        sites = {Site(pk=i, name=f'Site {i}', slug=f'site-{i}') for i in range(1, 53)}
+
+        with patch('utilities.error_handlers.messages') as messages:
+            handle_protectederror([Site(name='Site 0')], request, ProtectedError('', sites))
+
+        message = messages.error.call_args.args[1]
+        self.assertIn('52 dependent objects were found', message)
+        self.assertEqual(message.count('<a href='), 50)
+        self.assertTrue(message.endswith(', 2 more objects'))
 
 
 class HandleRestApiExceptionTestCase(TestCase):