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

Return a 403 when attempting to alter non-permitted objects

This commit is contained in:
Jeremy Stretch 2026-08-11 11:22:17 -04:00
parent 8aed00afdf
commit d384136045
5 changed files with 251 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
netbox/core/api/schema.py
View File

@ -204,6 +204,13 @@ class NetBoxAutoSchema(AutoSchema):
"modified: a bulk update is an all-or-none operation."
),
),
'403': OpenApiResponse(
response=BulkOperationErrorSerializer,
description=_(
"The requesting user is not permitted to apply one or more of the "
"modifications specified. No objects were modified."
),
),
}
if action == 'bulk_destroy':
@ -216,6 +223,13 @@ class NetBoxAutoSchema(AutoSchema):
"rule. No objects were deleted."
),
),
'403': OpenApiResponse(
response=BulkOperationErrorSerializer,
description=_(
"The requesting user is not permitted to delete one or more of the objects "
"specified. No objects were deleted."
),
),
'409': OpenApiResponse(
response=BulkOperationErrorSerializer,
description=_(
@ -244,6 +258,15 @@ class NetBoxAutoSchema(AutoSchema):
"were created: a bulk creation is an all-or-none operation."
),
),
# A 403 always carries a `detail`, and BulkOperationError's `errors` is optional, so
# the one component covers both the single-object and the bulk shape here.
'403': OpenApiResponse(
response=BulkOperationErrorSerializer,
description=_(
"The requesting user is not permitted to create one or more of the objects "
"specified. No objects were created."
),
),
}
return {}

20
netbox/core/tests/test_openapi_schema.py
View File

