[Backport 5.0.x] [Fixes #14217 #14219] Permissons diffing and fix to permissions removal

geonode/base/api/tests.py

@@ -1301,6 +1301,87 @@ def test_perms_resources(self):
         response = self.client.put(set_perms_url, data=resource_perm_spec, format="json")
         self.assertEqual(response.status_code, 200)
 
+    @override_settings(
+        EDITORS_CAN_MANAGE_ANONYMOUS_PERMISSIONS=True,
+        EDITORS_CAN_MANAGE_REGISTERED_MEMBERS_PERMISSIONS=True,
+    )
+    def test_resource_service_permissions_patch_merges_permissions(self):
+        self.assertTrue(self.client.login(username="admin", password="admin"))
+        admin = get_user_model().objects.get(username="admin")
+        bobby = get_user_model().objects.get(username="bobby")
+        norman = get_user_model().objects.get(username="norman")
+        resource = dataset_manager.create(
+            str(uuid4()), resource_type=Dataset, defaults={"title": "api_perms_patch_merge", "owner": admin}
+        )
+        resource.set_permissions({"users": {bobby: ["base.view_resourcebase"]}})
+
+        set_perms_url = urljoin(f"{reverse('base-resources-detail', kwargs={'pk': resource.pk})}/", "permissions")
+        response = self.client.patch(
+            set_perms_url,
+            data={
+                "uuid": resource.uuid,
+                "users": [{"id": norman.id, "permissions": "edit"}],
+            },
+            format="json",
+        )
+        self.assertEqual(response.status_code, 200)
+        resp_js = json.loads(response.content.decode("utf-8"))
+        resouce_service_dispatcher.apply((resp_js.get("execution_id"),))
+        response = self.client.get(resp_js.get("status_url"))
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.json().get("status"), "finished")
+
+        response = self.client.get(set_perms_url, format="json")
+        self.assertEqual(response.status_code, 200)
+        user_permissions = {u["username"]: u["permissions"] for u in response.data.get("users", [])}
+        self.assertEqual(user_permissions.get("bobby"), "view")
+        self.assertEqual(user_permissions.get("norman"), "edit")
+
+    @override_settings(
+        EDITORS_CAN_MANAGE_ANONYMOUS_PERMISSIONS=True,
+        EDITORS_CAN_MANAGE_REGISTERED_MEMBERS_PERMISSIONS=True,
+    )
+    def test_resource_service_permissions_put_replaces_permissions(self):
+        self.assertTrue(self.client.login(username="admin", password="admin"))
+        admin = get_user_model().objects.get(username="admin")
+        bobby = get_user_model().objects.get(username="bobby")
+        norman = get_user_model().objects.get(username="norman")
+        resource = dataset_manager.create(
+            str(uuid4()), resource_type=Dataset, defaults={"title": "api_perms_put_replace", "owner": admin}
+        )
+        resource.set_permissions(
+            {
+                "users": {
+                    bobby: ["base.view_resourcebase"],
+                    norman: ["base.view_resourcebase"],
+                }
+            }
+        )
+
+        set_perms_url = urljoin(f"{reverse('base-resources-detail', kwargs={'pk': resource.pk})}/", "permissions")
+        response = self.client.put(
+            set_perms_url,
+            data={
+                "uuid": resource.uuid,
+                "users": [{"id": norman.id, "permissions": "edit"}],
+                "organizations": [],
+                "groups": [],
+            },
+            format="json",
+        )
+        self.assertEqual(response.status_code, 200)
+        resp_js = json.loads(response.content.decode("utf-8"))
+        resouce_service_dispatcher.apply((resp_js.get("execution_id"),))
+        response = self.client.get(resp_js.get("status_url"))
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.json().get("status"), "finished")
+
+        response = self.client.get(set_perms_url, format="json")
+        self.assertEqual(response.status_code, 200)
+        user_permissions = {u["username"]: u["permissions"] for u in response.data.get("users", [])}
+        self.assertNotIn("bobby", user_permissions)
+        self.assertEqual(user_permissions.get("norman"), "edit")
+
     def test_featured_and_published_resources(self):
         """
         Ensure we can Get & Set Permissions across the Resource Base list.
@@ -2392,21 +2473,22 @@ def test_resource_service_permissions_with_restricted_settings(self):
 
         # Try to update permissions including anonymous and registered members groups
         set_perms_url = urljoin(f"{reverse('base-resources-detail', kwargs={'pk': resource.pk})}/", "permissions")
-        perm_spec = {
-            "uuid": resource.uuid,
-            "groups": [
-                {
-                    "id": anonymous_group.id,
-                    "name": "anonymous",
-                    "permissions": "view",
-                },
-                {
-                    "id": registered_group.id,
-                    "name": "registered-members",
-                    "permissions": "download",
-                },
-            ],
-        }
+        response = self.client.get(set_perms_url, format="json")
+        self.assertEqual(response.status_code, 200)
+        perm_spec = response.data
+        perm_spec["uuid"] = resource.uuid
+        perm_spec["groups"] = [
+            {
+                "id": anonymous_group.id,
+                "name": "anonymous",
+                "permissions": "view",
+            },
+            {
+                "id": registered_group.id,
+                "name": "registered-members",
+                "permissions": "download",
+            },
+        ]
 
         response = self.client.put(set_perms_url, data=perm_spec, format="json")
         self.assertEqual(response.status_code, 200)
@@ -2454,21 +2536,22 @@ def test_resource_service_permissions_with_restricted_settings(self):
         # login as admin (staff user) and verify admin can modify these permissions
         self.assertTrue(self.client.login(username="admin", password="admin"))
 
-        perm_spec_admin = {
-            "uuid": resource.uuid,
-            "groups": [
-                {
-                    "id": anonymous_group.id,
-                    "name": "anonymous",
-                    "permissions": "view",
-                },
-                {
-                    "id": registered_group.id,
-                    "name": "registered-members",
-                    "permissions": "view",
-                },
-            ],
-        }
+        response = self.client.get(set_perms_url, format="json")
+        self.assertEqual(response.status_code, 200)
+        perm_spec_admin = response.data
+        perm_spec_admin["uuid"] = resource.uuid
+        perm_spec_admin["groups"] = [
+            {
+                "id": anonymous_group.id,
+                "name": "anonymous",
+                "permissions": "view",
+            },
+            {
+                "id": registered_group.id,
+                "name": "registered-members",
+                "permissions": "view",
+            },
+        ]
 
         response = self.client.put(set_perms_url, data=perm_spec_admin, format="json")
         self.assertEqual(response.status_code, 200)
@@ -2539,21 +2622,22 @@ def test_resource_service_permissions_with_partial_restriction(self):
         registered_group = Group.objects.get(name="registered-members")
 
         set_perms_url = urljoin(f"{reverse('base-resources-detail', kwargs={'pk': resource.pk})}/", "permissions")
-        perm_spec = {
-            "uuid": resource.uuid,
-            "groups": [
-                {
-                    "id": anonymous_group.id,
-                    "name": "anonymous",
-                    "permissions": "view",
-                },
-                {
-                    "id": registered_group.id,
-                    "name": "registered-members",
-                    "permissions": "download",
-                },
-            ],
-        }
+        response = self.client.get(set_perms_url, format="json")
+        self.assertEqual(response.status_code, 200)
+        perm_spec = response.data
+        perm_spec["uuid"] = resource.uuid
+        perm_spec["groups"] = [
+            {
+                "id": anonymous_group.id,
+                "name": "anonymous",
+                "permissions": "view",
+            },
+            {
+                "id": registered_group.id,
+                "name": "registered-members",
+                "permissions": "download",
+            },
+        ]
 
         response = self.client.put(set_perms_url, data=perm_spec, format="json")
         self.assertEqual(response.status_code, 200)
@@ -2599,6 +2683,79 @@ def test_resource_service_permissions_with_partial_restriction(self):
             "Bobby should not have been able to set registered-members to download",
         )
 
+    @override_settings(
+        EDITORS_CAN_MANAGE_ANONYMOUS_PERMISSIONS=False,
+        EDITORS_CAN_MANAGE_REGISTERED_MEMBERS_PERMISSIONS=False,
+    )
+    def test_resource_service_permissions_patch_with_restricted_settings(self):
+        resource = Dataset.objects.filter(owner__username="admin").first()
+        bobby = get_user_model().objects.get(username="bobby")
+        resource.set_permissions(
+            {
+                "users": {
+                    bobby: [
+                        "base.change_resourcebase",
+                        "base.change_resourcebase_metadata",
+                        "base.change_resourcebase_permissions",
+                    ]
+                }
+            }
+        )
+
+        self.assertTrue(self.client.login(username="bobby", password="bob"))
+
+        anonymous_group = Group.objects.get(name="anonymous")
+        registered_group = Group.objects.get(name="registered-members")
+        set_perms_url = urljoin(f"{reverse('base-resources-detail', kwargs={'pk': resource.pk})}/", "permissions")
+
+        response = self.client.get(set_perms_url, format="json")
+        self.assertEqual(response.status_code, 200)
+        initial_group_permissions = {g["name"]: g["permissions"] for g in response.data.get("groups", [])}
+        anonymous_target = "download" if initial_group_permissions.get("anonymous") != "download" else "view"
+        registered_target = "download" if initial_group_permissions.get("registered-members") != "download" else "view"
+
+        response = self.client.patch(
+            set_perms_url,
+            data={
+                "uuid": resource.uuid,
+                "groups": [
+                    {
+                        "id": anonymous_group.id,
+                        "name": "anonymous",
+                        "permissions": anonymous_target,
+                    },
+                    {
+                        "id": registered_group.id,
+                        "name": "registered-members",
+                        "permissions": registered_target,
+                    },
+                ],
+            },
+            format="json",
+        )
+        self.assertEqual(response.status_code, 200)
+
+        resp_js = json.loads(response.content.decode("utf-8"))
+        execution_id = resp_js.get("execution_id", "")
+        status_url = resp_js.get("status_url", None)
+        for _cnt in range(0, 10):
+            response = self.client.get(f"{status_url}")
+            self.assertEqual(response.status_code, 200)
+            resp_js = json.loads(response.content.decode("utf-8"))
+            if resp_js.get("status", "") == "finished":
+                break
+            else:
+                resouce_service_dispatcher.apply((execution_id,))
+                sleep(3.0)
+
+        response = self.client.get(set_perms_url, format="json")
+        self.assertEqual(response.status_code, 200)
+        group_permissions = {g["name"]: g["permissions"] for g in response.data.get("groups", [])}
+        self.assertEqual(group_permissions.get("anonymous"), initial_group_permissions.get("anonymous"))
+        self.assertEqual(
+            group_permissions.get("registered-members"), initial_group_permissions.get("registered-members")
+        )
+
     def test_resource_service_copy(self):
         files = os.path.join(gisdata.GOOD_DATA, "vector/single_point.shp")
         files_as_dict, _ = get_files(files)

geonode/base/api/views.py

@@ -68,7 +68,7 @@
     AdvertisedFilter,
 )
 from geonode.groups.models import GroupProfile, Group
-from geonode.security.permissions import get_compact_perms_list, PermSpec
+from geonode.security.permissions import get_compact_perms_list, PermSpec, PermSpecCompact
 from geonode.security.utils import (
     get_visible_resources,
     get_resources_with_perms,
@@ -105,7 +105,7 @@
 )
 from geonode.people.api.serializers import UserSerializer
 from .pagination import GeoNodeApiPagination
-from geonode.base.utils import validate_extra_metadata
+from geonode.base.utils import validate_extra_metadata, patch_perms
 from geonode.assets.models import Asset
 from geonode.assets.utils import create_asset_and_link, unlink_asset
 from geonode.assets.handlers import asset_handler_registry
@@ -620,23 +620,33 @@ def resource_service_permissions(self, request, pk, *args, **kwargs):
                 )
             elif request.method in ["PUT", "PATCH"]:
                 user_perms = permissions_registry.get_perms(instance=resource, user=request.user)
-                if request.data.get("groups"):
-                    excluded_ids = []
-                    if "can_manage_anonymous_permissions" not in user_perms:
-                        anonymous_group = Group.objects.get(name="anonymous")
-                        excluded_ids.append(anonymous_group.id)
-                        logger.info(
-                            f"User {request.user.username} cannot manage anonymous permissions on resource {resource.pk}"
-                        )
-                    if "can_manage_registered_member_permissions" not in user_perms:
-                        registered_group = Group.objects.get(name=groups_settings.REGISTERED_MEMBERS_GROUP_NAME)
-                        excluded_ids.append(registered_group.id)
-                        logger.info(
-                            f"User {request.user.username} cannot manage registered members permissions on resource {resource.pk}"
-                        )
-                    if excluded_ids:
-                        request.data["groups"] = [g for g in request.data["groups"] if g.get("id") not in excluded_ids]
-                perms_spec_compact_resource = patch_perms(request.data, perms_spec.compact, resource)
+                current_compact = PermSpecCompact(perms_spec.compact, resource)
+                if request.method == "PATCH":
+                    proposed_compact = PermSpecCompact(perms_spec.compact, resource)
+                    proposed_compact.merge(PermSpecCompact(request.data, resource))
+                else:
+                    proposed_compact = PermSpecCompact(request.data, resource)
+
+                perms_diff = current_compact.diff(proposed_compact)
+                excluded_group_ids = []
+                if "can_manage_anonymous_permissions" not in user_perms:
+                    anonymous_group = Group.objects.get(name="anonymous")
+                    excluded_group_ids.append(anonymous_group.id)
+                    logger.info(
+                        f"User {request.user.username} cannot manage anonymous permissions on resource {resource.pk}"
+                    )
+                if "can_manage_registered_member_permissions" not in user_perms:
+                    registered_group = Group.objects.get(name=groups_settings.REGISTERED_MEMBERS_GROUP_NAME)
+                    excluded_group_ids.append(registered_group.id)
+                    logger.info(
+                        f"User {request.user.username} cannot manage registered members permissions on resource {resource.pk}"
+                    )
+                if excluded_group_ids:
+                    for diff_action in ("added", "removed", "changed"):
+                        perms_diff.groups[diff_action] = [
+                            item for item in perms_diff.groups[diff_action] if item.get("id") not in excluded_group_ids
+                        ]
+                perms_spec_compact_resource = patch_perms(perms_spec.compact, perms_diff, resource)
 
                 if resource.dirty_state:
                     raise Exception("Cannot update if the resource is in dirty state")

geonode/base/utils.py

@@ -39,7 +39,7 @@
 from geonode.security.utils import AdvancedSecurityWorkflowManager
 from geonode.thumbs.utils import get_thumbs, remove_thumb
 from geonode.utils import get_legend_url
-from geonode.security.permissions import PermSpecCompact
+from geonode.security.permissions import PermSpecCompactDiff
 
 logger = logging.getLogger("geonode.base.utils")
 
@@ -216,11 +216,14 @@ def remove_country_from_languagecode(language: str):
     return lang
 
 
-def patch_perms(updated_perms_compact, current_perms_compact, resource):
+def patch_perms(current_perms_compact, perms_diff, resource):
     """
-    Patch updated permission changes with current permissions.
+    Apply a permission diff to a current compact spec.
+
+    ``perms_diff`` may be a :class:`PermSpecCompactDiff` instance or its dict
+    representation (as produced by :meth:`PermSpecCompactDiff.to_dict` or
+    :meth:`PermSpecCompact.diff`). Returns the resulting ``PermSpecCompact``.
     """
-    perms_spec_compact_patch = PermSpecCompact(updated_perms_compact, resource)
-    perms_spec_compact_resource = PermSpecCompact(current_perms_compact, resource)
-    perms_spec_compact_resource.merge(perms_spec_compact_patch)
-    return perms_spec_compact_resource
+    if not isinstance(perms_diff, PermSpecCompactDiff):
+        perms_diff = PermSpecCompactDiff.from_dict(perms_diff)
+    return perms_diff.apply(current_perms_compact, resource)