[Fixes #12713] Permissions Registry and permissions handler

geonode/base/api/permissions.py

@@ -35,8 +35,7 @@
 from geonode.groups.models import GroupProfile
 from rest_framework.permissions import DjangoModelPermissions
 from guardian.shortcuts import get_objects_for_user
-from itertools import chain
-from guardian.shortcuts import get_groups_with_perms
+from geonode.security.registry import permissions_registry
 
 logger = logging.getLogger(__name__)
 
@@ -251,19 +250,10 @@ def has_permission(self, request, view):
             )
 
             # getting the user permission for that resource
-            resource_perms = res.get_user_perms(request.user)
-
-            groups = get_groups_with_perms(res, attach_perms=True)
-            # we are making this because the request.user.groups sometimes returns empty si is not fully reliable
-            for group, perm in groups.items():
-                # checking if the user is in that group
-                if group.user_set.filter(username=request.user).exists():
-                    resource_perms = list(chain(resource_perms, perm))
+            available_perms = permissions_registry.get_perms(instance=res, user=request.user)
 
             if request.user.has_perm("base.add_resourcebase"):
-                resource_perms.append("add_resourcebase")
-            # merging all available permissions into a single list
-            available_perms = list(set(resource_perms))
+                available_perms.append("add_resourcebase")
             # fixup the permissions name
             perms_without_base = [x.replace("base.", "") for x in perms]
             # if at least one of the permissions is available the request is True

geonode/base/api/serializers.py

@@ -69,6 +69,7 @@
 from geonode.security.utils import get_resources_with_perms, get_geoapp_subtypes
 from geonode.resource.models import ExecutionRequest
 from django.contrib.gis.geos import Polygon
+from geonode.security.registry import permissions_registry
 
 logger = logging.getLogger(__name__)
 
@@ -523,7 +524,11 @@ class Meta:
     def to_representation(self, instance):
         request = self.context.get("request", None)
         resource = ResourceBase.objects.get(pk=instance)
-        return resource.get_user_perms(request.user) if request and request.user and resource else []
+        return (
+            permissions_registry.get_perms(instance=resource, user=request.user)
+            if request and request.user and resource
+            else []
+        )
 
 
 class LinksSerializer(DynamicModelSerializer):

geonode/base/api/tests.py

@@ -93,6 +93,7 @@
 )
 from geonode.resource.api.tasks import resouce_service_dispatcher
 from guardian.shortcuts import assign_perm
+from geonode.security.registry import permissions_registry
 
 logger = logging.getLogger(__name__)
 
@@ -2373,7 +2374,11 @@ def test_resource_service_copy_with_perms_dataset_set_default_perms(self):
             }
             resource.set_permissions(_perms)
             # checking that bobby is in the original dataset perms list
-            self.assertTrue("bobby" in "bobby" in [x.username for x in resource.get_all_level_info().get("users", [])])
+            self.assertTrue(
+                "bobby"
+                in "bobby"
+                in [x.username for x in permissions_registry.get_perms(instance=resource).get("users", [])]
+            )
             # copying the resource, should remove the perms for bobby
             # only the default perms should be available
             copy_url = reverse("importer_resource_copy", kwargs={"pk": resource.pk})
@@ -2388,8 +2393,14 @@ def test_resource_service_copy_with_perms_dataset_set_default_perms(self):
         self.assertEqual("finished", self.client.get(response.json().get("status_url")).json().get("status"))
         _resource = Dataset.objects.filter(title__icontains="test_copy_with_perms").last()
         self.assertIsNotNone(_resource)
-        self.assertNotIn("bobby", [x.username for x in _resource.get_all_level_info().get("users", [])])
-        self.assertIn("admin", [x.username for x in _resource.get_all_level_info().get("users", [])])
+        self.assertNotIn(
+            "bobby",
+            [x.username for x in permissions_registry.get_perms(instance=_resource).get("users", [])],
+        )
+        self.assertIn(
+            "admin",
+            [x.username for x in permissions_registry.get_perms(instance=_resource).get("users", [])],
+        )
 
     def test_resource_service_copy_with_perms_doc(self):
         files = os.path.join(gisdata.GOOD_DATA, "vector/single_point.shp")
@@ -3427,7 +3438,7 @@ def test_simple_resourcebase_can_be_created_by_resourcemanager(self):
             "groups": {anonymous_group: set(["view_resourcebase"])},
         }
 
-        actual_perms = resource.get_all_level_info().copy()
+        actual_perms = permissions_registry.get_perms(instance=resource).copy()
         self.assertIsNotNone(actual_perms)
         self.assertTrue(self.user in actual_perms["users"].keys())
         self.assertTrue(anonymous_group in actual_perms["groups"].keys())

geonode/base/views.py

@@ -72,6 +72,7 @@
 
 from geonode.base.forms import CategoryForm, TKeywordForm, ThesaurusAvailableForm
 from geonode.base.models import Thesaurus, TopicCategory
+from geonode.security.registry import permissions_registry
 
 from .forms import ResourceBaseForm
 
@@ -509,8 +510,7 @@
 
     # Call this first in order to be sure "perms_list" is correct
     permissions_json = _perms_info_json(resourcebase_obj)
-
-    perms_list = resourcebase_obj.get_user_perms(request.user)
+    perms_list = permissions_registry.get_perms(instance=resourcebase_obj, user=request.user)
 
     group = None
     if resourcebase_obj.group:

geonode/documents/tests.py

@@ -60,6 +60,7 @@
 from geonode.upload.api.exceptions import FileUploadLimitException
 
 from .forms import DocumentCreateForm
+from geonode.security.registry import permissions_registry
 
 
 TEST_GIF = os.path.join(os.path.dirname(__file__), "tests/data/img.gif")
@@ -401,8 +402,8 @@ def test_set_document_permissions(self):
         self.assertFalse(self.anonymous_user.has_perm("view_resourcebase", document.get_self_resource()))
 
         # Test that previous permissions for users other than ones specified in
-        # the perm_spec (and the document owner) were removed
-        current_perms = document.get_all_level_info()
+        # the perm_spec (and the document owner) were
+        current_perms = permissions_registry.get_perms(instance=document)
         self.assertEqual(len(current_perms["users"]), 1)
 
         # Test that the User permissions specified in the perm_spec were

geonode/geoapps/views.py

@@ -46,6 +46,7 @@
 from geonode.base.forms import CategoryForm, TKeywordForm, ThesaurusAvailableForm
 from geonode.base.models import Thesaurus, TopicCategory
 from geonode.utils import resolve_object
+from geonode.security.registry import permissions_registry
 
 from .forms import GeoAppForm
 
@@ -106,7 +107,7 @@
     # Call this first in order to be sure "perms_list" is correct
     permissions_json = _perms_info_json(geoapp_obj)
 
-    perms_list = geoapp_obj.get_user_perms(request.user)
+    perms_list = permissions_registry.get_perms(instance=geoapp_obj, user=request.user)
 
     group = None
     if geoapp_obj.group:

geonode/geoserver/security.py

@@ -37,6 +37,7 @@
 from geonode.geoserver.helpers import geofence, gf_utils, gs_catalog
 from geonode.groups.models import GroupProfile
 from geonode.utils import get_dataset_workspace
+from geonode.security.registry import permissions_registry
 
 
 logger = logging.getLogger(__name__)
@@ -349,8 +350,7 @@ def sync_resources_with_guardian(resource=None, force=False):
                 batch = AutoPriorityBatch(gf_utils.get_first_available_priority(), f"Sync resources {dataset}")
 
                 gf_utils.collect_delete_layer_rules(get_dataset_workspace(dataset), dataset.name, batch)
-
-                perm_spec = dataset.get_all_level_info()
+                perm_spec = permissions_registry.get_perms(instance=dataset)
                 # All the other users
                 if "users" in perm_spec:
                     for user, perms in perm_spec["users"].items():

geonode/geoserver/tests/integration.py

@@ -42,6 +42,7 @@
 from geonode.decorators import on_ogc_backend
 from geonode.base.models import TopicCategory, Link
 from geonode.geoserver.helpers import set_attributes_from_geoserver
+from geonode.security.registry import permissions_registry
 
 LOCAL_TIMEOUT = 300
 
@@ -351,4 +352,5 @@
             saved_dataset.delete()
 
     def get_user_resource_perms(self, instance, user):
-        return list(instance.get_user_perms(user).union(instance.get_self_resource().get_user_perms(user)))
+        return permissions_registry.get_perms(instance=instance, user=user)
+        # return list(instance.get_user_perms(user).union(instance.get_self_resource().get_user_perms(user)))

geonode/groups/tests.py

@@ -37,6 +37,7 @@
 from geonode.groups.models import GroupProfile, GroupMember, GroupCategory
 
 from geonode.base.populate_test_data import all_public, create_models, remove_models, create_single_dataset
+from geonode.security.registry import permissions_registry
 
 logger = logging.getLogger(__name__)
 
@@ -334,7 +335,7 @@ def test_perms_info(self):
         # Add test to test perms being sent to the front end.
         layer = Dataset.objects.first()
         layer.set_default_permissions()
-        perms_info = layer.get_all_level_info()
+        perms_info = permissions_registry.get_perms(instance=layer)
 
         # Ensure there is only one group 'anonymous' by default
         self.assertEqual(len(perms_info["groups"].keys()), 1)
@@ -695,7 +696,7 @@ def test_group_activity_pages_render(self):
         try:
             # Add test to test perms being sent to the front end.
             dataset.set_default_permissions()
-            perms_info = dataset.get_all_level_info()
+            perms_info = permissions_registry.get_perms(instance=dataset)
 
             # Ensure there is only one group 'anonymous' by default
             self.assertEqual(len(perms_info["groups"].keys()), 1)

geonode/layers/tests.py

@@ -76,6 +76,7 @@
 
 from geonode.base.populate_test_data import all_public, create_models, remove_models, create_single_dataset
 from geonode.layers.download_handler import DatasetDownloadHandler
+from geonode.security.registry import permissions_registry
 
 logger = logging.getLogger(__name__)
 
@@ -641,7 +642,7 @@ def test_assign_change_dataset_data_perm(self):
         layer = Dataset.objects.first()
         user = get_anonymous_user()
         layer.set_permissions({"users": {user.username: ["change_dataset_data"]}})
-        perms = layer.get_all_level_info()
+        perms = permissions_registry.get_perms(instance=layer)
         self.assertNotIn(user, perms["users"])
         self.assertNotIn(user.username, perms["users"])
 
@@ -736,13 +737,13 @@ def test_surrogate_escape_string(self):
     def test_assign_remove_permissions(self):
         # Assing
         layer = Dataset.objects.all().first()
-        perm_spec = layer.get_all_level_info()
+        perm_spec = permissions_registry.get_perms(instance=layer)
         self.assertNotIn(get_user_model().objects.get(username="norman"), perm_spec["users"])
 
         utils.set_datasets_permissions(
             "edit", resources_names=[layer.name], users_usernames=["norman"], delete_flag=False, verbose=True
         )
-        perm_spec = layer.get_all_level_info()
+        perm_spec = permissions_registry.get_perms(instance=layer)
         _c = 0
         if "users" in perm_spec:
             for _u in perm_spec["users"]:
@@ -755,7 +756,7 @@ def test_assign_remove_permissions(self):
         utils.set_datasets_permissions(
             "read", resources_names=[layer.name], users_usernames=["norman"], delete_flag=True, verbose=True
         )
-        perm_spec = layer.get_all_level_info()
+        perm_spec = permissions_registry.get_perms(instance=layer)
         _c = 0
         if "users" in perm_spec:
             for _u in perm_spec["users"]:
@@ -768,15 +769,15 @@ def test_assign_remove_permissions(self):
     def test_assign_remove_permissions_for_groups(self):
         # Assing
         layer = Dataset.objects.all().first()
-        perm_spec = layer.get_all_level_info()
+        perm_spec = permissions_registry.get_perms(instance=layer)
         group_profile = GroupProfile.objects.create(slug="group1", title="group1", access="public")
         self.assertNotIn(group_profile, perm_spec["groups"])
 
         # giving manage permissions to the group
         utils.set_datasets_permissions(
             "manage", resources_names=[layer.name], groups_names=["group1"], delete_flag=False, verbose=True
         )
