From 79f940f3627ad2c6ce02340d7bfc150d40233ae3 Mon Sep 17 00:00:00 2001 From: Jeremy Stretch Date: Mon, 24 Aug 2026 11:47:25 -0400 Subject: [PATCH] Fixes #22978: Discard queued events when a REST API write is rolled back MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The REST API's write paths are transactional, but the in-memory events queue is not: change logging receivers queue events eagerly (for deletions in pre_delete, before the row is removed), and event_tracking() flushes that queue only after the response has been rendered. Where a rolled-back write is caught and converted into a normal response — the ProtectedError/RestrictedError and AbortRequest handlers in NetBoxModelViewSet.dispatch(), and the ObjectDoesNotExist to PermissionDenied conversions in the perform_*() methods — the flush therefore dispatched events for objects that were never created, updated, or deleted. Backport discard_events_on_rollback() from the feature branch and enter it inside the transaction guarding each API write: perform_create(), perform_update(), and perform_destroy(); the SequentialBulkCreatesMixin, BulkUpdateModelMixin, and BulkDestroyModelMixin bulk actions; and AvailableObjectsView.post(). The context manager sends clear_events on the way out whenever the wrapped transaction is rolled back, whether by an exception escaping the block or by an explicit set_rollback(), which also covers the exceptions that reach dispatch(). This mirrors what the UI views now do (#22934). Co-Authored-By: Claude Opus 5 --- netbox/extras/tests/test_event_rules.py | 247 +++++++++++++++++++++++- netbox/ipam/api/views.py | 5 +- netbox/netbox/api/viewsets/__init__.py | 9 +- netbox/netbox/api/viewsets/mixins.py | 51 ++++- 4 files changed, 300 insertions(+), 12 deletions(-) diff --git a/netbox/extras/tests/test_event_rules.py b/netbox/extras/tests/test_event_rules.py index d75bf5bf6..fd6c306a9 100644 --- a/netbox/extras/tests/test_event_rules.py +++ b/netbox/extras/tests/test_event_rules.py @@ -8,7 +8,7 @@ from unittest.mock import Mock, patch import django_rq from django.conf import settings from django.http import HttpResponse -from django.test import RequestFactory, TestCase, tag +from django.test import RequestFactory, TestCase, override_settings, tag from django.urls import reverse from PIL import Image from requests import Session @@ -17,16 +17,19 @@ from rest_framework import status from core.choices import JobNotificationChoices, ManagedFileRootPathChoices from core.events import * from core.models import Job, ObjectType -from dcim.choices import SiteStatusChoices -from dcim.models import DeviceType, Interface, Manufacturer, Site +from dcim.choices import DeviceStatusChoices, InterfaceTypeChoices, SiteStatusChoices +from dcim.models import Device, DeviceRole, DeviceType, Interface, Manufacturer, Site from extras.choices import EventRuleActionChoices from extras.events import enqueue_event, flush_events, serialize_for_event from extras.models import EventRule, Notification, Script, ScriptModule, Tag, Webhook from extras.scripts import Script as ScriptBase from extras.signals import process_job_end_event_rules from extras.webhooks import generate_signature, send_webhook +from ipam.choices import IPAddressStatusChoices +from ipam.models import IPAddress, Prefix from netbox.context_managers import event_tracking -from utilities.testing import APITestCase, create_test_device +from users.models import ObjectPermission +from utilities.testing import APITestCase, create_test_device, disable_warnings from utilities.testing.mixins import RQQueueTestMixin @@ -167,6 +170,32 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase): self.assertEqual(job.kwargs['snapshots']['postchange']['name'], 'Site 1') self.assertEqual(job.kwargs['snapshots']['postchange']['tags'], ['Bar', 'Foo']) + def test_single_create_rollback_discards_events(self): + """ + Check that creating an object which is then rolled back by the object-level permission check + in perform_create() queues no background task. + """ + # Permit the creation of active sites only. The new object is saved (queueing its event) + # before _validate_objects() rejects it and the transaction is rolled back. + 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(Site)) + + data = {'name': 'Site 1', 'slug': 'site-1', 'status': SiteStatusChoices.STATUS_PLANNED} + url = reverse('dcim-api:site-list') + with disable_warnings('django.request'): + response = self.client.post(url, data, format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN) + self.assertEqual(Site.objects.count(), 0) + + # No task may be queued for a creation that was rolled back + self.assertEqual(self.queue.count, 0) + def test_bulk_create_process_eventrule(self): """ Check that bulk creating multiple objects with an applicable EventRule queues a background task for each @@ -218,6 +247,72 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase): self.assertEqual(job.kwargs['snapshots']['postchange']['name'], response.data[i]['name']) self.assertEqual(job.kwargs['snapshots']['postchange']['tags'], ['Bar', 'Foo']) + def test_bulk_create_rollback_discards_events(self): + """ + Check that a sequential bulk create which is rolled back queues no background tasks for the + objects that were provisionally created before the failure. + """ + manufacturer = Manufacturer.objects.create(name='Manufacturer 1', slug='manufacturer-1') + device_type = DeviceType.objects.create(manufacturer=manufacturer, model='Device Type 1', slug='device-type-1') + role = DeviceRole.objects.create(name='Device Role 1', slug='device-role-1') + site = Site.objects.create(name='Site 1', slug='site-1') + + # DeviceViewSet uses SequentialBulkCreatesMixin, so each valid object is provisionally + # created (and its event queued) before a later object fails validation. + event_rule = EventRule.objects.get(name='Event Rule 1') + event_rule.object_types.set([ObjectType.objects.get_for_model(Device)]) + + data = [ + { + 'name': 'Device 1', + 'device_type': device_type.pk, + 'role': role.pk, + 'site': site.pk, + 'status': DeviceStatusChoices.STATUS_ACTIVE, + }, + {}, # Missing all required fields + ] + url = reverse('dcim-api:device-list') + self.add_permissions('dcim.add_device', 'dcim.view_site', 'dcim.view_devicetype', 'dcim.view_devicerole') + response = self.client.post(url, data, format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST) + self.assertEqual(Device.objects.count(), 0) + + # No task may be queued for a creation that was rolled back + self.assertEqual(self.queue.count, 0) + + def test_available_objects_create_rollback_discards_events(self): + """ + Check that creating an object via an available-objects endpoint (e.g. available-ips) queues + no background task when the object-level permission check rolls the transaction back. + """ + prefix = Prefix.objects.create(prefix='192.0.2.0/24') + + event_rule = EventRule.objects.get(name='Event Rule 1') + event_rule.object_types.set([ObjectType.objects.get_for_model(IPAddress)]) + + # Permit the creation of active IP addresses only. The new object is saved (queueing its + # event) before _validate_objects() rejects it and the transaction is rolled back. + obj_perm = ObjectPermission( + name='Test permission', + actions=['add'], + constraints={'status': IPAddressStatusChoices.STATUS_ACTIVE}, + ) + obj_perm.save() + obj_perm.users.add(self.user) + obj_perm.object_types.add(ObjectType.objects.get_for_model(IPAddress)) + self.add_permissions('ipam.view_prefix') + + url = reverse('ipam-api:prefix-available-ips', kwargs={'pk': prefix.pk}) + data = {'status': IPAddressStatusChoices.STATUS_RESERVED} + with disable_warnings('django.request'): + response = self.client.post(url, data, format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN) + self.assertEqual(IPAddress.objects.count(), 0) + + # No task may be queued for a creation that was rolled back + self.assertEqual(self.queue.count, 0) + def test_single_update_process_eventrule(self): """ Check that updating an object with an applicable EventRule queues a background task for the rule's action. @@ -252,6 +347,37 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase): self.assertEqual(job.kwargs['snapshots']['postchange']['name'], 'Site X') self.assertEqual(job.kwargs['snapshots']['postchange']['tags'], ['Baz']) + def test_single_update_rollback_discards_events(self): + """ + Check that updating an object which is then rolled back by the object-level permission check + in perform_update() queues no background task. + """ + site = Site.objects.create(name='Site 1', slug='site-1', status=SiteStatusChoices.STATUS_ACTIVE) + + # Permit the modification of active sites only. Setting the status to "planned" takes the + # object outside the permission's scope, so it is saved (queueing its event) and then + # rejected by _validate_objects(), rolling the transaction back. + 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(Site)) + + url = reverse('dcim-api:site-detail', kwargs={'pk': site.pk}) + with disable_warnings('django.request'): + response = self.client.patch( + url, {'status': SiteStatusChoices.STATUS_PLANNED}, format='json', **self.header + ) + self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN) + site.refresh_from_db() + self.assertEqual(site.status, SiteStatusChoices.STATUS_ACTIVE) + + # No task may be queued for an update that was rolled back + self.assertEqual(self.queue.count, 0) + def test_bulk_update_process_eventrule(self): """ Check that bulk updating multiple objects with an applicable EventRule queues a background task for each @@ -309,6 +435,38 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase): self.assertEqual(job.kwargs['snapshots']['postchange']['name'], response.data[i]['name']) self.assertEqual(job.kwargs['snapshots']['postchange']['tags'], ['Baz']) + def test_bulk_update_rollback_discards_events(self): + """ + Check that a bulk update which is rolled back because one object failed validation queues no + background tasks for the objects that were provisionally updated. + """ + 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(sites) + + # The first two objects are valid and will be provisionally updated; the third fails + # validation, rolling the entire batch back. + data = [ + {'id': sites[0].pk, 'name': 'Site X'}, + {'id': sites[1].pk, 'name': 'Site Y'}, + {'id': sites[2].pk, 'status': 'not-a-valid-status'}, + ] + url = reverse('dcim-api:site-list') + self.add_permissions('dcim.change_site') + response = self.client.patch(url, data, format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST) + + # No object may have been modified + for site in sites: + site.refresh_from_db() + self.assertListEqual([site.name for site in sites], ['Site 1', 'Site 2', 'Site 3']) + + # No task may be queued for an update that was rolled back + self.assertEqual(self.queue.count, 0) + def test_single_delete_process_eventrule(self): """ Check that deleting an object with an applicable EventRule queues a background task for the rule's action. @@ -333,6 +491,35 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase): self.assertEqual(job.kwargs['snapshots']['prechange']['name'], 'Site 1') self.assertEqual(job.kwargs['snapshots']['prechange']['tags'], ['Bar', 'Foo']) + def test_single_delete_rollback_discards_events(self): + """ + Check that deleting an object whose cascading deletion is aborted queues no background task + for the dependent objects that were already processed. + """ + device = create_test_device('Device 1') + Interface.objects.create( + device=device, name='Interface 1', type=InterfaceTypeChoices.TYPE_1GE_FIXED, description='Has one' + ) + Interface.objects.create(device=device, name='Interface 2', type=InterfaceTypeChoices.TYPE_1GE_FIXED) + + event_rule = EventRule.objects.get(name='Event Rule 3') + event_rule.object_types.set([ObjectType.objects.get_for_model(Interface)]) + + url = reverse('dcim-api:device-detail', kwargs={'pk': device.pk}) + self.add_permissions('dcim.delete_device') + + # Deleting the Device cascades to both Interfaces. The first satisfies the protection rule + # and so is processed (queueing its event); the second does not, aborting the request. + protection_rules = {'dcim.interface': [{'description': {'required': True}}]} + with override_settings(PROTECTION_RULES=protection_rules): + response = self.client.delete(url, **self.header) + self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST) + self.assertTrue(Device.objects.filter(pk=device.pk).exists()) + self.assertEqual(Interface.objects.filter(device=device).count(), 2) + + # No task may be queued for a deletion that was rolled back + self.assertEqual(self.queue.count, 0) + def test_bulk_delete_process_eventrule(self): """ Check that bulk deleting multiple objects with an applicable EventRule queues a background task for each @@ -367,6 +554,58 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase): self.assertEqual(job.kwargs['snapshots']['prechange']['name'], sites[i].name) self.assertEqual(job.kwargs['snapshots']['prechange']['tags'], ['Bar', 'Foo']) + def test_bulk_delete_rollback_discards_events(self): + """ + Check that a bulk delete which is rolled back because one object is protected queues no + background tasks for the objects that were provisionally deleted. + """ + 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(sites) + + # A Device references the third Site, whose deletion will therefore raise a ProtectedError + # and roll the entire batch back. + create_test_device('Device 1', site=sites[2]) + + data = [{'id': site.pk} for site in sites] + url = reverse('dcim-api:site-list') + self.add_permissions('dcim.delete_site') + response = self.client.delete(url, data, format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_409_CONFLICT) + self.assertEqual(Site.objects.count(), 3) + + # No task may be queued for a deletion that was rolled back + self.assertEqual(self.queue.count, 0) + + def test_bulk_delete_abort_discards_events(self): + """ + Check that a bulk delete aborted by an exception (rather than by a per-object error) also + queues no background tasks. A protection rule raises AbortRequest from a signal receiver, + which propagates out of the per-object loop. + """ + sites = ( + Site(name='Site 1', slug='site-1', description='Has a description'), + Site(name='Site 2', slug='site-2'), + ) + Site.objects.bulk_create(sites) + + data = [{'id': site.pk} for site in sites] + url = reverse('dcim-api:site-list') + self.add_permissions('dcim.delete_site') + + # Site 2 has no description, so its deletion is blocked once Site 1 has already been deleted + protection_rules = {'dcim.site': [{'description': {'required': True}}]} + with override_settings(PROTECTION_RULES=protection_rules): + response = self.client.delete(url, data, format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST) + self.assertEqual(Site.objects.count(), 2) + + # No task may be queued for a deletion that was rolled back + self.assertEqual(self.queue.count, 0) + @skipIf('netbox.tests.dummy_plugin' not in settings.PLUGINS, 'dummy_plugin not in settings.PLUGINS') def test_send_webhook(self): request_id = uuid.uuid4() diff --git a/netbox/ipam/api/views.py b/netbox/ipam/api/views.py index cdc6786f0..9024dbe5f 100644 --- a/netbox/ipam/api/views.py +++ b/netbox/ipam/api/views.py @@ -19,7 +19,7 @@ from ipam import filtersets from ipam.models import * from ipam.utils import get_next_available_prefix from netbox.api.viewsets import NetBoxModelViewSet -from netbox.api.viewsets.mixins import ObjectValidationMixin +from netbox.api.viewsets.mixins import ObjectValidationMixin, discard_events_on_rollback from netbox.config import get_config from netbox.constants import ADVISORY_LOCK_KEYS from utilities.api import get_serializer_for_model @@ -295,8 +295,9 @@ class AvailableObjectsView(ObjectValidationMixin, APIView): return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) # Create the new IP address(es) + using = router.db_for_write(self.queryset.model) try: - with transaction.atomic(using=router.db_for_write(self.queryset.model)): + with transaction.atomic(using=using), discard_events_on_rollback(self, using=using): created = serializer.save() self._validate_objects(created) except ObjectDoesNotExist: diff --git a/netbox/netbox/api/viewsets/__init__.py b/netbox/netbox/api/viewsets/__init__.py index 36f330f54..24a0000aa 100644 --- a/netbox/netbox/api/viewsets/__init__.py +++ b/netbox/netbox/api/viewsets/__init__.py @@ -260,8 +260,9 @@ class NetBoxModelViewSet( logger.info(f"Creating new {model._meta.verbose_name}") # Enforce object-level permissions on save() + using = router.db_for_write(model) try: - with transaction.atomic(using=router.db_for_write(model)): + with transaction.atomic(using=using), mixins.discard_events_on_rollback(self, using=using): instance = serializer.save() self._validate_objects(instance) except ObjectDoesNotExist: @@ -299,8 +300,9 @@ class NetBoxModelViewSet( logger.info(f"Updating {model._meta.verbose_name} {serializer.instance} (PK: {serializer.instance.pk})") # Enforce object-level permissions on save() + using = router.db_for_write(model) try: - with transaction.atomic(using=router.db_for_write(model)): + with transaction.atomic(using=using), mixins.discard_events_on_rollback(self, using=using): # Re-check the If-Match ETag under a row-level lock to close the TOCTOU window # between the initial check in update() and the actual write. if self._get_if_match(self.request): @@ -333,8 +335,9 @@ class NetBoxModelViewSet( logger = logging.getLogger(f'netbox.api.views.{self.__class__.__name__}') logger.info(f"Deleting {model._meta.verbose_name} {instance} (PK: {instance.pk})") + using = router.db_for_write(model) try: - with transaction.atomic(using=router.db_for_write(model)): + with transaction.atomic(using=using), mixins.discard_events_on_rollback(self, using=using): # Re-check the If-Match ETag under a row-level lock to close the TOCTOU window # between the initial check in destroy() and the actual delete. if self._get_if_match(self.request): diff --git a/netbox/netbox/api/viewsets/mixins.py b/netbox/netbox/api/viewsets/mixins.py index 7caf38480..3dcba7a37 100644 --- a/netbox/netbox/api/viewsets/mixins.py +++ b/netbox/netbox/api/viewsets/mixins.py @@ -1,3 +1,5 @@ +from contextlib import contextmanager + from django.core.exceptions import ObjectDoesNotExist from django.db import router, transaction from django.http import Http404 @@ -5,6 +7,7 @@ from rest_framework import status from rest_framework.response import Response from core.models import ObjectType +from core.signals import clear_events from extras.models import ExportTemplate from netbox.api.serializers import BulkOperationSerializer from netbox.api.serializers.bulk import get_bulk_update_serializer_class @@ -16,9 +19,48 @@ __all__ = ( 'ExportTemplatesMixin', 'ObjectValidationMixin', 'SequentialBulkCreatesMixin', + 'discard_events_on_rollback', ) +@contextmanager +def discard_events_on_rollback(sender, using=None): + """ + Discard any queued events if the transaction wrapping this block is rolled back. + + The change logging signal receivers queue events eagerly, as the payload for a deleted object + must be captured while that object and its related rows are still reachable. The queue is not + flushed to the events pipeline until after the response has been rendered, however, so events + queued for writes which were subsequently rolled back would otherwise still be dispatched, + firing webhooks and event rules for changes that were never committed. + + Bulk operations need this because they provisionally write every valid object in a batch and + then roll the entire batch back if any one object failed. Single-object writes need it because + a write can be undone after it has been saved (for instance by the object-level permission + check in perform_create()/perform_update(), or by a signal receiver raising AbortRequest). The + UI's views send the same signal when they abandon a transaction. + + Must be entered *inside* the transaction whose rollback it guards, so that the rollback flag is + still set when this block exits. + + Note that this discards the entire request's queue, not only the events queued within the + guarded block. Nesting is therefore safe only because every rollback guarded here aborts the + whole request, making the two equivalent: the bulk actions guard the whole batch while the + per-object perform_*() calls they make guard each write, and a failure in either case abandons + the request. Do not use this in a loop which catches a per-object failure and continues, as + the events for objects which were successfully written would be discarded as well. + """ + try: + yield + except Exception: + # An exception escaping the block (e.g. AbortRequest raised by a signal receiver) rolls + # the transaction back just as an explicit set_rollback() does. + clear_events.send(sender=sender) + raise + if transaction.get_connection(using).needs_rollback: + clear_events.send(sender=sender) + + class CustomFieldsMixin: """ For models which support custom fields, populate the `custom_fields` context. @@ -61,7 +103,8 @@ class SequentialBulkCreatesMixin: appropriately. """ def create(self, request, *args, **kwargs): - with transaction.atomic(using=router.db_for_write(self.queryset.model)): + using = router.db_for_write(self.queryset.model) + with transaction.atomic(using=using), discard_events_on_rollback(self, using=using): if not isinstance(request.data, list): # Creating a single object return super().create(request, *args, **kwargs) @@ -122,7 +165,8 @@ class BulkUpdateModelMixin: def perform_bulk_update(self, objects, update_data, partial): updated_pks = [] - with transaction.atomic(using=router.db_for_write(self.queryset.model)): + using = router.db_for_write(self.queryset.model) + with transaction.atomic(using=using), discard_events_on_rollback(self, using=using): for obj in objects: data = update_data.get(obj.id) if hasattr(obj, 'snapshot'): @@ -185,7 +229,8 @@ class BulkDestroyModelMixin: def perform_bulk_destroy(self, objects, changelog_messages=None): changelog_messages = changelog_messages or {} - with transaction.atomic(using=router.db_for_write(self.queryset.model)): + using = router.db_for_write(self.queryset.model) + with transaction.atomic(using=using), discard_events_on_rollback(self, using=using): for obj in objects: if hasattr(obj, 'snapshot'): obj.snapshot()