Closes #22872: Validate custom script Meta values before enqueueing (#23068)

Validate the effective timeout and notification settings at the
ScriptJob enqueue boundary so invalid script configuration is reported
consistently across all execution paths instead of raising an unhandled
exception.

Preserve the inherited positional enqueue contract and prevent tests
from interfering through shared RQ queue state during parallel runs.

netbox/extras/api/views.py

@@ -1,3 +1,4 @@
+from django.core.exceptions import ValidationError as DjangoValidationError
 from django.http import Http404
 from django.shortcuts import get_object_or_404
 from django.utils.translation import gettext_lazy as _
@@ -404,17 +405,23 @@ class ScriptViewSet(ListModelMixin, RetrieveModelMixin, BaseViewSet):
             raise RQWorkerNotRunningException()
 
         if input_serializer.is_valid():
-            ScriptJob.enqueue(
-                instance=script,
-                user=request.user,
-                data=input_serializer.data['data'],
-                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'),
-            )
+            try:
+                ScriptJob.enqueue(
+                    instance=script,
+                    user=request.user,
+                    data=input_serializer.data['data'],
+                    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'),
+                )
+            except DjangoValidationError as e:
+                # The script's execution configuration is invalid (see #22872). Surface it as a 400 rather than
+                # allowing the exception to bubble up as an HTTP 500. These are script-level config errors, not
+                # request-field errors, so report them under the non-field "detail" key.
+                raise ValidationError({'detail': e.messages}) from e
             serializer = serializers.ScriptDetailSerializer(script, context={'request': request})
 
             return Response(serializer.data)

netbox/extras/events.py

@@ -2,6 +2,7 @@ import logging
 from collections import UserDict, defaultdict
 
 from django.conf import settings
+from django.core.exceptions import ValidationError
 from django.utils import timezone
 from django.utils.module_loading import import_string
 from django.utils.translation import gettext as _
@@ -261,8 +262,18 @@ def process_event_rules(event_rules, object_type, event):
             if 'request' in event:
                 params['request'] = copy_safe_request(event['request'], include_files=False)
 
-            # Enqueue the job
-            ScriptJob.enqueue(**params)
+            # Enqueue the job. If the script's Meta configuration is invalid (see #22872), log the error and skip this
+            # action rather than allowing the exception to abort the event pipeline (and, since events are processed
+            # in-request, the originating object change). Note this is intentionally asymmetric with the webhook
+            # branch above, which lets enqueue failures propagate: script Meta is validated eagerly at enqueue and a
+            # misconfigured script must not take down an unrelated object change.
+            try:
+                ScriptJob.enqueue(**params)
+            except ValidationError as e:
+                logger.error(
+                    "Skipping script action for event rule %s: invalid script configuration: %s",
+                    event_rule, '; '.join(e.messages)
+                )
 
         # Notification groups
         elif event_rule.action_type == EventRuleActionChoices.NOTIFICATION:

netbox/extras/jobs.py

@@ -8,6 +8,7 @@ from django.utils.translation import gettext as _
 from core.signals import clear_events
 from dcim.models import Device
 from extras.models import Script as ScriptModel
+from extras.scripts import _UNSET
 from netbox.context_managers import event_tracking
 from netbox.jobs import JobRunner
 from netbox.registry import registry
@@ -27,6 +28,30 @@ class ScriptJob(JobRunner):
     class Meta:
         name = 'Run Script'
 
