Unify permission checks

src/georama/core/common/managers.py

@@ -1,5 +1,6 @@
 from django.core.exceptions import ImproperlyConfigured
 from django.db import models
+from guardian.shortcuts import get_objects_for_user
 
 from georama.core.common.querysets import OrganisationalQuerySet
 from georama.core.models.organisation import Organisation
@@ -53,3 +54,44 @@ def validate_organisational(model: models.Model):
                 "An OrganisationalModelAdmin has to be configured with models bound to"
                 "an Organisational georama.core.common.managers.OrganisationalManager"
             )
+
+
+class LayerManager(OrganisationalManager):
+    def get_public_or_permitted(
+        self, organisation: Organisation, user, perms
+    ) -> OrganisationalQuerySet:
+        qs = self.get_queryset().organisation_objects(organisation)
+        return get_objects_for_user(user, perms, qs) | qs.filter(public=True)
+
+    def get_permitted(self, organisation: Organisation, user, perms) -> OrganisationalQuerySet:
+        qs = self.get_queryset().organisation_objects(organisation)
+        return get_objects_for_user(user, perms, qs)
+
+    def accessible_layers(
+        self,
+        organisation: Organisation,
+        user,
+        perms: list[str],
+        layer_ids: list[str] | None = None,
+        include_public: bool = False,
+    ) -> OrganisationalQuerySet:
+        perm_qs = self.get_public_or_permitted if include_public else self.get_permitted
+        if layer_ids is None:
+            # most notably this is the case on capability requests
+            return perm_qs(organisation, user, perms)
+
+        # first check is about the layer names (raising if missmatch is found)
+        qs = self.get_queryset().organisation_objects(organisation).filter(id__in=layer_ids)
+        found_difference = set(layer_ids) - {layer.identifier for layer in qs}
+        if len(found_difference) > 0:
+            raise qs.model.DoesNotExist(f"Layer(s) not found: {list(found_difference)}")
+
+        # continue with the available list checking for permissions
+        accessible_layers = {}
+        qs = perm_qs(organisation, user, perms).filter(id__in=layer_ids)
+        for layer in qs:
+            accessible_layers[layer.identifier] = layer
+        permission_difference = set(layer_ids) - set(accessible_layers)
+        if len(permission_difference) > 0:
+            raise PermissionError(f"Layer(s) not permitted: {list(permission_difference)}")
+        return qs

src/georama/features/factories.py

@@ -23,7 +23,6 @@ class Meta:
 class FeatureLayerFactory(factory.django.DjangoModelFactory):
     class Meta:
         model = FeatureLayer
-        django_get_or_create = ("datasource",)
 
     public = factory.Iterator([True, False])
     metadata = factory.SubFactory(MetadataFactory)

src/georama/features/managers/feature_layer.py

@@ -1,10 +1,10 @@
 from django.db import models, transaction
 
-from georama.core.common.managers import OrganisationalManager
+from georama.core.common.managers import LayerManager
 from georama.features.models.field import Field
 
 
-class FeatureLayerManager(OrganisationalManager):
+class FeatureLayerManager(LayerManager):
     def get_queryset(self) -> models.QuerySet:
         """Always prefetch bound fields to reduce queries.
 

src/georama/features/models/feature_layer.py

@@ -121,6 +121,10 @@ def datasource_related_fields(self):
                 name=datasource_field.name,
             )
 
+    @property
+    def identifier(self):
+        return str(self.id)
+
     @property
     def title(self):
         return self.metadata.title

src/georama/features/views/pygeoapi.py

@@ -6,7 +6,7 @@
 from django.http import Http404, HttpRequest, HttpResponse
 from django.urls import reverse
 from django.views import View
-from guardian.shortcuts import get_objects_for_user, get_perms
+from guardian.shortcuts import get_perms
 from pygeoapi import l10n
 from pygeoapi.api import API, APIRequest, apply_gzip
 from pygeoapi.openapi import get_oas
@@ -66,11 +66,21 @@ def execute_from_django(
             response[key] = value
         return response
 
-    def get_collection_or_404(self, collection_id: str):
+    def get_collection_or_404_or_403(
+        self, collection_id: str, perm: str, include_public: bool = False
+    ):
         try:
-            return self.model.objects.get(id=collection_id)
+            return self.model.objects.accessible_layers(
+                self.request.georama_organisation,
+                self.request.user,
+                [perm],
+                [collection_id],
+                include_public,
+            )
         except self.model.DoesNotExist as e:
-            raise Http404("Collection not found") from e
+            raise Http404(e) from e
+        except PermissionError as e:
+            raise PermissionDenied(e) from e
 
     def landing(self, request: HttpRequest) -> HttpResponse:
         """
@@ -124,11 +134,10 @@ def collection_schema(
 
         :returns: Django HTTP Response
         """
-        feature_layer = self.get_collection_or_404(collection_id)
-        if "view_featurelayer" in get_perms(request.user, feature_layer):
-            return self.execute_from_django(core_api.get_collection_schema, request, collection_id)
-        else:
-            raise PermissionDenied()
+        _ = self.get_collection_or_404_or_403(
+            collection_id, "view_objects_on_published_layer", include_public=True
+        )
+        return self.execute_from_django(core_api.get_collection_schema, request, collection_id)
 
     def collection_queryables(
         self, request: HttpRequest, collection_id: str | None = None
@@ -141,13 +150,12 @@ def collection_queryables(
 
         :returns: Django HTTP Response
         """
-        feature_layer = self.get_collection_or_404(collection_id)
-        if "view_featurelayer" in get_perms(request.user, feature_layer):
-            return self.execute_from_django(
-                itemtypes_api.get_collection_queryables, request, collection_id
-            )
-        else:
-            raise PermissionDenied()
+        _ = self.get_collection_or_404_or_403(
+            collection_id, "view_objects_on_published_layer", include_public=True
+        )
+        return self.execute_from_django(
+            itemtypes_api.get_collection_queryables, request, collection_id
+        )
 
     def collection_items(self, request: HttpRequest, collection_id: str) -> HttpResponse:
         """
@@ -159,49 +167,50 @@ def collection_items(self, request: HttpRequest, collection_id: str) -> HttpResp
         :returns: Django HTTP response
         """
 
-        feature_layer = self.get_collection_or_404(collection_id)
-
         if request.method == "GET":
-            if "view_featurelayer" in get_perms(request.user, feature_layer):
-                response_ = self.execute_from_django(
-                    itemtypes_api.get_collection_items,
-                    request,
-                    collection_id,
-                    skip_valid_check=True,
-                )
-            else:
-                raise PermissionDenied()
+            _ = self.get_collection_or_404_or_403(
+                collection_id, "view_objects_on_published_layer", include_public=True
+            )
+            response_ = self.execute_from_django(
+                itemtypes_api.get_collection_items,
+                request,
+                collection_id,
+                skip_valid_check=True,
+            )
         elif request.method == "POST":
-            if "add_featurelayer" in get_perms(request.user, feature_layer):
-                if request.content_type is not None:
-                    if request.content_type == "application/geo+json":
-                        response_ = self.execute_from_django(
-                            itemtypes_api.manage_collection_item,
-                            request,
-                            "create",
-                            collection_id,
-                            skip_valid_check=True,
-                        )
-                    else:
-                        response_ = self.execute_from_django(
-                            itemtypes_api.post_collection_items,
-                            request,
-                            collection_id,
-                            skip_valid_check=True,
-                        )
+            _ = self.get_collection_or_404_or_403(
+                collection_id, "create_objects_on_published_layer"
+            )
+            if request.content_type is not None:
+                if request.content_type == "application/geo+json":
+                    response_ = self.execute_from_django(
+                        itemtypes_api.manage_collection_item,
+                        request,
+                        "create",
+                        collection_id,
+                        skip_valid_check=True,
+                    )
+                else:
+                    response_ = self.execute_from_django(
+                        itemtypes_api.manage_collection_item,
+                        request,
+                        "create",
+                        collection_id,
+                        skip_valid_check=True,
+                    )
             else:
-                raise PermissionDenied()
+                raise BadRequest()
         elif request.method == "OPTIONS":
-            if "view_featurelayer" in get_perms(request.user, feature_layer):
-                response_ = self.execute_from_django(
-                    itemtypes_api.manage_collection_item,
-                    request,
-                    "options",
-                    collection_id,
-                    skip_valid_check=True,
-                )
-            else:
-                raise PermissionDenied()
+            _ = self.get_collection_or_404_or_403(
+                collection_id, "view_objects_on_published_layer", include_public=True
+            )
+            response_ = self.execute_from_django(
+                itemtypes_api.manage_collection_item,
+                request,
+                "options",
+                collection_id,
+                skip_valid_check=True,
+            )
         else:
             raise BadRequest()
 
@@ -219,50 +228,50 @@ def collection_item(
 
         :returns: Django HTTP response
         """
-        feature_layer = self.get_collection_or_404(collection_id)
+
         if request.method == "GET":
-            if "view_featurelayer" in get_perms(request.user, feature_layer):
-                response_ = self.execute_from_django(
-                    itemtypes_api.get_collection_item, request, collection_id, item_id
-                )
-            else:
-                raise PermissionDenied()
+            _ = self.get_collection_or_404_or_403(
+                collection_id, "view_objects_on_published_layer", include_public=True
+            )
+            response_ = self.execute_from_django(
+                itemtypes_api.get_collection_item, request, collection_id, item_id
+            )
         elif request.method == "PUT":
-            if "change_featurelayer" in get_perms(request.user, feature_layer):
-                response_ = self.execute_from_django(
-                    itemtypes_api.manage_collection_item,
-                    request,
-                    "update",
-                    collection_id,
-                    item_id,
-                    skip_valid_check=True,
-                )
-            else:
-                raise PermissionDenied()
+            _ = self.get_collection_or_404_or_403(
+                collection_id, "update_objects_on_published_layer"
+            )
+            response_ = self.execute_from_django(
+                itemtypes_api.manage_collection_item,
+                request,
+                "update",
+                collection_id,
+                item_id,
+                skip_valid_check=True,
+            )
         elif request.method == "DELETE":
-            if "delete_featurelayer" in get_perms(request.user, feature_layer):
-                response_ = self.execute_from_django(
-                    itemtypes_api.manage_collection_item,
-                    request,
-                    "delete",
-                    collection_id,
-                    item_id,
-                    skip_valid_check=True,
-                )
-            else:
-                raise PermissionDenied()
+            _ = self.get_collection_or_404_or_403(
+                collection_id, "delete_objects_on_published_layer"
+            )
+            response_ = self.execute_from_django(
+                itemtypes_api.manage_collection_item,
+                request,
+                "delete",
+                collection_id,
+                item_id,
+                skip_valid_check=True,
+            )
         elif request.method == "OPTIONS":
-            if "view_featurelayer" in get_perms(request.user, feature_layer):
-                response_ = self.execute_from_django(
-                    itemtypes_api.manage_collection_item,
-                    request,
-                    "options",
-                    collection_id,
-                    item_id,
-                    skip_valid_check=True,
-                )
-            else:
-                raise PermissionDenied()
+            _ = self.get_collection_or_404_or_403(
+                collection_id, "view_objects_on_published_layer", include_public=True
+            )
+            response_ = self.execute_from_django(
+                itemtypes_api.manage_collection_item,
+                request,
+                "options",
+                collection_id,
+                item_id,
+                skip_valid_check=True,
+            )
         else:
             raise BadRequest()
 
@@ -273,8 +282,13 @@ def handle_runtime_config(self, request: HttpRequest) -> tuple[dict, dict]:
         server_config["server"]["url"] = (
             f"{request.scheme}://{request.get_host()}{reverse('features:landing')}"
         )
-        for feature_layer in get_objects_for_user(request.user, ["view_featurelayer"], self.model):
-            server_config["resources"][str(feature_layer.id)] = self.create_resource(
+        for feature_layer in self.model.objects.accessible_layers(
+            self.request.georama_organisation,
+            request.user,
+            ["view_objects_on_published_layer"],
+            include_public=True,
+        ):
+            server_config["resources"][feature_layer.identifier] = self.create_resource(
                 feature_layer, request
             )
         return server_config, get_oas(server_config)
@@ -377,7 +391,11 @@ def create_postgres_provider(
     def create_resource(self, feature_layer: FeatureLayer, request: HttpRequest) -> dict:
         editable = any(
             p in get_perms(request.user, feature_layer)
-            for p in ("add_featurelayer", "change_featurelayer", "delete_featurelayer")
+            for p in (
+                "create_objects_on_published_layer",
+                "update_objects_on_published_layer",
+                "delete_objects_on_published_layer",
+            )
         )
 
         features_properties = [p for p in feature_layer.fields.all() if p.visible]