22745 - Enforce object permissions on Script REST API write operations (#22777)

netbox/extras/api/routers.py

@@ -0,0 +1,34 @@
+from rest_framework.routers import Route
+
+from netbox.api.routers import NetBoxRouter
+
+from .views import ScriptViewSet
+
+__all__ = (
+    'ScriptRouter',
+)
+
+
+class ScriptRouter(NetBoxRouter):
+    """
+    Extend NetBoxRouter to map POST on the script detail route to ScriptViewSet.run(). DRF's detail route
+    maps only the standard CRUD methods; absent this, run() must be declared as a raw post() method, which
+    binds to every route of the ViewSet and is invisible to per-action permissions & schema generation.
+    """
+    def get_routes(self, viewset):
+        if not issubclass(viewset, ScriptViewSet):
+            return super().get_routes(viewset)
+
+        # Extend the detail route template. Applied before super() expands the templates so that any
+        # @action routes are untouched; _replace() avoids mutating the templates shared by all routers.
+        routes = self.routes
+        self.routes = [
+            route._replace(mapping={**route.mapping, 'post': 'run'})
+            if isinstance(route, Route) and route.detail else route
+            for route in routes
+        ]
+
+        try:
+            return super().get_routes(viewset)
+        finally:
+            self.routes = routes

netbox/extras/api/urls.py

@@ -1,10 +1,9 @@
 from django.urls import include, path
 
-from netbox.api.routers import NetBoxRouter
-
 from . import views
+from .routers import ScriptRouter
 
-router = NetBoxRouter()
+router = ScriptRouter()
 router.APIRootView = views.ExtrasRootView
 
 router.register('event-rules', views.EventRuleViewSet)

netbox/extras/api/views.py

@@ -1,16 +1,15 @@
 from django.http import Http404
 from django.shortcuts import get_object_or_404
 from django.utils.translation import gettext_lazy as _
-from drf_spectacular.utils import OpenApiResponse, OpenApiTypes, extend_schema, extend_schema_view
+from drf_spectacular.utils import OpenApiResponse, OpenApiTypes, extend_schema
 from rest_framework import status
 from rest_framework.decorators import action
-from rest_framework.exceptions import PermissionDenied
+from rest_framework.exceptions import PermissionDenied, ValidationError
 from rest_framework.generics import RetrieveUpdateDestroyAPIView
 from rest_framework.mixins import CreateModelMixin, ListModelMixin, RetrieveModelMixin, UpdateModelMixin
 from rest_framework.renderers import JSONRenderer
 from rest_framework.response import Response
 from rest_framework.routers import APIRootView
-from rest_framework.viewsets import ModelViewSet
 
 from core.choices import ManagedFileRootPathChoices
 from extras import filtersets
@@ -22,6 +21,7 @@ from netbox.api.metadata import ContentTypeMetadata
 from netbox.api.renderers import TextRenderer
 from netbox.api.viewsets import BaseViewSet, NetBoxModelViewSet
 from netbox.api.viewsets.mixins import ObjectValidationMixin
+from users.models import Token
 from utilities.exceptions import RQWorkerNotRunningException
 from utilities.request import copy_safe_request
 from utilities.rqworker import any_workers_for_queue
@@ -307,30 +307,44 @@ class ScriptModuleViewSet(ObjectValidationMixin, CreateModelMixin, UpdateModelMi
         return obj
 
 
-@extend_schema_view(
-    update=extend_schema(request=serializers.ScriptInputSerializer),
-    partial_update=extend_schema(request=serializers.ScriptInputSerializer),
-)
-class ScriptViewSet(ModelViewSet):
+class ScriptViewSet(ListModelMixin, RetrieveModelMixin, BaseViewSet):
+    # Individual scripts are created, modified, and deleted through their module (see ScriptModuleViewSet),
+    # so the standard write actions are intentionally omitted here. Only listing/retrieving a script (GET)
+    # and running one (POST to the detail route) are supported.
     permission_classes = [IsAuthenticatedOrLoginNotRequired]
     queryset = Script.objects.all()
     serializer_class = serializers.ScriptSerializer
     filterset_class = filtersets.ScriptFilterSet
 
-    _ignore_model_permissions = True
     lookup_value_regex = '[^/]+'  # Allow dots
 
-    def initial(self, request, *args, **kwargs):
-        super().initial(request, *args, **kwargs)
+    def get_serializer(self, *args, **kwargs):
+        # A POST to the detail route runs the script, taking ScriptInputSerializer as its request body.
+        # (This is keyed on the request method rather than on self.action, which is unset when generating
+        # OPTIONS metadata.) ScriptInputSerializer is instantiated directly rather than via BaseViewSet,
+        # which would pass it the fields/omit kwargs supported only by BaseModelSerializer.
+        if getattr(self.request, 'method', None) == 'POST':
+            kwargs.setdefault('context', self.get_serializer_context())
+            return serializers.ScriptInputSerializer(*args, **kwargs)
+        return super().get_serializer(*args, **kwargs)
 
-        # Restrict the view's QuerySet to allow only the permitted objects
-        if request.user.is_authenticated:
-            action = 'run' if request.method == 'POST' else 'view'
-            self.queryset = self.queryset.restrict(request.user, action)
+    def get_serializer_context(self):
+        context = super().get_serializer_context()
+
+        # ScriptInputSerializer resolves its field defaults and validates scheduling against the script
+        # being run (set by run() below).
+        context['script'] = getattr(self, 'script', None)
+
+        return context
 
     def _get_script(self, pk):
-        # If pk is numeric, retrieve script by ID
-        if pk.isnumeric():
+        # Retrieve the script by ID if the PK is all decimal digits. (isdecimal() rather than isnumeric(),
+        # as the latter also matches characters which cannot be cast to an integer.)
+        if pk.isdecimal():
+            try:
+                pk = int(pk)
+            except ValueError:
+                raise Http404
             return get_object_or_404(self.queryset, pk=pk)
 
         # Default to retrieval by module & name
@@ -341,26 +355,49 @@ class ScriptViewSet(ModelViewSet):
 
         return get_object_or_404(self.queryset, module__file_path=f'{module_name}.py', name=script_name)
 
-    def retrieve(self, request, pk):
+    def retrieve(self, request, pk, **kwargs):
         script = self._get_script(pk)
         serializer = serializers.ScriptDetailSerializer(script, context={'request': request})
 
         return Response(serializer.data)
 
-    def post(self, request, pk):
+    @extend_schema(
+        operation_id='extras_scripts_run',
+        request=serializers.ScriptInputSerializer,
+        responses={
+            200: OpenApiResponse(
+                response=serializers.ScriptDetailSerializer,
+                description=_("The script has been enqueued for execution."),
+            ),
+        },
+    )
+    def run(self, request, pk, **kwargs):
         """
         Run a Script identified by its numeric PK or module & name and return the pending Job as the result
         """
+        # Bound to POST on the detail route by ScriptRouter
 
-        script = self._get_script(pk)
+        # Reject read-only tokens before resolving the script, so that an insufficient token is always
+        # reported as such. (Not via TokenWritePermission, which permits token auth only.)
+        if isinstance(request.auth, Token) and not request.auth.write_enabled:
+            raise PermissionDenied(_("This token does not permit write operations (running a script)."))
+
+        # An unauthenticated user can never run a script; report that explicitly, as restrict() below would
+        # match no scripts and yield a misleading 404.
+        if not request.user.is_authenticated:
+            raise PermissionDenied(_("This user does not have permission to run this script."))
+
+        # Running a script is a 'run' operation (not the 'add' that BaseViewSet maps to POST), so restrict
+        # the QuerySet on 'run' before resolving the script. A script the user cannot run yields a 404.
+        self.queryset = self.queryset.model.objects.restrict(request.user, 'run')
+        self.script = script = self._get_script(pk)
 
-        if not request.user.has_perm('extras.run_script', obj=script):
-            raise PermissionDenied("This user does not have permission to run this script.")
+        # A script whose Python class cannot be resolved (e.g. its module has been modified or the script has
+        # been deleted, retaining the record for its jobs) cannot be run
+        if not script.is_executable or script.python_class is None:
+            raise ValidationError(_("This script is not currently executable."))
 
-        input_serializer = serializers.ScriptInputSerializer(
-            data=request.data,
-            context={'script': script}
-        )
+        input_serializer = self.get_serializer(data=request.data)
 
         # Check that at least one RQ worker is running
         if not any_workers_for_queue('default'):

netbox/extras/tests/test_api.py

@@ -8,13 +8,14 @@ from unittest.mock import MagicMock, patch
 from django.contrib.contenttypes.models import ContentType
 from django.core.files.uploadedfile import SimpleUploadedFile
 from django.db import IntegrityError
+from django.test import override_settings
 from django.urls import reverse
 from django.utils.timezone import make_aware, now
 from rest_framework import status
 
 from core.choices import ManagedFileRootPathChoices
 from core.events import *
-from core.models import DataFile, DataSource, ObjectType
+from core.models import DataFile, DataSource, Job, ObjectType
 from dcim.models import Device, DeviceRole, DeviceType, Location, Manufacturer, Rack, RackRole, Site
 from extras.choices import *
 from extras.models import *
@@ -23,7 +24,7 @@ from extras.scripts import Script as PythonClass
 from users.constants import TOKEN_PREFIX
 from users.models import Group, ObjectPermission, Token, User
 from utilities.tables import get_table_for_model
-from utilities.testing import APITestCase, APIViewTestCases
+from utilities.testing import APITestCase, APIViewTestCases, disable_warnings
 
 
 class AppTestCase(APITestCase):
@@ -1403,6 +1404,34 @@ class ScriptTestCase(APITestCase):
         self.assertEqual(response.data['vars']['var2'], 'IntegerVar')
         self.assertEqual(response.data['vars']['var3'], 'BooleanVar')
 
+    def test_list_scripts(self):
+        """
+        The list route is served by BaseViewSet, which resolves the QuerySet's prefetches & annotations (and
+        any fields/omit request parameters) from the serializer.
+        """
+        url = reverse('extras-api:script-list')
+
+        response = self.client.get(url, **self.header)
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+        self.assertEqual(response.data['count'], 1)
+        self.assertEqual(response.data['results'][0]['name'], self.TestScriptClass.Meta.name)
+
+        response = self.client.get(f'{url}?fields=id,name', **self.header)
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+        self.assertEqual(sorted(response.data['results'][0]), ['id', 'name'])
+
+    def test_get_script_by_module_and_name(self):
+        """
+        A script may also be identified by its module & name, e.g. /api/extras/scripts/example.MyReport/.
+        """
+        script = Script.objects.first()
+        url = reverse('extras-api:script-detail', kwargs={'pk': f'script.{script.name}'})
+
+        response = self.client.get(url, **self.header)
+
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+        self.assertEqual(response.data['id'], script.pk)
+
     def test_schedule_script_past_time_rejected(self):
         """
         Scheduling with past schedule_at should fail.
@@ -1473,6 +1502,203 @@ class ScriptTestCase(APITestCase):
             # Restore the original setting for other tests
             self.TestScriptClass.Meta.scheduling_enabled = original
 
+    def test_run_script_without_permission(self):
+        """
+        A user permitted to view a script but not to run it must not be able to enqueue it. (The script is
+        excluded from the restricted QuerySet, so the request yields a 404.)
+        """
+        payload = {'data': {'var1': 'hello', 'var2': 1, 'var3': False}, 'commit': True}
+
+        # setUp() grants only extras.view_script
+        with disable_warnings('django.request'):
+            response = self.client.post(self.url, payload, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_404_NOT_FOUND)
+        self.assertFalse(Job.objects.exists())
+
+        # Granting the run permission permits the same request
+        self.add_permissions('extras.run_script')
+        response = self.client.post(self.url, payload, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+        self.assertTrue(Job.objects.exists())
+
+    @override_settings(LOGIN_REQUIRED=False, EXEMPT_VIEW_PERMISSIONS=['*'])
+    def test_run_script_anonymous(self):
+        """
+        An unauthenticated user must be told that running a script is not permitted, rather than that the
+        script does not exist.
+        """
+        payload = {'data': {'var1': 'hello', 'var2': 1, 'var3': False}, 'commit': True}
+
+        with disable_warnings('django.request'):
+            response = self.client.post(self.url, payload, format='json')
+        self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)
+        self.assertFalse(Job.objects.exists())
+
+    def test_run_script_read_only_token(self):
+        """
+        Running a script is a write operation and must be rejected for a read-only token.
+        """
+        self.add_permissions('extras.run_script')
+        payload = {'data': {'var1': 'hello', 'var2': 1, 'var3': False}, 'commit': True}
+
+        # A write-disabled token should be rejected
+        ro_token = Token.objects.create(version=2, user=self.user, write_enabled=False)
+        ro_header = {'HTTP_AUTHORIZATION': f'Bearer {TOKEN_PREFIX}{ro_token.key}.{ro_token.token}'}
+        response = self.client.post(self.url, payload, format='json', **ro_header)
+        self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)
+
+        # The default (write-enabled) token should succeed
+        response = self.client.post(self.url, payload, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+
+    def test_run_script_read_only_token_without_permission(self):
+        """
+        A read-only token is rejected before the script is resolved, so an insufficient token is reported as
+        such regardless of the user's permission to run the script.
+        """
+        payload = {'data': {'var1': 'hello', 'var2': 1, 'var3': False}, 'commit': True}
+
+        # setUp() grants only extras.view_script
+        ro_token = Token.objects.create(version=2, user=self.user, write_enabled=False)
+        ro_header = {'HTTP_AUTHORIZATION': f'Bearer {TOKEN_PREFIX}{ro_token.key}.{ro_token.token}'}
+        response = self.client.post(self.url, payload, format='json', **ro_header)
+        self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)
+
+    def test_run_script_not_executable(self):
+        """
+        A script whose Python class cannot be resolved must be rejected, not raise an exception.
+        """
+        self.add_permissions('extras.run_script')
+        payload = {'data': {'var1': 'hello', 'var2': 1, 'var3': False}, 'commit': True}
+
+        # Simulate a script whose class can no longer be found in its module
+        class_patch = patch.object(Script, 'python_class', None)
+        class_patch.start()
+        self.addCleanup(class_patch.stop)
+
+        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_by_module_and_name(self):
+        """
+        A script identified by its module & name (rather than by its PK) must also be runnable.
+        """
+        self.add_permissions('extras.run_script')
+        payload = {'data': {'var1': 'hello', 'var2': 1, 'var3': False}, 'commit': True}
+        script = Script.objects.first()
+        url = reverse('extras-api:script-detail', kwargs={'pk': f'script.{script.name}'})
+
+        response = self.client.post(url, payload, format='json', **self.header)
+
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+        self.assertEqual(response.data['id'], script.pk)
+        self.assertTrue(Job.objects.exists())
+
+    def test_run_script_format_suffix(self):
+        """
+        The format-suffix variants of the detail route (e.g. /1.json) must dispatch to run().
+        """
+        self.add_permissions('extras.run_script')
+        payload = {'data': {'var1': 'hello', 'var2': 1, 'var3': False}, 'commit': True}
+        script = Script.objects.first()
+        lookups = (script.pk, f'script.{script.name}')
+
+        for lookup in lookups:
+            with self.subTest(lookup=lookup):
+                url = reverse('extras-api:script-detail', kwargs={'pk': lookup, 'format': 'json'})
+
+                response = self.client.post(url, payload, format='json', **self.header)
+
+                self.assertHttpStatus(response, status.HTTP_200_OK)
+                self.assertEqual(response.data['id'], script.pk)
+
+        self.assertEqual(Job.objects.count(), len(lookups))
+
+    def test_modify_script_methods_disabled(self):
+        """
+        Individual scripts are created, modified, and deleted through their module, so PUT/PATCH/DELETE on
+        the script endpoint are not supported (even for a user holding the corresponding permissions).
+        """
+        self.add_permissions('extras.change_script', 'extras.delete_script')
+        script = Script.objects.first()
+
+        for method in ('put', 'patch', 'delete'):
+            with self.subTest(method=method):
+                with disable_warnings('django.request'):
+                    response = getattr(self.client, method)(self.url, {}, format='json', **self.header)
+                self.assertHttpStatus(response, status.HTTP_405_METHOD_NOT_ALLOWED)
+
+        # The script must remain untouched
+        self.assertTrue(Script.objects.filter(pk=script.pk).exists())
+
+    def test_create_script_disabled(self):
+        """
+        Scripts cannot be created via the API: POST is mapped only on the detail route (to run a script),
+        and must be neither permitted nor advertised on the list route.
+        """
+        self.add_permissions('extras.add_script')
+        list_url = reverse('extras-api:script-list')
+
+        with disable_warnings('django.request'):
+            response = self.client.post(list_url, {}, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_405_METHOD_NOT_ALLOWED)
+
+        # OPTIONS must not advertise a create action for the list route
+        response = self.client.options(list_url, **self.header)
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+        self.assertNotIn('POST', response.data.get('actions', {}))
+
+    def test_options_detail_route(self):
+        """
+        POST on the detail route runs a script, so its OPTIONS metadata must describe the run input
+        rather than the Script model's own fields.
+        """
+        self.add_permissions('extras.run_script')
+
+        response = self.client.options(self.url, **self.header)
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+        post_fields = response.data['actions']['POST']
+        self.assertIn('data', post_fields)
+        self.assertIn('commit', post_fields)
+        self.assertNotIn('module', post_fields)
+        self.assertNotIn('name', post_fields)
+
+    def test_options_detail_route_dynamic_fields(self):
+        """
+        The run input serializer does not support the fields/omit query parameters, but their presence must
+        not break the generation of OPTIONS metadata.
+        """
+        self.add_permissions('extras.run_script')
+
+        for query in ('fields=id', 'omit=id'):
+            with self.subTest(query=query):
+                response = self.client.options(f'{self.url}?{query}', **self.header)
+
+                self.assertHttpStatus(response, status.HTTP_200_OK)
+                self.assertIn('data', response.data['actions']['POST'])
+
+    def test_unsupported_method(self):
+        """
+        A request using an HTTP method which maps to no action must be rejected with a 405.
+        """
+        with disable_warnings('django.request'):
+            response = self.client.trace(self.url, **self.header)
+        self.assertHttpStatus(response, status.HTTP_405_METHOD_NOT_ALLOWED)
+
+    def test_get_script_invalid_pk(self):
+        """
+        A PK which cannot be cast to an integer must yield a 404, not a server error. This covers numeric (but
+        non-decimal) characters, as well as a decimal value too long for Python to convert.
+        """
+        for pk in ('½', '1' * 5000):
+            with self.subTest(pk=pk[:10]):
+                url = reverse('extras-api:script-detail', kwargs={'pk': pk})
+
+                with disable_warnings('django.request'):
+                    response = self.client.get(url, **self.header)
+                self.assertHttpStatus(response, status.HTTP_404_NOT_FOUND)
+
 
 class CreatedUpdatedFilterTestCase(APITestCase):
 

netbox/extras/tests/test_api_routers.py

@@ -0,0 +1,52 @@
+from django.test import TestCase
+
+from extras.api.routers import ScriptRouter
+from extras.api.views import CustomFieldChoiceSetViewSet, ScriptViewSet, WebhookViewSet
+
+
+class ScriptRouterTestCase(TestCase):
+    """
+    Verify the routes generated by ScriptRouter.
+    """
+    @staticmethod
+    def get_actions(viewset):
+        """
+        Return a mapping of route name to the HTTP methods bound on it for the given ViewSet.
+        """
+        router = ScriptRouter()
+        router.register('dummy', viewset, basename='dummy')
+
+        return {
+            url.name: url.callback.actions
+            for url in router.urls if hasattr(url.callback, 'actions')
+        }
+
+    def test_script_routes(self):
+        actions = self.get_actions(ScriptViewSet)
+
+        # POST on the detail route runs the script; the list route accepts only GET
+        self.assertEqual(actions['dummy-detail'], {'get': 'retrieve', 'post': 'run'})
+        self.assertEqual(actions['dummy-list'], {'get': 'list'})
+
+    def test_script_viewset_subclass(self):
+        # A subclass of ScriptViewSet (e.g. as registered by a plugin) gets the same route mapping
+        class MyScriptViewSet(ScriptViewSet):
+            pass
+
+        actions = self.get_actions(MyScriptViewSet)
+
+        self.assertEqual(actions['dummy-detail'], {'get': 'retrieve', 'post': 'run'})
+        self.assertEqual(actions['dummy-list'], {'get': 'list'})
+
+    def test_other_viewsets_unaffected(self):
+        # Standard ViewSets keep the stock detail route mapping
+        self.assertNotIn('post', self.get_actions(WebhookViewSet)['dummy-detail'])
+
+        # Routes generated for @action methods are untouched
+        self.assertEqual(self.get_actions(CustomFieldChoiceSetViewSet)['dummy-choices'], {'get': 'choices'})
+
+    def test_route_templates_not_mutated(self):
+        router = ScriptRouter()
+        router.get_routes(ScriptViewSet)
+
+        self.assertNotIn('post', router.routes[2].mapping)

netbox/netbox/api/viewsets/__init__.py

@@ -7,6 +7,7 @@ from django.db.models import ProtectedError, RestrictedError
 from django_pglocks import advisory_lock
 from rest_framework import mixins as drf_mixins
 from rest_framework import status
+from rest_framework.exceptions import MethodNotAllowed
 from rest_framework.response import Response
 from rest_framework.viewsets import GenericViewSet
 
@@ -87,6 +88,13 @@ class BaseViewSet(GenericViewSet):
     def initial(self, request, *args, **kwargs):
         super().initial(request, *args, **kwargs)
 
+        # Reject any method for which no action has been declared, rather than proceeding against an
+        # unrestricted QuerySet. (A method mapped to None, e.g. OPTIONS, is permitted: it needs no
+        # restriction.) This is the same 405 DRF would return when resolving the handler for an unmapped
+        # method, but it also covers a handler bound to such a method (e.g. @action(methods=['trace'])).
+        if request.method not in HTTP_ACTIONS:
+            raise MethodNotAllowed(request.method)
+
         # Restrict the view's QuerySet to allow only the permitted objects
         if request.user.is_authenticated:
             if action := HTTP_ACTIONS[request.method]: