Browse Source

Fixes #22872: Validate custom script Meta values before enqueueing

A custom script could declare an invalid job_timeout or notifications_default
in its Meta class. Both values were passed through to the job unvalidated: an
invalid notifications_default reached Job.full_clean() and an invalid job_timeout
reached RQ, each raising an unhandled exception that surfaced as an HTTP 500.

Validation now happens once at the ScriptJob.enqueue choke point, via a new
BaseScript.validate_meta() classmethod. Because every way of running a script
funnels through ScriptJob.enqueue (interactive runs, the REST API, the runscript
command, event-rule actions, and recurring reschedules), a misconfigured script
is caught before any job is created and each caller surfaces the error in its own
idiom: the UI re-renders the form with a message, the REST API returns a 400, the
runscript command raises a CommandError, an event-rule action is logged and
skipped so it cannot abort the triggering object change, and a recurring
reschedule is logged against the completed job without altering its outcome.

Unset values keep their existing valid defaults and are not rejected, so scripts
that run today are unaffected.
Jason Novinger 1 ngày trước cách đây
mục cha
commit
f576782f5a

+ 8 - 0
docs/release-notes/version-4.6.md

@@ -1,5 +1,13 @@
 # NetBox v4.6
 
+## v4.6.10
+
+### Bug Fixes
+
+* [#22872](https://github.com/netbox-community/netbox/issues/22872) - Report an actionable error instead of a server error when a custom script declares an invalid `job_timeout` or `notifications_default` in its `Meta` class
+
+---
+
 ## v4.6.9 (2026-08-25)
 
 ### Enhancements

+ 18 - 11
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 Meta 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 as non-field errors.
+                raise ValidationError(e.messages)
             serializer = serializers.ScriptDetailSerializer(script, context={'request': request})
 
             return Response(serializer.data)

+ 13 - 2
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:

+ 14 - 0
netbox/extras/jobs.py

@@ -27,6 +27,20 @@ class ScriptJob(JobRunner):
     class Meta:
         name = 'Run Script'
 
+    @classmethod
+    def enqueue(cls, instance, *args, **kwargs):
+        """
+        Validate the script's Meta 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).
+        """
+        script_class = getattr(instance, 'python_class', None)
+        if script_class is not None:
+            script_class.validate_meta()
+
+        return super().enqueue(instance, *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

+ 25 - 19
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}")

+ 41 - 0
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
@@ -403,6 +406,44 @@ class BaseScript:
     def notifications_default(self):
         return getattr(self.Meta, 'notifications_default', JobNotificationChoices.NOTIFICATION_ALWAYS)
 
+    @classmethod
+    def validate_meta(cls):
+        """
+        Validate the script's execution-related Meta parameters. 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). Unset values fall back to valid defaults and are not rejected.
+        """
+        errors = {}
+
+        job_timeout = cls.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)
+
+        notifications_default = cls.notifications_default
+        if notifications_default not in JobNotificationChoices.values():
+            valid = ', '.join(JobNotificationChoices.values())
+            errors['notifications_default'] = _(
+                "Invalid notifications_default value '{value}': must be one of {valid}."
+            ).format(value=notifications_default, valid=valid)
+
+        if errors:
+            raise ValidationError(errors)
+
     @property
     def filename(self):
         return inspect.getfile(self.__class__)

+ 49 - 1
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

+ 55 - 1
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,60 @@ 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)
+
+        site_type = ObjectType.objects.get_for_model(Site)
+        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([site_type])
+
+        site = Site.objects.create(name='Site 1', slug='site-1')
+        self.add_permissions('dcim.change_site')
+        url = reverse('dcim-api:site-detail', kwargs={'pk': site.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)
+        site.refresh_from_db()
+        self.assertEqual(site.description, 'updated')
+
+        # No script job was enqueued, and the misconfiguration was logged
+        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."""

+ 233 - 1
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,228 @@ 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)
+        with self.captureOnCommitCallbacks(execute=True):
+            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_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))

+ 57 - 0
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']
 

+ 20 - 13
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 = {

+ 42 - 21
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):