[admin/api] Make batch user creation organization field readonly #609

This change prevents modification of the organization field for existing
RadiusBatch objects in both admin interface and REST API. When editing
existing batch objects, the organization field is now readonly to maintain
data integrity and prevent accidental organization changes.

Changes:
- Added organization field to readonly_fields in RadiusBatchAdmin for existing objects
- Created RadiusBatchUpdateSerializer with organization as readonly field
- Added BatchUpdateView for handling batch detail API operations
- Added comprehensive tests for both admin and API readonly functionality

Fixes #609

Author: Eeshu-Yadav

openwisp_radius/admin.py

@@ -463,18 +463,17 @@ def delete_selected_batches(self, request, queryset):
         )
 
     def get_readonly_fields(self, request, obj=None):
-        readonly_fields = super(RadiusBatchAdmin, self).get_readonly_fields(
-            request, obj
-        )
+        readonly_fields = super().get_readonly_fields(request, obj)
         if obj:
-            return (
+            return readonly_fields + (
                 "strategy",
+                "organization",
                 "prefix",
                 "csvfile",
                 "number_of_users",
                 "users",
                 "expiration_date",
-            ) + readonly_fields
+            )
         return readonly_fields
 
 

openwisp_radius/api/serializers.py

@@ -438,6 +438,34 @@ class Meta:
         read_only_fields = ("created", "modified", "user_credentials")
 
 
+class RadiusBatchUpdateSerializer(RadiusBatchSerializer):
+    """
+    Serializer for updating RadiusBatch objects.
+    Makes the organization field readonly for existing objects.
+    """
+    
+    def validate(self, data):
+        """
+        Simplified validation for partial updates.
+        Only validates essential fields based on strategy.
+        """
+        strategy = data.get("strategy") or (self.instance and self.instance.strategy)
+        
+        if (strategy == "prefix" and 
+            "number_of_users" in data and 
+            not data.get("number_of_users")):
+            raise serializers.ValidationError(
+                {"number_of_users": _("The field number_of_users cannot be empty")}
+            )
+        
+        return serializers.ModelSerializer.validate(self, data)
+    
+    class Meta:
+        model = RadiusBatch
+        fields = "__all__"
+        read_only_fields = ("created", "modified", "user_credentials", "organization")
+
+
 class PasswordResetSerializer(BasePasswordResetSerializer):
     input = serializers.CharField()
     email = None

openwisp_radius/api/urls.py

@@ -78,6 +78,7 @@ def get_api_urls(api_views=None):
                 name="phone_number_change",
             ),
             path("radius/batch/", api_views.batch, name="batch"),
+            path("radius/batch/<uuid:pk>/", api_views.batch_detail, name="batch_detail"),
             path(
                 "radius/organization/<slug:slug>/batch/<uuid:pk>/pdf/",
                 api_views.download_rad_batch_pdf,

openwisp_radius/api/views.py

@@ -31,6 +31,7 @@
     GenericAPIView,
     ListAPIView,
     RetrieveAPIView,
+    RetrieveUpdateAPIView,
 )
 from rest_framework.permissions import (
     DjangoModelPermissions,
@@ -62,6 +63,7 @@
     ChangePhoneNumberSerializer,
     RadiusAccountingSerializer,
     RadiusBatchSerializer,
+    RadiusBatchUpdateSerializer,
     UserRadiusUsageSerializer,
     ValidatePhoneTokenSerializer,
 )
@@ -144,6 +146,35 @@ def validate_membership(self, user):
             raise serializers.ValidationError({"non_field_errors": [message]})
 
 
+class BatchUpdateView(ThrottledAPIMixin, RetrieveUpdateAPIView):
+    """
+    API view for updating existing RadiusBatch objects.
+    Organization field is readonly for existing objects.
+    """
+    authentication_classes = (BearerAuthentication, SessionAuthentication)
+    permission_classes = (IsOrganizationManager, IsAdminUser, DjangoModelPermissions)
+    queryset = RadiusBatch.objects.none()
+    serializer_class = RadiusBatchSerializer
+
+    def get_queryset(self):
+        """Filter batches by user's organization membership"""
+        user = self.request.user
+        if user.is_superuser:
+            return RadiusBatch.objects.all()
+        return RadiusBatch.objects.filter(
+            organization__in=user.organizations_managed
+        )
+
+    def get_serializer_class(self):
+        """Use RadiusBatchUpdateSerializer for PUT/PATCH requests"""
+        if self.request.method in ('PUT', 'PATCH'):
+            return RadiusBatchUpdateSerializer
+        return RadiusBatchSerializer
+
+
+batch_detail = BatchUpdateView.as_view()
+
+
 class DownloadRadiusBatchPdfView(ThrottledAPIMixin, DispatchOrgMixin, RetrieveAPIView):
     authentication_classes = (BearerAuthentication, SessionAuthentication)
     permission_classes = (IsOrganizationManager, IsAdminUser, DjangoModelPermissions)

openwisp_radius/tests/test_admin.py

