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

Fixes #22617: Editing objects via bulk import form requires "change" permission (#22618)

This commit is contained in:
Jeremy Stretch 2026-07-08 11:23:00 -04:00 committed by GitHub
parent 3561de3d56
commit 9ee38b6c1a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 115 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
netbox/dcim/tests/test_views.py
View File

@ -1303,6 +1303,22 @@ class ModuleTypeTestCase(ViewTestCases.PrimaryObjectViewTestCase):
# run base test
super().test_bulk_update_objects_with_permission()
def test_bulk_update_objects_without_change_permission(self):
# ModuleTypeImportView declares these as additional_permissions, so they're required to reach the view
self.add_permissions(
'dcim.add_consoleporttemplate',
'dcim.add_consoleserverporttemplate',
'dcim.add_powerporttemplate',
'dcim.add_poweroutlettemplate',
'dcim.add_interfacetemplate',
'dcim.add_frontporttemplate',
'dcim.add_rearporttemplate',
'dcim.add_modulebaytemplate',
)
# run base test
super().test_bulk_update_objects_without_change_permission()
@tag('regression')
def test_bulk_import_objects_with_permission(self):
self.add_permissions(

46
netbox/netbox/views/generic/bulk_views.py
View File

@ -489,13 +489,10 @@ class BulkImportView(GetReturnURLMixin, BaseMultiObjectView):
def _save_object(self, model_form, request, parent_idx):
_action = 'Updated' if model_form.instance.pk else 'Created'
# Save the primary object
# Save the primary object. Object-level permissions are enforced in aggregate by
# create_and_update_objects() once all records have been processed.
obj = self.save_object(model_form, request)
# Enforce object-level permissions
if not self.queryset.filter(pk=obj.pk).first():
raise PermissionsViolation()
# Iterate through the related object forms (if any), validating and saving each instance.
for field_name, related_object_form in self.related_object_forms.items():
@ -640,9 +637,28 @@ class BulkImportView(GetReturnURLMixin, BaseMultiObjectView):
)
raise ValidationError(error_msg)
# A record which references an existing object by ID performs an update rather than a creation. The bulk
# import view is gated only on the 'add' permission, but updating an existing object requires 'change' (as
# enforced by the REST API). Require the 'change' permission at the model level before permitting any updates,
# and restrict the prefetched objects to those the user is permitted to change (object-level enforcement).
update_pks = set(prefetch_ids)
if prefetch_ids:
change_permission = get_permission_for_model(self.queryset.model, 'change')
if not request.user.has_perm(change_permission):
raise ValidationError(
_(
"This import includes {count} record(s) that reference an existing object by ID and would "
"update it, which requires the {permission} permission. Remove the ID column to create new "
"objects instead."
).format(count=len(prefetch_ids), permission=change_permission)
)
change_queryset = self.queryset.model.objects.restrict(request.user, 'change')
else:
change_queryset = self.queryset.model.objects
prefetched_objects = {
obj.pk: obj
for obj in self.queryset.model.objects.filter(id__in=prefetch_ids)
for obj in change_queryset.filter(id__in=prefetch_ids)
} if prefetch_ids else {}
# For MPTT models, delay tree updates until all saves are complete
@ -652,6 +668,17 @@ class BulkImportView(GetReturnURLMixin, BaseMultiObjectView):
else:
saved_objects = self._process_import_records(form, request, records, prefetched_objects)
# Enforce object-level permissions in aggregate. Newly created objects are constrained by the 'add'
# permission (self.queryset is already restricted to 'add'); updated objects by 'change' (reusing the
# queryset built above, so no additional per-record work). This runs inside the caller's atomic
# transaction, so any violation rolls back the entire import.
created_pks = [obj.pk for obj in saved_objects if obj.pk not in update_pks]
if self.queryset.filter(pk__in=created_pks).count() != len(created_pks):
raise PermissionsViolation()
updated_pks = [obj.pk for obj in saved_objects if obj.pk in update_pks]
if updated_pks and change_queryset.filter(pk__in=updated_pks).count() != len(updated_pks):
raise PermissionsViolation()
return saved_objects
#
@ -693,14 +720,11 @@ class BulkImportView(GetReturnURLMixin, BaseMultiObjectView):
return redirect(redirect_url)
try:
# Iterate through data and bind each record to a new model form instance.
# Iterate through data and bind each record to a new model form instance. Object-level
# permissions are enforced within create_and_update_objects().
with transaction.atomic(using=router.db_for_write(model)):
new_objects = self.create_and_update_objects(form, request)
# Enforce object-level permissions
if self.queryset.filter(pk__in=[obj.pk for obj in new_objects]).count() != len(new_objects):
raise PermissionsViolation
msg = _('Imported {count} {object_type}').format(
count=len(new_objects),
object_type=model._meta.verbose_name_plural

74
netbox/utilities/testing/views.py
View File

@ -3,7 +3,6 @@ import csv
from django.conf import settings
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ObjectDoesNotExist
from django.db.models import ForeignKey
from django.test import override_settings
from django.urls import reverse
from django.utils.translation import gettext as _
@ -747,6 +746,53 @@ class ViewTestCases:
self.assertObjectChange(oc, action=ObjectChangeActionChoices.ACTION_CREATE,
message=data['changelog_message'])
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'])
def test_bulk_update_objects_without_change_permission(self):
# Bulk import rows carrying an object ID update existing objects. This must require the 'change'
# permission, matching the REST API; the 'add' permission alone must not permit updates.
if not hasattr(self, 'csv_update_data'):
raise NotImplementedError(_("The test must define csv_update_data."))
initial_count = self._get_queryset().count()
array, csv_data = self._get_update_csv_data()
data = {
'format': ImportFormatChoices.CSV,
'data': csv_data,
'csv_delimiter': CSVDelimiterChoices.AUTO,
}
# Assign only the 'add' permission
obj_perm = ObjectPermission(
name='Test permission',
actions=['add']
)
obj_perm.save()
obj_perm.users.add(self.user)
obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
# Take a snapshot of the objects targeted for update
reader = csv.DictReader(array, delimiter=',')
check_data = list(reader)
before = {
line['id']: self.model.objects.get(id=line['id'])
for line in check_data
}
# The import must be rejected with a permissions error (form re-rendered) and no object modified
response = self.client.post(self._get_url('bulk_import'), data)
self.assertHttpStatus(response, 200)
self.assertContains(response, 'Remove the ID column to create new objects instead.')
self.assertEqual(initial_count, self._get_queryset().count())
for line in check_data:
obj = self.model.objects.get(id=line['id'])
for attr in line:
if attr == 'id':
continue
# Skip relational fields (FK/M2M), consistent with test_bulk_update_objects_with_permission
if self.model._meta.get_field(attr).is_relation:
continue
self.assertEqual(getattr(obj, attr), getattr(before[line['id']], attr))
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'])
def test_bulk_update_objects_with_permission(self):
if not hasattr(self, 'csv_update_data'):
@ -760,10 +806,10 @@ class ViewTestCases:
'csv_delimiter': CSVDelimiterChoices.AUTO,
}
# Assign model-level permission
# Updating existing objects requires both 'add' (to reach the view) and 'change' (to update)
obj_perm = ObjectPermission(
name='Test permission',
actions=['add']
actions=['add', 'change']
)
obj_perm.save()
obj_perm.users.add(self.user)
@ -773,17 +819,25 @@ class ViewTestCases:
self.assertHttpStatus(self.client.post(self._get_url('bulk_import'), data), 302)
self.assertEqual(initial_count, self._get_queryset().count())
# Verify that each object was actually updated to match the value specified in the CSV
reader = csv.DictReader(array, delimiter=',')
check_data = list(reader)
for line in check_data:
obj = self.model.objects.get(id=line["id"])
for attr, value in line.items():
if attr != "id":
field = self.model._meta.get_field(attr)
value = getattr(obj, attr)
# cannot verify FK fields as don't know what name the CSV maps to
if value is not None and not isinstance(field, ForeignKey):
self.assertEqual(value, value)
for attr, expected in line.items():
if attr == "id":
continue
field = self.model._meta.get_field(attr)
# Skip relational fields (FK/M2M): the CSV value can't be mapped to a comparable attribute
if field.is_relation:
continue
actual = getattr(obj, attr)
# Only verify simple scalar values against the raw CSV string; skip lists and other complex
# representations that the import form transforms (e.g. choice-set extra_choices).
if not isinstance(actual, (str, int, float)):
continue
# Compare case-insensitively to tolerate values normalized on save (e.g. MAC addresses)
self.assertEqual(str(actual).lower(), str(expected).lower())
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], EXEMPT_EXCLUDE_MODELS=[])
def test_bulk_import_objects_with_constrained_permission(self, post_import_callback=None):