merge with master

Author: mattiagiupponi

.env.sample

@@ -127,6 +127,7 @@ DEFAULT_FROM_EMAIL='{email}' # eg Company <[email protected]>
 # Session/Access Control
 LOCKDOWN_GEONODE=False
 X_FRAME_OPTIONS="SAMEORIGIN"
+SESSION_ENGINE=django.contrib.sessions.backends.cached_db
 SESSION_EXPIRED_CONTROL_ENABLED=True
 DEFAULT_ANONYMOUS_VIEW_PERMISSION=True
 DEFAULT_ANONYMOUS_DOWNLOAD_PERMISSION=True

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))
-
-            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 = permissions_registry.get_perms(
+                instance=res, user=request.user, include_user_add_resource=True
+            )
+
             # 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

@@ -94,6 +94,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__)
 
@@ -2374,7 +2375,9 @@ 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.assertIn(
+                "bobby", [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})
@@ -2389,8 +2392,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")
@@ -3428,7 +3437,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

@@ -58,6 +58,7 @@
 from geonode.base.auth import get_or_create_token
 from geonode.security.views import _perms_info_json
 
+from geonode.security.registry import permissions_registry
 
 logger = logging.getLogger(__name__)
 
@@ -415,8 +416,7 @@ def resourcebase_embed(request, resourcebaseid, template="base/base_edit.html"):
 
     # 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/context_processors.py

@@ -46,6 +46,7 @@ def resource_urls(request):
         STATIC_URL=settings.STATIC_URL,
         CATALOGUE_BASE_URL=default_catalogue_backend()["URL"],
         ACCOUNT_OPEN_SIGNUP=settings.ACCOUNT_OPEN_SIGNUP,
+        ACCOUNT_OPEN_SOCIALSIGNUP=settings.ACCOUNT_OPEN_SOCIALSIGNUP,
         ACCOUNT_APPROVAL_REQUIRED=settings.ACCOUNT_APPROVAL_REQUIRED,
         VERSION=get_version(),
         SITE_NAME=site.name,

geonode/documents/forms.py

@@ -129,7 +129,7 @@ def clean(self):
             raise forms.ValidationError(_("A document cannot have both a file and a url."))
 
         if extension:
-            cleaned_data["extension"] = extension.replace(".", "")
+            cleaned_data["extension"] = extension.replace(".", "").lower()
 
         return cleaned_data
 

geonode/documents/tests.py

@@ -32,6 +32,7 @@
 
 from unittest.mock import patch
 from urllib.parse import urlparse
+from pathlib import Path
 
 from django.urls import reverse
 from django.conf import settings
@@ -41,7 +42,7 @@
 
 from guardian.shortcuts import get_anonymous_user
 
-from geonode.assets.utils import create_asset_and_link
+from geonode.assets.utils import create_asset_and_link, get_default_asset
 from geonode.maps.models import Map
 from geonode.compat import ensure_string
 from geonode.base.enumerations import SOURCE_TYPE_REMOTE
@@ -56,6 +57,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")
@@ -189,6 +191,25 @@ def test_create_document_url_view(self):
         d = Document.objects.get(title="GeoNode Map")
         self.assertEqual(d.doc_url, "http://www.geonode.org/map.pdf")
 
+    def test_uploaded_csv_with_uppercase_extension(self):
+        """
+        The extension of the file should always be lowercase
+        """
+
+        self.client.login(username="admin", password="admin")
+        try:
+            with open(os.path.join(os.path.dirname(__file__), "tests/data/test.CSV"), "rb") as f:
+                data = {"title": "CSV with uppercase extension", "doc_file": f, "extension": "CSV"}
+                self.client.post(reverse("document_upload"), data=data)
+            d = Document.objects.get(title="CSV with uppercase extension")
+            # verify that the extension is not lowercase
+            self.assertEqual(d.extension, "csv")
+            # be sure that also the file extension is not lowercase
+            asset = get_default_asset(d)
+            self.assertEqual(Path(asset.location[0]).suffix, ".csv")
+        finally:
+            Document.objects.filter(title="CSV with uppercase extension").delete()
+
     def test_upload_document_form(self):
         """
         Tests the Upload form.
@@ -397,8 +418,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/documents/tests/data/test.CSV

@@ -0,0 +1,6 @@
+Username; Identifier;First name;Last name
+booker12;9012;Rachel;Booker
+grey07;2070;Laura;Grey
+johnson81;4081;Craig;Johnson
+jenkins46;9346;Mary;Jenkins
+smith79;5079;Jamie;Smith

geonode/documents/views.py

@@ -41,6 +41,8 @@
 from geonode.resource.manager import resource_manager
 from geonode.base import enumerations
 
+from pathlib import Path
+
 from .utils import get_download_response
 
 from .models import Document
@@ -137,7 +139,8 @@ def form_valid(self, form):
         if file:
             tempdir = mkdtemp()
             dirname = os.path.basename(tempdir)
-            filepath = storage_manager.save(f"{dirname}/{file.name}", file)
+            name = Path(file.name)
+            filepath = storage_manager.save(f"{dirname}/{name.stem}{name.suffix.lower()}", file)
             storage_path = storage_manager.path(filepath)
             self.object = resource_manager.create(
                 None,