[Backport 4.1.x] SSRF Bypass to return internal host data

.env_test

@@ -47,7 +47,7 @@ ASYNC_SIGNALS=True
 
 SITEURL=http://localhost:8001/
 
-ALLOWED_HOSTS="['django', '*']"
+ALLOWED_HOSTS="['django', 'localhost']"
 
 # Data Uploader
 DEFAULT_BACKEND_UPLOADER=geonode.importer

geonode/proxy/tests.py

@@ -178,6 +178,62 @@ class Response:
             },
         )
 
+    def test_proxy_url_forgery(self):
+        """The GeoNode Proxy should preserve the original request headers."""
+        import geonode.proxy.views
+        from urllib.parse import urlsplit
+
+        class Response:
+            status_code = 200
+            content = "Hello World"
+            headers = {
+                "Content-Type": "text/plain",
+                "Vary": "Authorization, Accept-Language, Cookie, origin",
+                "X-Content-Type-Options": "nosniff",
+                "X-XSS-Protection": "1; mode=block",
+                "Referrer-Policy": "same-origin",
+                "X-Frame-Options": "SAMEORIGIN",
+                "Content-Language": "en-us",
+                "Content-Length": "119",
+                "Content-Disposition": 'attachment; filename="filename.tif"',
+            }
+
+        request_mock = MagicMock()
+        request_mock.return_value = (Response(), None)
+
+        # Non-Legit requests attempting SSRF
+        geonode.proxy.views.http_client.request = request_mock
+        url = f"http://example.org\\@%23{urlsplit(settings.SITEURL).hostname}"
+
+        response = self.client.get(f"{self.proxy_url}?url={url}")
+        self.assertEqual(response.status_code, 403)
+
+        url = f"http://125.126.127.128\\@%23{urlsplit(settings.SITEURL).hostname}"
+
+        response = self.client.get(f"{self.proxy_url}?url={url}")
+        self.assertEqual(response.status_code, 403)
+
+        url = f"{settings.SITEURL.rstrip('/')}@db/"
+
+        response = self.client.get(f"{self.proxy_url}?url={url}")
+        self.assertEqual(response.status_code, 403)
+
+        url = f"{settings.SITEURL.rstrip('/')}%40db/"
+
+        response = self.client.get(f"{self.proxy_url}?url={url}")
+        self.assertEqual(response.status_code, 403)
+
+        # Legit requests using the local host (SITEURL)
+        url = f"/\\@%23{urlsplit(settings.SITEURL).hostname}"
+
+        response = self.client.get(f"{self.proxy_url}?url={url}")
+        self.assertEqual(response.status_code, 200)
+
+        url = f"{settings.SITEURL}\\@%23{urlsplit(settings.SITEURL).hostname}"
+
+        response = self.client.get(f"{self.proxy_url}?url={url}")
+        self.assertEqual(response.status_code, 200)
+
 
 class DownloadResourceTestCase(GeoNodeBaseTestSupport):
     def setUp(self):

geonode/proxy/views.py

@@ -40,7 +40,14 @@
 from geonode.upload.models import Upload
 from geonode.base.models import ResourceBase
 from geonode.storage.manager import storage_manager
-from geonode.utils import resolve_object, check_ogc_backend, get_headers, http_client, json_response
+from geonode.utils import (
+    resolve_object,
+    check_ogc_backend,
+    get_headers,
+    http_client,
+    json_response,
+    extract_ip_or_domain,
+)
 from geonode.base.enumerations import LINK_TYPES as _LT
 
 from geonode import geoserver  # noqa
@@ -130,7 +137,7 @@ def proxy(
             _remote_host = urlsplit(_s.base_url).hostname
             PROXY_ALLOWED_HOSTS += (_remote_host,)
 
-        if not validate_host(url.hostname, PROXY_ALLOWED_HOSTS):
+        if not validate_host(extract_ip_or_domain(raw_url), PROXY_ALLOWED_HOSTS):
             return HttpResponse(
                 "DEBUG is set to False but the host of the path provided to the proxy service"
                 " is not in the PROXY_ALLOWED_HOSTS setting.",

geonode/utils.py

@@ -34,6 +34,7 @@
 import datetime
 import requests
 import tempfile
+import ipaddress
 import traceback
 import subprocess
 
@@ -87,6 +88,7 @@
     urlparse,
     urlsplit,
     urlencode,
+    urlunparse,
     parse_qsl,
     ParseResult,
 )
@@ -1894,6 +1896,42 @@ def build_absolute_uri(url):
     return url
 
 
+def remove_credentials_from_url(url):
+    # Parse the URL
+    parsed_url = urlparse(url)
+
+    # Remove the username and password from the parsed URL
+    parsed_url = parsed_url._replace(netloc=parsed_url.netloc.split("@")[-1])
+
+    # Reconstruct the URL without credentials
+    cleaned_url = urlunparse(parsed_url)
+
+    return cleaned_url
+
+
+def extract_ip_or_domain(url):
+    # Decode the URL to handle percent-encoded characters
+    _url = remove_credentials_from_url(unquote(url))
+
+    ip_regex = re.compile("^(?:http://|https://)(\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\.\\d{1,3})")
+    domain_regex = re.compile("^(?:http://|https://)([a-zA-Z0-9.-]+)")
+
+    match = ip_regex.findall(_url)
+    if len(match):
+        ip_address = match[0]
+        try:
+            ipaddress.ip_address(ip_address)  # Validate the IP address
+            return ip_address
+        except ValueError:
+            pass
+
+    match = domain_regex.findall(_url)
+    if len(match):
+        return match[0]
+
+    return None
+
+
 def get_xpath_value(
     element: etree.Element, xpath_expression: str, nsmap: typing.Optional[dict] = None
 ) -> typing.Optional[str]: