[Fixes #12996] Fix permissions reassign of API to transfer resources

Author: mattiagiupponi

geonode/base/api/views.py

@@ -618,6 +618,9 @@ def resource_service_permissions(self, request, pk, *args, **kwargs):
                 )
             elif request.method == "PUT":
                 perms_spec_compact = PermSpecCompact(request.data, resource)
+                if resource.dirty_state:
+                    raise Exception("Cannot update if the resource is in dirty state")
+                resource.set_dirty_state()
                 _exec_request = ExecutionRequest.objects.create(
                     user=request.user,
                     func_name="set_permissions",
@@ -634,6 +637,9 @@ def resource_service_permissions(self, request, pk, *args, **kwargs):
                 perms_spec_compact_patch = PermSpecCompact(request.data, resource)
                 perms_spec_compact_resource = PermSpecCompact(perms_spec.compact, resource)
                 perms_spec_compact_resource.merge(perms_spec_compact_patch)
+                if resource.dirty_state:
+                    raise Exception("Cannot update if the resource is in dirty state")
+                resource.set_dirty_state()
                 _exec_request = ExecutionRequest.objects.create(
                     user=request.user,
                     func_name="set_permissions",

geonode/people/api/views.py

@@ -1,3 +1,23 @@
+#########################################################################
+#
+# Copyright (C) 2025 OSGeo
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+#
+#########################################################################
+
+import logging
 from django.conf import settings
 from dynamic_rest.filters import DynamicFilterBackend, DynamicSortingFilter
 from oauth2_provider.contrib.rest_framework import OAuth2Authentication
@@ -25,6 +45,9 @@
 from geonode.security.registry import permissions_registry
 
 
+logger = logging.getLogger()
+
+
 class UserViewSet(DynamicModelViewSet):
     """
     API endpoint that allows users to be viewed or edited.
@@ -151,9 +174,16 @@ def transfer_resources(self, request, pk=None):
         admin = get_user_model().objects.filter(is_superuser=True, is_staff=True).first()
         target_user = request.data.get("newOwner")  # the new owner
         previous_owner = request.data.get("currentOwner")  # the previous owner, usually it match the user
-        transfer_resource_subset = request.data.get("resources", None)
+        transfer_resource_subset = (
+            request.data.get("resources", None)
+            if not hasattr(request.data, "getlist")
+            else request.data.getlist("resources", None)
+        )
         target = None
 
+        if not target_user and previous_owner:
+            return Response("Payload not passed", status=400)
+
         if user.is_superuser or (
             not user.is_superuser
             and ResourceBase.objects.filter(owner=user, pk__in=transfer_resource_subset).count()
@@ -183,13 +213,22 @@ def transfer_resources(self, request, pk=None):
                 we can use the resource manager because inside it will automatically update
                 the owner
                 """
-                perms = permissions_registry.get_perms(instance=instance, include_virtual=False)
-                prev_owner = get_user_model().objects.filter(pk=previous_owner).first()
-
-                if prev_owner and not prev_owner.is_superuser:
-                    perms["users"].pop(prev_owner)
-
-                resource_manager.set_permissions(instance.uuid, instance, owner=target or user, permissions=perms)
+                try:
+                    # putting the resource in dirty state
+                    instance.set_dirty_state()
+                    # updating the perms with the new owner
+                    perms = permissions_registry.get_perms(instance=instance, include_virtual=False)
+                    prev_owner = get_user_model().objects.filter(pk=previous_owner).first()
+
+                    if prev_owner and not prev_owner.is_superuser:
+                        perms["users"].pop(prev_owner)
+                    # calling the registry to update the perms
+                    resource_manager.set_permissions(instance.uuid, instance, owner=target or user, permissions=perms)
+                except Exception as e:
+                    logger.exeption(e)
+                finally:
+                    # clearing the dirty state
+                    instance.clear_dirty_state()
 
             return Response("Resources transfered successfully", status=200)
 

geonode/people/tests.py

@@ -35,7 +35,7 @@
 from geonode.layers.models import Dataset
 from geonode.people import profileextractors
 
-from geonode.base.populate_test_data import all_public, create_models, remove_models
+from geonode.base.populate_test_data import all_public, create_models, create_single_dataset, remove_models
 from django.db.models import Q
 from geonode.security.registry import permissions_registry
 
@@ -1150,7 +1150,8 @@ def test_transfer_resources_all(self):
         self.assertTrue(bobby_resources.exists())
         # call api
         response = self.client.post(
-            path=f"{reverse('users-list')}/{bobby.pk}/transfer_resources", data={"owner": norman.id}
+            path=f"{reverse('users-list')}/{bobby.pk}/transfer_resources",
+            data={"currentOwner": bobby.id, "newOwner": norman.id},
         )
         # check that bobby owns the resources no more
         self.assertFalse(bobby_resources.exists())
@@ -1174,7 +1175,8 @@ def test_transfer_resources_invalid_user(self):
         self.assertTrue(bobby_resources.exists())
         # call api
         response = self.client.post(
-            path=f"{reverse('users-list')}/{bobby}/transfer_resources", data={"owner": invalid_user_id}
+            path=f"{reverse('users-list')}/{bobby}/transfer_resources",
+            data={"currentOwner": bobby.id, "newOwner": invalid_user_id},
         )
         # response should be 404
         self.assertEqual(response.status_code, 404)
@@ -1200,7 +1202,8 @@ def test_transfer_resources_default(self):
         self.assertTrue(bobby_resources.exists())
         # call api
         response = self.client.post(
-            path=f"{reverse('users-list')}/{bobby.pk}/transfer_resources", data={"owner": "DEFAULT"}
+            path=f"{reverse('users-list')}/{bobby.pk}/transfer_resources",
+            data={"currentOwner": bobby.id, "newOwner": "DEFAULT"},
         )
         self.assertTrue(response.status_code == 200)
         # check that bobby owns the resources no more
@@ -1232,7 +1235,8 @@ def test_transfer_resources_to_missing_default(self):
         self.assertTrue(bobby_resources.exists())
         # call api
         response = self.client.post(
-            path=f"{reverse('users-list')}/{bobby.pk}/transfer_resources", data={"owner": "DEFAULT"}
+            path=f"{reverse('users-list')}/{bobby.pk}/transfer_resources",
+            data={"newOwner": "DEFAULT", "previousOwner": bobby.pk},
         )
         self.assertTrue(response.status_code == 500)
         self.assertEqual(response.data, "Principal User not found")
@@ -1257,7 +1261,8 @@ def test_transfer_resources_to_self(self):
 
         # call api
         response = self.client.post(
-            path=f"{reverse('users-list')}/{bobby.pk}/transfer_resources", data={"owner": bobby.pk}
+            path=f"{reverse('users-list')}/{bobby.pk}/transfer_resources",
+            data={"previousOwner": bobby.pk, "newOwner": bobby.pk},
         )
         self.assertTrue(response.status_code == 400)
         self.assertEqual(response.data, "Cannot reassign to self")
@@ -1277,7 +1282,6 @@ def test_transfer_resources_nopayload(self):
         bobby_resources = ResourceBase.objects.filter(owner=bobby)
         prior_bobby_resources = bobby_resources.all()
         self.assertTrue(bobby_resources.exists())
-
         # call api
         response = self.client.post(path=f"{reverse('users-list')}/{bobby.pk}/transfer_resources", data={})
         # response should be 404
@@ -1296,13 +1300,17 @@ def test_transfer_resource_subset(self):
         self.assertTrue(bobby.is_authenticated)
         # check bobbys resources
         resource_to_transfer = ResourceBase.objects.filter(owner=bobby).first()
-        second_resource = ResourceBase.objects.filter(owner=bobby).last()
+        second_resource = create_single_dataset("test for api", owner=bobby)
         new_owner = get_user_model().objects.exclude(username__in=["bobby", "AnonymousUser"]).first()
 
         # call api to transfer bobby resource to the other user
         response = self.client.post(
             path=f"{reverse('users-list')}/{bobby.pk}/transfer_resources",
-            data={"owner": new_owner.pk, "resources": [resource_to_transfer.pk, second_resource.pk]},
+            data={
+                "newOwner": new_owner.pk,
+                "currentOwner": bobby.pk,
+                "resources": [resource_to_transfer.pk, second_resource.pk],
+            },
         )
         # response should be 200
         self.assertEqual(response.status_code, 200)