Merge pull request from GHSA-rmxg-6qqf-x8mr

* - Fixes "Server Side Request forgery"

* Fixes IP or domain extraction

* fixed formatting

* - Bump ipaddress to version 1.0.23

---------

Co-authored-by: G. Allegri <giohappy@gmail.com>

geonode/proxy/tests.py

@@ -178,6 +178,52 @@ 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)
+
+        # 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 requests
 import tempfile
 import importlib
+import ipaddress
 import itertools
 import traceback
 import subprocess
@@ -1930,6 +1931,26 @@ def build_absolute_uri(url):
     return url
 
 
+def extract_ip_or_domain(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]:

requirements.txt

@@ -27,6 +27,7 @@ rdflib==6.3.2
 smart_open==6.3.0
 PyMuPDF==1.22.5
 pathvalidate==3.1.0
+ipaddress==1.0.23
 
 # Django Apps
 django-allauth==0.54.0

setup.cfg

@@ -53,6 +53,7 @@ install_requires =
     smart_open==6.3.0
     PyMuPDF==1.22.5
     pathvalidate==3.1.0
+    ipaddress==1.0.23
 
     # Django Apps
     django-allauth==0.54.0