Move script input validation to REST API and revert ScriptJob form-cleaning (fixes #22750)

* Perform Script input validation and convert ObjectVar/MultiObjectVar IDs -> model instances in extras/api/views.py: ScriptViewSet.post instead of in ScriptJob.run.
* Return HTTP 400 for invalid script input (form errors) so API clients receive immediate feedback instead of enqueuing failing background jobs.
* Revert ScriptJob.run to its original behavior so UI, management command, and EventRule callers keep their existing contracts (avoids breaking event-driven scripts).
* Explicitly pop known execution parameters (_schedule_at, _interval, _notifications, _commit) rather than generically stripping underscore-prefixed keys.
* Preserve uploaded files by adding them to cleaned data when the form did not claim them (so legacy scripts that expect files remain compatible).
* Update unit tests to reflect the new contract (ScriptJob.run receives cleaned data).
* Adds recommended follow-up tests (API integration, EventRule, MultiObjectVar) as follow-ups.

Fixes: #22750

Portions of this PR (initial code and tests) were drafted with assistance from an AI assistant (GitHub Copilot).

netbox/extras/api/views.py

@@ -367,17 +367,56 @@ class ScriptViewSet(ModelViewSet):
             raise RQWorkerNotRunningException()
 
         if input_serializer.is_valid():
+            # Instantiate the script class so we can validate/clean the input via its form.
+            script_class = script.python_class
+            script_instance = script_class()
+
+            # Prepare payload and files
+            payload = input_serializer.validated_data.get('data', {}) or {}
+            files = request.FILES if request else None
+
+            # Validate via the script's form so ObjectVar/MultiObjectVar IDs get converted
+            try:
+                form = script_instance.as_form(data=payload, files=files)
+            except Exception as e:
+                # Defensive: if form construction raises, respond 400 with a helpful message.
+                return Response({'detail': f"Error preparing script form: {e}"}, status=status.HTTP_400_BAD_REQUEST)
+
+            if not form.is_valid():
+                # Return form errors as a 400 so clients get immediate feedback (instead of a failed background job)
+                return Response(form.errors, status=status.HTTP_400_BAD_REQUEST)
+
+            # Use cleaned_data for execution parameters and script variables
+            cleaned = dict(form.cleaned_data)
+
+            # Pop known execution parameters explicitly (do not generically strip _-prefixed names)
+            schedule_at = cleaned.pop('_schedule_at', input_serializer.validated_data.get('schedule_at'))
+            interval = cleaned.pop('_interval', input_serializer.validated_data.get('interval'))
+            notifications = cleaned.pop('_notifications', input_serializer.validated_data.get('notifications'))
+            commit = cleaned.pop(
+                '_commit',
+                input_serializer.validated_data.get('commit', script_instance.commit_default)
+            )
+
+            # Ensure any uploaded files are preserved if not claimed by the form
+            if files:
+                for fname, fobj in files.items():
+                    if fname not in cleaned:
+                        cleaned[fname] = fobj
+
+            # Enqueue the job with cleaned data (model instances/QuerySets where appropriate)
             ScriptJob.enqueue(
                 instance=script,
                 user=request.user,
-                data=input_serializer.data['data'],
+                data=cleaned,
                 request=copy_safe_request(request),
-                commit=input_serializer.data['commit'],
-                job_timeout=script.python_class.job_timeout,
-                schedule_at=input_serializer.validated_data.get('schedule_at'),
-                interval=input_serializer.validated_data.get('interval'),
-                notifications=input_serializer.validated_data.get('notifications'),
+                commit=commit,
+                job_timeout=script_class.job_timeout,
+                schedule_at=schedule_at,
+                interval=interval,
+                notifications=notifications,
             )
+
             serializer = serializers.ScriptDetailSerializer(script, context={'request': request})
 
             return Response(serializer.data)

netbox/extras/jobs.py

@@ -2,7 +2,6 @@ import logging
 import traceback
 from contextlib import ExitStack
 
-from django.core.exceptions import ValidationError
 from django.db import DEFAULT_DB_ALIAS, router, transaction
 from django.utils.translation import gettext as _
 
@@ -113,53 +112,16 @@ class ScriptJob(JobRunner):
         script = script_model.python_class()
         self.logger.debug(f"Loaded script {script.full_name}")
 
+        # Add files to form data
+        if request:
+            files = request.FILES
+            for field_name, fileobj in files.items():
+                data[field_name] = fileobj
+
         # Add the current request as a property of the script
         script.request = request
         self.logger.debug(f"Request ID: {request.id if request else None}")
 
-        # Normalize incoming payload for the form: API callers submit variables under "data".
-        payload = data or {}
-        if isinstance(payload, dict) and 'data' in payload:
-            payload = payload['data'] or {}
-
-        files = request.FILES if request else None
-        if files:
-            for field_name, fileobj in files.items():
-                # merge into payload so script.run receives the uploaded files in data
-                payload[field_name] = fileobj
-
-        # Validate & clean using the script's form so ObjectVar/MultiObjectVar IDs become model instances
-        if hasattr(script, 'as_form') and callable(getattr(script, 'as_form')):
-            try:
-                form = script.as_form(data=payload, files=files)
-                if not form.is_valid():
-                    raise AbortScript(f"Script input validation failed: {form.errors.as_json()}")
-
-                cleaned = form.cleaned_data
-
-                # Remove execution parameters
-                for key in list(cleaned.keys()):
-                    if key.startswith('_'):
-                        cleaned.pop(key)
-
-                # Preserve uploaded files that were merged into the payload so scripts still see them
-                # even if the Script's form doesn't declare file fields.
-                if files:
-                    for fname, fobj in files.items():
-                        if fname not in cleaned:
-                            cleaned[fname] = fobj
-
-                # Use cleaned form data as the data passed into the script
-                data = cleaned
-            except AbortScript:
-                # Re-raise for run_script() to log/handle
-                raise
-            except (ValidationError, TypeError, ValueError) as e:
-                raise AbortScript(f"Error validating script input: {e!s}")
-        else:
-            # Script doesn't provide as_form (e.g., lightweight test double); keep `data` as-is.
-            data = payload
-
         if commit:
             self.logger.info("Executing script (commit enabled)")
         else:

netbox/extras/tests/test_jobs.py

@@ -369,8 +369,18 @@ class ScriptJobFormCleaningTestCase(TestCase):
 
         runner = _make_runner(object_id=1)
         with patch.object(ScriptModel.objects, 'get', return_value=script_model_stub):
-            # Pass nested payload shape like the REST API: {"data": {"site": <pk>}}
-            runner.run(data={'data': {'site': site.pk}}, request=None, commit=True)
+            # Simulate what the API view does: build the form and use cleaned_data
+            form = script_instance.as_form(data={'site': site.pk}, files=None)
+            assert form.is_valid(), f"test setup: form invalid: {form.errors}"
+            cleaned = dict(form.cleaned_data)
+            # Pop execution params if present (API does this)
+            cleaned.pop('_commit', None)
+            cleaned.pop('_schedule_at', None)
+            cleaned.pop('_interval', None)
+            cleaned.pop('_notifications', None)
+
+            # Pass the cleaned dict to run(), not the nested {"data": {...}} shape
+            runner.run(data=cleaned, request=None, commit=True)
 
         # Assert the script received a model instance for 'site'
         self.assertIsNotNone(script_instance.received)
@@ -391,8 +401,7 @@ class ScriptJobFormCleaningTestCase(TestCase):
 
             def run(self, data, commit=True):
                 self.received = data
-                actual_type = type(data['site'])
-                return f"got type: {actual_type}"
+                return "ok"
 
         script_instance = TestScript()
         script_model_stub = MagicMock()
@@ -405,7 +414,17 @@ class ScriptJobFormCleaningTestCase(TestCase):
 
         runner = _make_runner(object_id=1)
         with patch.object(ScriptModel.objects, 'get', return_value=script_model_stub):
-            runner.run(data={'data': {'site': site.pk}}, request=fake_request, commit=True)
+            # Build the cleaned data as the API would
+            form = script_instance.as_form(data={'site': site.pk}, files=fake_request.FILES)
+            assert form.is_valid(), f"test setup: form invalid: {form.errors}"
+            cleaned = dict(form.cleaned_data)
+
+            # API merges any uploaded files that the form didn't declare
+            for fname, fobj in fake_request.FILES.items():
+                if fname not in cleaned:
+                    cleaned[fname] = fobj
+
+            runner.run(data=cleaned, request=fake_request, commit=True)
 
         # Assert both file merged and ObjectVar conversion happened
         self.assertIsNotNone(script_instance.received)