From f732a8e878889974ed4f86e90c408e9b8c1afdde Mon Sep 17 00:00:00 2001 From: mburggraf Date: Tue, 9 Jun 2026 19:55:20 +0200 Subject: [PATCH] Fixes #22376: Remove files from request for script action event rules --- netbox/extras/events.py | 2 +- netbox/extras/tests/test_event_rules.py | 153 +++++++++++++++++++++++- netbox/utilities/request.py | 3 + 3 files changed, 154 insertions(+), 4 deletions(-) diff --git a/netbox/extras/events.py b/netbox/extras/events.py index a2f3aac78..c01e02ed5 100644 --- a/netbox/extras/events.py +++ b/netbox/extras/events.py @@ -257,7 +257,7 @@ def process_event_rules(event_rules, object_type, event): if 'snapshots' in event: params['snapshots'] = event['snapshots'] if 'request' in event: - params['request'] = copy_safe_request(event['request']) + params['request'] = copy_safe_request(event['request'], include_files=False) # Enqueue the job ScriptJob.enqueue(**params) diff --git a/netbox/extras/tests/test_event_rules.py b/netbox/extras/tests/test_event_rules.py index 42faf5bdc..f2d753be5 100644 --- a/netbox/extras/tests/test_event_rules.py +++ b/netbox/extras/tests/test_event_rules.py @@ -1,5 +1,7 @@ import json +import logging import uuid +from io import BytesIO from unittest import skipIf from unittest.mock import Mock, patch @@ -8,16 +10,19 @@ from django.conf import settings from django.http import HttpResponse from django.test import RequestFactory, tag from django.urls import reverse +from PIL import Image from requests import Session from rest_framework import status +from core.choices import ManagedFileRootPathChoices from core.events import * -from core.models import ObjectType +from core.models import Job, ObjectType from dcim.choices import SiteStatusChoices -from dcim.models import Interface, Site +from dcim.models import 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, Script, Tag, Webhook +from extras.models import EventRule, 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 netbox.context_managers import event_tracking @@ -653,3 +658,145 @@ class EventRuleTestCase(APITestCase): self.add_permissions('dcim.add_site') response = self.client.post(url, {'name': 'Site X', 'slug': 'site-x'}, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_201_CREATED) + + @tag('regression') + def test_eventrule_script_action_with_object_image_files(self): + """ + Verify that a Script event-rule action can be enqueued and executed cleanly when the + triggering object carries uploaded files (e.g. DeviceType images). + This is a regression test for issue #22376. + + """ + # Create a dummy script class and an instance of it + class DummyScript(ScriptBase): + class Meta: + name = "Dummy Script" + + def run(self, data, commit=True): + return "finished successfully" + + dummy_script = DummyScript() + + # Create ScriptModule and Script + with patch.object(ScriptModule, 'sync_classes'): + module = ScriptModule.objects.create( + file_root=ManagedFileRootPathChoices.SCRIPTS, + file_path='dummy_script.py', + ) + script = Script.objects.create( + module=module, + name='Dummy Script', + is_executable=True, + ) + script_type = ObjectType.objects.get_for_model(Script) + + # Create an event rule that triggers on DeviceType update with Script action + devicetype_type = ObjectType.objects.get_for_model(DeviceType) + event_rule = EventRule.objects.create( + name='Test Script Event Rule with Files', + event_types=[OBJECT_UPDATED], + action_type=EventRuleActionChoices.SCRIPT, + action_object_type=script_type, + action_object_id=script.pk, + ) + event_rule.object_types.set([devicetype_type]) + + # Create a manufacturer and DeviceType + manufacturer = Manufacturer.objects.create( + name='Test Manufacturer', + slug='test-manufacturer', + ) + devicetype = DeviceType.objects.create( + model='Test DeviceType', + slug="test-devicetype", + manufacturer=manufacturer, + ) + + # Create an image file + image = BytesIO() + Image.new('RGB', (1, 1)).save(image, format='PNG') + image.name = 'test_image.png' + image.seek(0) + + # PATCH the DeviceType via REST API to add the image + data = { + 'front_image': image, + } + url = reverse('dcim-api:devicetype-detail', kwargs={'pk': devicetype.pk}) + self.add_permissions('dcim.change_devicetype') + + # Mock the script's python_class to prevent the test from trying to load from disk + with patch.object(Script, 'python_class') as mock: + mock.return_value = dummy_script + # Since in core/models/jobs.py Jobs are enqueued with a transaction.on_commit-handler + # we simulate commit by using captureOnCommitCallbacks context manager + with self.captureOnCommitCallbacks(execute=True): + response = self.client.patch(url, data, format='multipart', **self.header) + self.assertHttpStatus(response, status.HTTP_200_OK) + + # Assert that the script job was enqueued cleanly and is waiting for execution + self.assertEqual(self.queue.count, 1) + script_job = Job.objects.filter(name=dummy_script.name).last() + self.assertEqual(script_job.status, "pending") + + # silence rqworker (cleaner output) and trigger job execution + logging.getLogger('rq.worker').setLevel(logging.ERROR) + django_rq.get_worker().work(burst=True) + + # Assert that our script was executed without any errors + script_job.refresh_from_db() + self.assertEqual(script_job.status, "completed") + self.assertEqual(script_job.data.get('output', ''), "finished successfully") + + @tag('regression') + def test_eventrule_webhook_action_with_object_image_files(self): + """ + Verify that a Webhook event-rule action can be enqueued and executed cleanly when + the triggering object carries uploaded files (e.g. DeviceType images). + This is a regression test for issue #20873. + """ + # Create an event rule that triggers on DeviceType update with Script action + webhook = Webhook.objects.get(name='Webhook 1') + webhook_type = ObjectType.objects.get_for_model(Webhook) + devicetype_type = ObjectType.objects.get_for_model(DeviceType) + event_rule = EventRule.objects.create( + name='Test Webhook Event Rule with Files', + event_types=[OBJECT_UPDATED], + action_type=EventRuleActionChoices.WEBHOOK, + action_object_type=webhook_type, + action_object_id=webhook.pk, + ) + event_rule.object_types.set([devicetype_type]) + + # Create a manufacturer and DeviceType + manufacturer = Manufacturer.objects.create( + name='Test Manufacturer', + slug='test-manufacturer', + ) + devicetype = DeviceType.objects.create( + model='Test DeviceType', + slug="test-devicetype", + manufacturer=manufacturer, + ) + + # Create an image file + image = BytesIO() + Image.new('RGB', (1, 1)).save(image, format='PNG') + image.name = 'test_image.png' + image.seek(0) + + # PATCH the DeviceType via REST API to add the image + data = { + 'front_image': image, + } + url = reverse('dcim-api:devicetype-detail', kwargs={'pk': devicetype.pk}) + self.add_permissions('dcim.change_devicetype') + + response = self.client.patch(url, data, format='multipart', **self.header) + self.assertHttpStatus(response, status.HTTP_200_OK) + + # Assert that the webhook job was enqueued cleanly + self.assertEqual(self.queue.count, 1) + job = self.queue.jobs[0] + self.assertEqual(job.kwargs['event_rule'], event_rule) + self.assertEqual(job.kwargs['event_type'], OBJECT_UPDATED) diff --git a/netbox/utilities/request.py b/netbox/utilities/request.py index 6471a4dec..f57e2e4b9 100644 --- a/netbox/utilities/request.py +++ b/netbox/utilities/request.py @@ -3,6 +3,7 @@ from contextlib import ExitStack, contextmanager from urllib.parse import urlparse from django.conf import settings +from django.utils.datastructures import MultiValueDict from django.utils.http import url_has_allowed_host_and_scheme from django.utils.translation import gettext_lazy as _ from netaddr import AddrFormatError, IPAddress @@ -66,6 +67,8 @@ def copy_safe_request(request, include_files=True): } if include_files: data['FILES'] = request.FILES + else: + data['FILES'] = MultiValueDict() return NetBoxFakeRequest(data)