@@ -1494,3 +1494,33 @@ def test_admin_menu_groups(self):
         with self.subTest("test radius group is registered"):
             html = '<div class="mg-dropdown-label">RADIUS </div>'
             self.assertContains(response, html, html=True)
+
+    def test_radiusbatch_organization_readonly_for_existing_objects(self):
+        """
+        Test that organization field is readonly for existing RadiusBatch objects
+        """
+        batch = self._create_radius_batch(
+            name="test-batch",
+            strategy="prefix",
+            prefix="test-prefix"
+        )
+        url = reverse(f"admin:{self.app_label}_radiusbatch_change", args=[batch.pk])
+        
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, 200)
+        
+        self.assertContains(response, 'readonly')
+        self.assertContains(response, batch.organization.name)
+
+    def test_radiusbatch_organization_editable_for_new_objects(self):
+        """
+        Test that organization field is editable for new RadiusBatch objects
+        """
+        url = reverse(f"admin:{self.app_label}_radiusbatch_add")
+        
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, 200)
+        
+        self.assertContains(response, 'name="organization"')
+        form_html = response.content.decode()
+        self.assertIn('name="organization"', form_html)

openwisp_radius/tests/test_api/test_api.py

@@ -1053,6 +1053,111 @@ def test_user_group_check_serializer_counter_does_not_exist(self):
             },
         )
 
+    def _get_admin_auth_header(self):
+        """Helper method to get admin authentication header"""
+        login_payload = {"username": "admin", "password": "tester"}
+        login_url = reverse("radius:user_auth_token", args=[self.default_org.slug])
+        login_response = self.client.post(login_url, data=login_payload)
+        return f"Bearer {login_response.json()['key']}"
+
+    def test_batch_update_organization_readonly(self):
+        """
+        Test that organization field is readonly when updating RadiusBatch objects
+        """
+        data = self._radius_batch_prefix_data()
+        response = self._radius_batch_post_request(data)
+        self.assertEqual(response.status_code, 201)
+        batch = RadiusBatch.objects.get()
+        original_org = batch.organization
+
+        new_org = self._create_org(**{"name": "new-org", "slug": "new-org"})
+
+        header = self._get_admin_auth_header()
+
+        url = reverse("radius:batch_detail", args=[batch.pk])
+        update_data = {
+            "name": "updated-batch-name",
+            "organization": str(new_org.pk),
+        }
+        response = self.client.patch(
+            url, 
+            json.dumps(update_data), 
+            HTTP_AUTHORIZATION=header,
+            content_type="application/json"
+        )
+        self.assertEqual(response.status_code, 200)
+
+        batch.refresh_from_db()
+        self.assertEqual(batch.organization, original_org)
+        self.assertEqual(batch.name, "updated-batch-name")
+
+    def test_batch_retrieve_and_update_api(self):
+        """
+        Test retrieving and updating RadiusBatch objects via API
+        """
+        data = self._radius_batch_prefix_data()
+        response = self._radius_batch_post_request(data)
+        self.assertEqual(response.status_code, 201)
+        batch = RadiusBatch.objects.get()
+
+        header = self._get_admin_auth_header()
+
+        url = reverse("radius:batch_detail", args=[batch.pk])
+        response = self.client.get(url, HTTP_AUTHORIZATION=header)
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.data["name"], batch.name)
+        self.assertEqual(str(response.data["organization"]), str(batch.organization.pk))
+
+        update_data = {
+            "name": "updated-batch-name",
+            "strategy": "prefix",
+            "prefix": batch.prefix,
+            "organization_slug": batch.organization.slug,
+        }
+        response = self.client.put(
+            url, 
+            json.dumps(update_data), 
+            HTTP_AUTHORIZATION=header,
+            content_type="application/json"
+        )
+        self.assertEqual(response.status_code, 200)
+        batch.refresh_from_db()
+        self.assertEqual(batch.name, "updated-batch-name")
+
+    def test_batch_update_permissions(self):
+        """
+        Test that proper permissions are required for updating RadiusBatch objects
+        """
+        data = self._radius_batch_prefix_data()
+        response = self._radius_batch_post_request(data)
+        self.assertEqual(response.status_code, 201)
+        batch = RadiusBatch.objects.get()
+
+        url = reverse("radius:batch_detail", args=[batch.pk])
+
+        response = self.client.patch(url, {"name": "new-name"})
+        self.assertEqual(response.status_code, 401)
+
+        user = self._get_user()
+        user_token = Token.objects.create(user=user)
+        header = f"Bearer {user_token.key}"
+        response = self.client.patch(
+            url, 
+            json.dumps({"name": "new-name"}), 
+            HTTP_AUTHORIZATION=header,
+            content_type="application/json"
+        )
+        self.assertEqual(response.status_code, 403)
+
+        header = self._get_admin_auth_header()
+        response = self.client.patch(
+            url, 
+            json.dumps({"name": "new-name"}), 
+            HTTP_AUTHORIZATION=header,
+            content_type="application/json"
+        )
+        self.assertEqual(response.status_code, 200)
+
 
 class TestTransactionApi(AcctMixin, ApiTokenMixin, BaseTransactionTestCase):
     def test_user_radius_usage_view(self):