Rebase into...

lib/galaxy/managers/jobs.py

@@ -106,6 +106,11 @@ def __init__(self, app: StructuredApp):
         self.dataset_manager = DatasetManager(app)
 
     def index_query(self, trans, payload: JobIndexQueryPayload) -> sqlalchemy.engine.Result:
+        """The caller is responsible for security checks on the resulting job if
+        history_id, invocation_id, or implicit_collection_jobs_id is set.
+        Otherwise this will only return the user's jobs or all jobs if the requesting
+        user is acting as an admin.
+        """
         is_admin = trans.user_is_admin
         user_details = payload.user_details
         decoded_user_id = payload.user_id
@@ -202,7 +207,9 @@ def add_search_criteria(stmt):
             if user_details:
                 stmt = stmt.outerjoin(Job.user)
         else:
-            stmt = stmt.where(Job.user_id == trans.user.id)
+            if history_id is None and invocation_id is None and implicit_collection_jobs_id is None:
+                stmt = stmt.where(Job.user_id == trans.user.id)
+            # caller better check security
 
         stmt = build_and_apply_filters(stmt, payload.states, lambda s: model.Job.state == s)
         stmt = build_and_apply_filters(stmt, payload.tool_ids, lambda t: model.Job.tool_id == t)

lib/galaxy/webapps/galaxy/api/jobs.py

@@ -529,6 +529,25 @@ def delete(
             message = None
         return self.service.job_manager.stop(job, message=message)
 
+    @router.get(
+        "/api/implicit_collection_jobs/{implicit_collection_job_id}",
+        name="get implicit collection jobs",
+        operation_id="implicit_collection_jobs__get",
+        summary="Get a list of implicit collection jobs associated with specified ID",
+    )
+    def get_implicit_collection_jobs(
+        self,
+        job_id: Annotated[DecodedDatabaseIdField, JobIdPathParam],
+        trans: ProvidesUserContext = DependsOnTrans,
+        payload: Annotated[Optional[DeleteJobPayload], DeleteJobBody] = None,
+    ) -> bool:
+        job = self.service.get_job(trans=trans, job_id=job_id)
+        if payload:
+            message = payload.message
+        else:
+            message = None
+        return self.service.job_manager.stop(job, message=message)
+
 
 class JobController(BaseGalaxyAPIController, UsesVisualizationMixin):
     job_manager = depends(JobManager)

lib/galaxy/webapps/galaxy/services/jobs.py

@@ -10,6 +10,7 @@
     exceptions,
     model,
 )
+from galaxy.managers.base import security_check
 from galaxy.managers import hdas
 from galaxy.managers.context import ProvidesUserContext
 from galaxy.managers.jobs import (
@@ -74,13 +75,16 @@ def index(
             payload.user_details = True
         user_details = payload.user_details
         decoded_user_id = payload.user_id
-
         if not is_admin:
             self._check_nonadmin_access(view, user_details, decoded_user_id, trans.user.id)
 
+        check_security_of_jobs = payload.invocation_id is not None or payload.implicit_collection_jobs_id is not None or payload.history_id is not None
         jobs = self.job_manager.index_query(trans, payload)
         out = []
         for job in jobs.yield_per(model.YIELD_PER_ROWS):
+            # TODO: optimize if this crucial
+            if check_security_of_jobs and not security_check(trans, job.history, check_accessible=True):
+                raise exceptions.ItemAccessibilityException("Cannot access the request job objects.")
             job_dict = job.to_dict(view, system_details=is_admin)
             j = security.encode_all_ids(job_dict, True)
             if view == JobIndexViewEnum.admin_job_list: