[Fixes #14152] Restrict the creation of remote resources to administrators only by default

geonode/documents/tests.py

@@ -35,6 +35,7 @@
 
 from django.urls import reverse
 from django.conf import settings
+from django.test import override_settings
 from django.contrib.auth import get_user_model
 from django.core.files.uploadedfile import SimpleUploadedFile
 from django.template.defaultfilters import filesizeformat
@@ -156,6 +157,45 @@ def test_download_is_ajax_safe(self):
         d = create_single_doc("example_doc_name")
         self.assertTrue(d.download_is_ajax_safe)
 
+    @override_settings(REGISTERED_USERS_CAN_ADD_REMOTE_RESOURCES=False)
+    def test_remote_document_add_allowed_for_admin(self):
+        self.assertTrue(self.client.login(username="admin", password="admin"))
+        title = "Remote doc allowed for admin user"
+        form_data = {
+            "title": title,
+            "doc_url": "http://www.geonode.org/map.pdf",
+        }
+
+        response = self.client.post(reverse("document_upload"), data=form_data)
+        self.assertEqual(response.status_code, 302)
+
+    @override_settings(REGISTERED_USERS_CAN_ADD_REMOTE_RESOURCES=False)
+    def test_remote_document_add_forbidden_for_regular_user(self):
+        self.assertTrue(self.client.login(username="bobby", password="bob"))
+        title = "Remote doc denied for regular user"
+        form_data = {
+            "title": title,
+            "doc_url": "http://www.geonode.org/map.pdf",
+        }
+
+        response = self.client.post(reverse("document_upload"), data=form_data)
+        self.assertEqual(response.status_code, 403)
+        self.assertFalse(Document.objects.filter(title=title).exists())
+
+    @override_settings(REGISTERED_USERS_CAN_ADD_REMOTE_RESOURCES=True)
+    def test_remote_document_add_allowed_for_regular_user_when_enabled(self):
+        self.assertTrue(self.client.login(username="bobby", password="bob"))
+        title = "Remote doc allowed for regular user"
+
+        form_data = {
+            "title": title,
+            "doc_url": "https://raw.githubusercontent.com/GeoNode/geonode/master/geonode/documents/tests/data/test.CSV",
+        }
+
+        response = self.client.post(reverse("document_upload"), data=form_data)
+        self.assertEqual(response.status_code, 302)
+        self.assertTrue(Document.objects.filter(title=title).exists())
+
     def test_create_document_url(self):
         """Tests creating an external document instead of a file."""
 

geonode/documents/views.py

@@ -29,6 +29,7 @@
 from django.views.generic.edit import CreateView
 from django.http import HttpResponse, HttpResponseRedirect
 
+from geonode.security.utils import check_add_remote_resource_perm
 from geonode.base.api.exceptions import geonode_exception_handler
 from geonode.client.hooks import hookset
 from geonode.utils import mkdtemp
@@ -162,6 +163,8 @@ def form_valid(self, form):
             shutil.rmtree(tempdir, ignore_errors=True)
 
         else:
+            check_add_remote_resource_perm(self.request.user)
+
             self.object = document_manager.create(
                 None,
                 resource_type=Document,

geonode/security/registry.py

@@ -162,6 +162,10 @@ def get_db_perms_by_user(self, user):
                 # add custom permissions
                 perms.add(p)
 
+        # Create a synthetic permission for adding remote resources
+        if user.is_superuser or user.is_staff or getattr(settings, "REGISTERED_USERS_CAN_ADD_REMOTE_RESOURCES", False):
+            perms.add("add_remote_resource")
+
         # check READ_ONLY mode
         config = Configuration.load()
         if config.read_only:

geonode/security/tests.py

@@ -629,6 +629,26 @@ def test_special_groups_flags_per_setting_independence(self):
         finally:
             staff_user.delete()
 
+    def test_add_remote_resource_perm_admin_always_has_it(self):
+        """Superusers always have the add_remote_resource permission."""
+        admin = get_user_model().objects.get(username="admin")
+        perms = permissions_registry.get_db_perms_by_user(admin)
+        self.assertIn("add_remote_resource", perms)
+
+    @override_settings(REGISTERED_USERS_CAN_ADD_REMOTE_RESOURCES=False)
+    def test_add_remote_resource_perm_regular_user_default(self):
+        """Regular users do NOT have add_remote_resource when setting is False (default)."""
+        bobby = get_user_model().objects.get(username="bobby")
+        perms = permissions_registry.get_db_perms_by_user(bobby)
+        self.assertNotIn("add_remote_resource", perms)
+
+    @override_settings(REGISTERED_USERS_CAN_ADD_REMOTE_RESOURCES=True)
+    def test_add_remote_resource_perm_regular_user_enabled(self):
+        """Regular users DO have add_remote_resource when setting is True."""
+        bobby = get_user_model().objects.get(username="bobby")
+        perms = permissions_registry.get_db_perms_by_user(bobby)
+        self.assertIn("add_remote_resource", perms)
+
     @on_ogc_backend(geoserver.BACKEND_PACKAGE)
     def test_perm_specs_synchronization(self):
         """Test that Dataset is correctly synchronized with guardian:

geonode/security/utils.py

@@ -24,6 +24,7 @@
 
 from django.conf import settings
 from django.contrib.auth.models import Group
+from django.core.exceptions import PermissionDenied
 from guardian.shortcuts import get_objects_for_user, get_objects_for_group
 
 from geonode.groups.conf import settings as groups_settings
@@ -162,6 +163,15 @@ def get_user_visible_groups(user, include_public_invite: bool = False):
 )
 
 
+def check_add_remote_resource_perm(user):
+    """
+    Checks whether the given user has permission to add remote resources.
+    """
+    perms = permissions_registry.get_db_perms_by_user(user)
+    if "add_remote_resource" not in perms:
+        raise PermissionDenied("You do not have permission to add remote resources.")
+
+
 class AdvancedSecurityWorkflowManager:
     @staticmethod
     def is_anonymous_can_view():

geonode/services/tests.py

@@ -340,6 +340,7 @@ def test_get_service_handler_arcgis(self, mock_map_service):
 
     @skip("test to be revisioned")
     @mock.patch("arcrest.MapService", autospec=True)
+    @override_settings(REGISTERED_USERS_CAN_ADD_REMOTE_RESOURCES=True)
     def test_get_arcgis_alternative_structure(self, mock_map_service):
         LayerESRIExtent = namedtuple("LayerESRIExtent", "spatialReference xmin ymin ymax xmax")
         LayerESRIExtentSpatialReference = namedtuple("LayerESRIExtentSpatialReference", "wkid latestWkid")
@@ -664,6 +665,7 @@ def test_get_resource(self, mock_wms_parsed_service, mock_wms):
     @mock.patch("geonode.services.serviceprocessors.wms.WmsServiceHandler.parsed_service", autospec=True)
     @mock.patch("geonode.services.serviceprocessors.wms.WmsServiceHandler.get_resources", autospec=True)
     @mock.patch("geonode.services.serviceprocessors.wms.WmsServiceHandler.get_resource", autospec=True)
+    @override_settings(REGISTERED_USERS_CAN_ADD_REMOTE_RESOURCES=True)
     def test_get_resources(self, mock_wms_get_resource, mock_wms_get_resources, mock_wms_parsed_service, mock_wms):
         mock_wms.return_value = (self.phony_url, self.parsed_wms)
         mock_wms_parsed_service.return_value = self.parsed_wms
@@ -735,6 +737,7 @@ def test_get_store(self, mock_get_gs_cascading_store, mock_wms_parsed_service, m
         )
 
     @flaky(max_runs=3)
+    @override_settings(REGISTERED_USERS_CAN_ADD_REMOTE_RESOURCES=True)
     def test_local_user_cant_delete_service(self):
         self.client.logout()
         response = self.client.get(reverse("register_service"))
@@ -805,6 +808,7 @@ def test_removing_the_service_delete_also_the_harvester(self):
         self.assertFalse(Harvester.objects.filter(id=harvester.id).exists())
 
     @flaky(max_runs=3)
+    @override_settings(REGISTERED_USERS_CAN_ADD_REMOTE_RESOURCES=True)
     def test_add_duplicate_remote_service_url(self):
         form_data = {
             "url": "https://gs-stable.geo-solutions.it/geoserver/wms?service=wms&version=1.3.0&request=GetCapabilities",
@@ -933,6 +937,9 @@ def setUp(self):
             metadata_only=True,
             base_url="http://bogus.pocus.com/ows",
         )
+        self.test_user = get_user_model().objects.create_user(
+            username="test_user12", email="testuser@example.com", password="testpass123"
+        )
         self.sut.clear_dirty_state()
 
     def test_user_admin_can_access_to_page(self):
@@ -944,6 +951,24 @@ def test_anonymous_user_can_see_the_services(self):
         response = self.client.get(reverse("services"))
         self.assertEqual(response.status_code, 200)
 
+    @override_settings(REGISTERED_USERS_CAN_ADD_REMOTE_RESOURCES=False)
+    def test_register_service_allowed_for_admin(self):
+        self.client.login(username="admin", password="admin")
+        response = self.client.get(reverse("register_service"))
+        self.assertEqual(response.status_code, 200)
+
+    @override_settings(REGISTERED_USERS_CAN_ADD_REMOTE_RESOURCES=False)
+    def test_register_service_denied_for_regular_user(self):
+        self.client.force_login(self.test_user)
+        response = self.client.get(reverse("register_service"))
+        self.assertEqual(response.status_code, 403)
+
+    @override_settings(REGISTERED_USERS_CAN_ADD_REMOTE_RESOURCES=True)
+    def test_register_service_allowed_for_regular_user(self):
+        self.client.force_login(self.test_user)
+        response = self.client.get(reverse("register_service"))
+        self.assertEqual(response.status_code, 200)
+
     @override_settings(SERVICES_TYPE_MODULES=SERVICES_TYPE_MODULES)
     def test_will_use_multiple_service_types_defined(self):
         elems = parse_services_types()

geonode/services/views.py

@@ -34,7 +34,7 @@
 from geonode.base.models import ResourceBase
 from geonode.harvesting.models import Harvester
 from geonode.security.views import _perms_info_json
-from geonode.security.utils import get_visible_resources
+from geonode.security.utils import get_visible_resources, check_add_remote_resource_perm
 from django.core.cache import caches
 
 from .models import Service
@@ -59,6 +59,8 @@ def services(request):
 
 @login_required
 def register_service(request):
+    check_add_remote_resource_perm(request.user)
+
     service_register_template = "services/service_register.html"
     if request.method == "POST":
         form = forms.CreateServiceForm(request.POST)

geonode/settings.py

@@ -1878,6 +1878,10 @@ def get_geonode_catalogue_service():
     os.getenv("EDITORS_CAN_MANAGE_REGISTERED_MEMBERS_PERMISSIONS", "True")
 )
 
+REGISTERED_USERS_CAN_ADD_REMOTE_RESOURCES = ast.literal_eval(
+    os.getenv("REGISTERED_USERS_CAN_ADD_REMOTE_RESOURCES", "False")
+)
+
 PERMISSIONS_HANDLERS = [
     "geonode.security.handlers.GroupManagersPermissionsHandler",
     "geonode.security.handlers.SpecialGroupsPermissionsHandler",

geonode/upload/api/tests.py

@@ -19,6 +19,7 @@
 from django.contrib.auth import get_user_model
 from django.core.files.uploadedfile import SimpleUploadedFile
 from geonode.layers.models import Dataset
+from django.test import override_settings
 from django.urls import reverse
 from unittest.mock import MagicMock, patch
 
@@ -36,6 +37,9 @@ class TestImporterViewSet(ImporterBaseTestSupport):
     def setUpClass(cls):
         super().setUpClass()
         cls.url = reverse("importer_upload")
+        cls.test_user = get_user_model().objects.create_user(
+            username="test_user12", email="testuser@example.com", password="testpass123"
+        )
 
     def setUp(self):
         self.dataset = create_single_dataset(name="test_dataset_copy")
@@ -84,6 +88,45 @@ def test_gpkg_raise_error_with_invalid_payload(self):
         self.assertEqual(400, response.status_code)
         self.assertEqual(expected, response.json())
 
+    @override_settings(REGISTERED_USERS_CAN_ADD_REMOTE_RESOURCES=False)
+    @patch("geonode.upload.handlers.remote.cog.RemoteCOGResourceHandler.is_valid_url")
+    @patch("geonode.upload.handlers.remote.cog.RemoteCOGResourceHandler.can_handle")
+    @patch("geonode.upload.api.views.import_orchestrator.s")
+    def test_remote_dataset_add_allowed_for_admin(
+        self,
+        mock_sig,
+        mock_can_handle,
+        mock_is_valid_url,
+    ):
+        mock_is_valid_url.return_value = True
+        mock_can_handle.return_value = True
+
+        self.client.force_login(get_user_model().objects.get(username="admin"))
+
+        payload = {
+            "url": "https://example.com/data.tif",
+            "title": "Remote dataset",
+            "type": "cog",
+            "action": "upload",
+        }
+
+        response = self.client.post(self.url, data=payload)
+        self.assertEqual(201, response.status_code)
+
+    @override_settings(REGISTERED_USERS_CAN_ADD_REMOTE_RESOURCES=False)
+    def test_remote_dataset_add_forbidden_for_regular_user_by_default(self):
+        self.client.force_login(self.test_user)
+
+        payload = {
+            "url": "https://example.com/data.tif",
+            "title": "Remote dataset denied",
+            "type": "cog",
+            "action": "upload",
+        }
+
+        response = self.client.post(self.url, data=payload)
+        self.assertEqual(403, response.status_code)
+
     @patch("geonode.upload.api.views.import_orchestrator")
     def test_gpkg_task_is_called(self, patch_upload):
         patch_upload.apply_async.side_effect = MagicMock()

geonode/upload/api/views.py

@@ -50,6 +50,7 @@
 from rest_framework.response import Response
 from geonode.proxy.utils import proxy_urls_registry
 from geonode.storage.manager import FileSystemStorageManager
+from geonode.security.utils import check_add_remote_resource_perm
 
 from geonode.upload.api.serializer import (
     UploadParallelismLimitSerializer,
@@ -148,6 +149,9 @@ def create(self, request, *args, **kwargs):
             **{key: value[0] if isinstance(value, list) else value for key, value in request.FILES.items()},
         }
 
+        if "url" in _data:
+            check_add_remote_resource_perm(request.user)
+
         # clone the memory files into local file system
         if "url" not in _data and not _data.get("is_empty", False):
             storage_manager = StorageManager(