Merge pull request #20311 from davelopez/25.0_increase_proxy_api_robustness

[25.0] Increase proxy API robustness by validating URL before use

lib/galaxy/webapps/galaxy/api/proxy.py

@@ -3,6 +3,7 @@ API Controller to handle remote zip operations.
 """
 
 import logging
+from urllib.parse import urlparse
 
 import httpx
 from fastapi import (
@@ -34,6 +35,18 @@ URLQueryParam: str = Query(
     description="The URL of the remote file",
 )
 
+ALLOWED_SCHEMES = ("https", "http")
+
+
+def is_valid_url(url: str) -> bool:
+    if url.count("://") != 1:
+        return False  # Ensure there is exactly one scheme
+    try:
+        parsed = urlparse(url)
+        return all([parsed.scheme in ALLOWED_SCHEMES, parsed.netloc])
+    except ValueError:
+        return False
+
 
 @router.cbv
 class FastAPIProxy:
@@ -47,6 +60,9 @@ class FastAPIProxy:
         if trans.anonymous:
             raise UserRequiredException("Anonymous users are not allowed to access this endpoint")
 
+        if not is_valid_url(url):
+            raise RequestParameterInvalidException("Invalid URL format.")
+
         validate_uri_access(url, trans.user_is_admin, trans.app.config.fetch_url_allowlist_ips)
 
         headers = {}

lib/galaxy_test/api/test_proxy.py

@@ -1,3 +1,5 @@
+import pytest
+
 from galaxy.util.unittest_utils import skip_if_github_down
 from galaxy_test.base.populators import DatasetPopulator
 from ._framework import ApiTestCase
@@ -50,8 +52,22 @@ class TestProxyApi(ApiTestCase):
         self._assert_status_code_is(response, 403)
         assert response.json()["err_msg"] == "Anonymous users are not allowed to access this endpoint"
 
-    def test_user_cannot_access_local_file(self):
-        local_file_url = "file://etc/passwd"
-        response = self._get(f"proxy?url={local_file_url}")
-        self._assert_status_code_is(response, 403)
-        assert response.json()["err_msg"] == "Action requires admin account."
+    @pytest.mark.parametrize(
+        "invalid_url",
+        [
+            "htp://invalid-url",
+            "http:/missing-slash.com",
+            "//missing-scheme.com",
+            "invalid-url",
+            "https://first.url\nhttps://second.url",
+            "https://first.url https://second.url",
+            "https://first.urlhttps://second.url",
+            # Schemes other than http and https are not allowed
+            "ftp://not-allowed.com",
+            "file://etc/passwd",
+        ],
+    )
+    def test_invalid_url_format(self, invalid_url: str):
+        response = self._get(f"proxy?url={invalid_url}")
+        self._assert_status_code_is(response, 400)
+        assert response.json()["err_msg"] == "Invalid URL format."