harden proxy

geonode/proxy/tests.py

@@ -222,8 +222,9 @@ class Response:
         request_mock = MagicMock()
         request_mock.return_value = (Response(), None)
 
-        # Non-Legit requests attempting SSRF
         geonode.proxy.views.http_client.request = request_mock
+
+        # Non-legit requests attempting SSRF
         url = f"http://example.org\\@%23{urlsplit(settings.SITEURL).hostname}"
 
         response = self.client.get(f"{self.proxy_url}?url={url}")
@@ -255,6 +256,25 @@ class Response:
         response = self.client.get(f"{self.proxy_url}?url={url}")
         self.assertEqual(response.status_code, 200)
 
+        # Special variable attacks
+        # Test with special variable attacks like "$", "*", "?" or Unicode escape sequences
+        special_var_url = f"http://example.org/test/{urlsplit(settings.SITEURL).hostname}?var=$HOME"
+
+        response = self.client.get(f"{self.proxy_url}?url={special_var_url}")
+        self.assertEqual(response.status_code, 403, "Request with special variable should be blocked")
+
+        # Test with a URL containing a wildcard (*)
+        wildcard_url = f"http://example.org/test/{urlsplit(settings.SITEURL).hostname}*"
+
+        response = self.client.get(f"{self.proxy_url}?url={wildcard_url}")
+        self.assertEqual(response.status_code, 403, "Request with wildcard should be blocked")
+
+        # Test with percent-encoded special characters
+        encoded_url = f"http://example.org/test/%24%40%7E"
+
+        response = self.client.get(f"{self.proxy_url}?url={encoded_url}")
+        self.assertEqual(response.status_code, 403, "Request with encoded special characters should be blocked")
+
 
 class DownloadResourceTestCase(GeoNodeBaseTestSupport):
     def setUp(self):

geonode/proxy/views.py

@@ -57,6 +57,8 @@
 from geonode.assets.utils import get_default_asset
 from zipstream import ZipStream
 from .utils import proxy_urls_registry
+from urllib.parse import unquote
+
 
 logger = logging.getLogger(__name__)
 
@@ -79,7 +81,6 @@ def proxy(
     access_token=None,
     **kwargs,
 ):
-
     if not timeout:
         timeout = getattr(ogc_server_settings, "TIMEOUT", TIMEOUT)
 
@@ -92,8 +93,11 @@ def proxy(
         )
 
     raw_url = url or request.GET["url"]
-    raw_url = urljoin(settings.SITEURL, raw_url) if raw_url.startswith("/") else raw_url
-    url = urlsplit(raw_url)
+
+    # Ensure URL is fully decoded before validation
+    decoded_raw_url = unquote(raw_url)
+    raw_url = urljoin(settings.SITEURL, decoded_raw_url) if decoded_raw_url.startswith("/") else decoded_raw_url
+    url = urlsplit(decoded_raw_url)
     scheme = str(url.scheme)
     locator = str(url.path)
     if url.query != "":
@@ -108,7 +112,16 @@ def proxy(
         ):
             proxy_allowed_hosts.append(url.hostname)
 
-        if not validate_host(extract_ip_or_domain(raw_url), proxy_allowed_hosts):
+        # Ensure the hostname is not None before validating
+        if url.hostname is None or not re.match(r"^[a-zA-Z0-9\.\-]+$", url.hostname):
+            return HttpResponse(
+                "Invalid request: malformed hostname.",
+                status=403,
+                content_type="text/plain",
+            )
+
+        # Validate the decoded hostname
+        if not validate_host(extract_ip_or_domain(decoded_raw_url), proxy_allowed_hosts):
             return HttpResponse(
                 "The url provided to the proxy service is not a valid hostname.",
                 status=403,
@@ -117,7 +130,7 @@ def proxy(
 
     # Collecting headers and cookies
     if not headers:
-        headers, access_token = get_headers(request, url, raw_url, allowed_hosts=allowed_hosts)
+        headers, access_token = get_headers(request, url, decoded_raw_url, allowed_hosts=allowed_hosts)
     if not access_token:
         auth_header = None
         if "Authorization" in headers:
@@ -129,7 +142,7 @@ def proxy(
     user = get_auth_user(access_token)
 
     # Inject access_token if necessary
-    parsed = urlparse(raw_url)
+    parsed = urlparse(decoded_raw_url)
     parsed._replace(path=locator.encode("utf8"))
     if parsed.netloc == site_url.netloc and scheme != site_url.scheme:
         parsed = parsed._replace(scheme=site_url.scheme)
@@ -151,7 +164,6 @@ def proxy(
 
     # Avoid translating local geoserver calls into external ones
     if check_ogc_backend(geoserver.BACKEND_PACKAGE):
-
         _url = _url.replace(f"{settings.SITEURL}geoserver", ogc_server_settings.LOCATION.rstrip("/"))
         _data = _data.replace(f"{settings.SITEURL}geoserver", ogc_server_settings.LOCATION.rstrip("/"))