[Fixes #14481] Enhancement of Resource cloning logic

geonode/base/api/serializers.py

@@ -647,7 +647,7 @@ class ResourceBaseSerializer(MultiLangOutputMixin, DynamicModelSerializer):
     license = ComplexDynamicRelationField(LicenseSerializer, embed=True)
     spatial_representation_type = ComplexDynamicRelationField(SpatialRepresentationTypeSerializer, embed=True)
     blob = serializers.JSONField(required=False, write_only=True)
-    is_copyable = serializers.BooleanField(read_only=True)
+    is_copyable = serializers.SerializerMethodField(read_only=True)
     download_url = DownloadLinkField(read_only=True)
     favorite = FavoriteField(read_only=True)
     download_urls = DownloadArrayLinkField(read_only=True)
@@ -791,6 +791,18 @@ def get_perms(self, instance):
         )
         return permissions
 
+    def get_is_copyable(self, instance):
+        from geonode.resource.registry import resource_manager_registry
+
+        request = self.context.get("request")
+        try:
+            return resource_manager_registry.get_for_instance(instance).is_copyable(
+                instance, user=request.user if request else None
+            )
+        except Exception as e:
+            logger.warning(f"Cannot evaluate is_copyable for resource {instance.pk}: {e}")
+            return False
+
     def save(self, **kwargs):
         extent = self.validated_data.pop("extent", None)
         instance = super().save(**kwargs)

geonode/base/api/tests.py

@@ -2868,28 +2868,63 @@ def test_resource_service_copy(self):
         second_cloned = Dataset.objects.exclude(pk__in=[resource.pk, cloned_resource.pk]).latest("id")
         self.assertFalse(second_cloned.featured, msg="Cloned resource should have featured=False")
 
-        # clone dataset with invalid file
-        # resource.files = ["/path/invalid_file.wrong"]
-        # resource.save()
-        asset.location = ["/path/invalid_file.wrong"]
-        asset.save()
-        response = self.client.put(copy_url)
-
-        self.assertEqual(response.status_code, 400)
-        self.assertEqual(response.json()["message"], "Resource can not be cloned.")
-        # clone dataset with no files
-        link.delete()
-        asset.delete()
-        response = self.client.put(copy_url)
-
-        self.assertEqual(response.status_code, 400)
-        self.assertEqual(response.json()["message"], "Resource can not be cloned.")
-        # clean
+        # the file/asset based checks do not apply to non remote vector datasets anymore,
+        # as the data can also be only stored on db level and cloning takes care of cloning db table.
         try:
             resource.delete()
         except Exception as e:
             logger.warning(f"Can't delete test resource {resource}", exc_info=e)
 
+    def test_is_copyable_document(self):
+        owner = get_user_model().objects.get(username="admin")
+        raster_file = os.path.join(gisdata.GOOD_DATA, "raster", "relief_san_andres.tif")
+        resource = Document.objects.create(
+            owner=get_user_model().objects.get(username="admin"),
+            subtype="image",
+            alternate="geonode:test_copy",
+            resource_type="document",
+            uuid=str(uuid4()),
+        )
+        try:
+            asset, _ = create_asset_and_link(resource, owner, [raster_file])
+            self.assertTrue(resource_manager_registry.get_for_instance(resource).is_copyable(resource, user=owner))
+
+            asset.delete()
+            self.assertFalse(resource_manager_registry.get_for_instance(resource).is_copyable(resource, user=owner))
+        finally:
+            try:
+                resource.delete()
+            except Exception as e:
+                logger.warning(f"Can't delete test resource {resource}", exc_info=e)
+
+    def test_is_copyable_raster_dataset(self):
+        owner = get_user_model().objects.get(username="admin")
+        raster_file = os.path.join(gisdata.GOOD_DATA, "raster", "relief_san_andres.tif")
+        resource = Dataset.objects.create(
+            owner=owner,
+            name="test_raster_copy",
+            store="geonode_data",
+            subtype="raster",
+            alternate="geonode:test_raster_copy",
+            resource_type="dataset",
+            uuid=str(uuid4()),
+        )
+        try:
+            asset, _ = create_asset_and_link(resource, owner, [raster_file], title="Original")
+            self.assertTrue(resource_manager_registry.get_for_instance(resource).is_copyable(resource, user=owner))
+
+            asset.title = "not_the_original"
+            asset.save()
+            self.assertFalse(resource_manager_registry.get_for_instance(resource).is_copyable(resource, user=owner))
+
+            asset.delete()
+            self.assertFalse(resource_manager_registry.get_for_instance(resource).is_copyable(resource, user=owner))
+        finally:
+            try:
+                resource.delete()
+            except Exception as e:
+                logger.warning(f"Can't delete test resource {resource}", exc_info=e)
+
     def test_resource_service_copy_with_perms_dataset(self):
         files = os.path.join(gisdata.GOOD_DATA, "vector/single_point.shp")
         files_as_dict, _ = get_files(files)
