From a94878aa08cf425866f56850e5d35996bf292cc3 Mon Sep 17 00:00:00 2001 From: Arthur Hanson Date: Tue, 11 Aug 2026 05:09:13 -0700 Subject: [PATCH] 22745 - Enforce object permissions on Script REST API write operations (#22777) --- netbox/extras/api/routers.py | 34 ++++ netbox/extras/api/urls.py | 5 +- netbox/extras/api/views.py | 89 ++++++--- netbox/extras/tests/test_api.py | 230 +++++++++++++++++++++++- netbox/extras/tests/test_api_routers.py | 52 ++++++ netbox/netbox/api/viewsets/__init__.py | 8 + 6 files changed, 387 insertions(+), 31 deletions(-) create mode 100644 netbox/extras/api/routers.py create mode 100644 netbox/extras/tests/test_api_routers.py diff --git a/netbox/extras/api/routers.py b/netbox/extras/api/routers.py new file mode 100644 index 000000000..e27797d7b --- /dev/null +++ b/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 diff --git a/netbox/extras/api/urls.py b/netbox/extras/api/urls.py index cd1a9f683..dcc359c7a 100644 --- a/netbox/extras/api/urls.py +++ b/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) diff --git a/netbox/extras/api/views.py b/netbox/extras/api/views.py index d7c21c86b..12827b938 100644 --- a/netbox/extras/api/views.py +++ b/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).")) - if not request.user.has_perm('extras.run_script', obj=script): - raise PermissionDenied("This user does not have permission to run this 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.")) - input_serializer = serializers.ScriptInputSerializer( - data=request.data, - context={'script': 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) + + # 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 = self.get_serializer(data=request.data) # Check that at least one RQ worker is running if not any_workers_for_queue('default'): diff --git a/netbox/extras/tests/test_api.py b/netbox/extras/tests/test_api.py index 89af94be6..cc94bd09c 100644 --- a/netbox/extras/tests/test_api.py +++ b/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): diff --git a/netbox/extras/tests/test_api_routers.py b/netbox/extras/tests/test_api_routers.py new file mode 100644 index 000000000..25c791511 --- /dev/null +++ b/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) diff --git a/netbox/netbox/api/viewsets/__init__.py b/netbox/netbox/api/viewsets/__init__.py index 95b2351d3..36f330f54 100644 --- a/netbox/netbox/api/viewsets/__init__.py +++ b/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]: