feat(extras): Allow updating uploaded Script Modules via API

Add PUT/PATCH support to ScriptModuleViewSet for replacing Script Module
content in place. Modules can be addressed by numeric ID or file name,
and the uploaded file name must match the existing file path.
The module's scripts are re-synchronized from the new content after
successful update.

Fixes #22544

docs/customization/custom-scripts.md

@@ -443,6 +443,18 @@ curl -X POST \
 http://netbox/api/extras/scripts/upload/
 ```
 
+### Updating an Uploaded Script
+
+An existing script module can be replaced in place by sending a `multipart/form-data` PUT or PATCH request to the module's detail URL. The module may be identified by its numeric ID or by its file name. The uploaded file name must match the existing module's file path, and the caller must have the `extras.change_scriptmodule` and `core.change_managedfile` permissions. The module's scripts are re-synchronized from the new content.
+
+```no-highlight
+curl -X PUT \
+-H "Authorization: Bearer $TOKEN" \
+-H "Accept: application/json; indent=4" \
+-F "file=@/path/to/myscript.py" \
+http://netbox/api/extras/scripts/upload/myscript.py/
+```
+
 ## Running Custom Scripts
 
 !!! note

netbox/extras/api/serializers_/scripts.py

@@ -1,7 +1,7 @@
 import logging
 
 from django.core.files.storage import storages
-from django.db import IntegrityError
+from django.db import IntegrityError, router, transaction
 from django.utils.translation import gettext_lazy as _
 from drf_spectacular.utils import extend_schema_field
 from rest_framework import serializers
@@ -38,11 +38,21 @@ class ScriptModuleSerializer(ValidatedModelSerializer):
         file = data.pop('file', None)
         data['file_root'] = ManagedFileRootPathChoices.SCRIPTS
 
-        # Reject duplicates before writing to storage so a failed upload can't touch the existing file
-        if file is not None and ScriptModule.objects.filter(
-            file_root=ManagedFileRootPathChoices.SCRIPTS, file_path=file.name
-        ).exists():
-            raise serializers.ValidationError(_("A script module with this file name already exists."))
+        if self.instance is None:
+            # Reject duplicates before writing to storage so a failed upload can't touch the existing file
+            if file is not None and ScriptModule.objects.filter(
+                file_root=ManagedFileRootPathChoices.SCRIPTS, file_path=file.name
+            ).exists():
+                raise serializers.ValidationError(_("A script module with this file name already exists."))
+        elif file is None:
+            # Replacing a module's content requires a file upload, even for a partial update
+            raise serializers.ValidationError({'file': _("This field is required.")})
+        elif file.name != self.instance.file_path:
+            raise serializers.ValidationError({
+                'file': _(
+                    "The uploaded file name must match the existing file path ({path})."
+                ).format(path=self.instance.file_path)
+            })
 
         data = super().validate(data)
         data.pop('file_root', None)
@@ -85,6 +95,35 @@ class ScriptModuleSerializer(ValidatedModelSerializer):
                 except Exception:
                     logger.warning(f"Failed to delete orphaned script file '{file_path}' from storage.")
 
+    def update(self, instance, validated_data):
+        file = validated_data.pop('file')
+        storage = storages.create_storage(storages.backends["scripts"])
+
+        # Overwrite the existing file in place, keeping file_path stable
+        file.seek(0)
+        saved_path = storage.save(instance.file_path, file)
+        if saved_path != instance.file_path:
+            # The backend saved under an alternate name instead of overwriting; drop the orphan and reject
+            try:
+                storage.delete(saved_path)
+            except Exception:
+                logger.warning(f"Failed to delete orphaned script file '{saved_path}' from storage.")
+            raise serializers.ValidationError({
+                'file': _(
+                    "The scripts storage backend did not overwrite the existing file. Ensure the "
+                    "backend is configured to allow overwrites."
+                )
+            })
+
+        # Discard any cached class discovery so save() re-syncs from the new content
+        instance.__dict__.pop('module_scripts', None)
+        instance.last_updated = local_now()
+        # Keep Script row sync all-or-nothing; the storage write above cannot join the transaction
+        with transaction.atomic(using=router.db_for_write(ScriptModule)):
+            instance.save()
+
+        return instance
+
 
 class ScriptSerializer(ValidatedModelSerializer):
     description = serializers.SerializerMethodField(read_only=True)

netbox/extras/api/views.py

@@ -6,12 +6,13 @@ from rest_framework import status
 from rest_framework.decorators import action
 from rest_framework.exceptions import PermissionDenied
 from rest_framework.generics import RetrieveUpdateDestroyAPIView
-from rest_framework.mixins import CreateModelMixin, ListModelMixin, RetrieveModelMixin
+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
 from extras.jobs import ScriptJob
 from extras.models import *
@@ -282,9 +283,28 @@ class ConfigTemplateViewSet(SyncedDataMixin, ConfigTemplateRenderMixin, NetBoxMo
 # Scripts
 #
 
-class ScriptModuleViewSet(ObjectValidationMixin, CreateModelMixin, BaseViewSet):
-    queryset = ScriptModule.objects.all()
+class ScriptModuleViewSet(ObjectValidationMixin, CreateModelMixin, UpdateModelMixin, BaseViewSet):
+    queryset = ScriptModule.objects.filter(file_root=ManagedFileRootPathChoices.SCRIPTS)
     serializer_class = serializers.ScriptModuleSerializer
+    lookup_value_regex = '[^/]+'  # Allow dots
+
+    def get_object(self):
+        """
+        Retrieve a ScriptModule by numeric ID or by file name (e.g. my_script.py).
+        """
+        queryset = self.filter_queryset(self.get_queryset())
+        lookup = self.kwargs.get(self.lookup_url_kwarg or self.lookup_field, '')
+
+        # Support lookup by numeric PK or by file_path. Treat all-decimal values as PKs
+        # to preserve normal detail-route behavior; otherwise resolve the value as a
+        # script module filename, e.g. "myscript.py".
+        if lookup.isdecimal():
+            obj = get_object_or_404(queryset, pk=int(lookup))
+        else:
+            obj = get_object_or_404(queryset, file_path=lookup)
+
+        self.check_object_permissions(self.request, obj)
+        return obj
 
 
 @extend_schema_view(

netbox/extras/tests/test_api.py

@@ -2,10 +2,12 @@ import datetime
 import hashlib
 import io
 import json
+from contextlib import contextmanager
 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.urls import reverse
 from django.utils.timezone import make_aware, now
 from rest_framework import status
@@ -1698,17 +1700,22 @@ class NotificationTestCase(APIViewTestCases.APIViewTestCase):
 
 
 class _InMemoryScriptStorage:
-    """Stateful stand-in for the scripts storage backend; mimics allow_overwrite=True."""
+    """Stateful stand-in for the scripts storage backend; mirrors its allow_overwrite option."""
 
-    def __init__(self):
+    def __init__(self, allow_overwrite=True):
         self.files = {}
+        self.allow_overwrite = allow_overwrite
 
     def save(self, name, content):
+        if not self.allow_overwrite and name in self.files:
+            name = f'{name}.1'
         content.seek(0)
         self.files[name] = content.read()
         return name
 
     def open(self, name, mode='rb'):
+        if name not in self.files:
+            raise FileNotFoundError(name)
         return io.BytesIO(self.files[name])
 
     def delete(self, name):
@@ -1731,6 +1738,18 @@ class ScriptModuleTestCase(APITestCase):
         super().setUp()
         self.url = reverse('extras-api:scriptmodule-list')  # /api/extras/scripts/upload/
 
+    @contextmanager
+    def _patched_script_storage(self, fake_storage):
+        """Patch both storage entry points (serializer writes, module imports) to the given fake."""
+        with (
+            patch('extras.api.serializers_.scripts.storages') as serializer_storages,
+            patch('extras.models.mixins.storages') as module_storages,
+        ):
+            serializer_storages.create_storage.return_value = fake_storage
+            serializer_storages.backends = {'scripts': {}}
+            module_storages.__getitem__.return_value = fake_storage
+            yield
+
     def test_upload_script_module_without_permission(self):
         script_content = b"from extras.scripts import Script\nclass TestScript(Script):\n    pass\n"
         upload_file = SimpleUploadedFile('test_upload.py', script_content, content_type='text/plain')
@@ -1840,3 +1859,354 @@ class ScriptModuleTestCase(APITestCase):
         self.add_permissions('extras.add_scriptmodule', 'core.add_managedfile')
         response = self.client.post(self.url, {}, format='json', **self.header)
         self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+
+    def test_update_script_module(self):
+        """A PATCH with a new file replaces the stored content and re-syncs the module's scripts."""
+        self.add_permissions(
+            'extras.add_scriptmodule', 'core.add_managedfile',
+            'extras.change_scriptmodule', 'core.change_managedfile',
+        )
+        original_content = (
+            b"from extras.scripts import Script\n\n\n"
+            b"class ProbeScript(Script):\n    def run(self, data, commit):\n        return 'v1'\n"
+        )
+        updated_content = (
+            b"from extras.scripts import Script\n\n\n"
+            b"class ProbeScriptV2(Script):\n    def run(self, data, commit):\n        return 'v2'\n"
+        )
+        fake_storage = _InMemoryScriptStorage()
+
+        with self._patched_script_storage(fake_storage):
+            response = self.client.post(
+                self.url,
+                {'file': SimpleUploadedFile('zz_update.py', original_content, content_type='text/plain')},
+                format='multipart',
+                **self.header,
+            )
+            self.assertHttpStatus(response, status.HTTP_201_CREATED)
+            module_id = response.data['id']
+            detail_url = reverse('extras-api:scriptmodule-detail', kwargs={'pk': module_id})
+
+            response = self.client.patch(
+                detail_url,
+                {'file': SimpleUploadedFile('zz_update.py', updated_content, content_type='text/plain')},
+                format='multipart',
+                **self.header,
+            )
+            self.assertHttpStatus(response, status.HTTP_200_OK)
+
+        self.assertEqual(response.data['id'], module_id)
+        self.assertEqual(response.data['file_path'], 'zz_update.py')
+        self.assertIsNotNone(response.data['last_updated'])
+        self.assertEqual(fake_storage.files['zz_update.py'], updated_content)
+        # Script classes are re-synced from the new content
+        self.assertFalse(Script.objects.filter(module_id=module_id, name='ProbeScript').exists())
+        self.assertTrue(Script.objects.filter(module_id=module_id, name='ProbeScriptV2').exists())
+        self.assertEqual(ScriptModule.objects.filter(file_path='zz_update.py').count(), 1)
+
+    def test_update_script_module_without_file_fails(self):
+        """A PATCH without a file upload is rejected and leaves the stored content unchanged."""
+        self.add_permissions(
+            'extras.add_scriptmodule', 'core.add_managedfile',
+            'extras.change_scriptmodule', 'core.change_managedfile',
+        )
+        script_content = b"from extras.scripts import Script\nclass TestScript(Script):\n    pass\n"
+        fake_storage = _InMemoryScriptStorage()
+
+        with self._patched_script_storage(fake_storage):
+            response = self.client.post(
+                self.url,
+                {'file': SimpleUploadedFile('zz_nofile.py', script_content, content_type='text/plain')},
+                format='multipart',
+                **self.header,
+            )
+            self.assertHttpStatus(response, status.HTTP_201_CREATED)
+            detail_url = reverse('extras-api:scriptmodule-detail', kwargs={'pk': response.data['id']})
+
+            response = self.client.patch(detail_url, {}, format='json', **self.header)
+
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertIn('file', response.data)
+        self.assertEqual(fake_storage.files['zz_nofile.py'], script_content)
+
+    def test_update_script_module_without_permission(self):
+        """A PATCH without change permissions returns 403 and leaves the stored content unchanged."""
+        self.add_permissions('extras.add_scriptmodule', 'core.add_managedfile')
+        script_content = b"from extras.scripts import Script\nclass TestScript(Script):\n    pass\n"
+        fake_storage = _InMemoryScriptStorage()
+
+        with self._patched_script_storage(fake_storage):
+            response = self.client.post(
+                self.url,
+                {'file': SimpleUploadedFile('zz_forbidden.py', script_content, content_type='text/plain')},
+                format='multipart',
+                **self.header,
+            )
+            self.assertHttpStatus(response, status.HTTP_201_CREATED)
+            detail_url = reverse('extras-api:scriptmodule-detail', kwargs={'pk': response.data['id']})
+
+            response = self.client.patch(
+                detail_url,
+                {'file': SimpleUploadedFile('zz_forbidden.py', script_content, content_type='text/plain')},
+                format='multipart',
+                **self.header,
+            )
+
+        self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)
+        self.assertEqual(fake_storage.files['zz_forbidden.py'], script_content)
+
+    def test_update_script_module_by_file_name(self):
+        """A module can be updated by addressing it with its file name instead of its numeric ID."""
+        self.add_permissions(
+            'extras.add_scriptmodule', 'core.add_managedfile',
+            'extras.change_scriptmodule', 'core.change_managedfile',
+        )
+        original_content = (
+            b"from extras.scripts import Script\n\n\n"
+            b"class ProbeScript(Script):\n    def run(self, data, commit):\n        return 'v1'\n"
+        )
+        updated_content = original_content.replace(b"'v1'", b"'v2'")
+        fake_storage = _InMemoryScriptStorage()
+
+        with self._patched_script_storage(fake_storage):
+            response = self.client.post(
+                self.url,
+                {'file': SimpleUploadedFile('zz_by_name.py', original_content, content_type='text/plain')},
+                format='multipart',
+                **self.header,
+            )
+            self.assertHttpStatus(response, status.HTTP_201_CREATED)
+            module_id = response.data['id']
+            detail_url = reverse('extras-api:scriptmodule-detail', kwargs={'pk': 'zz_by_name.py'})
+
+            response = self.client.put(
+                detail_url,
+                {'file': SimpleUploadedFile('zz_by_name.py', updated_content, content_type='text/plain')},
+                format='multipart',
+                **self.header,
+            )
+            self.assertHttpStatus(response, status.HTTP_200_OK)
+
+        self.assertEqual(response.data['id'], module_id)
+        self.assertEqual(fake_storage.files['zz_by_name.py'], updated_content)
+
+    def test_update_script_module_rejects_mismatched_file_name(self):
+        """An update whose uploaded file name differs from the module's file path is rejected."""
+        self.add_permissions(
+            'extras.add_scriptmodule', 'core.add_managedfile',
+            'extras.change_scriptmodule', 'core.change_managedfile',
+        )
+        original_content = (
+            b"from extras.scripts import Script\n\n\n"
+            b"class ProbeScript(Script):\n    def run(self, data, commit):\n        return 'v1'\n"
+        )
+        fake_storage = _InMemoryScriptStorage()
+
+        with self._patched_script_storage(fake_storage):
+            response = self.client.post(
+                self.url,
+                {'file': SimpleUploadedFile('zz_mismatch.py', original_content, content_type='text/plain')},
+                format='multipart',
+                **self.header,
+            )
+            self.assertHttpStatus(response, status.HTTP_201_CREATED)
+            detail_url = reverse('extras-api:scriptmodule-detail', kwargs={'pk': response.data['id']})
+
+            response = self.client.patch(
+                detail_url,
+                {'file': SimpleUploadedFile('zz_other.py', original_content, content_type='text/plain')},
+                format='multipart',
+                **self.header,
+            )
+
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(fake_storage.files['zz_mismatch.py'], original_content)
+        self.assertNotIn('zz_other.py', fake_storage.files)
+
+    def test_update_script_module_storage_name_mismatch_fails(self):
+        """If the storage backend saves under an alternate name, the update is rejected and rolled back."""
+        self.add_permissions(
+            'extras.add_scriptmodule', 'core.add_managedfile',
+            'extras.change_scriptmodule', 'core.change_managedfile',
+        )
+        original_content = (
+            b"from extras.scripts import Script\n\n\n"
+            b"class ProbeScript(Script):\n    def run(self, data, commit):\n        return 'v1'\n"
+        )
+        updated_content = original_content.replace(b"'v1'", b"'v2'")
+        fake_storage = _InMemoryScriptStorage(allow_overwrite=False)
+
+        with self._patched_script_storage(fake_storage):
+            response = self.client.post(
+                self.url,
+                {'file': SimpleUploadedFile('zz_suffix.py', original_content, content_type='text/plain')},
+                format='multipart',
+                **self.header,
+            )
+            self.assertHttpStatus(response, status.HTTP_201_CREATED)
+            module = ScriptModule.objects.get(file_path='zz_suffix.py')
+            detail_url = reverse('extras-api:scriptmodule-detail', kwargs={'pk': module.pk})
+
+            response = self.client.patch(
+                detail_url,
+                {'file': SimpleUploadedFile('zz_suffix.py', updated_content, content_type='text/plain')},
+                format='multipart',
+                **self.header,
+            )
+
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        # Original file intact, alternate-name orphan removed
+        self.assertEqual(fake_storage.files['zz_suffix.py'], original_content)
+        self.assertNotIn('zz_suffix.py.1', fake_storage.files)
+        module.refresh_from_db()
+        self.assertEqual(module.file_path, 'zz_suffix.py')
+
+    def test_update_faulty_script_module_preserves_existing_module(self):
+        """An update with invalid script content is rejected before storage or Script rows change."""
+        self.add_permissions(
+            'extras.add_scriptmodule', 'core.add_managedfile',
+            'extras.change_scriptmodule', 'core.change_managedfile',
+        )
+        original_content = (
+            b"from extras.scripts import Script\n\n\n"
+            b"class ProbeScript(Script):\n    def run(self, data, commit):\n        return 'v1'\n"
+        )
+        # 'extras.script' is invalid; the correct module is 'extras.scripts'
+        faulty_content = b"from extras.script import Script\nclass TestScript(Script):\n    pass\n"
+        fake_storage = _InMemoryScriptStorage()
+
+        with self._patched_script_storage(fake_storage):
+            response = self.client.post(
+                self.url,
+                {'file': SimpleUploadedFile('zz_faulty_update.py', original_content, content_type='text/plain')},
+                format='multipart',
+                **self.header,
+            )
+            self.assertHttpStatus(response, status.HTTP_201_CREATED)
+            module_id = response.data['id']
+            detail_url = reverse('extras-api:scriptmodule-detail', kwargs={'pk': module_id})
+
+            response = self.client.patch(
+                detail_url,
+                {'file': SimpleUploadedFile('zz_faulty_update.py', faulty_content, content_type='text/plain')},
+                format='multipart',
+                **self.header,
+            )
+
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(fake_storage.files['zz_faulty_update.py'], original_content)
+        self.assertTrue(Script.objects.filter(module_id=module_id, name='ProbeScript').exists())
+
+    def test_update_script_module_not_found(self):
+        """An update addressing a nonexistent module returns 404 for both lookup styles."""
+        self.add_permissions('extras.change_scriptmodule', 'core.change_managedfile')
+        for lookup in ('999999', 'zz_missing.py', '½'):
+            detail_url = reverse('extras-api:scriptmodule-detail', kwargs={'pk': lookup})
+            response = self.client.patch(detail_url, {}, format='json', **self.header)
+            self.assertHttpStatus(response, status.HTTP_404_NOT_FOUND)
+
+    def test_update_script_module_via_put_without_file_fails(self):
+        """A PUT without a file upload is rejected by the field-level required check."""
+        self.add_permissions(
+            'extras.add_scriptmodule', 'core.add_managedfile',
+            'extras.change_scriptmodule', 'core.change_managedfile',
+        )
+        script_content = b"from extras.scripts import Script\nclass TestScript(Script):\n    pass\n"
+        fake_storage = _InMemoryScriptStorage()
+
+        with self._patched_script_storage(fake_storage):
+            response = self.client.post(
+                self.url,
+                {'file': SimpleUploadedFile('zz_put_nofile.py', script_content, content_type='text/plain')},
+                format='multipart',
+                **self.header,
+            )
+            self.assertHttpStatus(response, status.HTTP_201_CREATED)
+            detail_url = reverse('extras-api:scriptmodule-detail', kwargs={'pk': response.data['id']})
+
+            response = self.client.put(detail_url, {}, format='json', **self.header)
+
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertIn('file', response.data)
+        self.assertEqual(fake_storage.files['zz_put_nofile.py'], script_content)
+
+    def test_update_script_module_rolls_back_scripts_on_save_failure(self):
+        """A failed Script sync during an update rolls back all Script row changes."""
+        self.add_permissions(
+            'extras.add_scriptmodule', 'core.add_managedfile',
+            'extras.change_scriptmodule', 'core.change_managedfile',
+        )
+        original_content = (
+            b"from extras.scripts import Script\n\n\n"
+            b"class ProbeScript(Script):\n    def run(self, data, commit):\n        return 'v1'\n"
+        )
+        updated_content = (
+            b"from extras.scripts import Script\n\n\n"
+            b"class ProbeScriptV2(Script):\n    def run(self, data, commit):\n        return 'v2'\n"
+        )
+        fake_storage = _InMemoryScriptStorage()
+
+        with self._patched_script_storage(fake_storage):
+            response = self.client.post(
+                self.url,
+                {'file': SimpleUploadedFile('zz_rollback.py', original_content, content_type='text/plain')},
+                format='multipart',
+                **self.header,
+            )
+            self.assertHttpStatus(response, status.HTTP_201_CREATED)
+            module_id = response.data['id']
+            detail_url = reverse('extras-api:scriptmodule-detail', kwargs={'pk': module_id})
+
+            # Fail the sync mid-way: the old Script row is already deleted when create() raises
+            with patch(
+                'extras.models.scripts.Script.objects.create',
+                side_effect=IntegrityError('Simulated database error'),
+            ):
+                with self.assertRaises(IntegrityError):
+                    self.client.patch(
+                        detail_url,
+                        {'file': SimpleUploadedFile('zz_rollback.py', updated_content, content_type='text/plain')},
+                        format='multipart',
+                        **self.header,
+                    )
+
+        # DB is all-or-nothing; storage keeps the new content and a retried update re-syncs the rows
+        self.assertTrue(Script.objects.filter(module_id=module_id, name='ProbeScript').exists())
+        self.assertFalse(Script.objects.filter(module_id=module_id, name='ProbeScriptV2').exists())
+        self.assertEqual(fake_storage.files['zz_rollback.py'], updated_content)
+
+    def test_update_script_module_with_missing_stored_file(self):
+        """An update succeeds when the stored file is missing, re-creating it from the upload."""
+        self.add_permissions(
+            'extras.add_scriptmodule', 'core.add_managedfile',
+            'extras.change_scriptmodule', 'core.change_managedfile',
+        )
+        original_content = (
+            b"from extras.scripts import Script\n\n\n"
+            b"class ProbeScript(Script):\n    def run(self, data, commit):\n        return 'v1'\n"
+        )
+        updated_content = original_content.replace(b"'v1'", b"'v2'")
+        fake_storage = _InMemoryScriptStorage()
+
+        with self._patched_script_storage(fake_storage):
+            response = self.client.post(
+                self.url,
+                {'file': SimpleUploadedFile('zz_lost_file.py', original_content, content_type='text/plain')},
+                format='multipart',
+                **self.header,
+            )
+            self.assertHttpStatus(response, status.HTTP_201_CREATED)
+            detail_url = reverse('extras-api:scriptmodule-detail', kwargs={'pk': response.data['id']})
+
+            # Simulate a file removed from storage outside NetBox
+            del fake_storage.files['zz_lost_file.py']
+
+            response = self.client.patch(
+                detail_url,
+                {'file': SimpleUploadedFile('zz_lost_file.py', updated_content, content_type='text/plain')},
+                format='multipart',
+                **self.header,
+            )
+
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+        self.assertEqual(fake_storage.files['zz_lost_file.py'], updated_content)