[Fixes #14309] Generalize remote service type registration and auth handling

geonode/security/auth_handlers.py

@@ -16,6 +16,7 @@
 # along with this program. If not, see <http://www.gnu.org/licenses/>.
 #
 #########################################################################
+import base64
 from abc import ABC, abstractmethod
 
 from django.core.exceptions import ValidationError
@@ -39,6 +40,9 @@ def _init_from_config(self):
     def get_request_auth(self) -> AuthBase:
         raise NotImplementedError
 
+    def get_gdal_config(self, url):
+        raise NotImplementedError
+
     def auth_request(self, request, **kwargs):
         raise NotImplementedError
 
@@ -109,6 +113,11 @@ def _init_from_config(self):
     def get_request_auth(self) -> AuthBase:
         return HashableAuthBase(HTTPBasicAuth(self.username, self.password))
 
+    def get_gdal_config(self, url):
+        credentials = f"{self.username}:{self.password}".encode()
+        token = base64.b64encode(credentials).decode()
+        return url, {"GDAL_HTTP_HEADERS": f"Authorization: Basic {token}"}
+
     def auth_request(self, request, **kwargs):
         request.auth = self.get_request_auth()
         return request

geonode/security/tests.py

@@ -4248,6 +4248,13 @@ def test_basic_auth_handler_auth_request_sets_request_auth(self):
         self.assertEqual(request.auth.auth.username, "test_user")
         self.assertEqual(request.auth.auth.password, "test_password")
 
+    def test_basic_auth_handler_get_gdal_config(self):
+        auth_handler = auth_handler_registry.build(self.auth_config)
+        expected_token = base64.b64encode(b"test_user:test_password").decode()
+        url, options = auth_handler.get_gdal_config("https://example.com/data.tif")
+        self.assertEqual("https://example.com/data.tif", url)
+        self.assertEqual({"GDAL_HTTP_HEADERS": f"Authorization: Basic {expected_token}"}, options)
+
 
 class AuthHandlerRegistryTests(TestCase):
     class SampleAuthHandler(AuthHandler):

geonode/services/apps.py

@@ -63,6 +63,9 @@ def ready(self):
         super().ready()
         # Let's make sure the signals are connected to the App
         from . import signals  # noqa
+        from geonode.services.serviceprocessors.registry import service_type_registry
+
+        service_type_registry.init_registry()
 
         post_migrate.connect(run_setup_hooks, sender=self)
         # settings.CELERY_BEAT_SCHEDULE['probe_services'] = {

geonode/services/forms.py

@@ -29,9 +29,8 @@
 from geonode.security.models import AuthConfig
 
 from . import enumerations
-from .models import Service
+from .models import Service, get_service_type_choices
 from .serviceprocessors import get_service_handler
-from geonode.services.serviceprocessors import get_available_service_types
 from geonode.utils import is_safe_url
 
 logger = logging.getLogger(__name__)
@@ -48,7 +47,7 @@ class CreateServiceForm(forms.Form):
     )
     type = forms.ChoiceField(
         label=_("Service Type"),
-        choices=[(k, v["label"]) for k, v in get_available_service_types().items()],  # from dictionary to tuple
+        choices=get_service_type_choices,
         initial="AUTO",
     )
 

geonode/services/migrations/0060_alter_service_type.py

@@ -0,0 +1,19 @@
+# Generated by Django 5.2.15 on 2026-06-09 12:56
+
+import geonode.services.models
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("services", "0059_remove_service_password_remove_service_username"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="service",
+            name="type",
+            field=models.CharField(choices=geonode.services.models.get_service_type_choices, max_length=10),
+        ),
+    ]

geonode/services/models.py

@@ -32,15 +32,18 @@
 from geonode.services.serviceprocessors import get_available_service_types
 from . import enumerations
 
-service_type_as_tuple = [(k, v["label"]) for k, v in get_available_service_types().items()]
+
+def get_service_type_choices():
+    return [(k, v["label"]) for k, v in get_available_service_types().items()]
+
 
 logger = logging.getLogger("geonode.services")
 
 
 class Service(ResourceBase):
     """Service Class to represent remote Geo Web Services"""
 
-    type = models.CharField(max_length=10, choices=service_type_as_tuple)
+    type = models.CharField(max_length=10, choices=get_service_type_choices)
     method = models.CharField(
         max_length=1,
         choices=(
@@ -102,12 +105,15 @@ def service_url(self):
     @property
     def ptype(self):
         # Return the gxp ptype that should be used to display layers
-        return GXP_PTYPES[self.type] if self.type else None
+        return GXP_PTYPES.get(self.type) if self.type else None
 
     @property
     def service_type(self):
         # Return the gxp ptype that should be used to display layers
-        return [x for x in service_type_as_tuple if x[0] == self.type][0][1]
+        service_type = get_available_service_types().get(self.type)
+        if service_type:
+            return service_type["label"]
+        return self.type
 
     def get_absolute_url(self):
         return "/services/%i" % self.id

geonode/services/serviceprocessors/__init__.py

@@ -18,45 +18,17 @@
 #########################################################################
 import logging
 
-from collections import OrderedDict
-from django.utils.translation import gettext_lazy as _
 from django.conf import settings
 from geonode.services import enumerations
-from geonode.services.utils import parse_services_types
 from django.core.cache import caches
+from geonode.services.serviceprocessors.registry import service_type_registry
 
 service_cache = caches["services"]
 logger = logging.getLogger(__name__)
 
 
-def get_available_service_types():
-    # LGTM: Fixes - Module uses member of cyclically imported module, which can lead to failure at import time.
-    from geonode.services.serviceprocessors.wms import GeoNodeServiceHandler, WmsServiceHandler
-    from geonode.services.serviceprocessors.arcgis import ArcImageServiceHandler, ArcMapServiceHandler
-
-    default = OrderedDict(
-        {
-            enumerations.WMS: {"OWS": True, "handler": WmsServiceHandler, "label": _("Web Map Service")},
-            enumerations.GN_WMS: {
-                "OWS": True,
-                "handler": GeoNodeServiceHandler,
-                "label": _("GeoNode (Web Map Service)"),
-            },
-            # enumerations.WFS: {"OWS": True, "handler": ServiceHandlerBase, "label": _('Paired WMS/WFS/WCS'},
-            # enumerations.TMS: {"OWS": False, "handler": ServiceHandlerBase, "label": _('Paired WMS/WFS/WCS'},
-            enumerations.REST_MAP: {"OWS": False, "handler": ArcMapServiceHandler, "label": _("ArcGIS REST MapServer")},
-            enumerations.REST_IMG: {
-                "OWS": False,
-                "handler": ArcImageServiceHandler,
-                "label": _("ArcGIS REST ImageServer"),
-            },
-            # enumerations.CSW: {"OWS": False, "handler": ServiceHandlerBase, "label": _('Catalogue Service')},
-            # enumerations.OGP: {"OWS": True, "handler": ServiceHandlerBase, "label": _('OpenGeoPortal')},  # TODO: verify this
-            # enumerations.HGL: {"OWS": False, "handler": ServiceHandlerBase, "label": _('Harvard Geospatial Library')},  # TODO: verify this
-        }
-    )
-
-    return OrderedDict({**default, **parse_services_types()})
+def get_available_service_types():
+    return service_type_registry.get_available_service_types()
 
 
 def get_service_handler(base_url, service_type=enumerations.AUTO, service_id=None, *args, **kwargs):
@@ -67,9 +39,7 @@ def get_service_handler(base_url, service_type=enumerations.AUTO, service_id=Non
     if entry := service_cache.get(base_url):
         return entry
 
-    handlers = get_available_service_types()
-
-    handler = handlers.get(service_type, {}).get("handler")
+    handler = service_type_registry.get_handler_class(service_type)
     try:
         service_handler = handler(base_url, service_id, *args, **kwargs)
         service_cache.set(service_handler.url, service_handler, settings.SERVICE_CACHE_EXPIRATION_TIME)

geonode/services/serviceprocessors/registry.py

@@ -0,0 +1,85 @@
+#########################################################################
+#
+# Copyright (C) 2026 OSGeo
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+#
+#########################################################################
+from collections import OrderedDict
+
+from django.conf import settings
+from django.utils.module_loading import import_string
+from django.utils.translation import gettext_lazy as _
+
+from geonode.services import enumerations
+
+
+class ServiceTypeRegistry:
+    def __init__(self):
+        self.registry = OrderedDict()
+        self._initialized = False
+
+    def register(self, service_type, handler, label, OWS=False, **kwargs):
+        self.registry[service_type] = {
+            "OWS": OWS,
+            "handler": handler,
+            "label": label,
+            **kwargs,
+        }
+
+    def unregister(self, service_type):
+        self.registry.pop(service_type, None)
+
+    def init_registry(self):
+        if self._initialized:
+            return
+
+        self.registry = OrderedDict()
+        self._register_default_service_types()
+        self._register_configured_service_types()
+        self._initialized = True
+
+    def reset(self):
+        self.registry = OrderedDict()
+        self._initialized = False
+
+    def get_available_service_types(self):
+        self.init_registry()
+        return OrderedDict(self.registry)
+
+    def get_handler_class(self, service_type):
+        service_type_config = self.get_available_service_types().get(service_type, {})
+        handler = service_type_config.get("handler")
+        if isinstance(handler, str):
+            return import_string(handler)
+        return handler
+
+    def _register_default_service_types(self):
+        # Keep imports lazy to avoid circular imports during Django app loading.
+        from geonode.services.serviceprocessors.arcgis import ArcImageServiceHandler, ArcMapServiceHandler
+        from geonode.services.serviceprocessors.wms import GeoNodeServiceHandler, WmsServiceHandler
+
+        self.register(enumerations.WMS, WmsServiceHandler, _("Web Map Service"), OWS=True)
+        self.register(enumerations.GN_WMS, GeoNodeServiceHandler, _("GeoNode (Web Map Service)"), OWS=True)
+        self.register(enumerations.REST_MAP, ArcMapServiceHandler, _("ArcGIS REST MapServer"))
+        self.register(enumerations.REST_IMG, ArcImageServiceHandler, _("ArcGIS REST ImageServer"))
+
+    def _register_configured_service_types(self):
+        for service_type_module_path in getattr(settings, "SERVICES_TYPE_MODULES", []):
+            custom_service_type_module = import_string(service_type_module_path)
+            for service_type, service_type_config in custom_service_type_module.services_type.items():
+                self.register(service_type, **service_type_config)
+
+
+service_type_registry = ServiceTypeRegistry()