Merge branch 'dev' of https://github.com/galaxyproject/galaxy into metrics

lib/galaxy/datatypes/sniff.py

@@ -12,8 +12,8 @@ import shutil
 import struct
 import sys
 import tempfile
-import urllib.request
 import zipfile
+from functools import partial
 from typing import (
     Dict,
     IO,
@@ -25,12 +25,11 @@ from typing import (
 from typing_extensions import Protocol
 
 from galaxy import util
-from galaxy.files import ConfiguredFileSources
+from galaxy.files.uris import stream_url_to_file as files_stream_url_to_file
 from galaxy.util import (
     compression_utils,
     file_reader,
     is_binary,
-    stream_to_open_named_file,
 )
 from galaxy.util.checkers import (
     check_html,
@@ -63,28 +62,7 @@ def sniff_with_cls(cls, fname):
         return False
 
 
-def stream_url_to_file(path: str, file_sources: Optional[ConfiguredFileSources] = None):
-    prefix = "url_paste"
-    if file_sources and file_sources.looks_like_uri(path):
-        file_source_path = file_sources.get_file_source_path(path)
-        with tempfile.NamedTemporaryFile(prefix=prefix, delete=False) as temp:
-            temp_name = temp.name
-        file_source_path.file_source.realize_to(file_source_path.path, temp_name)
-        return temp_name
-    else:
-        page = urllib.request.urlopen(
-            path, timeout=util.DEFAULT_SOCKET_TIMEOUT
-        )  # page will be .close()ed in stream_to_file
-        temp_name = stream_to_file(
-            page, prefix=prefix, source_encoding=util.get_charset_from_http_headers(page.headers)
-        )
-        return temp_name
-
-
-def stream_to_file(stream, suffix="", prefix="", dir=None, text=False, **kwd):
-    """Writes a stream to a temporary file, returns the temporary file's name"""
-    fd, temp_name = tempfile.mkstemp(suffix=suffix, prefix=prefix, dir=dir, text=text)
-    return stream_to_open_named_file(stream, fd, temp_name, **kwd)
+stream_url_to_file = partial(files_stream_url_to_file, prefix="gx_url_paste")
 
 
 def handle_composite_file(datatype, src_path, extra_files, name, is_binary, tmp_dir, tmp_prefix, upload_opts):

lib/galaxy/files/uris.py

@@ -0,0 +1,168 @@
+import base64
+import ipaddress
+import logging
+import os
+import socket
+import tempfile
+import urllib.request
+from typing import (
+    List,
+    Optional,
+    TYPE_CHECKING,
+    Union,
+)
+from urllib.parse import urlparse
+
+from galaxy.exceptions import (
+    AdminRequiredException,
+    ConfigDoesNotAllowException,
+)
+from galaxy.util import (
+    DEFAULT_SOCKET_TIMEOUT,
+    get_charset_from_http_headers,
+    stream_to_open_named_file,
+    unicodify,
+)
+
+if TYPE_CHECKING:
+    from galaxy.files import ConfiguredFileSources
+
+
+log = logging.getLogger(__name__)
+
+
+def stream_url_to_str(
+    path: str, file_sources: Optional["ConfiguredFileSources"] = None, prefix: str = "gx_file_stream"
+) -> str:
+    tmp_file = stream_url_to_file(path, file_sources=file_sources, prefix=prefix)
+    try:
+        with open(tmp_file, "r") as f:
+            return f.read()
+    finally:
+        os.remove(tmp_file)
+
+
+def stream_url_to_file(
+    path: str, file_sources: Optional["ConfiguredFileSources"] = None, prefix: str = "gx_file_stream"
+) -> str:
+    temp_name: str
+    if file_sources and file_sources.looks_like_uri(path):
+        file_source_path = file_sources.get_file_source_path(path)
+        with tempfile.NamedTemporaryFile(prefix=prefix, delete=False) as temp:
+            temp_name = temp.name
+        file_source_path.file_source.realize_to(file_source_path.path, temp_name)
+    elif path.startswith("base64://"):
+        with tempfile.NamedTemporaryFile(prefix=prefix, delete=False) as temp:
+            temp_name = temp.name
+            temp.write(base64.b64decode(path[len("base64://") :]))
+            temp.flush()
+    else:
+        page = urllib.request.urlopen(path, timeout=DEFAULT_SOCKET_TIMEOUT)  # page will be .close()ed in stream_to_file
+        temp_name = stream_to_file(page, prefix=prefix, source_encoding=get_charset_from_http_headers(page.headers))
+    return temp_name
+
+
+def stream_to_file(stream, suffix="", prefix="", dir=None, text=False, **kwd):
+    """Writes a stream to a temporary file, returns the temporary file's name"""
+    fd, temp_name = tempfile.mkstemp(suffix=suffix, prefix=prefix, dir=dir, text=text)
+    return stream_to_open_named_file(stream, fd, temp_name, **kwd)
+
+
+IpAddressT = Union[ipaddress.IPv4Address, ipaddress.IPv6Address]
+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://")):
+        return uri
+
+    # Strip leading whitespace before passing url to urlparse()
+    url = uri.lstrip()
+    # Extract hostname component
+    parsed_url = urlparse(url).netloc
+    # If credentials are in this URL, we need to strip those.
+    if parsed_url.count("@") > 0:
+        # credentials.
+        parsed_url = parsed_url[parsed_url.rindex("@") + 1 :]
+    # Percent encoded colons and other characters will not be resolved as such
+    # so we don't have to either.
+
+    # Sometimes the netloc will contain the port which is not desired, so we
+    # need to extract that.
+    port = None
+    # However, it could ALSO be an IPv6 address they've supplied.
+    if ":" in parsed_url:
+        # IPv6 addresses have colons in them already (it seems like always more than two)
+        if parsed_url.count(":") >= 2:
+            # Since IPv6 already use colons extensively, they wrap it in
+            # brackets when there is a port, e.g. http://[2001:db8:1f70::999:de8:7648:6e8]:100/
+            # However if it ends with a ']' then there is no port after it and
+            # they've wrapped it in brackets just for fun.
+            if "]" in parsed_url and not parsed_url.endswith("]"):
+                # If this +1 throws a range error, we don't care, their url
+                # shouldn't end with a colon.
+                idx = parsed_url.rindex(":")
+                # We parse as an int and let this fail ungracefully if parsing
+                # fails because we desire to fail closed rather than open.
+                port = int(parsed_url[idx + 1 :])
+                parsed_url = parsed_url[:idx]
+            else:
+                # Plain ipv6 without port
+                pass
+        else:
+            # This should finally be ipv4 with port. It cannot be IPv6 as that
+            # was caught by earlier cases, and it cannot be due to credentials.
+            idx = parsed_url.rindex(":")
+            port = int(parsed_url[idx + 1 :])
+            parsed_url = parsed_url[:idx]
+
+    # safe to log out, no credentials/request path, just an IP + port
+    log.debug("parsed url, port: %s : %s", parsed_url, port)
+    # Call getaddrinfo to resolve hostname into tuples containing IPs.
+    addrinfo = socket.getaddrinfo(parsed_url, port)
+    # Get the IP addresses that this entry resolves to (uniquely)
+    # We drop:
+    #   AF_* family: It will resolve to AF_INET or AF_INET6, getaddrinfo(3) doesn't even mention AF_UNIX,
+    #   socktype: We don't care if a stream/dgram/raw protocol
+    #   protocol: we don't care if it is tcp or udp.
+    addrinfo_results = {info[4][0] for info in addrinfo}
+    # There may be multiple (e.g. IPv4 + IPv6 or DNS round robin). Any one of these
+    # could resolve to a local addresses (and could be returned by chance),
+    # therefore we must check them all.
+    for raw_ip in addrinfo_results:
+        # Convert to an IP object so we can tell if it is in private space.
+        ip = ipaddress.ip_address(unicodify(raw_ip))
+        # If this is a private address
+        if ip.is_private:
+            results = []
+            # If this IP is not anywhere in the allowlist
+            for allowlisted in ip_allowlist:
+                # If it's an IP address range (rather than a single one...)
+                if isinstance(allowlisted, (ipaddress.IPv4Network, ipaddress.IPv6Network)):
+                    results.append(ip in allowlisted)
+                else:
+                    results.append(ip == allowlisted)
+
+            if any(results):
+                # If we had any True, then THIS (and ONLY THIS) IP address that
+                # that specific DNS entry resolved to is in allowlisted and
+                # safe to access. But we cannot exit here, we must ensure that
+                # all IPs that that DNS entry resolves to are likewise safe.
+                pass
+            else:
+                # Otherwise, we deny access.
+                raise ConfigDoesNotAllowException("Access to this address in not permitted by server configuration")
+    return url

lib/galaxy/tools/actions/upload_common.py

@@ -1,7 +1,5 @@
-import ipaddress
 import logging
 import os
-import socket
 import tempfile
 from dataclasses import dataclass
 from io import StringIO
@@ -14,16 +12,15 @@ from typing import (
     List,
     Optional,
 )
-from urllib.parse import urlparse
 
 from sqlalchemy.orm import joinedload
 from webob.compat import cgi_FieldStorage
 
 from galaxy import util
-from galaxy.datatypes.sniff import stream_to_file
-from galaxy.exceptions import (
-    ConfigDoesNotAllowException,
-    RequestParameterInvalidException,
+from galaxy.exceptions import RequestParameterInvalidException
+from galaxy.files.uris import (
+    stream_to_file,
+    validate_non_local,
 )
 from galaxy.model import (
     FormDefinition,
@@ -32,10 +29,7 @@ from galaxy.model import (
     Role,
     tags,
 )
-from galaxy.util import (
-    is_url,
-    unicodify,
-)
+from galaxy.util import is_url
 from galaxy.util.path import external_chown
 
 log = logging.getLogger(__name__)
@@ -46,90 +40,6 @@ def validate_datatype_extension(datatypes_registry, ext):
         raise RequestParameterInvalidException(f"Requested extension '{ext}' unknown, cannot upload dataset.")
 
 
-def validate_url(url, ip_allowlist):
-    # If it doesn't look like a URL, ignore it.
-    if not (url.lstrip().startswith("http://") or url.lstrip().startswith("https://")):
-        return url
-
-    # Strip leading whitespace before passing url to urlparse()
-    url = url.lstrip()
-    # Extract hostname component
-    parsed_url = urlparse(url).netloc
-    # If credentials are in this URL, we need to strip those.
-    if parsed_url.count("@") > 0:
-        # credentials.
-        parsed_url = parsed_url[parsed_url.rindex("@") + 1 :]
-    # Percent encoded colons and other characters will not be resolved as such
-    # so we don't have to either.
-
-    # Sometimes the netloc will contain the port which is not desired, so we
-    # need to extract that.
-    port = None
-    # However, it could ALSO be an IPv6 address they've supplied.
-    if ":" in parsed_url:
-        # IPv6 addresses have colons in them already (it seems like always more than two)
-        if parsed_url.count(":") >= 2:
-            # Since IPv6 already use colons extensively, they wrap it in
-            # brackets when there is a port, e.g. http://[2001:db8:1f70::999:de8:7648:6e8]:100/
-            # However if it ends with a ']' then there is no port after it and
-            # they've wrapped it in brackets just for fun.
-            if "]" in parsed_url and not parsed_url.endswith("]"):
-                # If this +1 throws a range error, we don't care, their url
-                # shouldn't end with a colon.
-                idx = parsed_url.rindex(":")
-                # We parse as an int and let this fail ungracefully if parsing
-                # fails because we desire to fail closed rather than open.
-                port = int(parsed_url[idx + 1 :])
-                parsed_url = parsed_url[:idx]
-            else:
-                # Plain ipv6 without port
-                pass
-        else:
-            # This should finally be ipv4 with port. It cannot be IPv6 as that
-            # was caught by earlier cases, and it cannot be due to credentials.
-            idx = parsed_url.rindex(":")
-            port = int(parsed_url[idx + 1 :])
-            parsed_url = parsed_url[:idx]
-
-    # safe to log out, no credentials/request path, just an IP + port
-    log.debug("parsed url, port: %s : %s", parsed_url, port)
-    # Call getaddrinfo to resolve hostname into tuples containing IPs.
-    addrinfo = socket.getaddrinfo(parsed_url, port)
-    # Get the IP addresses that this entry resolves to (uniquely)
-    # We drop:
-    #   AF_* family: It will resolve to AF_INET or AF_INET6, getaddrinfo(3) doesn't even mention AF_UNIX,
-    #   socktype: We don't care if a stream/dgram/raw protocol
-    #   protocol: we don't care if it is tcp or udp.
-    addrinfo_results = {info[4][0] for info in addrinfo}
-    # There may be multiple (e.g. IPv4 + IPv6 or DNS round robin). Any one of these
-    # could resolve to a local addresses (and could be returned by chance),
-    # therefore we must check them all.
-    for raw_ip in addrinfo_results:
-        # Convert to an IP object so we can tell if it is in private space.
-        ip = ipaddress.ip_address(unicodify(raw_ip))
-        # If this is a private address
-        if ip.is_private:
-            results = []
-            # If this IP is not anywhere in the allowlist
-            for allowlisted in ip_allowlist:
-                # If it's an IP address range (rather than a single one...)
-                if hasattr(allowlisted, "subnets"):
-                    results.append(ip in allowlisted)
-                else:
-                    results.append(ip == allowlisted)
-
-            if any(results):
-                # If we had any True, then THIS (and ONLY THIS) IP address that
-                # that specific DNS entry resolved to is in allowlisted and
-                # safe to access. But we cannot exit here, we must ensure that
-                # all IPs that that DNS entry resolves to are likewise safe.
-                pass
-            else:
-                # Otherwise, we deny access.
-                raise ConfigDoesNotAllowException("Access to this address in not permitted by server configuration")
-    return url
-
-
 def persist_uploads(params, trans):
     """
     Turn any uploads in the submitted form to persisted files.
@@ -152,7 +62,7 @@ def persist_uploads(params, trans):
                 and upload_dataset["url_paste"].strip() != ""
             ):
                 upload_dataset["url_paste"] = stream_to_file(
-                    StringIO(validate_url(upload_dataset["url_paste"], trans.app.config.fetch_url_allowlist_ips)),
+                    StringIO(validate_non_local(upload_dataset["url_paste"], trans.app.config.fetch_url_allowlist_ips)),
                     prefix="strio_url_paste_",
                 )
             else:

lib/galaxy/tools/data_fetch.py

@@ -21,6 +21,10 @@ from galaxy.datatypes.upload_util import (
     handle_upload,
     UploadProblemException,
 )
+from galaxy.files.uris import (
+    stream_to_file,
+    stream_url_to_file,
+)
 from galaxy.util import (
     in_directory,
     safe_makedirs,
@@ -154,7 +158,7 @@ def _fetch_target(upload_config, target):
             dataset_bunch = Bunch(
                 name=name,
             )
-            primary_file = sniff.stream_to_file(
+            primary_file = stream_to_file(
                 StringIO(datatype.generate_primary_file(dataset_bunch)), prefix="upload_auto_primary_file", dir="."
             )
             extra_files_path = f"{primary_file}_extra"
@@ -422,7 +426,7 @@ def _has_src_to_path(upload_config, item, is_dataset=False) -> Tuple[str, str]:
     if src == "url":
         url = item.get("url")
         try:
-            path = sniff.stream_url_to_file(url, file_sources=get_file_sources(upload_config.working_directory))
+            path = stream_url_to_file(url, file_sources=get_file_sources(upload_config.working_directory))
         except Exception as e:
             raise Exception(f"Failed to fetch url {url}. {str(e)}")
 
@@ -436,7 +440,7 @@ def _has_src_to_path(upload_config, item, is_dataset=False) -> Tuple[str, str]:
         if name is None:
             name = url.split("/")[-1]
     elif src == "pasted":
-        path = sniff.stream_to_file(StringIO(item["paste_content"]))
+        path = stream_to_file(StringIO(item["paste_content"]))
         if name is None:
             name = "Pasted Entry"
     else:

lib/galaxy/tools/imp_exp/unpack_tar_gz_archive.py

@@ -13,16 +13,16 @@ import os
 import tarfile
 from base64 import b64decode
 
-from galaxy.datatypes import sniff
+from galaxy.files import ConfiguredFileSources
+from galaxy.files.uris import stream_url_to_file
 
 # Set max size of archive/file that will be handled to be 100 GB. This is
 # arbitrary and should be adjusted as needed.
 MAX_SIZE = 100 * math.pow(2, 30)
 
 
-def get_file_sources(file_sources_path):
+def get_file_sources(file_sources_path) -> ConfiguredFileSources:
     assert os.path.exists(file_sources_path), f"file sources path [{file_sources_path}] does not exist"
-    from galaxy.files import ConfiguredFileSources
 
     with open(file_sources_path) as f:
         file_sources_as_dict = json.load(f)
@@ -63,7 +63,9 @@ def main(options, args):
 
     # Get archive from URL.
     if is_url:
-        archive_file = sniff.stream_url_to_file(archive_source, file_sources=get_file_sources(options.file_sources))
+        archive_file = stream_url_to_file(
+            archive_source, file_sources=get_file_sources(options.file_sources), prefix="gx_history_archive"
+        )
     elif is_file:
         archive_file = archive_source
 

lib/galaxy/tools/parameters/grouping.py

@@ -15,14 +15,12 @@ from typing import (
     TYPE_CHECKING,
 )
 
-from galaxy.datatypes import (
-    data,
-    sniff,
-)
+from galaxy.datatypes import data
 from galaxy.exceptions import (
     AdminRequiredException,
     ConfigDoesNotAllowException,
 )
+from galaxy.files.uris import stream_to_file
 from galaxy.util import (
     asbool,
     inflector,
@@ -656,7 +654,7 @@ class UploadDataset(Group):
             dataset.name = self.get_composite_dataset_name(context)
             if dataset.datatype.composite_type == "auto_primary_file":
                 # replace sniff here with just creating an empty file
-                temp_name = sniff.stream_to_file(
+                temp_name = stream_to_file(
                     io.StringIO(d_type.generate_primary_file(dataset)), prefix="upload_auto_primary_file"
                 )
                 dataset.primary_file = temp_name

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

@@ -13,7 +13,6 @@ from typing import (
     Optional,
 )
 
-import requests
 from fastapi import (
     Body,
     Path,
@@ -29,6 +28,10 @@ from galaxy import (
     model,
     util,
 )
+from galaxy.files.uris import (
+    stream_url_to_str,
+    validate_uri_access,
+)
 from galaxy.managers.context import ProvidesUserContext
 from galaxy.managers.jobs import (
     fetch_job_states,
@@ -305,9 +308,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)
@@ -319,7 +321,9 @@ class WorkflowsAPIController(BaseGalaxyAPIController, UsesStoredWorkflowMixin, U
                     archive_data = self.app.trs_proxy.get_version_descriptor(trs_server, trs_tool_id, trs_version_id)
                 else:
                     try:
-                        archive_data = requests.get(archive_source, timeout=util.DEFAULT_SOCKET_TIMEOUT).text
+                        archive_data = stream_url_to_str(
+                            archive_source, trans.app.file_sources, prefix="gx_workflow_download"
+                        )
                         import_source = "URL"
                     except Exception:
                         raise exceptions.MessageException(f"Failed to open URL '{escape(archive_source)}'.")

lib/galaxy/webapps/galaxy/services/_fetch_util.py

@@ -6,14 +6,12 @@ from galaxy.actions.library import (
     validate_server_directory_upload,
 )
 from galaxy.exceptions import RequestParameterInvalidException
+from galaxy.files.uris import validate_non_local
 from galaxy.model.store.discover import (
     get_required_item,
     replace_request_syntax_sugar,
 )
-from galaxy.tools.actions.upload_common import (
-    validate_datatype_extension,
-    validate_url,
-)
+from galaxy.tools.actions.upload_common import validate_datatype_extension
 from galaxy.util import relpath
 
 log = logging.getLogger(__name__)
@@ -163,7 +161,7 @@ def validate_and_normalize_targets(trans, payload):
             if not looks_like_url:
                 raise RequestParameterInvalidException(f"Invalid URL [{url}] found in src definition.")
 
-            validate_url(url, trans.app.config.fetch_url_allowlist_ips)
+            validate_non_local(url, trans.app.config.fetch_url_allowlist_ips)
             item["in_place"] = run_as_real_user
         elif src == "files":
             item["in_place"] = run_as_real_user

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/api/test_workflows.py

@@ -1,3 +1,4 @@
+import base64
 import json
 import os
 import shutil
@@ -29,6 +30,7 @@ from galaxy_test.base.populators import (
     RunJobsSummary,
     skip_without_tool,
     wait_on,
+    workflow_str,
     WorkflowPopulator,
 )
 from galaxy_test.base.workflow_fixtures import (
@@ -967,6 +969,15 @@ steps:
             workflow.get("source_metadata").get("url") == url
         )  # disappearance of source_metadata on modification is tested in test_trs_import
 
+    def test_base64_import(self):
+        base64_url = "base64://" + base64.b64encode(workflow_str.encode("utf-8")).decode("utf-8")
+        response = self._post("workflows", data={"archive_source": base64_url})
+        print(response.content)
+        response.raise_for_status()
+        workflow_id = response.json()["id"]
+        workflow = self._download_workflow(workflow_id)
+        assert "TestWorkflow1" in workflow["name"]
+
     def test_trs_import(self):
         trs_payload = {
             "archive_source": "trs_tool",

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

@@ -0,0 +1,55 @@
+import ipaddress
+
+from galaxy.exceptions import (
+    AdminRequiredException,
+    ConfigDoesNotAllowException,
+)
+from galaxy.files.uris import (
+    validate_non_local,
+    validate_uri_access,
+)
+
+
+def test_validate_local_ip_access():
+    assert validates_as_non_local("http://google.com", [])
+    assert not validates_as_non_local("http://127.0.0.1/secrets.txt", [])
+    assert validates_as_non_local("http://127.0.0.1/secrets.txt", [ipaddress.ip_address("127.0.0.1")])
+    assert validates_as_non_local("http://127.0.0.1/secrets.txt", [ipaddress.ip_network("127.0.0.0/24")])
+    assert not validates_as_non_local("http://127.0.0.1/secrets.txt", [ipaddress.ip_network("127.0.0.0/32")])
+    assert validates_as_non_local(
+        "http://127.0.0.1/secrets.txt", [ipaddress.ip_network("127.0.0.0/32"), ipaddress.ip_network("127.0.0.1/32")]
+    )
+
+
+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

tools/data_source/data_source.py

@@ -19,6 +19,7 @@ from galaxy.jobs import TOOL_PROVIDED_JOB_METADATA_FILE
 from galaxy.util import (
     DEFAULT_SOCKET_TIMEOUT,
     get_charset_from_http_headers,
+    stream_to_open_named_file,
 )
 
 GALAXY_PARAM_PREFIX = "GALAXY"
@@ -107,7 +108,7 @@ def __main__():
                     % (file_size, max_file_size)
                 )
         try:
-            cur_filename = sniff.stream_to_open_named_file(
+            cur_filename = stream_to_open_named_file(
                 page,
                 os.open(cur_filename, os.O_WRONLY | os.O_CREAT),
                 cur_filename,