Enforce file source access during dataset materialization

Previously, DatasetInstanceMaterializer._stream_source called
stream_url_to_file without a user_context, which caused
FilesSource._check_user_access to silently skip the roles/groups check
for gxfiles:// URIs (see the comment on _check_user_access: "if the
user_context is None, then the check is skipped"). As a result, a
workflow invoked by a user without access to a group-restricted file
source would still successfully materialize data from it.

Thread user_context through DatasetInstanceMaterializer (and
materializer_factory) and construct one in both callers:

- HDAManager.materialize wraps the RequestUser in a WorkRequestContext
  and ProvidesFileSourcesUserContext. This is the path exercised by
  the workflow scheduler when undeferring workflow inputs.
- ToolEvaluator._materialize_objects does the same using its own
  _user / _history. This is the path exercised when a tool
  consumes a deferred input at job run time.

lib/galaxy/managers/hdas.py

@@ -36,6 +36,7 @@ from galaxy import (
     exceptions,
     model,
 )
+from galaxy.files import ProvidesFileSourcesUserContext
 from galaxy.managers import (
     annotatable,
     base,
@@ -81,6 +82,7 @@ from galaxy.tool_util_models.parameters import (
     FileRequestUri,
 )
 from galaxy.util.compression_utils import get_fileobj
+from galaxy.work.context import WorkRequestContext
 
 if TYPE_CHECKING:
     from galaxy.model import LibraryDatasetDatasetAssociation
@@ -176,13 +178,15 @@ class HDAManager(
         self, request: MaterializeDatasetInstanceTaskRequest, session: Session, in_place: bool = False
     ) -> bool:
         request_user: RequestUser = request.user
+        user = self.user_manager.by_id(request_user.user_id)
+        user_context = ProvidesFileSourcesUserContext(WorkRequestContext(app=self.app, user=user))
         materializer = materializer_factory(
             True,  # attached...
             object_store=self.app.object_store,
             file_sources=self.app.file_sources,
             sa_session=session,
+            user_context=user_context,
         )
-        user = self.user_manager.by_id(request_user.user_id)
         if request.source == DatasetSourceType.hda:
             dataset_instance: Union[HistoryDatasetAssociation, LibraryDatasetDatasetAssociation] = self.get_accessible(
                 request.content, user

lib/galaxy/model/deferred.py

@@ -16,7 +16,10 @@ from galaxy.datatypes.sniff import (
     stream_url_to_file,
 )
 from galaxy.exceptions import ObjectAttributeInvalidException
-from galaxy.files import ConfiguredFileSources
+from galaxy.files import (
+    ConfiguredFileSources,
+    OptionalUserContext,
+)
 from galaxy.model import (
     Dataset,
     DatasetCollection,
@@ -79,18 +82,24 @@ class DatasetInstanceMaterializer:
         transient_path_mapper: Optional[TransientPathMapper] = None,
         file_sources: Optional[ConfiguredFileSources] = None,
         sa_session: Optional[Session] = None,
+        user_context: OptionalUserContext = None,
     ):
         """Constructor for DatasetInstanceMaterializer.
 
         If attached is true, these objects should be created in a supplied object store.
         If not, this class produces transient HDAs with external_filename and
         external_extra_files_path set.
+
+        ``user_context`` is forwarded to file source operations so that access
+        controls (``requires_roles`` / ``requires_groups``) are enforced when
+        materializing from ``gxfiles://`` URIs.
         """
         self._attached = attached
         self._transient_path_mapper = transient_path_mapper
         self._object_store_populator = object_store_populator
         self._file_sources = file_sources
         self._sa_session = sa_session
+        self._user_context = user_context
         self._previously_materialized: dict[int, HistoryDatasetAssociation] = {}
 
     def ensure_materialized(
@@ -252,7 +261,7 @@ class DatasetInstanceMaterializer:
         source_uri = target_source.source_uri
         if source_uri is None:
             raise Exception("Cannot stream from dataset source without specified source_uri")
-        path = stream_url_to_file(source_uri, file_sources=self._file_sources)
+        path = stream_url_to_file(source_uri, file_sources=self._file_sources, user_context=self._user_context)
         if target_source.hashes:
             for source_hash in target_source.hashes:
                 _validate_hash(path, source_hash, "downloaded file")
@@ -385,6 +394,7 @@ def materializer_factory(
     transient_directory: Optional[str] = None,
     file_sources: Optional[ConfiguredFileSources] = None,
     sa_session: Optional[Session] = None,
+    user_context: OptionalUserContext = None,
 ) -> DatasetInstanceMaterializer:
     if object_store_populator is None and object_store is not None:
         object_store_populator = ObjectStorePopulator(object_store, None)
@@ -396,6 +406,7 @@ def materializer_factory(
         transient_path_mapper=transient_path_mapper,
         file_sources=file_sources,
         sa_session=sa_session,
+        user_context=user_context,
     )
 
 

lib/galaxy/tools/evaluation.py

@@ -19,6 +19,7 @@ from packaging.version import Version
 
 from galaxy import model
 from galaxy.authnz.util import provider_name_to_backend
+from galaxy.files import ProvidesFileSourcesUserContext
 from galaxy.job_execution.compute_environment import ComputeEnvironment
 from galaxy.job_execution.datasets import DeferrableObjectsT
 from galaxy.job_execution.setup import ensure_configs_directory
@@ -295,10 +296,14 @@ class ToolEvaluator:
         undeferred_objects: dict[str, DeferrableObjectsT] = {}
         transient_directory = os.path.join(job_working_directory, "inputs")
         safe_makedirs(transient_directory)
+        user_context = ProvidesFileSourcesUserContext(
+            WorkRequestContext(app=self.app, user=self._user, history=self._history)
+        )
         dataset_materializer = materializer_factory(
             False,  # unattached to a session.
             transient_directory=transient_directory,
             file_sources=self.app.file_sources,
+            user_context=user_context,
         )
         for key, value in deferred_objects.items():
             if isinstance(value, model.DatasetInstance):

test/integration/test_remote_files_posix.py

@@ -4,7 +4,11 @@ from sqlalchemy import select
 
 from galaxy.model import Dataset
 from galaxy_test.base import api_asserts
-from galaxy_test.base.populators import DatasetPopulator
+from galaxy_test.base.populators import (
+    DatasetPopulator,
+    WorkflowPopulator,
+)
+from galaxy_test.base.workflow_fixtures import WORKFLOW_SIMPLE_CAT_TWICE
 from galaxy_test.driver import integration_util
 from galaxy_test.driver.integration_setup import (
     GROUP_A,
@@ -17,6 +21,7 @@ from galaxy_test.driver.integration_setup import (
 
 class TestPosixFileSourceIntegration(PosixFileSourceSetup, integration_util.IntegrationTestCase):
     dataset_populator: DatasetPopulator
+    framework_tool_and_types = True
     required_role_expression = REQUIRED_ROLE_EXPRESSION
     required_group_expression = REQUIRED_GROUP_EXPRESSION
 
@@ -24,6 +29,7 @@ class TestPosixFileSourceIntegration(PosixFileSourceSetup, integration_util.Inte
         super().setUp()
         self._write_file_fixtures()
         self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
+        self.workflow_populator = WorkflowPopulator(self.galaxy_interactor)
 
     def test_plugin_config(self):
         # Default user has required role but not required group, so cannot see plugin
@@ -68,6 +74,92 @@ class TestPosixFileSourceIntegration(PosixFileSourceSetup, integration_util.Inte
         list_response = self.galaxy_interactor.get("remote_files", data)
         self._assert_list_response_matches_fixtures(list_response)
 
+    def test_workflow_input_materialized_from_group_file_source(self):
+        # Ensure the default non-admin user is in the required group so the
+        # file source is accessible.
+        group_a_id = self._ensure_group(GROUP_A)
+        self._add_user_to_group(group_a_id, self.dataset_populator.user_id())
+
+        with self.dataset_populator.test_history() as history_id:
+            # Pass the workflow input as a URL at invocation time so the scheduler
+            # creates a deferred HDA and then materializes it from the
+            # group-restricted file source (requires_materialization path in
+            # galaxy.workflow.run_request).
+            workflow_id, response = self._invoke_gxfiles_posix_test_workflow(history_id)
+            invocation_id = response.json()["id"]
+            self.workflow_populator.wait_for_workflow(workflow_id, invocation_id, history_id, assert_ok=True)
+
+            # The input was materialized from gxfiles://posix_test/a; its fixture
+            # content is "a\n", so cat of it concatenated with itself is "a\na\n".
+            input_hda = self.dataset_populator.get_history_dataset_details(history_id, hid=1)
+            assert input_hda["state"] == "ok", input_hda
+            cat_output = self.dataset_populator.get_history_dataset_content(history_id, hid=2)
+            assert cat_output == "a\na\n", cat_output
+
+    def test_workflow_input_materialization_denied_without_group(self):
+        # Counterpart to test_workflow_input_materialized_from_group_file_source:
+        # a user who is not a member of any of the groups required by posix_test
+        # cannot see the file source, and any workflow invocation that tries to
+        # materialize a gxfiles://posix_test/... input must fail rather than
+        # silently leak data across the group boundary.
+        with self._different_user("wf_no_fs_access_user@bx.psu.edu"):
+            # The restricted plugin is not advertised to this user.
+            plugin_config_response = self.galaxy_interactor.get("remote_files/plugins")
+            api_asserts.assert_status_code_is_ok(plugin_config_response)
+            plugins = plugin_config_response.json()
+            assert all(p["uri_root"] != "gxfiles://posix_test" for p in plugins), plugins
+
+            # And direct access to the file source is forbidden.
+            list_response = self.galaxy_interactor.get("remote_files", {"target": "gxfiles://posix_test"})
+            self._assert_access_forbidden_response(list_response)
+
+            # Invoking a workflow with a gxfiles://posix_test/... input must not
+            # result in the file being materialized into this user's history.
+            with self.dataset_populator.test_history() as history_id:
+                workflow_id, response = self._invoke_gxfiles_posix_test_workflow(history_id)
+                invocation_id = response.json()["id"]
+                # Wait for the invocation to reach a terminal state; materialization
+                # should fail because the file source is not accessible to this user.
+                self.workflow_populator.wait_for_invocation(workflow_id, invocation_id, assert_ok=False)
+                invocation = self.workflow_populator.get_invocation(invocation_id)
+                assert invocation["state"] == "failed", invocation
+                # The scheduler records a `dataset_failed` message pointing at the
+                # HDA whose materialization failed. The ItemAccessibilityException
+                # from _check_user_access is surfaced on that HDA's `info` field.
+                messages = invocation["messages"]
+                dataset_failed = [m for m in messages if m["reason"] == "dataset_failed"]
+                assert dataset_failed, messages
+                failed_hda = self.dataset_populator.get_history_dataset_details(history_id, hid=1, assert_ok=False)
+                assert failed_hda["state"] == "error", failed_hda
+                assert "no access to file source" in failed_hda["misc_info"], failed_hda
+
+    def _invoke_gxfiles_posix_test_workflow(self, history_id: str):
+        """Submit WORKFLOW_SIMPLE_CAT_TWICE with a single gxfiles://posix_test/a
+        URL input and return (workflow_id, raw invocation response)."""
+        workflow_id = self.workflow_populator.upload_yaml_workflow(WORKFLOW_SIMPLE_CAT_TWICE)
+        workflow_request = {
+            "history": f"hist_id={history_id}",
+            "inputs": {
+                "input1": {
+                    "src": "url",
+                    "url": "gxfiles://posix_test/a",
+                    "ext": "txt",
+                    "deferred": False,
+                },
+            },
+            "inputs_by": "name",
+        }
+        return workflow_id, self.workflow_populator.invoke_workflow_raw(workflow_id, workflow_request, assert_ok=True)
+
+    def _ensure_group(self, group_name: str) -> str:
+        """Idempotent: return the id of an existing group with this name, or create it."""
+        groups_response = self._get("groups", admin=True)
+        api_asserts.assert_status_code_is_ok(groups_response)
+        for group in groups_response.json():
+            if group["name"] == group_name:
+                return group["id"]
+        return self._create_group(group_name)
+
     def _create_group(self, group_name: str):
         payload = {
             "name": group_name,