Fixes #23364: Restrict script API job results to jobs the user can view (#23409)

* Fixes #23364: Restrict script API job results to jobs the user can view

The script list and detail endpoints reported each script's latest job
without applying the requesting user's core.view_job permission, exposing
other users' jobs (including output and logs on the detail view). Both now
report the newest job the user is permitted to view, or none.

Running a script still returns the job just enqueued, regardless of
view_job, so existing run-only clients keep working. It also no longer
risks returning another user's concurrently created job.

* Return null for script detail result when no job is visible

The detail serializer previously fell back to JobSerializer(None).data, a
partial stub missing id, url and status. Return null instead, matching the
list endpoint, and document the change for API clients.

docs/customization/custom-scripts.md

@@ -552,6 +552,9 @@ Optionally `schedule_at` can be passed in the form data with a datetime string t
 !!! note
     Script input submitted through the REST API is validated against the variables declared by the script. Missing required variables or invalid values result in an HTTP 400 response, and undeclared keys are discarded rather than passed to `run()`. Existing API clients that relied on the previous pass-through behavior may need to update their requests. Scripts declaring a `FileVar` must be run via a `multipart/form-data` request, passing `data` as a JSON string alongside the uploaded file.
 
+!!! note "Changed in NetBox v4.7.4"
+    A script's `result` in the REST API reports only jobs the requesting user is permitted to view (`core.view_job`), and is `null` when there are none. The response to running a script always includes the job just created. To let users poll the results of their own script runs, grant them `core.view_job` with the constraint `{"user": "$user"}`.
+
 ### Via the CLI
 
 Scripts can be run on the CLI by invoking the management command:

netbox/extras/api/serializers_/scripts.py

@@ -128,7 +128,7 @@ class ScriptModuleSerializer(ValidatedModelSerializer):
 class ScriptSerializer(ValidatedModelSerializer):
     description = serializers.SerializerMethodField(read_only=True)
     vars = serializers.SerializerMethodField(read_only=True)
-    result = JobSerializer(nested=True, read_only=True)
+    result = serializers.SerializerMethodField(read_only=True)
 
     class Meta:
         model = Script
@@ -155,18 +155,31 @@ class ScriptSerializer(ValidatedModelSerializer):
             return obj.python_class().description
         return None
 
+    @extend_schema_field(JobSerializer(nested=True, allow_null=True))
+    def get_result(self, obj):
+        if job := self._get_latest_job(obj):
+            return JobSerializer(job, nested=True, context=self.context).data
+        return None
+
+    def _get_latest_job(self, obj):
+        """
+        Return the most recent Job for the script which the requesting user is permitted to view.
+        """
+        request = self.context.get('request')
+        # No request (e.g. serialize_for_event()) means no user to check permissions against
+        if request is None:
+            return None
+        return obj.jobs.restrict(request.user, 'view').order_by('-created').first()
+
 
 class ScriptDetailSerializer(ScriptSerializer):
-    result = serializers.SerializerMethodField(read_only=True)
 
-    @extend_schema_field(JobSerializer())
+    @extend_schema_field(JobSerializer(allow_null=True))
     def get_result(self, obj):
-        job = obj.jobs.all().order_by('-created').first()
-        context = {
-            'request': self.context['request']
-        }
-        data = JobSerializer(job, context=context).data
-        return data
+        # A job passed in by the run action takes precedence over the latest visible job
+        if job := self.context.get('job') or self._get_latest_job(obj):
+            return JobSerializer(job, context={'request': self.context.get('request')}).data
+        return None
 
 
 class ScriptInputSerializer(serializers.Serializer):

netbox/extras/api/views.py

@@ -436,7 +436,7 @@ class ScriptViewSet(ListModelMixin, RetrieveModelMixin, BaseViewSet):
             data.pop(k, None)
 
         try:
