Loading lib/galaxy/managers/workflows.py +3 −2 Original line number Diff line number Diff line Loading @@ -44,6 +44,7 @@ from galaxy import ( from galaxy.job_execution.actions.post import ActionBox from galaxy.managers import sharable from galaxy.managers.context import ProvidesUserContext from galaxy.model import StoredWorkflow from galaxy.model.index_filter_util import ( append_user_filter, raw_text_column_filter, Loading Loading @@ -228,7 +229,7 @@ class WorkflowsManager(sharable.SharableModelManager): query = query.offset(payload.offset) return query, total_matches def get_stored_workflow(self, trans, workflow_id, by_stored_id=True): def get_stored_workflow(self, trans, workflow_id, by_stored_id=True) -> StoredWorkflow: """Use a supplied ID (UUID or encoded stored workflow ID) to find a workflow. """ Loading Loading @@ -276,7 +277,7 @@ class WorkflowsManager(sharable.SharableModelManager): stored_workflow = self.get_stored_workflow(trans, workflow_id, by_stored_id=by_stored_id) # check to see if user has permissions to selected workflow if stored_workflow.user != trans.user and not trans.user_is_admin and not stored_workflow.published: if stored_workflow.user != trans.user and not trans.user_is_admin and not stored_workflow.importable: if ( trans.sa_session.query(trans.app.model.StoredWorkflowUserShareAssociation) .filter_by(user=trans.user, stored_workflow=stored_workflow) Loading lib/galaxy/structured_app.py +2 −1 Original line number Diff line number Diff line Loading @@ -38,6 +38,7 @@ from galaxy.workflow.trs_proxy import TrsProxy if TYPE_CHECKING: from galaxy.jobs import JobConfiguration from galaxy.managers.workflows import WorkflowsManager from galaxy.tools.data import ToolDataTableManager Loading Loading @@ -136,7 +137,7 @@ class StructuredApp(MinimalManagerApp): queue_worker: Any # 'galaxy.queue_worker.GalaxyQueueWorker' history_manager: Any # 'galaxy.managers.histories.HistoryManager' hda_manager: Any # 'galaxy.managers.hdas.HDAManager' workflow_manager: Any # 'galaxy.managers.workflows.WorkflowsManager' workflow_manager: "WorkflowsManager" workflow_contents_manager: Any # 'galaxy.managers.workflows.WorkflowContentsManager' library_folder_manager: Any # 'galaxy.managers.folders.FolderManager' library_manager: Any # 'galaxy.managers.libraries.LibraryManager' Loading lib/galaxy_test/api/test_workflows.py +7 −0 Original line number Diff line number Diff line Loading @@ -347,6 +347,13 @@ class WorkflowsApiTestCase(BaseWorkflowsApiTestCase, ChangeDatatypeTestCase): workflows_url = self._api_url(f"workflows/{workflow_id}/download") assert get(workflows_url).status_code == 403 def test_anon_can_download_importable_workflow(self): workflow_id = self.workflow_populator.simple_workflow("test_downloadable", importable=True) workflows_url = self._api_url(f"workflows/{workflow_id}/download") response = get(workflows_url) response.raise_for_status() assert response.json()["a_galaxy_workflow"] == "true" def test_anon_can_download_public_workflow(self): workflow_id = self.workflow_populator.simple_workflow("test_downloadable", publish=True) workflows_url = self._api_url(f"workflows/{workflow_id}/download") Loading Loading
lib/galaxy/managers/workflows.py +3 −2 Original line number Diff line number Diff line Loading @@ -44,6 +44,7 @@ from galaxy import ( from galaxy.job_execution.actions.post import ActionBox from galaxy.managers import sharable from galaxy.managers.context import ProvidesUserContext from galaxy.model import StoredWorkflow from galaxy.model.index_filter_util import ( append_user_filter, raw_text_column_filter, Loading Loading @@ -228,7 +229,7 @@ class WorkflowsManager(sharable.SharableModelManager): query = query.offset(payload.offset) return query, total_matches def get_stored_workflow(self, trans, workflow_id, by_stored_id=True): def get_stored_workflow(self, trans, workflow_id, by_stored_id=True) -> StoredWorkflow: """Use a supplied ID (UUID or encoded stored workflow ID) to find a workflow. """ Loading Loading @@ -276,7 +277,7 @@ class WorkflowsManager(sharable.SharableModelManager): stored_workflow = self.get_stored_workflow(trans, workflow_id, by_stored_id=by_stored_id) # check to see if user has permissions to selected workflow if stored_workflow.user != trans.user and not trans.user_is_admin and not stored_workflow.published: if stored_workflow.user != trans.user and not trans.user_is_admin and not stored_workflow.importable: if ( trans.sa_session.query(trans.app.model.StoredWorkflowUserShareAssociation) .filter_by(user=trans.user, stored_workflow=stored_workflow) Loading
lib/galaxy/structured_app.py +2 −1 Original line number Diff line number Diff line Loading @@ -38,6 +38,7 @@ from galaxy.workflow.trs_proxy import TrsProxy if TYPE_CHECKING: from galaxy.jobs import JobConfiguration from galaxy.managers.workflows import WorkflowsManager from galaxy.tools.data import ToolDataTableManager Loading Loading @@ -136,7 +137,7 @@ class StructuredApp(MinimalManagerApp): queue_worker: Any # 'galaxy.queue_worker.GalaxyQueueWorker' history_manager: Any # 'galaxy.managers.histories.HistoryManager' hda_manager: Any # 'galaxy.managers.hdas.HDAManager' workflow_manager: Any # 'galaxy.managers.workflows.WorkflowsManager' workflow_manager: "WorkflowsManager" workflow_contents_manager: Any # 'galaxy.managers.workflows.WorkflowContentsManager' library_folder_manager: Any # 'galaxy.managers.folders.FolderManager' library_manager: Any # 'galaxy.managers.libraries.LibraryManager' Loading
lib/galaxy_test/api/test_workflows.py +7 −0 Original line number Diff line number Diff line Loading @@ -347,6 +347,13 @@ class WorkflowsApiTestCase(BaseWorkflowsApiTestCase, ChangeDatatypeTestCase): workflows_url = self._api_url(f"workflows/{workflow_id}/download") assert get(workflows_url).status_code == 403 def test_anon_can_download_importable_workflow(self): workflow_id = self.workflow_populator.simple_workflow("test_downloadable", importable=True) workflows_url = self._api_url(f"workflows/{workflow_id}/download") response = get(workflows_url) response.raise_for_status() assert response.json()["a_galaxy_workflow"] == "true" def test_anon_can_download_public_workflow(self): workflow_id = self.workflow_populator.simple_workflow("test_downloadable", publish=True) workflows_url = self._api_url(f"workflows/{workflow_id}/download") Loading