@ -154,8 +154,8 @@ class OpenAPISchemaTestCase(TestCase):
def test_bulk_delete_documents_error_responses(self):
"""
Bulk delete operations should document both the 400 (unresolvable request) and the 409
(dependency or protection rule) responses.
Bulk delete operations should document the 400 (unresolvable request or protection rule), the
403 (not permitted) and the 409 (dependent object) responses.
Refs: #20054
"""
@ -164,8 +164,23 @@ class OpenAPISchemaTestCase(TestCase):
for path in ('/api/dcim/sites/', '/api/ipam/prefixes/', '/api/users/users/'):
with self.subTest(path=path):
self.assertEqual(self._get_response_schema(path, 'delete', '400'), ref)
self.assertEqual(self._get_response_schema(path, 'delete', '403'), ref)
self.assertEqual(self._get_response_schema(path, 'delete', '409'), ref)
def test_bulk_write_operations_document_forbidden_response(self):
"""
Every bulk write should document the 403 returned when an object-level permission refuses one
of the objects specified.
Refs: #20054
"""
ref = {'$ref': '#/components/schemas/BulkOperationError'}
for path in ('/api/dcim/sites/', '/api/ipam/prefixes/', '/api/users/users/'):
for method in ('post', 'put', 'patch', 'delete'):
with self.subTest(path=path, method=method):
self.assertEqual(self._get_response_schema(path, method, '403'), ref)
def test_create_documents_error_response_for_either_shape(self):
"""
A POST to a list endpoint accepts either a single object or a list, so its 400 response
@ -196,6 +211,7 @@ class OpenAPISchemaTestCase(TestCase):
with self.subTest(method=method):
responses = self.schema['paths'][path][method]['responses']
self.assertNotIn('409', responses)
self.assertNotIn('403', responses)
for code, response in responses.items():
schema = response.get('content', {}).get('application/json', {}).get('schema', {})
self.assertNotEqual(

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

@ -25,6 +25,7 @@ from utilities.testing import (
create_test_device,
create_test_nat_ip_pair,
disable_logging,
disable_warnings,
)
from virtualization.models import Cluster, ClusterType
from wireless.choices import WirelessChannelChoices
@ -739,6 +740,111 @@ 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_permission_constraint(self):
"""
PATCH a set of objects where the update would move one of them outside the requesting user's
object-level permissions. The offending object must be named, rather than the whole batch
failing with an opaque 403, and nothing may be modified.
"""
site1 = Site.objects.get(slug='site-1')
site2 = Site.objects.get(slug='site-2')
Site.objects.filter(pk__in=(site1.pk, site2.pk)).update(status=SiteStatusChoices.STATUS_ACTIVE)
# Only active sites may be changed, so setting Site 2's status to "planned" saves the object
# and then fails _validate_objects(), which perform_update() reports as PermissionDenied.
obj_perm = ObjectPermission(
name='Test permission',
actions=['change'],
constraints={'status': SiteStatusChoices.STATUS_ACTIVE},
)
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, 'status': SiteStatusChoices.STATUS_PLANNED},
]
with disable_warnings('django.request'):
response = self.client.patch(self._get_list_url(), data, format='json', **self.header)
# Still a 403, as the single-object endpoint returns, but now correlated
self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)
self.assertIn('detail', response.data)
self.assertEqual([e['id'] for e in response.data['errors']], [site2.pk])
self.assertIsInstance(response.data['errors'][0]['errors']['__all__'], list)
# Neither site may have been modified, including the permitted one
site1.refresh_from_db()
site2.refresh_from_db()
self.assertEqual(site1.description, '', 'Site 1 should not have been updated')
self.assertEqual(site2.status, SiteStatusChoices.STATUS_ACTIVE, 'Site 2 should not have been updated')
def test_bulk_update_objects_permission_constraint_and_validation_error(self):
"""
PATCH a set of objects where one entry is invalid and another is refused by object-level
permissions. Both must be reported, and the authorization failure must determine the status
code: it is the failure which would remain were the invalid entry corrected.
"""
site1 = Site.objects.get(slug='site-1')
site2 = Site.objects.get(slug='site-2')
Site.objects.filter(pk__in=(site1.pk, site2.pk)).update(status=SiteStatusChoices.STATUS_ACTIVE)
obj_perm = ObjectPermission(
name='Test permission',
actions=['change'],
constraints={'status': SiteStatusChoices.STATUS_ACTIVE},
)
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, 'status': 'not-a-valid-status'}, # Fails validation (400)
{'id': site2.pk, 'status': SiteStatusChoices.STATUS_PLANNED}, # Not permitted (403)
]
with disable_warnings('django.request'):
response = self.client.patch(self._get_list_url(), data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)
self.assertEqual([e['id'] for e in response.data['errors']], [site1.pk, site2.pk])
site1.refresh_from_db()
site2.refresh_from_db()
self.assertEqual(site1.status, SiteStatusChoices.STATUS_ACTIVE)
self.assertEqual(site2.status, SiteStatusChoices.STATUS_ACTIVE)
def test_bulk_create_objects_permission_constraint(self):
"""
POST a set of objects where one falls outside the requesting user's object-level permissions.
The offending object must be correlated by its position, and nothing may be created.
"""
obj_perm = ObjectPermission(
name='Test permission',
actions=['add'],
constraints={'status': SiteStatusChoices.STATUS_ACTIVE},
)
obj_perm.save()
obj_perm.users.add(self.user)
obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
initial_count = self._get_queryset().count()
data = [
{'name': 'Site 20', 'slug': 'site-20', 'status': SiteStatusChoices.STATUS_ACTIVE},
{'name': 'Site 21', 'slug': 'site-21', 'status': SiteStatusChoices.STATUS_PLANNED},
]
with disable_warnings('django.request'):
response = self.client.post(self._get_list_url(), data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)
self.assertEqual([e['index'] for e in response.data['errors']], [1])
self.assertIsInstance(response.data['errors'][0]['errors']['__all__'], list)
self.assertEqual(
self._get_queryset().count(), initial_count,
'No objects should be created when any sibling is not permitted',
)
def test_bulk_delete_objects_conflict_and_abort_request(self):
"""
DELETE a set of objects where one is blocked by a dependent object and another by a