-            ScriptJob.enqueue(
+            job = ScriptJob.enqueue(
                 instance=script,
                 user=request.user,
                 data=data,
@@ -453,7 +453,8 @@ class ScriptViewSet(ListModelMixin, RetrieveModelMixin, BaseViewSet):
             # 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 the job just enqueued (regardless of view_job permission), never another user's concurrent job
+        serializer = serializers.ScriptDetailSerializer(script, context={'request': request, 'job': job})
         return Response(serializer.data)
 
 

netbox/extras/tests/test_api.py

@@ -2,6 +2,7 @@ import datetime
 import hashlib
 import io
 import json
+import uuid
 from contextlib import contextmanager
 from unittest.mock import MagicMock, PropertyMock, patch
 
@@ -31,7 +32,7 @@ from dcim.models import (
     RearPort,
     Site,
 )
-from extras.api.serializers import EventRuleSerializer
+from extras.api.serializers import EventRuleSerializer, ScriptSerializer
 from extras.choices import *
 from extras.models import *
 from extras.scripts import BooleanVar, IntegerVar, MultiObjectVar, ObjectVar, StringVar
@@ -2623,17 +2624,26 @@ class ScriptRunExecutionTestCase(APITestCase):
                 file_root=ManagedFileRootPathChoices.SCRIPTS,
                 file_path='run_script.py',
             )
-        script = Script.objects.create(
+        cls.script = Script.objects.create(
             module=module,
             name='Test run script',
             is_executable=True,
         )
-        cls.url = reverse('extras-api:script-detail', kwargs={'pk': script.pk})
+        cls.url = reverse('extras-api:script-detail', kwargs={'pk': cls.script.pk})
 
     def setUp(self):
         super().setUp()
         self.add_permissions('extras.run_script')
 
+        # A saved Job for the mocked ScriptJob.enqueue() to return, as run() serializes the enqueued job
+        self.job = Job.objects.create(
+            object_type=ObjectType.objects.get_for_model(Script),
+            object_id=self.script.pk,
+            name='Test run script',
+            job_id=uuid.uuid4(),
+            user=self.user,
+        )
+
         # Monkey-patch the Script model to return our TestScriptClass above, restoring
         # the real property afterwards so later tests aren't left with our stub.
         python_class_patch = patch.object(Script, 'python_class', new=self.TestScriptClass)
@@ -2648,6 +2658,7 @@ class ScriptRunExecutionTestCase(APITestCase):
 
     @patch('extras.jobs.ScriptJob.enqueue')
     def test_run_forwards_commit_value(self, mock_enqueue):
+        mock_enqueue.return_value = self.job
         for commit_value in (True, False):
             with self.subTest(commit=commit_value):
                 mock_enqueue.reset_mock()
@@ -2661,6 +2672,7 @@ class ScriptRunExecutionTestCase(APITestCase):
 
     @patch('extras.jobs.ScriptJob.enqueue')
     def test_run_forwards_notifications_value(self, mock_enqueue):
+        mock_enqueue.return_value = self.job
         # Regression: ScriptForm.clean() overwrites an empty '_notifications' with the
         # field's own initial, so a client-supplied value never reached ScriptJob.enqueue.
         payload = {
@@ -2679,6 +2691,7 @@ class ScriptRunExecutionTestCase(APITestCase):
 
     @patch('extras.jobs.ScriptJob.enqueue')
     def test_run_forwards_schedule_at_and_interval(self, mock_enqueue):
+        mock_enqueue.return_value = self.job
         # Regression: schedule_at/interval were likewise read from the form (always
         # absent there) instead of the validated request
         schedule_at = now() + datetime.timedelta(hours=1)
@@ -2698,6 +2711,7 @@ class ScriptRunExecutionTestCase(APITestCase):
 
     @patch('extras.jobs.ScriptJob.enqueue')
     def test_run_converts_objectvar_and_multiobjectvar_ids(self, mock_enqueue):
+        mock_enqueue.return_value = self.job
         payload = {
             'data': {
                 'site': self.sites[0].pk,
@@ -2718,6 +2732,7 @@ class ScriptRunExecutionTestCase(APITestCase):
 
     @patch('extras.jobs.ScriptJob.enqueue')
     def test_run_backfills_default_for_omitted_required_var(self, mock_enqueue):
+        mock_enqueue.return_value = self.job
         # Regression: required vars declaring `default=` were not back-filled before
         # binding the form on the API path (unlike the UI path), so they 400'd even
         # though the client legitimately omitted them.
@@ -2750,6 +2765,7 @@ class ScriptRunExecutionTestCase(APITestCase):
 
     @patch('extras.jobs.ScriptJob.enqueue')
     def test_run_ignores_undeclared_keys(self, mock_enqueue):
+        mock_enqueue.return_value = self.job
         # Binding 'data' to the script's form means keys which don't correspond to a declared
         # variable are dropped rather than forwarded to run(). This is the contract documented
         # under "Running Custom Scripts > Via the API"; pin it so it can't regress silently.
@@ -2780,6 +2796,7 @@ class ScriptRunExecutionTestCase(APITestCase):
 
     @patch('extras.jobs.ScriptJob.enqueue')
     def test_run_resolves_booleanvar_default_and_explicit_values(self, mock_enqueue):
+        mock_enqueue.return_value = self.job
         # BooleanVar renders as a checkbox, and CheckboxInput reads a missing key as False.
         # An omitted BooleanVar must therefore pick up its declared default, while an
         # explicitly supplied False must not be overwritten by that default.
@@ -2798,6 +2815,134 @@ class ScriptRunExecutionTestCase(APITestCase):
                 self.assertIs(mock_enqueue.call_args.kwargs['data']['flag'], expected)
 
 
+@tag('regression')  # Ref: #23364
+class ScriptJobRestrictionTestCase(APITestCase):
+    """
+    A script's result reports the latest job the requesting user is permitted to view (see #23364).
+    """
+
+    class TestScriptClass(PythonClass):
+        class Meta:
+            name = 'Restricted script'
+
+        def run(self, data, commit=True):
+            return 'ok'
+
+    @classmethod
+    def setUpTestData(cls):
+        with patch.object(ScriptModule, 'sync_classes'):
+            module = ScriptModule.objects.create(
+                file_root=ManagedFileRootPathChoices.SCRIPTS,
+                file_path='restricted_script.py',
+            )
+        cls.script = Script.objects.create(
+            module=module,
+            name='Restricted script',
+            is_executable=True,
+        )
+        cls.list_url = reverse('extras-api:script-list')
+        cls.detail_url = reverse('extras-api:script-detail', kwargs={'pk': cls.script.pk})
+
+        cls.other_job = Job.objects.create(
+            object_type=ObjectType.objects.get_for_model(Script),
+            object_id=cls.script.pk,
+            name='Other user job',
+            job_id=uuid.uuid4(),
+            user=User.objects.create_user(username='script-job-other-user'),
+        )
+
+    def setUp(self):
+        super().setUp()
+        self.add_permissions('extras.view_script')
+
+        # The requesting user's own job predates the other user's, so a restricted lookup must fall
+        # through to it rather than report the newest job or none at all
+        self.own_job = self._create_job('Own job', self.other_job.created - datetime.timedelta(hours=1))
+
+        python_class_patch = patch.object(Script, 'python_class', new=self.TestScriptClass)
+        python_class_patch.start()
+        self.addCleanup(python_class_patch.stop)
+
+    def _create_job(self, name, created):
+        job = Job.objects.create(
+            object_type=ObjectType.objects.get_for_model(Script),
+            object_id=self.script.pk,
+            name=name,
+            job_id=uuid.uuid4(),
+            user=self.user,
+        )
+        Job.objects.filter(pk=job.pk).update(created=created)
+        return job
+
+    def _grant_view_job(self, constraints=None):
+        obj_perm = ObjectPermission(name='View jobs', actions=['view'], constraints=constraints)
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(Job))
+
+    @staticmethod
+    def _job_path(job):
+        # The list representation of a job is brief, identifying it only by URL
+        return reverse('core-api:job-detail', kwargs={'pk': job.pk})
+
+    def _get_results(self):
+        list_response = self.client.get(self.list_url, **self.header)
+        self.assertHttpStatus(list_response, status.HTTP_200_OK)
+        detail_response = self.client.get(self.detail_url, **self.header)
+        self.assertHttpStatus(detail_response, status.HTTP_200_OK)
+        return list_response.data['results'][0]['result'], detail_response.data['result']
+
+    def test_unconstrained_view_job(self):
+        self._grant_view_job()
+
+        list_result, detail_result = self._get_results()
+
+        self.assertTrue(list_result['url'].endswith(self._job_path(self.other_job)))
+        self.assertEqual(detail_result['id'], self.other_job.pk)
+
+    def test_constrained_view_job(self):
+        self._grant_view_job(constraints={'user': '$user'})
+
+        list_result, detail_result = self._get_results()
+
+        self.assertTrue(list_result['url'].endswith(self._job_path(self.own_job)))
+        self.assertEqual(detail_result['id'], self.own_job.pk)
+
+    def test_without_view_job(self):
+        list_result, detail_result = self._get_results()
+
+        self.assertIsNone(list_result)
+        self.assertIsNone(detail_result)
+
+    def test_serialize_without_request(self):
+        # e.g. serialize_for_event(), which has no user to check permissions against
+        data = ScriptSerializer(self.script, context={'request': None}).data
+
+        self.assertIsNone(data['result'])
+
+    @patch('extras.api.views.any_workers_for_queue', return_value=True)
+    @patch('extras.jobs.ScriptJob.enqueue')
+    def test_run_returns_enqueued_job(self, mock_enqueue, mock_workers):
+        """
+        Running a script returns the job just enqueued, even without view_job permission and when another
+        user's job is newer. Polling the script thereafter reports no result without view_job permission.
+        """
+        self.add_permissions('extras.run_script')
+        mock_enqueue.return_value = self._create_job(
+            'Enqueued job', self.other_job.created - datetime.timedelta(minutes=1)
+        )
+
+        response = self.client.post(self.detail_url, {'data': {}, 'commit': True}, format='json', **self.header)
+
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+        self.assertEqual(response.data['result']['id'], mock_enqueue.return_value.pk)
+
+        response = self.client.get(self.detail_url, **self.header)
+
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+        self.assertIsNone(response.data['result'])
+
+
 class CreatedUpdatedFilterTestCase(APITestCase):
 
     @classmethod