Apply URI access checks more uniformly.

This doesn't resolve explicit security problems I can imagine a way to take advantage of - but if we have these checks we should apply them uniformly - not just for file uploads.

lib/galaxy/files/uris.py

@@ -7,7 +7,10 @@ from typing import (
 )
 from urllib.parse import urlparse
 
-from galaxy.exceptions import ConfigDoesNotAllowException
+from galaxy.exceptions import (
+    AdminRequiredException,
+    ConfigDoesNotAllowException,
+)
 from galaxy.util import unicodify
 
 log = logging.getLogger(__name__)
@@ -17,6 +20,17 @@ IpNetwrokT = Union[ipaddress.IPv4Network, ipaddress.IPv6Network]
 IpAllowedListEntryT = Union[IpAddressT, IpNetwrokT]
 
 
+def validate_uri_access(uri: str, is_admin: bool, ip_allowlist: List[IpAllowedListEntryT]) -> None:
+    """Perform uniform checks on supplied URIs.
+
+    - Prevent access to local IPs not found in ip_allowlist.
+    - Don't allow non-admins to access file:// URIs.
+    """
+    validate_non_local(uri, ip_allowlist)
+    if not is_admin and uri.lstrip().startswith("file://"):
+        raise AdminRequiredException()
+
+
 def validate_non_local(uri: str, ip_allowlist: List[IpAllowedListEntryT]) -> str:
     # If it doesn't look like a URL, ignore it.
     if not (uri.lstrip().startswith("http://") or uri.lstrip().startswith("https://")):

lib/galaxy/webapps/galaxy/api/workflows.py

@@ -29,6 +29,7 @@ from galaxy import (
     model,
     util,
 )
+from galaxy.files.uris import validate_uri_access
 from galaxy.managers.context import ProvidesUserContext
 from galaxy.managers.jobs import (
     fetch_job_states,
@@ -305,9 +306,8 @@ class WorkflowsAPIController(BaseGalaxyAPIController, UsesStoredWorkflowMixin, U
             archive_file = payload.get("archive_file")
             archive_data = None
             if archive_source:
+                validate_uri_access(archive_source, trans.user_is_admin, trans.app.config.fetch_url_allowlist_ips)
                 if archive_source.startswith("file://"):
-                    if not trans.user_is_admin:
-                        raise exceptions.AdminRequiredException()
                     workflow_src = {"src": "from_path", "path": archive_source[len("file://") :]}
                     payload["workflow"] = workflow_src
                     return self.__api_import_new_workflow(trans, payload, **kwd)

lib/galaxy/webapps/galaxy/services/histories.py

@@ -21,6 +21,7 @@ from sqlalchemy import (
 
 from galaxy import exceptions as glx_exceptions
 from galaxy import model
+from galaxy.files.uris import validate_uri_access
 from galaxy.managers.citations import CitationsManager
 from galaxy.managers.context import ProvidesHistoryContext
 from galaxy.managers.histories import (
@@ -194,6 +195,9 @@ class HistoriesService(ServiceBase):
                     archive_source = self._save_upload_file_tmp(archive_file)
             else:
                 raise glx_exceptions.MessageException("Please provide a url or file.")
+            if archive_type == HistoryImportArchiveSourceType.url:
+                assert archive_source
+                validate_uri_access(archive_source, trans.user_is_admin, trans.app.config.fetch_url_allowlist_ips)
             job = self.manager.queue_history_import(trans, archive_type=archive_type, archive_source=archive_source)
             job_dict = job.to_dict()
             job_dict[

lib/galaxy_test/driver/driver_util.py

@@ -254,6 +254,7 @@ def setup_galaxy_config(
         logging=LOGGING_CONFIG_DEFAULT,
         monitor_thread_join_timeout=5,
         object_store_store_by="uuid",
+        fetch_url_allowlist="127.0.0.1",
     )
     if default_shed_tool_data_table_config:
         config["shed_tool_data_table_config"] = default_shed_tool_data_table_config

test/integration/test_config_defaults.py

@@ -178,6 +178,7 @@ DO_NOT_TEST = [
     "disable_library_comptypes",  # broken: default overridden with empty string
     "enable_per_request_sql_debugging",  # overridden for tests
     "expose_dataset_path",  # broken: default overridden
+    "fetch_url_allowlist",  # specified in driver_util to allow history export tests to target localhost
     "ftp_upload_purge",  # broken: default overridden
     "ftp_upload_dir_template",  # dynamically sets os.path.sep
     "galaxy_data_manager_data_path",  # broken: review config/, possibly refactor

test/unit/files/test_uris.py

@@ -1,7 +1,13 @@
 import ipaddress
 
-from galaxy.exceptions import ConfigDoesNotAllowException
-from galaxy.files.uris import validate_non_local
+from galaxy.exceptions import (
+    AdminRequiredException,
+    ConfigDoesNotAllowException,
+)
+from galaxy.files.uris import (
+    validate_non_local,
+    validate_uri_access,
+)
 
 
 def test_validate_local_ip_access():
@@ -15,9 +21,35 @@ def test_validate_local_ip_access():
     )
 
 
+def test_validate():
+    # Check the non local ip address checks from above...
+    assert validates("http://google.com", False, [])
+    assert not validates("http://127.0.0.1/secrets.txt", False, [])
+    assert validates("http://127.0.0.1/secrets.txt", False, [ipaddress.ip_address("127.0.0.1")])
+    assert validates("http://127.0.0.1/secrets.txt", False, [ipaddress.ip_network("127.0.0.0/24")])
+    assert not validates("http://127.0.0.1/secrets.txt", False, [ipaddress.ip_network("127.0.0.0/32")])
+    assert validates(
+        "http://127.0.0.1/secrets.txt",
+        False,
+        [ipaddress.ip_network("127.0.0.0/32"), ipaddress.ip_network("127.0.0.1/32")],
+    )
+
+    # check that only admins can acess files...
+    assert not validates("file://foo/bar", False, [])
+    assert validates("file://foo/bar", True, [])
+
+
 def validates_as_non_local(uri: str, allow_list):
     try:
         validate_non_local(uri, allow_list)
     except ConfigDoesNotAllowException:
         return False
     return True
+
+
+def validates(uri: str, is_admin, allow_list):
+    try:
+        validate_uri_access(uri, is_admin, allow_list)
+    except (ConfigDoesNotAllowException, AdminRequiredException):
+        return False
+    return True