+    @classmethod
+    def enqueue(cls, *args, **kwargs):
+        """
+        Validate the script's execution parameters before enqueueing. This is the single choke point through which
+        every script execution passes (interactive runs, the REST API, the runscript command, event-rule actions, and
+        recurring reschedules), so validating here surfaces a misconfigured script as an actionable error rather than
+        an unhandled exception at enqueue time (see #22872).
+
+        The values actually being enqueued are validated, not just the script's Meta defaults, so an explicit
+        job_timeout or notifications supplied by the caller is checked too.
+        """
+        # The instance may be passed positionally (JobRunner.enqueue() forwards it to Job.enqueue()'s first argument)
+        # or by keyword. Resolve it for validation without consuming it, so the original arguments are forwarded to
+        # super() unchanged and the inherited calling contract is preserved.
+        instance = args[0] if args else kwargs.get('instance')
+        script_class = getattr(instance, 'python_class', None)
+        if script_class is not None:
+            script_class.validate_meta(
+                job_timeout=kwargs.get('job_timeout', _UNSET),
+                notifications=kwargs.get('notifications', _UNSET),
+            )
+
+        return super().enqueue(*args, **kwargs)
+
     def run_script(self, script, request, data, commit):
         """
         Core script execution task. We capture this within a method to allow for conditionally wrapping it with the

netbox/extras/management/commands/runscript.py

@@ -3,6 +3,7 @@ import logging
 import sys
 import uuid
 
+from django.core.exceptions import ValidationError
 from django.core.management.base import BaseCommand, CommandError
 
 from extras.jobs import ScriptJob
@@ -88,24 +89,29 @@ class Command(BaseCommand):
         notifications = form.cleaned_data.pop('_notifications')
 
         # Execute the script.
-        job = ScriptJob.enqueue(
-            instance=script_obj,
-            user=user,
-            immediate=True,
-            data=form.cleaned_data,
-            notifications=notifications,
-            request=NetBoxFakeRequest({
-                'META': {},
-                'COOKIES': {},
-                'POST': data,
-                'GET': {},
-                'FILES': {},
-                'user': user,
-                'method': 'POST',
-                'path': '',
-                'id': uuid.uuid4()
-            }),
-            commit=commit,
-        )
+        try:
+            job = ScriptJob.enqueue(
+                instance=script_obj,
+                user=user,
+                immediate=True,
+                data=form.cleaned_data,
+                notifications=notifications,
+                request=NetBoxFakeRequest({
+                    'META': {},
+                    'COOKIES': {},
+                    'POST': data,
+                    'GET': {},
+                    'FILES': {},
+                    'user': user,
+                    'method': 'POST',
+                    'path': '',
+                    'id': uuid.uuid4()
+                }),
+                commit=commit,
+            )
+        except ValidationError as e:
+            # The script's Meta configuration is invalid (see #22872). Report it as a clean command error rather than
+            # an unhandled traceback.
+            raise CommandError('; '.join(e.messages))
 
         logger.info(f"Script completed in {job.duration}")

netbox/extras/scripts.py

@@ -4,11 +4,14 @@ import os
 import re
 
 from django import forms
+from django.core.exceptions import ValidationError
 from django.core.files.storage import storages
 from django.core.validators import RegexValidator
 from django.utils import timezone
 from django.utils.functional import classproperty
 from django.utils.translation import gettext as _
+from rq.exceptions import TimeoutFormatError
+from rq.utils import parse_timeout
 
 from core.choices import JobNotificationChoices
 from extras.choices import LogLevelChoices
@@ -43,6 +46,9 @@ __all__ = (
     'get_module_and_script',
 )
 
+# Sentinel distinguishing "argument not supplied" from an explicit None in validate_meta().
+_UNSET = object()
+
 
 #
 # Script variables
@@ -403,6 +409,51 @@ class BaseScript:
     def notifications_default(self):
         return getattr(self.Meta, 'notifications_default', JobNotificationChoices.NOTIFICATION_ALWAYS)
 
+    @classmethod
+    def validate_meta(cls, job_timeout=_UNSET, notifications=_UNSET):
+        """
+        Validate the execution parameters used to run this script. Raises a ValidationError if any value is invalid,
+        so that a misconfigured script surfaces an actionable error rather than an unhandled exception when the job is
+        enqueued (see #22872).
+
+        The values actually enqueued are validated, not the raw Meta values: a caller may supply an explicit
+        `job_timeout` or `notifications` (e.g. via the REST API), in which case that value is checked. When a caller
+        omits a value, the corresponding Meta default is validated instead. Unset values fall back to valid defaults
+        and are not rejected.
+        """
+        errors = {}
+
+        job_timeout = cls.job_timeout if job_timeout is _UNSET else job_timeout
+        if job_timeout is not None:
+            # parse_timeout() is what RQ applies to the timeout downstream. It raises TimeoutFormatError for
+            # malformed duration strings, but a job_timeout of an unexpected type (e.g. a list) instead raises
+            # TypeError/ValueError/AssertionError from its internal int()/assert. Catch them all so any invalid value
+            # surfaces as an actionable error rather than an unhandled 500.
+            try:
+                parsed_timeout = parse_timeout(job_timeout)
+            except (TimeoutFormatError, TypeError, ValueError, AssertionError):
+                parsed_timeout = None
+                errors['job_timeout'] = _(
+                    "Invalid job_timeout value '{value}': must be an integer (seconds) or a duration string such as "
+                    "'1h' or '30m'."
+                ).format(value=job_timeout)
+            if parsed_timeout is not None and parsed_timeout <= 0:
+                errors['job_timeout'] = _(
+                    "Invalid job_timeout value '{value}': must be a positive duration."
+                ).format(value=job_timeout)
+
+        # A caller may pass notifications=None to mean "use the script's default"; treat that as unset.
+        if notifications is _UNSET or notifications is None:
+            notifications = cls.notifications_default
+        if notifications not in JobNotificationChoices.values():
+            valid = ', '.join(JobNotificationChoices.values())
+            errors['notifications_default'] = _(
+                "Invalid notifications value '{value}': must be one of {valid}."
+            ).format(value=notifications, valid=valid)
+
+        if errors:
+            raise ValidationError(errors)
+
     @property
     def filename(self):
         return inspect.getfile(self.__class__)

netbox/extras/tests/test_api.py

@@ -3,7 +3,7 @@ import hashlib
 import io
 import json
 from contextlib import contextmanager
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock, PropertyMock, patch
 
 from django.contrib.contenttypes.models import ContentType
 from django.core.files.uploadedfile import SimpleUploadedFile
@@ -1615,6 +1615,54 @@ class ScriptTestCase(APITestCase):
 
         self.assertEqual(Job.objects.count(), len(lookups))
 
+    def test_run_script_invalid_job_timeout(self):
+        """
+        A script whose Meta.job_timeout is invalid must be rejected with a 400, not raise an unhandled exception
+        (#22872).
+        """
+        self.add_permissions('extras.run_script')
+
+        class BadTimeoutScript(PythonClass):
+            class Meta:
+                name = 'Bad Timeout'
+                job_timeout = 'not-a-timeout'
+
+            def run(self, data, commit=True):
+                pass
+
+        payload = {'data': {}, 'commit': True}
+        with patch.object(Script, 'python_class', new_callable=PropertyMock) as mock_python_class:
+            mock_python_class.return_value = BadTimeoutScript
+            with disable_warnings('django.request'):
+                response = self.client.post(self.url, payload, format='json', **self.header)
+
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertFalse(Job.objects.exists())
+
+    def test_run_script_invalid_notifications_default(self):
+        """
+        A script whose Meta.notifications_default is invalid must be rejected with a 400, not raise an unhandled
+        exception (#22872).
+        """
+        self.add_permissions('extras.run_script')
+
+        class BadNotificationsScript(PythonClass):
+            class Meta:
+                name = 'Bad Notifications'
+                notifications_default = 'on_error'
+
+            def run(self, data, commit=True):
+                pass
+
+        payload = {'data': {}, 'commit': True}
+        with patch.object(Script, 'python_class', new_callable=PropertyMock) as mock_python_class:
+            mock_python_class.return_value = BadNotificationsScript
+            with disable_warnings('django.request'):
+                response = self.client.post(self.url, payload, format='json', **self.header)
+
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertFalse(Job.objects.exists())
+
     def test_modify_script_methods_disabled(self):
         """
         Individual scripts are created, modified, and deleted through their module, so PUT/PATCH/DELETE on

netbox/extras/tests/test_event_rules.py

@@ -3,7 +3,7 @@ import logging
 import uuid
 from io import BytesIO
 from unittest import skipIf
-from unittest.mock import Mock, patch
+from unittest.mock import Mock, PropertyMock, patch
 
 import django_rq
 from django.conf import settings
@@ -995,6 +995,63 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase):
         self.assertEqual(script_job.status, "completed")
         self.assertEqual(script_job.data.get('output', ''), "finished successfully")
 
+    @tag('regression')  # Issue #22872
+    def test_eventrule_script_action_invalid_meta_does_not_abort_change(self):
+        """
+        A Script event-rule action whose Meta configuration is invalid must be logged and skipped without aborting
+        the triggering object change or raising an HTTP 500 (#22872). Because event rules are processed in-request,
+        an unhandled ValidationError here would fail the originating request.
+        """
+        class BadMetaScript(ScriptBase):
+            class Meta:
+                name = "Bad Meta Script"
+                job_timeout = 'not-a-timeout'
+
+            def run(self, data, commit=True):
+                return "never reached"
+
+        with patch.object(ScriptModule, 'sync_classes'):
+            module = ScriptModule.objects.create(
+                file_root=ManagedFileRootPathChoices.SCRIPTS,
+                file_path='bad_meta_script.py',
+            )
+        script = Script.objects.create(module=module, name='Bad Meta Script', is_executable=True)
+        script_type = ObjectType.objects.get_for_model(Script)
+
+        # Trigger on Manufacturer rather than Site: the class-level event rules all target Site, so a Site-based rule
+        # here would collide with them and perturb other tests' queue expectations.
+        manufacturer_type = ObjectType.objects.get_for_model(Manufacturer)
+        event_rule = EventRule.objects.create(
+            name='Bad Meta Script Rule',
+            event_types=[OBJECT_UPDATED],
+            action_type=EventRuleActionChoices.SCRIPT,
+            action_object_type=script_type,
+            action_object_id=script.pk,
+        )
+        event_rule.object_types.set([manufacturer_type])
+
+        manufacturer = Manufacturer.objects.create(name='Manufacturer 1', slug='manufacturer-1')
+        self.add_permissions('dcim.change_manufacturer')
+        url = reverse('dcim-api:manufacturer-detail', kwargs={'pk': manufacturer.pk})
+
+        # python_class is a property returning the script class; patch it to return our bad-Meta class so validate_meta
+        # (a classmethod on it) is exercised the way production reads it.
+        with patch.object(Script, 'python_class', new_callable=PropertyMock) as mock:
+            mock.return_value = BadMetaScript
+            with self.captureOnCommitCallbacks(execute=True):
+                with self.assertLogs('netbox.events_processor', 'ERROR') as captured:
+                    response = self.client.patch(url, {'description': 'updated'}, format='json', **self.header)
+
+        # The triggering object change succeeds despite the misconfigured script
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+        manufacturer.refresh_from_db()
+        self.assertEqual(manufacturer.description, 'updated')
+
+        # No script job was enqueued (nothing queued, no Job record), and the misconfiguration was logged
+        self.assertEqual(self.queue.count, 0)
+        self.assertEqual(Job.objects.filter(name=BadMetaScript.Meta.name).count(), 0)
+        self.assertTrue(any('Bad Meta Script Rule' in line for line in captured.output))
+
     @tag('regression')  # Issue #22852
     def test_eventrule_script_action_honors_script_defaults(self):
         """A script run from an event rule uses the notification policy and job timeout from its Meta class."""

netbox/extras/tests/test_management_commands.py

@@ -412,6 +412,31 @@ class RunScriptTestCase(TestCase):
 
         self.assertEqual(enqueue.call_args.kwargs['user'], self.user)
 
+    def test_invalid_meta_raises_command_error(self):
+        """
+        A script with an invalid Meta value must fail with a clean CommandError rather than an unhandled
+        exception (#22872).
+        """
+        class BadMetaScript(Script):
+            class Meta:
+                job_timeout = 'not-a-timeout'
+
+            def run(self, data, commit):
+                return None
+
+        script_obj = SimpleNamespace(python_class=BadMetaScript)
+
+        # Note: ScriptJob.enqueue is intentionally NOT mocked here, so validate_meta() runs and raises.
+        with (
+            patch(
+                'extras.management.commands.runscript.get_module_and_script',
+                return_value=(None, script_obj),
+            ),
+            patch('extras.management.commands.runscript.logging.getLogger'),
+        ):
+            with self.assertRaises(CommandError):
+                call_command('runscript', 'test.Script', user='admin', stdout=StringIO())
+
 
 class WebhookReceiverTestCase(TestCase):
     def test_starts_http_server(self):

netbox/extras/tests/test_scripts.py

@@ -1,15 +1,22 @@
 import io
 import sys
+import uuid
 from datetime import UTC, date, datetime
 from decimal import Decimal
-from unittest.mock import patch
+from unittest.mock import PropertyMock, patch
 
+from django.contrib.auth import get_user_model
+from django.core.exceptions import ValidationError
 from django.core.files.uploadedfile import SimpleUploadedFile
 from django.test import TestCase
 from netaddr import IPAddress, IPNetwork
 
+from core.choices import JobNotificationChoices, JobStatusChoices, ManagedFileRootPathChoices
+from core.models import Job
 from dcim.models import DeviceRole
 from extras.constants import SCRIPT_MODULE_NAME_PREFIX
+from extras.jobs import ScriptJob
+from extras.models import Script as ScriptModel
 from extras.models import ScriptModule
 from extras.scripts import *
 
@@ -469,3 +476,268 @@ class ScriptModuleLoadingTestCase(TestCase):
         with self.assertLogs(logger_name, 'INFO') as captured:
             script.log_success('Start')
         self.assertIn('Start', captured.output[0])
+
+
+class ScriptMetaValidationTestCase(TestCase):
+    """
+    Tests for BaseScript.validate_meta() (#22872): invalid execution-related Meta values must raise an actionable
+    ValidationError, while unset/valid values must not.
+    """
+
+    def test_valid_meta_passes(self):
+        class TestScript(Script):
+            class Meta:
+                job_timeout = 600
+                notifications_default = JobNotificationChoices.NOTIFICATION_ON_FAILURE
+
+            def run(self, data, commit):
+                pass
+
+        TestScript.validate_meta()  # should not raise
+
+    def test_job_timeout_duration_string_passes(self):
+        class TestScript(Script):
+            class Meta:
+                job_timeout = '1h'
+
+            def run(self, data, commit):
+                pass
+
+        TestScript.validate_meta()  # should not raise
+
+    def test_unset_meta_passes(self):
+        class TestScript(Script):
+            def run(self, data, commit):
+                pass
+
+        # job_timeout defaults to None and notifications_default to ALWAYS; neither should be rejected
+        TestScript.validate_meta()
+
+    def test_all_notification_choices_pass(self):
+        for choice in JobNotificationChoices.values():
+            class TestScript(Script):
+                class Meta:
+                    notifications_default = choice
+
+                def run(self, data, commit):
+                    pass
+
+            TestScript.validate_meta()  # should not raise
+
+    def test_invalid_job_timeout_raises(self):
+        class TestScript(Script):
+            class Meta:
+                job_timeout = 'not-a-timeout'
+
+            def run(self, data, commit):
+                pass
+
+        with self.assertRaises(ValidationError) as cm:
+            TestScript.validate_meta()
+        self.assertIn('job_timeout', cm.exception.message_dict)
+
+    def test_invalid_notifications_default_raises(self):
+        class TestScript(Script):
+            class Meta:
+                notifications_default = 'on_error'
+
+            def run(self, data, commit):
+                pass
+
+        with self.assertRaises(ValidationError) as cm:
+            TestScript.validate_meta()
+        self.assertIn('notifications_default', cm.exception.message_dict)
+
+    def test_non_string_job_timeout_raises(self):
+        # A job_timeout of an unexpected type must surface as a ValidationError, not an unhandled TypeError.
+        class TestScript(Script):
+            class Meta:
+                job_timeout = [60]
+
+            def run(self, data, commit):
+                pass
+
+        with self.assertRaises(ValidationError) as cm:
+            TestScript.validate_meta()
+        self.assertIn('job_timeout', cm.exception.message_dict)
+
+    def test_non_positive_job_timeout_raises(self):
+        # parse_timeout() accepts 0 and negatives, but a non-positive timeout is nonsensical and must be rejected.
+        for value in (0, -30):
+            class TestScript(Script):
+                class Meta:
+                    job_timeout = value
+
+                def run(self, data, commit):
+                    pass
+
+            with self.assertRaises(ValidationError) as cm:
+                TestScript.validate_meta()
+            self.assertIn('job_timeout', cm.exception.message_dict)
+
+
+class ScriptJobEnqueueValidationTestCase(TestCase):
+    """
+    Tests that ScriptJob.enqueue() validates Meta before creating a Job (#22872). This is the choke point exercised by
+    event-rule actions and recurring reschedules, which have no request layer to catch the error.
+    """
+
+    @classmethod
+    def setUpTestData(cls):
+        cls.user = get_user_model().objects.create_user('scriptrunner')
+
+    def _make_script(self, python_class):
+        with patch.object(ScriptModule, 'sync_classes'):
+            module = ScriptModule.objects.create(
+                file_root=ManagedFileRootPathChoices.SCRIPTS,
+                file_path=f'meta_validation_{id(python_class)}.py',
+            )
+        script = ScriptModel.objects.create(module=module, name=python_class.Meta.name, is_executable=True)
+        # Return the raw python_class regardless of on-disk module state
+        patcher = patch.object(ScriptModel, 'python_class', property(lambda self, pc=python_class: pc))
+        patcher.start()
+        self.addCleanup(patcher.stop)
+        return script
+
+    def test_enqueue_rejects_invalid_job_timeout(self):
+        class BadTimeout(Script):
+            class Meta:
+                name = 'Bad Timeout'
+                job_timeout = 'not-a-timeout'
+
+            def run(self, data, commit):
+                pass
+
+        script = self._make_script(BadTimeout)
+        with self.captureOnCommitCallbacks(execute=True):
+            with self.assertRaises(ValidationError):
+                ScriptJob.enqueue(
+                    instance=script, user=self.user, job_timeout=BadTimeout.job_timeout,
+                    notifications=BadTimeout.notifications_default, data={}, commit=True,
+                )
+        self.assertEqual(Job.objects.count(), 0)
+
+    def test_enqueue_rejects_invalid_notifications_default(self):
+        class BadNotifications(Script):
+            class Meta:
+                name = 'Bad Notifications'
+                notifications_default = 'on_error'
+
+            def run(self, data, commit):
+                pass
+
+        script = self._make_script(BadNotifications)
+        with self.captureOnCommitCallbacks(execute=True):
+            with self.assertRaises(ValidationError):
+                ScriptJob.enqueue(
+                    instance=script, user=self.user, job_timeout=BadNotifications.job_timeout,
+                    notifications=BadNotifications.notifications_default, data={}, commit=True,
+                )
+        self.assertEqual(Job.objects.count(), 0)
+
+    def test_enqueue_accepts_valid_meta(self):
+        class GoodScript(Script):
+            class Meta:
+                name = 'Good Script'
+                job_timeout = '1h'
+                notifications_default = JobNotificationChoices.NOTIFICATION_ALWAYS
+
+            def run(self, data, commit):
+                pass
+
+        script = self._make_script(GoodScript)
+        # Do not execute the on_commit callback: the Job row is created by Job.enqueue() before the RQ push is
+        # registered, so asserting the row exists needs no real enqueue. Executing it would leave a job in the shared
+        # Redis queue that races other tests under the parallel runner (see #22872).
+        with self.captureOnCommitCallbacks():
+            job = ScriptJob.enqueue(
+                instance=script, user=self.user, job_timeout=GoodScript.job_timeout,
+                notifications=GoodScript.notifications_default, data={}, commit=True,
+            )
+        self.assertIsNotNone(job)
+        self.assertEqual(Job.objects.count(), 1)
+
+    def test_enqueue_positional_instance_is_validated_and_forwarded(self):
+        """
+        The instance may be passed positionally (JobRunner.enqueue() forwards it to Job.enqueue()'s first argument).
+        The override must validate it without breaking that inherited calling contract (#22872).
+        """
+        class BadTimeout(Script):
+            class Meta:
+                name = 'Bad Timeout Positional'
+                job_timeout = 'not-a-timeout'
+
+            def run(self, data, commit):
+                pass
+
+        script = self._make_script(BadTimeout)
+        # Passed positionally, not instance=... — must still be validated and rejected.
+        with self.captureOnCommitCallbacks(execute=True):
+            with self.assertRaises(ValidationError):
+                ScriptJob.enqueue(script, user=self.user, data={}, commit=True)
+        self.assertEqual(Job.objects.count(), 0)
+
+    def test_enqueue_positional_instance_valid_meta_creates_job(self):
+        """A valid script passed positionally must enqueue cleanly, i.e. the override forwards args unchanged."""
+        class GoodScript(Script):
+            class Meta:
+                name = 'Good Positional'
+
+            def run(self, data, commit):
+                pass
+
+        script = self._make_script(GoodScript)
+        # Do not execute the on_commit callback (see the note in test_enqueue_accepts_valid_meta): asserting the Job
+        # row exists needs no real RQ push, and executing it would leak a job into the shared queue (see #22872).
+        with self.captureOnCommitCallbacks():
+            job = ScriptJob.enqueue(script, user=self.user, data={}, commit=True)
+        self.assertIsNotNone(job)
+        self.assertEqual(Job.objects.count(), 1)
+
+    def test_reschedule_with_invalid_meta_preserves_completed_run(self):
+        """
+        If a recurring script's Meta.job_timeout becomes invalid between runs, the occurrence that just ran to
+        completion must keep its COMPLETED status and not be re-terminated as ERRORED, no successor may be scheduled,
+        and the reschedule failure must be recorded on the job (#22872).
+        """
+        class RecurringScript(Script):
+            class Meta:
+                name = 'Recurring'
+                # No custom job_timeout at schedule time: valid.
+
+            def run(self, data, commit):
+                pass
+
+        script = self._make_script(RecurringScript)
+
+        # Create a completed, recurring job as if a scheduled occurrence had just finished successfully.
+        job = Job.objects.create(
+            object=script,
+            name='Recurring',
+            status=JobStatusChoices.STATUS_COMPLETED,
+            user=self.user,
+            interval=60,
+            job_id=uuid.uuid4(),
+        )
+
+        # The script's Meta is edited to an invalid job_timeout before the reschedule fires.
+        class RecurringScriptBadTimeout(RecurringScript):
+            class Meta(RecurringScript.Meta):
+                job_timeout = 'not-a-timeout'
+
+        with patch.object(ScriptModel, 'python_class', new_callable=PropertyMock) as mock_pc:
+            mock_pc.return_value = RecurringScriptBadTimeout
+            with self.captureOnCommitCallbacks(execute=True):
+                # handle() runs the script (which succeeds) and then reschedules in its finally block; the reschedule
+                # enqueue is what fails validation here.
+                ScriptJob.handle(job, data={}, commit=False)
+
+        job.refresh_from_db()
+        # The completed run's status is preserved (not flipped to ERRORED)
+        self.assertEqual(job.status, JobStatusChoices.STATUS_COMPLETED)
+        # No successor was scheduled
+        self.assertEqual(
+            Job.objects.filter(name='Recurring').exclude(pk=job.pk).count(), 0
+        )
+        # The reschedule failure was recorded on the job
+        self.assertTrue(any('not rescheduled' in entry.get('message', '') for entry in job.log_entries))

netbox/extras/tests/test_views.py

@@ -1347,6 +1347,63 @@ class ScriptValidationErrorTestCase(TestCase):
         self.assertEqual(len(messages), 0)
 
 
+class ScriptMetaValidationViewTestCase(TestCase):
+    """
+    A script whose Meta declares an invalid job_timeout or notifications_default must surface an actionable error on
+    the run view rather than returning an HTTP 500 (#22872).
+    """
+    user_permissions = ['extras.view_script', 'extras.run_script']
+
+    class BadTimeoutScript(PythonClass):
+        class Meta:
+            name = 'Bad Timeout'
+            job_timeout = 'not-a-timeout'
+
+        def run(self, data, commit):
+            return "Complete"
+
+    class BadNotificationsScript(PythonClass):
+        class Meta:
+            name = 'Bad Notifications'
+            notifications_default = 'on_error'
+
+        def run(self, data, commit):
+            return "Complete"
+
+    @classmethod
+    def setUpTestData(cls):
+        with patch.object(ScriptModule, 'sync_classes'):
+            module = ScriptModule.objects.create(
+                file_root=ManagedFileRootPathChoices.SCRIPTS,
+                file_path='bad_meta.py',
+            )
+        cls.script = Script.objects.create(module=module, name='Bad meta', is_executable=True)
+
+    def _run_and_assert(self, python_class):
+        url = reverse('extras:script', kwargs={'pk': self.script.pk})
+        with patch.object(Script, 'python_class', new_callable=PropertyMock) as mock_python_class:
+            mock_python_class.return_value = python_class
+            with patch('extras.views.any_workers_for_queue', return_value=True):
+                with self.captureOnCommitCallbacks(execute=True):
+                    # Quick-run style: omit _notifications
+                    response = self.client.post(url, {'_commit': 'true'})
+
+        # Re-render with an error message, not a 500, and no Job enqueued
+        self.assertEqual(response.status_code, 200)
+        messages = list(response.context['messages'])
+        self.assertEqual(len(messages), 1)
+        self.assertIn('Unable to run script', str(messages[0]))
+        self.assertEqual(Job.objects.count(), 0)
+
+    @tag('regression')
+    def test_invalid_job_timeout_shows_error(self):
+        self._run_and_assert(self.BadTimeoutScript)
+
+    @tag('regression')
+    def test_invalid_notifications_default_shows_error(self):
+        self._run_and_assert(self.BadNotificationsScript)
+
+
 class ScriptDefaultValuesTestCase(TestCase):
     user_permissions = ['extras.view_script', 'extras.run_script']
 

netbox/extras/views.py

@@ -3,6 +3,7 @@ from datetime import datetime
 from django.contrib import messages
 from django.contrib.auth.mixins import LoginRequiredMixin
 from django.contrib.contenttypes.models import ContentType
+from django.core.exceptions import ValidationError
 from django.core.paginator import EmptyPage
 from django.db.models import Count, Q
 from django.http import Http404, HttpResponse, HttpResponseBadRequest, HttpResponseForbidden
@@ -1751,19 +1752,25 @@ class ScriptView(BaseScriptView):
             messages.error(request, _("Unable to run script: RQ worker process not running."))
         elif form.is_valid():
             ScriptJob = import_string("extras.jobs.ScriptJob")
-            job = ScriptJob.enqueue(
-                instance=script,
-                user=request.user,
-                schedule_at=form.cleaned_data.pop('_schedule_at'),
-                interval=form.cleaned_data.pop('_interval'),
-                notifications=form.cleaned_data.pop('_notifications'),
-                data=form.cleaned_data,
-                request=copy_safe_request(request),
-                job_timeout=script.python_class.job_timeout,
-                commit=form.cleaned_data.pop('_commit'),
-            )
-
-            return redirect('extras:script_result', job_pk=job.pk)
+            try:
+                job = ScriptJob.enqueue(
+                    instance=script,
+                    user=request.user,
+                    schedule_at=form.cleaned_data.pop('_schedule_at'),
+                    interval=form.cleaned_data.pop('_interval'),
+                    notifications=form.cleaned_data.pop('_notifications'),
+                    data=form.cleaned_data,
+                    request=copy_safe_request(request),
+                    job_timeout=script.python_class.job_timeout,
+                    commit=form.cleaned_data.pop('_commit'),
+                )
+            except ValidationError as e:
+                # The script's Meta configuration is invalid (see #22872). Surface it as a form error rather than
+                # allowing the exception to bubble up as an HTTP 500.
+                for msg in e.messages:
+                    messages.error(request, _("Unable to run script: {error}").format(error=msg))
+            else:
+                return redirect('extras:script_result', job_pk=job.pk)
         else:
             fieldset_fields = {field for _, fields in script_class.get_fieldsets() for field in fields}
             hidden_errors = {

netbox/netbox/jobs.py

@@ -5,9 +5,10 @@ from abc import ABC, abstractmethod
 from datetime import timedelta
 from pathlib import Path
 
-from django.core.exceptions import ImproperlyConfigured
+from django.core.exceptions import ImproperlyConfigured, ValidationError
 from django.utils import timezone
 from django.utils.functional import classproperty
+from django.utils.translation import gettext_lazy as _
 from django_pglocks import advisory_lock
 from rq.timeouts import JobTimeoutException
 
@@ -147,26 +148,46 @@ class JobRunner(ABC):
                     **kwargs,
                 )
 
-                if cls in registry['system_jobs']:
-                    # System jobs are also scheduled by `enqueue_once()` at worker startup,
-                    # which races with this finally block and can produce duplicate schedules
-                    # (see #22232). Acquire the same advisory lock used by `enqueue_once()`
-                    # and skip rescheduling if a successor is already enqueued.
-                    #
-                    # This branch is limited to system jobs because generic recurring jobs
-                    # (e.g. scheduled scripts) may have multiple legitimate schedules sharing
-                    # the same runner/object/interval but differing in their runtime kwargs.
-                    with advisory_lock(ADVISORY_LOCK_KEYS['job-schedules']):
-                        successor_exists = Job.objects.filter(
-                            name=cls.name,
-                            object_id__isnull=True,
-                            status__in=JobStatusChoices.ENQUEUED_STATE_CHOICES,
-                            interval=job.interval,
-                        ).exclude(pk=job.pk).exists()
-                        if not successor_exists:
-                            cls.enqueue(**enqueue_kwargs)
-                else:
-                    cls.enqueue(**enqueue_kwargs)
+                # Reschedule the next occurrence. If the object's configuration has become invalid since this run was
+                # scheduled (e.g. a script's Meta.job_timeout was edited to an invalid value, see #22872), the enqueue
+                # will raise a ValidationError. Record it on this job and decline to reschedule rather than allowing an
+                # unhandled exception to escape the worker's finally block.
+                try:
+                    if cls in registry['system_jobs']:
+                        # System jobs are also scheduled by `enqueue_once()` at worker startup,
+                        # which races with this finally block and can produce duplicate schedules
+                        # (see #22232). Acquire the same advisory lock used by `enqueue_once()`
+                        # and skip rescheduling if a successor is already enqueued.
+                        #
+                        # This branch is limited to system jobs because generic recurring jobs
+                        # (e.g. scheduled scripts) may have multiple legitimate schedules sharing
+                        # the same runner/object/interval but differing in their runtime kwargs.
+                        with advisory_lock(ADVISORY_LOCK_KEYS['job-schedules']):
+                            successor_exists = Job.objects.filter(
+                                name=cls.name,
+                                object_id__isnull=True,
+                                status__in=JobStatusChoices.ENQUEUED_STATE_CHOICES,
+                                interval=job.interval,
+                            ).exclude(pk=job.pk).exists()
+                            if not successor_exists:
+                                cls.enqueue(**enqueue_kwargs)
+                    else:
+                        cls.enqueue(**enqueue_kwargs)
+                except ValidationError as e:
+                    # The successor could not be scheduled because the object's configuration is now invalid. Record
+                    # this against the (already-terminated) job without overwriting the outcome of the run that just
+                    # completed — re-running terminate() here would clobber a successful run's status and fire a
+                    # duplicate notification (see #22872).
+                    error = _("Recurring job not rescheduled due to invalid configuration: {error}").format(
+                        error='; '.join(e.messages)
+                    )
+                    logger.error(f"Job {job}: {error}")
+                    job.log(logging.makeLogRecord({
+                        'levelno': logging.ERROR,
+                        'levelname': 'ERROR',
+                        'msg': error,
+                    }))
+                    job.save()
 
     @classmethod
     def get_jobs(cls, instance=None):

netbox/utilities/testing/mixins.py

@@ -1,3 +1,4 @@
+from django.test.testcases import SerializeMixin
 from django_rq import get_queue
 from django_rq.workers import get_worker
 from rq import SimpleWorker
@@ -7,10 +8,17 @@ __all__ = (
 )
 
 
-class RQQueueTestMixin:
+class RQQueueTestMixin(SerializeMixin):
     """
     Clear RQ queues before and after each test.
+
+    Test classes using this mixin share a single RQ (Redis) instance. Under the parallel
+    test runner that Redis is not isolated per worker (unlike the database), so concurrent
+    classes that enqueue and assert exact queue counts race each other. SerializeMixin
+    holds an exclusive lock on `lockfile`, so no two classes using this mixin run at the
+    same time, which removes that cross-worker contention.
     """
+    lockfile = __file__
     rq_queue_names = ('default', 'high', 'low')
 
     @classmethod