RomanNum3ral
/
netbox
mirror of https://github.com/netbox-community/netbox.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Raise error when attempting to update/delete objects by unknown ID

This commit is contained in:
Jeremy Stretch 2026-08-10 14:58:20 -04:00
parent 02a350dc67
commit e17227d67d
4 changed files with 201 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

55
netbox/dcim/tests/test_api.py
View File

@ -498,6 +498,61 @@ class SiteTestCase(APIViewTestCases.APIViewTestCase):
self.assertTrue(Site.objects.filter(pk=site1.pk).exists(), 'Site 1 should not have been deleted')
self.assertTrue(Site.objects.filter(pk=site2.pk).exists(), 'Site 2 should not have been deleted')
def test_bulk_update_objects_unpermitted(self):
"""
PATCH a set of objects where the requesting user's object-level permissions exclude one of
them. The excluded object must be reported rather than silently omitted from the response.
"""
site1 = Site.objects.get(slug='site-1')
site2 = Site.objects.get(slug='site-2')
obj_perm = ObjectPermission(name='Test permission', actions=['change'], constraints={'slug': 'site-1'})
obj_perm.save()
obj_perm.users.add(self.user)
obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
data = [
{'id': site1.pk, 'description': 'Permitted'},
{'id': site2.pk, 'description': 'Not permitted'},
]
response = self.client.patch(self._get_list_url(), data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
self.assertIn('detail', response.data)
self.assertEqual(len(response.data['errors']), 1)
self.assertEqual(response.data['errors'][0]['id'], site2.pk)
# Neither site may have been updated, including the one the user is permitted to change
site1.refresh_from_db()
site2.refresh_from_db()
self.assertEqual(site1.description, '', 'Site 1 should not have been updated')
self.assertEqual(site2.description, '', 'Site 2 should not have been updated')
def test_bulk_delete_objects_unpermitted(self):
"""
DELETE a set of objects where the requesting user's object-level permissions exclude one of
them. The excluded object must be reported rather than the request reporting success.
"""
site1 = Site.objects.get(slug='site-1')
site2 = Site.objects.get(slug='site-2')
obj_perm = ObjectPermission(name='Test permission', actions=['delete'], constraints={'slug': 'site-1'})
obj_perm.save()
obj_perm.users.add(self.user)
obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
data = [{'id': site1.pk}, {'id': site2.pk}]
response = self.client.delete(self._get_list_url(), data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
self.assertIn('detail', response.data)
self.assertEqual(len(response.data['errors']), 1)
self.assertEqual(response.data['errors'][0]['id'], site2.pk)
# Neither site may have been deleted, including the one the user is permitted to delete
self.assertTrue(Site.objects.filter(pk=site1.pk).exists(), 'Site 1 should not have been deleted')
self.assertTrue(Site.objects.filter(pk=site2.pk).exists(), 'Site 2 should not have been deleted')
class LocationTestCase(APIViewTestCases.APIViewTestCase):
model = Location

65
netbox/netbox/api/viewsets/mixins.py
View File

@ -29,9 +29,54 @@ __all__ = (
'ObjectValidationMixin',
'SequentialBulkCreatesMixin',
'discard_events_on_rollback',
'get_missing_objects_response',
)
def get_missing_objects_response(object_ids, queryset):
"""
Return a structured error Response naming each of the given object IDs which the queryset does
not match, or None if it matches them all.
An ID goes unmatched either because no such object exists or because the requesting user's
object-level permissions exclude it. The two are deliberately not distinguished, consistent with
the single-object endpoints, which return a 404 in both cases.
Bulk operations call this before performing any work: an unresolvable ID means the request names
an object the client cannot act on, so there is nothing to be gained by attempting the batch (it
would only be rolled back). Note that the status is 400 rather than the 409 a bulk delete
returns for a dependency conflict, as this is a problem with the request itself rather than with
the current state of the database.
"""
found_pks = set(queryset.values_list('pk', flat=True))
errors = [
{
'id': object_id,
'errors': {
'id': [_("Object with ID {id} does not exist").format(id=object_id)],
},
}
# dict.fromkeys() de-duplicates while preserving the order of first appearance, so an ID
# repeated in the request is reported once rather than once per occurrence.
for object_id in dict.fromkeys(object_ids)
if object_id not in found_pks
]
if not errors:
return None
return Response(
{
'detail': _('{failed_count} of {total} objects could not be found.').format(
failed_count=len(errors),
total=len(object_ids),
),
'errors': errors,
},
status=status.HTTP_400_BAD_REQUEST,
)
@contextmanager
def discard_events_on_rollback(sender, using=None):
"""
@ -279,9 +324,13 @@ class BulkUpdateModelMixin:
serializer = BulkOperationSerializer(data=request.data, many=True)
serializer.is_valid(raise_exception=True)
qs = self.get_bulk_update_queryset().filter(
pk__in=[o['id'] for o in serializer.data]
)
object_ids = [o['id'] for o in serializer.validated_data]
qs = self.get_bulk_update_queryset().filter(pk__in=object_ids)
# Reject the batch if any of the objects to be updated could not be found, rather than
# silently omitting them from the response.
if (response := get_missing_objects_response(object_ids, qs)) is not None:
return response
# Map update data by object ID
update_data = {
@ -375,9 +424,13 @@ class BulkDestroyModelMixin:
serializer = BulkOperationSerializer(data=request.data, many=True)
serializer.is_valid(raise_exception=True)
qs = self.get_bulk_destroy_queryset().filter(
pk__in=[o['id'] for o in serializer.validated_data]
)
object_ids = [o['id'] for o in serializer.validated_data]
qs = self.get_bulk_destroy_queryset().filter(pk__in=object_ids)
# Reject the batch if any of the objects to be deleted could not be found, rather than
# silently omitting them and reporting success.
if (response := get_missing_objects_response(object_ids, qs)) is not None:
return response
# Compile any changelog messages to be recorded on the objects being deleted
changelog_messages = {

18
netbox/netbox/tests/test_api_background.py
View File

@ -154,8 +154,9 @@ class BackgroundBulkWriteTests(RQQueueTestMixin, APITestCase):
self.assertEqual(r.description, 'put-bg')
def test_background_bulk_update_object_permission_subset(self):
# Constrained to Region 1 only; bulk update of all three should update only the
# permitted subset and SUCCEED (matching synchronous behavior; no rollback).
# Constrained to Region 1 only; the other two regions cannot be resolved, so the whole
# batch is rejected rather than the permitted subset being updated silently (matching
# synchronous behavior).
self.grant('change', 'view', constraints={'name': 'Region 1'})
payload = [{'id': r.pk, 'description': 'subset'} for r in self.regions]
response = self.client.patch(
@ -163,11 +164,14 @@ class BackgroundBulkWriteTests(RQQueueTestMixin, APITestCase):
)
self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
job = Job.objects.get(pk=response.data['job']['id'])
self.assertEqual(job.status, JobStatusChoices.STATUS_COMPLETED)
# DB side effect AND captured result reflect the permitted subset only.
self.assertEqual(Region.objects.filter(description='subset').count(), 1)
self.assertEqual(job.data['status_code'], status.HTTP_200_OK)
self.assertEqual(len(job.data['data']), 1)
self.assertEqual(job.status, JobStatusChoices.STATUS_FAILED)
# Nothing was updated, and the unresolvable IDs are named in the captured result.
self.assertEqual(Region.objects.filter(description='subset').count(), 0)
self.assertEqual(job.data['status_code'], status.HTTP_400_BAD_REQUEST)
self.assertEqual(
[e['id'] for e in job.data['data']['errors']],
[self.regions[1].pk, self.regions[2].pk],
)
def test_background_bulk_update_all_or_nothing(self):
self.grant('change', 'view')

76
netbox/utilities/testing/api.py
View File

@ -583,6 +583,50 @@ class APIViewTestCases:
f'Field {field!r} of object {id_list[0]} was modified — atomic rollback may be broken',
)
def test_bulk_update_objects_nonexistent_id(self):
"""
PATCH a set of objects where one of the IDs does not identify an existing object. Verify
the structured per-object error response and that no objects are modified.
"""
if self.bulk_update_data is None:
self.skipTest('Bulk update data not set')
obj_perm = ObjectPermission(name='Test permission', actions=['change'])
obj_perm.save()
obj_perm.users.add(self.user)
obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
id_list = list(self._get_queryset().values_list('id', flat=True)[:2])
self.assertEqual(len(id_list), 2, 'Insufficient number of objects to test bulk update')
missing_id = self._get_queryset().order_by('-id').first().id + 1
data = [{'id': id, **self.bulk_update_data} for id in (*id_list, missing_id)]
# Snapshot the objects which would otherwise have been updated
instances_before = list(self._get_queryset().filter(pk__in=id_list))
response = self.client.patch(self._get_list_url(), data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
self.assertIn('detail', response.data)
self.assertIn('errors', response.data)
self.assertEqual(len(response.data['errors']), 1)
self.assertEqual(response.data['errors'][0]['id'], missing_id)
self.assertIn('id', response.data['errors'][0]['errors'])
# The objects named alongside the missing one must not have been updated
for instance_before in instances_before:
instance_after = self._get_queryset().get(pk=instance_before.pk)
for field in self.bulk_update_data:
if field in ('changelog_message', 'add_tags', 'remove_tags'):
continue
self.assertEqual(
getattr(instance_after, field, None),
getattr(instance_before, field, None),
f'Field {field!r} of object {instance_before.pk} was modified despite an unresolvable '
f'sibling ID',
)
class DeleteObjectViewTestCase(APITestCase):
def test_delete_object_without_permission(self):
@ -672,6 +716,38 @@ class APIViewTestCases:
self.assertObjectChange(oc, action=ObjectChangeActionChoices.ACTION_DELETE,
message=changelog_message)
def test_bulk_delete_objects_nonexistent_id(self):
"""
DELETE a set of objects where one of the IDs does not identify an existing object. Verify
the structured per-object error response and that no objects are deleted.
"""
obj_perm = ObjectPermission(
name='Test permission',
actions=['delete']
)
obj_perm.save()
obj_perm.users.add(self.user)
obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
# Target the most recently created objects to avoid triggering recursive deletions
id_list = list(self._get_queryset().order_by('-id').values_list('id', flat=True)[:3])
self.assertEqual(len(id_list), 3, 'Insufficient number of objects to test bulk deletion')
missing_id = max(id_list) + 1
data = [{'id': id} for id in (*id_list, missing_id)]
initial_count = self._get_queryset().count()
response = self.client.delete(self._get_list_url(), data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
self.assertIn('detail', response.data)
self.assertIn('errors', response.data)
self.assertEqual(len(response.data['errors']), 1)
self.assertEqual(response.data['errors'][0]['id'], missing_id)
self.assertIn('id', response.data['errors'][0]['errors'])
# The objects named alongside the missing one must not have been deleted
self.assertEqual(self._get_queryset().count(), initial_count)
class GraphQLTestCase(APITestCase):
graphql_auto_filter_tests = True
graphql_auto_filter_exclude = ()