CVE-2023-40017: Server-Side Request Forgery (SSRF) (CWE-918)
value: GeoNode is an open source platform that facilitates the creation, sharing, and collaborative use of geospatial data. In versions 3.2.0 through 4.1.2, the endpoint `/proxy/?url=` does not properly protect against server-side request forgery. This allows an attacker to port scan internal hosts and request information from internal hosts. A patch is available at commit a9eebae80cb362009660a1fd49e105e7cdb499b9.
CWE description: The web server receives a URL or similar request from an upstream component and retrieves the contents of this URL, but it does not sufficiently ensure that the request is being sent to the expected destination.
CVSS v3 base score: 7.5
Published: 2023-08-24T23:15Z
Repository: https://github.com/GeoNode/geonode
Language: Python
Files changed: geonode/proxy/tests.py, geonode/proxy/views.py, geonode/utils.py, requirements.txt, setup.cfg
Commit date: 2023-08-23 17:32:13 +0200
Commit message: 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>
Diff: diff --git a/geonode/proxy/tests.py b/geonode/proxy/tests.py
index 73ef8ba98..e0cd63854 100644
--- a/geonode/proxy/tests.py
+++ b/geonode/proxy/tests.py
@@ -176,10 +176,56 @@ class ProxyTest(GeoNodeBaseTestSupport):
                 "Content-Length": "119",
                 "Content-Disposition": 'attachment; filename="filename.tif"',
             },
         )
 
+    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):
         super().setUp()
         self.maxDiff = None
diff --git a/geonode/proxy/views.py b/geonode/proxy/views.py
index 10030f3dc..fc49163f0 100644
--- a/geonode/proxy/views.py
+++ b/geonode/proxy/views.py
@@ -38,11 +38,18 @@ from django.views.decorators.csrf import requires_csrf_token
 
 from geonode.layers.models import Dataset
 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
 from geonode.base import register_event
 from geonode.base.auth import get_auth_user, get_token_from_auth_header
@@ -128,11 +135,11 @@ def proxy(
 
         for _s in Service.objects.all():
             _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.",
                 status=403,
                 content_type="text/plain",
diff --git a/geonode/utils.py b/geonode/utils.py
index 5abab4b24..2ceaa2ff0 100755
--- a/geonode/utils.py
+++ b/geonode/utils.py
@@ -32,10 +32,11 @@ import logging
 import tarfile
 import datetime
 import requests
 import tempfile
 import importlib
+import ipaddress
 import itertools
 import traceback
 import subprocess
 
 from lxml import etree
@@ -1928,10 +1929,30 @@ def build_absolute_uri(url):
     if url and "http" not in url:
         url = urljoin(settings.SITEURL, 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]:
     if not nsmap:
         nsmap = element.nsmap
diff --git a/requirements.txt b/requirements.txt
index 05db52f8a..df52b4ebb 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -25,10 +25,11 @@ zipstream-new==1.1.8
 schema==0.7.5
 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
 django-appconf==1.0.5
 django-celery-results==2.5.1
diff --git a/setup.cfg b/setup.cfg
index 41a528efb..28a66b161 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -51,10 +51,11 @@ install_requires =
     schema==0.7.5
     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
     django-appconf==1.0.5
     django-celery-results==2.5.1

Vulnerable code: from geonode.utils import resolve_object, check_ogc_backend, get_headers, http_client, json_response
        if not validate_host(url.hostname, PROXY_ALLOWED_HOSTS):
Fixed code:     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)

from geonode.utils import (
    resolve_object,
    check_ogc_backend,
    get_headers,
    http_client,
    json_response,
    extract_ip_or_domain,
)
        if not validate_host(extract_ip_or_domain(raw_url), PROXY_ALLOWED_HOSTS):
import ipaddress
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


ipaddress==1.0.23
    ipaddress==1.0.23