-        perm_spec = layer.get_all_level_info()
+        perm_spec = permissions_registry.get_perms(instance=layer)
         expected = {
             "change_dataset_data",
             "change_dataset_style",
@@ -795,7 +796,7 @@ def test_assign_remove_permissions_for_groups(self):
         utils.set_datasets_permissions(
             "view", resources_names=[layer.name], groups_names=["group1"], delete_flag=False, verbose=True
         )
-        perm_spec = layer.get_all_level_info()
+        perm_spec = permissions_registry.get_perms(instance=layer)
         expected = {"view_resourcebase"}
         # checking the perms list
         self.assertSetEqual(expected, set(perm_spec["groups"][group_profile.group]))
@@ -804,7 +805,7 @@ def test_assign_remove_permissions_for_groups(self):
         utils.set_datasets_permissions(
             "view", resources_names=[layer.name], groups_names=["group1"], delete_flag=True, verbose=True
         )
-        perm_spec = layer.get_all_level_info()
+        perm_spec = permissions_registry.get_perms(instance=layer)
         # checking the perms list
         self.assertTrue(group_profile.group not in perm_spec["groups"])
 
@@ -1815,7 +1816,7 @@ def _create_arguments(self, perms_type, mode="set"):
     def _assert_perms(self, expected_perms, dataset, username, assertion=True):
         dataset.refresh_from_db()
 
-        perms = dataset.get_all_level_info()
+        perms = permissions_registry.get_perms(instance=dataset)
         if assertion:
             self.assertTrue(username in [user.username for user in perms["users"]])
             actual = set(

geonode/layers/utils.py

@@ -48,6 +48,7 @@
 from geonode.utils import check_ogc_backend
 from geonode import GeoNodeException, geoserver
 from geonode.layers.models import shp_exts, csv_exts, vec_exts, cov_exts, Dataset
+from geonode.security.registry import permissions_registry
 
 READ_PERMISSIONS = ["view_resourcebase"]
 WRITE_PERMISSIONS = ["change_dataset_data", "change_dataset_style", "change_resourcebase_metadata"]
@@ -357,7 +358,9 @@ def set_datasets_permissions(
             # for the selected resource
             resource = resource.first()
             # getting the actual permissions available for the dataset
-            original_perms = PermSpec(resource.get_all_level_info(), resource)
+            original_perms = PermSpec(
+                permissions_registry.get_perms(instance=resource, include_virtual=False), resource
+            )
             new_perms_payload = {"organizations": [], "users": [], "groups": []}
             # if the username is specified, we add them to the payload with the compact
             # perm value

geonode/layers/views.py

@@ -61,6 +61,7 @@
 from geonode.people.forms import ProfileForm
 from geonode.utils import check_ogc_backend, llbbox_to_mercator, resolve_object
 from geonode.geoserver.helpers import ogc_server_settings
+from geonode.security.registry import permissions_registry
 
 if check_ogc_backend(geoserver.BACKEND_PACKAGE):
     from geonode.geoserver.helpers import gs_catalog
@@ -646,7 +647,7 @@ def dataset_metadata_detail(request, layername, template="datasets/dataset_metad
     site_url = settings.SITEURL.rstrip("/") if settings.SITEURL.startswith("http") else settings.SITEURL
 
     register_event(request, "view_metadata", layer)
-    perms_list = layer.get_user_perms(request.user)
+    perms_list = permissions_registry.get_perms(instance=layer, user=request.user)
 
     return render(
         request,

geonode/people/tests.py

@@ -37,6 +37,7 @@
 
 from geonode.base.populate_test_data import all_public, create_models, remove_models
 from django.db.models import Q
+from geonode.security.registry import permissions_registry
 
 
 class PeopleAndProfileTests(GeoNodeBaseTestSupport):
@@ -133,7 +134,7 @@ def test_set_unset_user_dataset_permissions(self):
                 )
         for layer in self.layers:
             user = get_user_model().objects.first()
-            perm_spec = layer.get_all_level_info()
+            perm_spec = permissions_registry.get_perms(instance=layer)
             self.assertFalse(user in perm_spec["users"], f"{layer} - {user}")
 
     @override_settings(ASYNC_SIGNALS=False)
@@ -166,7 +167,7 @@ def test_set_unset_group_dataset_permissions(self):
                     verbose=True,
                 )
         for layer in self.layers:
-            perm_spec = layer.get_all_level_info()
+            perm_spec = permissions_registry.get_perms(instance=layer)
             self.assertTrue(self.groups[0] in perm_spec["groups"])
 
     @override_settings(ASYNC_SIGNALS=False)
@@ -214,7 +215,7 @@ def test_unset_group_dataset_perms(self):
                     verbose=True,
                 )
         for layer in self.layers:
-            perm_spec = layer.get_all_level_info()
+            perm_spec = permissions_registry.get_perms(instance=layer)
             self.assertTrue(user not in perm_spec["users"])
 
     def test_forgot_username(self):