2
netbox/netbox/api/serializers/bulk.py
View File

@ -61,7 +61,7 @@ class BulkOperationErrorSerializer(serializers.Serializer):
responsible for it.
"""
detail = serializers.CharField(
help_text=_('A summary of the failure, e.g. "1 of 3 objects failed validation."')
help_text=_('A summary of the failure, e.g. "1 of 3 objects could not be updated."')
)
errors = BulkOperationEntryErrorSerializer(
many=True,

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

@ -2,7 +2,7 @@ import warnings
from collections import Counter
from contextlib import contextmanager
from django.core.exceptions import ObjectDoesNotExist
from django.core.exceptions import ObjectDoesNotExist, PermissionDenied
from django.db import router, transaction
from django.db.models import ProtectedError, RestrictedError
from django.http import Http404
@ -24,6 +24,7 @@ from utilities.request import copy_safe_request
from utilities.rqworker import any_workers_for_queue
__all__ = (
'BULK_ERROR_STATUSES',
'BackgroundOperationMixin',
'BulkCreateModelMixin',
'BulkDestroyModelMixin',
@ -37,8 +38,44 @@ __all__ = (
'get_invalid_entries_response',
'get_missing_objects_response',
'get_non_list_response',
'resolve_bulk_error_status',
)
# The status codes with which a failed bulk operation may be reported, in order of precedence: where
# the per-object failures within one batch imply more than one of these, the earliest applies, being
# the one which would still stand were the others corrected. An authorization failure thus outranks a
# conflict with the current state of the database, which in turn outranks a rejection of the request.
BULK_ERROR_STATUSES = (
status.HTTP_403_FORBIDDEN,
status.HTTP_409_CONFLICT,
status.HTTP_400_BAD_REQUEST,
)
# Reported for an object whose write was undone because the object it produced falls outside the
# queryset permitted to the requesting user (see ObjectValidationMixin._validate_objects). Which
# constraint was violated is deliberately not disclosed, consistent with the single-object endpoints.
PERMISSION_DENIED_MESSAGE = _("You do not have permission to perform this action on this object.")
def resolve_bulk_error_status(error_statuses):
"""
Return the single status code with which to report a bulk operation whose per-object failures
imply the given ones, or None if there were no failures.
:param error_statuses: The set of status codes implied by the failures within one batch, each
drawn from BULK_ERROR_STATUSES (which documents how they are ranked).
"""
if not error_statuses:
return None
for error_status in BULK_ERROR_STATUSES:
if error_status in error_statuses:
return error_status
# A code with no defined precedence (a subclass may report its own) is not silently ranked;
# fall back to the generic client error.
return status.HTTP_400_BAD_REQUEST
def get_non_list_response(data):
"""
@ -357,18 +394,18 @@ class BulkCreateModelMixin:
]
"""
def bulk_create(self, request, *args, **kwargs):
created_pks, errors = self.perform_bulk_create(request.data)
created_pks, errors, error_status = self.perform_bulk_create(request.data)
if errors:
return Response(
{
'detail': _('{failed_count} of {total} objects failed validation.').format(
'detail': _('{failed_count} of {total} objects could not be created.').format(
failed_count=len(errors),
total=len(request.data),
),
'errors': errors,
},
status=status.HTTP_400_BAD_REQUEST,
status=error_status,
)
# Re-fetch the new objects to serialize them with their related objects prefetched. Order by PK
@ -380,8 +417,16 @@ class BulkCreateModelMixin:
return Response(serializer.data, status=status.HTTP_201_CREATED)
def perform_bulk_create(self, data):
"""
Validate and create each of the given objects, rolling the entire batch back if any one of
them could not be created.
Returns the PKs of the objects created, the per-object errors, and the status code with
which to report them (None if there were none). See resolve_bulk_error_status().
"""
created_pks = []
errors = []
error_statuses = set()
using = router.db_for_write(self.queryset.model)
with transaction.atomic(using=using), discard_events_on_rollback(self, using=using):
# Validate and save each object in turn, rather than validating the entire batch up front,
@ -404,10 +449,12 @@ class BulkCreateModelMixin:
],
},
})
error_statuses.add(status.HTTP_400_BAD_REQUEST)
continue
serializer = self.get_serializer(data=item)
if not serializer.is_valid():
errors.append({'index': i, 'errors': serializer.errors})
error_statuses.add(status.HTTP_400_BAD_REQUEST)
continue
try:
# Provisionally create even when a prior item failed, so subsequent
@ -421,11 +468,19 @@ class BulkCreateModelMixin:
# and the remaining objects in the batch can still be evaluated. The message is
# coerced to a string because a few receivers pass an exception rather than text.
errors.append({'index': i, 'errors': {'__all__': [str(e.message)]}})
error_statuses.add(status.HTTP_400_BAD_REQUEST)
except PermissionDenied:
# Raised by perform_create() when the object it saved falls outside the queryset
# permitted to the requesting user. Reported per object so that the offending
# entry is named, but still as a 403, which is what the single-object endpoint
# returns for the same rejection.
errors.append({'index': i, 'errors': {'__all__': [PERMISSION_DENIED_MESSAGE]}})
error_statuses.add(status.HTTP_403_FORBIDDEN)
else:
created_pks.append(serializer.instance.pk)
if errors:
transaction.set_rollback(True)
return created_pks, errors
return created_pks, errors, resolve_bulk_error_status(error_statuses)
# TODO: Remove this in NetBox v5.0
@ -516,18 +571,21 @@ class BulkUpdateModelMixin:
for object_id, item in zip(object_ids, request.data, strict=True)
}
object_pks, errors = self.perform_bulk_update(qs, update_data, partial=partial)
object_pks, errors, error_status = self.perform_bulk_update(qs, update_data, partial=partial)
if errors:
return Response(
{
'detail': _('{failed_count} of {total} objects failed validation.').format(
'detail': _('{failed_count} of {total} objects could not be updated.').format(
failed_count=len(errors),
# Every object named was matched and attempted, the duplicate and missing-ID
# checks above having rejected the batch otherwise, so this equals the number
# of objects submitted.
total=len(object_pks) + len(errors),
),
'errors': errors,
},
status=status.HTTP_400_BAD_REQUEST,
status=error_status,
)
# Prefetch related objects for all updated instances
@ -537,8 +595,16 @@ class BulkUpdateModelMixin:
return Response(serializer.data, status=status.HTTP_200_OK)
def perform_bulk_update(self, objects, update_data, partial):
"""
Validate and apply the given attributes to each of the given objects, rolling the entire
batch back if any one of them could not be updated.
Returns the PKs of the objects updated, the per-object errors, and the status code with
which to report them (None if there were none). See resolve_bulk_error_status().
"""
updated_pks = []
errors = []
error_statuses = set()
using = router.db_for_write(self.queryset.model)
with transaction.atomic(using=using), discard_events_on_rollback(self, using=using):
# Validate and save each object in turn so subsequent validations see the DB
@ -551,6 +617,7 @@ class BulkUpdateModelMixin:
serializer = self.get_serializer(obj, data=data, partial=partial)
if not serializer.is_valid():
errors.append({'id': obj.pk, 'errors': serializer.errors})
error_statuses.add(status.HTTP_400_BAD_REQUEST)
continue
try:
self.perform_update(serializer)
@ -561,11 +628,20 @@ class BulkUpdateModelMixin:
# and the remaining objects in the batch can still be evaluated. The message is
# coerced to a string because a few receivers pass an exception rather than text.
errors.append({'id': obj.pk, 'errors': {'__all__': [str(e.message)]}})
error_statuses.add(status.HTTP_400_BAD_REQUEST)
except PermissionDenied:
# Raised by perform_update() when the object, as modified, falls outside the
# queryset permitted to the requesting user -- so unlike the check made before
# the batch begins (see get_missing_objects_response), this depends on the
# attributes submitted. Reported per object so that the offending entry is
# named, but still as a 403, as the single-object endpoint returns.
errors.append({'id': obj.pk, 'errors': {'__all__': [PERMISSION_DENIED_MESSAGE]}})
error_statuses.add(status.HTTP_403_FORBIDDEN)
else:
updated_pks.append(obj.pk)
if errors:
transaction.set_rollback(True)
return updated_pks, errors
return updated_pks, errors, resolve_bulk_error_status(error_statuses)
def get_bulk_update_serializer_class(self, *, partial=False):
return get_bulk_update_serializer_class(
@ -637,14 +713,9 @@ class BulkDestroyModelMixin:
o['id']: o.get('changelog_message') for o in serializer.validated_data
}
errors, total, has_conflict = self.perform_bulk_destroy(qs, changelog_messages)
errors, total, error_status = self.perform_bulk_destroy(qs, changelog_messages)
if errors:
# A dependency conflict reports 409, as it is a conflict with the current state of the
# database; every other failure reports 400, as it is a rejection of the request. This
# matches the single-object endpoint, where dispatch() maps the same two exception
# classes to the same two status codes. Where a batch hit both, the conflict takes
# precedence: it is the failure which would remain were the request itself corrected.
return Response(
{
'detail': _('{failed_count} of {total} objects could not be deleted.').format(
@ -653,7 +724,7 @@ class BulkDestroyModelMixin:
),
'errors': errors,
},
status=status.HTTP_409_CONFLICT if has_conflict else status.HTTP_400_BAD_REQUEST,
status=error_status,
)
return Response(status=status.HTTP_204_NO_CONTENT)
@ -663,15 +734,18 @@ class BulkDestroyModelMixin:
Attempt to delete each of the given objects, rolling the entire batch back if any one of
them could not be deleted.
Returns the per-object errors, the number of objects processed, and whether any of the
failures was a conflict with the current state of the database (a dependent object) rather
than a rejection of the request (a protection rule, or any other signal receiver raising
AbortRequest). The caller uses the last of these to select a status code.
Returns the per-object errors, the number of objects processed, and the status code with
which to report the errors (None if there were none). A dependency conflict yields a 409, as
it is a conflict with the current state of the database, whereas a protection rule (or any
other signal receiver raising AbortRequest) yields a 400, being a rejection of the request:
this matches the single-object endpoint, where dispatch() maps the same exception classes to
the same status codes. See resolve_bulk_error_status() for how a batch hitting more than one
of these is resolved.
"""
changelog_messages = changelog_messages or {}
errors = []
total = 0
has_conflict = False
error_statuses = set()
using = router.db_for_write(self.queryset.model)
with transaction.atomic(using=using), discard_events_on_rollback(self, using=using):
for obj in objects:
@ -683,7 +757,7 @@ class BulkDestroyModelMixin:
try:
self.perform_destroy(obj)
except (ProtectedError, RestrictedError) as e:
has_conflict = True
error_statuses.add(status.HTTP_409_CONFLICT)
protected = list(
e.protected_objects if isinstance(e, ProtectedError) else e.restricted_objects
)
@ -709,9 +783,15 @@ class BulkDestroyModelMixin:
# is rolled back to that savepoint and the remaining objects in the batch can
# still be evaluated.
errors.append({'id': pk, 'errors': {'__all__': [str(e.message)]}})
error_statuses.add(status.HTTP_400_BAD_REQUEST)
except PermissionDenied:
# Raised by perform_destroy() when the object falls outside the queryset
# permitted to the requesting user (reachable via the If-Match re-check).
errors.append({'id': pk, 'errors': {'__all__': [PERMISSION_DENIED_MESSAGE]}})
error_statuses.add(status.HTTP_403_FORBIDDEN)
if errors:
transaction.set_rollback(True)
return errors, total, has_conflict
return errors, total, resolve_bulk_error_status(error_statuses)
class ObjectValidationMixin: