[Fixes #13449] Reorganize permission registry (#13452)

* [Fixes #13449] Reorganize permission registry
* [Fixes #13449] move get_visible_resource inside permissions registry
* [Fixes #13449] move utils inside the permissions_registry

Author: mattiagiupponi

geonode/api/tests.py

@@ -951,7 +951,7 @@ def _create_thesauri(cls):
 
     @classmethod
     def _create_resources(self):
-        public_perm_spec = {"users": {"AnonymousUser": ["view_resourcebase"]}, "groups": []}
+        public_perm_spec = {"users": {"AnonymousUser": ["view_resourcebase"]}, "groups": {}}
 
         for x in range(20):
             d: ResourceBase = ResourceBase.objects.create(

geonode/facets/tests.py

@@ -151,7 +151,7 @@ def _create_keywords(cls):
 
     @classmethod
     def _create_resources(self):
-        public_perm_spec = {"users": {"AnonymousUser": ["view_resourcebase"]}, "groups": []}
+        public_perm_spec = {"users": {"AnonymousUser": ["view_resourcebase"]}, "groups": {}}
         for x in range(20):
             d: ResourceBase = ResourceBase.objects.create(
                 title=f"dataset_{x:02}",

geonode/groups/tests.py

@@ -347,7 +347,8 @@ def test_perms_info(self):
 
         perms_info = _perms_info_json(layer)
         # Ensure foo is in the perms_info output
-        self.assertCountEqual(json.loads(perms_info)["groups"], {"bar": ["view_resourcebase"]})
+        self.assertTrue("bar" in json.loads(perms_info)["groups"])
+        self.assertListEqual(json.loads(perms_info)["groups"]["bar"], ["view_resourcebase"])
 
     def test_resource_permissions(self):
         """
@@ -409,7 +410,8 @@ def test_resource_permissions(self):
                 permissions = json.loads(permissions)
 
             # Make sure the bar group now has write permissions
-            self.assertCountEqual(permissions["groups"], {"bar": ["change_resourcebase"]})
+            self.assertTrue("bar" in permissions["groups"])
+            self.assertListEqual(permissions["groups"]["bar"], ["change_resourcebase"])
 
             # Remove group permissions
             permissions = {"users": {"admin": ["change_resourcebase"]}}
@@ -435,7 +437,7 @@ def test_resource_permissions(self):
                 permissions = json.loads(permissions)
 
             # Assert the bar group no longer has permissions
-            self.assertCountEqual(permissions["groups"], {})
+            self.assertTrue("bar" not in permissions["groups"])
 
     def test_create_new_group(self):
         """

geonode/maps/tests.py

@@ -239,6 +239,7 @@ def test_map_view(self, thumbnail_mock):
             name="base:nic_admin",
             ows_url="http://localhost:8080/geoserver/wms",
         )
+        # map_created.set_default_permissions()
         resource_manager.set_permissions(None, instance=map_created, permissions=None, created=True)
         map_id = map_created.id
 

geonode/people/models.py

@@ -30,21 +30,19 @@
 from django.urls import reverse
 from django.contrib.sites.models import Site
 from django.utils.translation import gettext_lazy as _
-from django.contrib.auth.models import AbstractUser, Permission, UserManager
+from django.contrib.auth.models import AbstractUser, UserManager
 from django.contrib.auth.signals import user_logged_in, user_logged_out
 
 from taggit.managers import TaggableManager
 
 from geonode.base.enumerations import COUNTRIES
-from geonode.base.models import Configuration, ResourceBase
+from geonode.base.models import ResourceBase
 from geonode.groups.models import GroupProfile
-from geonode.security.permissions import PERMISSIONS, READ_ONLY_AFFECTED_PERMISSIONS
 
 from allauth.account.signals import user_signed_up
 from allauth.socialaccount.signals import social_account_added
 
-from geonode.security.utils import can_approve, can_feature, can_publish
-
+from geonode.security.registry import permissions_registry
 from .utils import format_address
 from .signals import (
     do_login,
@@ -204,29 +202,7 @@ def location(self):
 
     @property
     def perms(self):
-        perms = set()
-        if self.is_superuser or self.is_staff:
-            # return all permissions for admins
-            perms.update(PERMISSIONS.values())
-
-        user_groups = self.groups.values_list("name", flat=True)
-        group_perms = (
-            Permission.objects.filter(group__name__in=user_groups).distinct().values_list("codename", flat=True)
-        )
-        for p in group_perms:
-            if p in PERMISSIONS:
-                # return constant names defined by GeoNode
-                perms.add(PERMISSIONS[p])
-            else:
-                # add custom permissions
-                perms.add(p)
-
-        # check READ_ONLY mode
-        config = Configuration.load()
-        if config.read_only:
-            # exclude permissions affected by readonly
-            perms = [perm for perm in perms if perm not in READ_ONLY_AFFECTED_PERMISSIONS]
-        return list(perms)
+        return permissions_registry.get_db_perms_by_user(self)
 
     def save(self, *args, **kwargs):
         super().save(*args, **kwargs)
@@ -284,13 +260,13 @@ def can_change_resource_field(self, resource, field):
                 return self.can_feature(resource)
 
     def can_approve(self, resource):
-        return can_approve(self, resource)
+        return permissions_registry.user_can_approve(self, resource)
 
     def can_publish(self, resource):
-        return can_publish(self, resource)
+        return permissions_registry.user_can_publish(self, resource)
 
     def can_feature(self, resource):
-        return can_feature(self, resource)
+        return permissions_registry.user_can_feature(self, resource)
 
 
 def get_anonymous_user_instance(user_model):

geonode/security/handlers.py

@@ -18,8 +18,6 @@
 #########################################################################
 from abc import ABC
 
-from geonode.security.utils import AdvancedSecurityWorkflowManager
-
 
 class BasePermissionsHandler(ABC):
     """
@@ -55,6 +53,8 @@ class AdvancedWorkflowPermissionsHandler(BasePermissionsHandler):
 
     @staticmethod
     def fixup_perms(instance, perms_payload, include_virtual=True, *args, **kwargs):
+        from geonode.security.utils import AdvancedSecurityWorkflowManager
+
         # Fixup Advanced Workflow permissions
         return AdvancedSecurityWorkflowManager.get_permissions(
             instance.uuid,

geonode/security/models.py

@@ -47,8 +47,9 @@
     DATASET_EDIT_STYLE_PERMISSIONS,
 )
 
-from .utils import get_users_with_perms, get_user_obj_perms_model, skip_registered_members_common_group
+from .utils import get_users_with_perms, skip_registered_members_common_group
 from geonode.security.registry import permissions_registry
+from guardian.utils import get_user_obj_perms_model
 
 logger = logging.getLogger(__name__)
 
@@ -455,11 +456,4 @@ def user_can(self, user, permission):
         """
         Checks if a has a given permission to the resource.
         """
-        user_perms = permissions_registry.get_perms(instance=self, user=user)
-
-        if permission not in user_perms:
-            # TODO cater for permissions with syntax base.permission_codename
-            # eg 'base.change_resourcebase'
-            return False
-
-        return True
+        return permissions_registry.user_has_perm(user, self, permission)