@@ -2907,6 +2942,47 @@ def test_resource_service_copy_with_perms_dataset(self):
         )
         self._assertCloningWithPerms(resource)
 
+    def test_is_copyable_remote_dataset(self):
+        from geonode.base import enumerations
+        from geonode.security.models import AuthConfig
+
+        owner = get_user_model().objects.get(username="bobby")
+        other = get_user_model().objects.get(username="norman")
+        admin = get_user_model().objects.get(username="admin")
+
+        resource = Dataset.objects.create(
+            owner=owner,
+            name="test_remote_copy",
+            store="geonode_data",
+            subtype="remote",
+            alternate="geonode:test_remote_copy",
+            resource_type="dataset",
+            sourcetype=enumerations.SOURCE_TYPE_REMOTE,
+            uuid=str(uuid4()),
+        )
+        try:
+            # no AuthConfig: owner and administrators may clone, nobody else
+            self.assertTrue(resource_manager_registry.get_for_instance(resource).is_copyable(resource, user=owner))
+            self.assertFalse(resource_manager_registry.get_for_instance(resource).is_copyable(resource, user=other))
+            self.assertTrue(resource_manager_registry.get_for_instance(resource).is_copyable(resource, user=admin))
+
+            # an AuthConfig restricts cloning to the owner, admins included
+            resource.auth_config = AuthConfig.objects.create(type="basic")
+            resource.save()
+            resource.refresh_from_db()
+            self.assertTrue(resource_manager_registry.get_for_instance(resource).is_copyable(resource, user=owner))
+            self.assertFalse(resource_manager_registry.get_for_instance(resource).is_copyable(resource, user=other))
+            self.assertFalse(resource_manager_registry.get_for_instance(resource).is_copyable(resource, user=admin))
+
+            # attaching an arbitrary file must not make it copyable
+            create_asset_and_link(resource, other, [os.path.join(gisdata.GOOD_DATA, "vector/single_point.shp")])
+            self.assertFalse(resource_manager_registry.get_for_instance(resource).is_copyable(resource, user=other))
+        finally:
+            try:
+                resource.delete()
+            except Exception as e:
+                logger.warning(f"Can't delete test resource {resource}", exc_info=e)
+
     @patch.dict(os.environ, {"ASYNC_SIGNALS": "False"})
     @override_settings(ASYNC_SIGNALS=False)
     def test_resource_service_copy_with_perms_dataset_set_default_perms(self):

geonode/base/api/views.py

@@ -1127,7 +1127,7 @@ def resource_service_copy(self, request, pk, *args, **kwargs):
             or not request.user.has_perm("view_resourcebase", resource.get_self_resource())
         ):
             return Response(status=status.HTTP_403_FORBIDDEN)
-        if not resource.is_copyable:
+        if not resource_manager_registry.get_for_instance(resource).is_copyable(resource, user=request.user):
             return Response({"message": "Resource can not be cloned."}, status=400)
         try:
             request_params = self._get_request_params(request)

geonode/base/models.py

@@ -73,7 +73,6 @@
     find_by_attr,
     bbox_to_projection,
     bbox_swap,
-    get_allowed_extensions,
     is_monochromatic_image,
 )
 from geonode.thumbs.utils import thumb_size, remove_thumbs, get_unique_upload_path, ThumbnailAlgorithms
@@ -1309,17 +1308,6 @@ def instance_is_processed(self):
         except Exception:
             return False
 
-    @property
-    def is_copyable(self):
-        if self.resource_type == "dataset":
-            from geonode.assets.utils import get_default_asset
-            from geonode.geoserver.helpers import select_relevant_files
-
-            asset = get_default_asset(self)  # TODO: maybe we need to filter by original files
-            allowed_file = select_relevant_files(get_allowed_extensions(), asset.location) if asset else []
-            return len(allowed_file) != 0
-        return True
-
     def keyword_list(self):
         return [kw.name for kw in self.keywords.all()]
 

geonode/geoapps/manager.py

@@ -76,3 +76,6 @@ def update(self, uuid: str, /, instance: ResourceBase = None, vals: dict = {}, *
         if payload:
             return self._create_and_update(payload, instance=instance, notify=notify, request_user=request_user)
         return super().update(uuid, instance=instance, vals=vals, notify=notify, **kwargs)
+
+    def _is_local_resource_copyable(self, instance: GeoApp, user=None) -> bool:
+        return True

geonode/layers/manager.py

@@ -28,3 +28,10 @@ class DatasetResourceManager(BaseResourceManager):
     """
 
     handled_model = Dataset
+
+    def _is_local_resource_copyable(self, instance: Dataset, user=None) -> bool:
+        if instance.subtype in ("vector", "vector_time"):
+            return True
+        if instance.subtype == "raster":
+            return self._has_asset(instance, original_title=True)
+        return super()._is_local_resource_copyable(instance, user=user)

geonode/maps/manager.py

@@ -101,3 +101,6 @@ def update(self, uuid: str, /, instance: ResourceBase = None, vals: dict = {}, *
             notify=notify,
         )
         return instance
+
+    def _is_local_resource_copyable(self, instance: Map, user=None) -> bool:
+        return True

geonode/resource/manager.py

@@ -986,6 +986,36 @@ def set_thumbnail(
                 logger.exception(e)
         return False
 
+    def _has_asset(self, instance: ResourceBase, original_title=False) -> bool:
+        from geonode.assets.models import Asset
+
+        filters = {"link__resource": instance}
+        if original_title:
+            filters["title"] = "Original"
+
+        return Asset.objects.filter(**filters).exists()
+
+    def _is_local_resource_copyable(self, instance: ResourceBase, user=None) -> bool:
+        return self._has_asset(instance)
+
+    def is_copyable(self, instance: ResourceBase, /, user=None) -> bool:
+        """
+        Whether ``user`` is allowed to clone ``instance``.
+        """
+        _resource = instance.get_real_instance()
+        if _resource.sourcetype == enumerations.SOURCE_TYPE_REMOTE:
+            if user is None:
+                return False
+            if user == _resource.owner:
+                return True
+            remote_service = getattr(_resource, "remote_service", None)
+            has_auth_config = _resource.auth_config_id or (remote_service and remote_service.auth_config_id)
+            if not has_auth_config and user.is_superuser:
+                return True
+            else:
+                return False
+        return self._is_local_resource_copyable(_resource, user=user)
+
     def _apply_extent_and_role_defaults(
         self, instance: ResourceBase, extent: dict = None, user: settings.AUTH_USER_MODEL = None
     ) -> None:

geonode/upload/api/views.py

@@ -249,8 +249,12 @@ class ResourceImporter(DynamicModelViewSet):
     pagination_class = GeoNodeApiPagination
 
     def copy(self, request, *args, **kwargs):
+        from geonode.resource.registry import resource_manager_registry
+
         try:
             resource = self.get_object()
+            if not resource_manager_registry.get_for_instance(resource).is_copyable(resource, user=request.user):
+                return Response({"message": "Resource can not be cloned."}, status=400)
             if resource.resourcehandlerinfo_set.exists():
                 handler_module_path = resource.resourcehandlerinfo_set.first().handler_module_path