Merge pull request #16143 from davelopez/improve_history_immutability_checks

Improve histories and datasets immutability checks

lib/galaxy/exceptions/__init__.py

@@ -177,6 +177,11 @@ class ItemOwnershipException(MessageException):
     err_code = error_codes_by_name["USER_DOES_NOT_OWN_ITEM"]
 
 
+class ItemImmutableException(MessageException):
+    status_code = 403
+    err_code = error_codes_by_name["ITEM_IS_IMMUTABLE"]
+
+
 class ConfigDoesNotAllowException(MessageException):
     status_code = 403
     err_code = error_codes_by_name["CONFIG_DOES_NOT_ALLOW"]

lib/galaxy/exceptions/error_codes.json

@@ -1,192 +1,197 @@
 [
-   {
-      "name": "UNKNOWN",
-      "code": 0,
-      "message": "Unknown error occurred while processing request."
-   },
-   {
-      "name": "ACCEPTED_RETRY_LATER",
-      "code": 202001,
-      "message": "Galaxy has accepted this request and is processing. A retry-after header indicates to the client when to retry."
-   },
-   {
-      "name": "NO_CONTENT_GENERIC",
-      "code": 204001,
-      "message": "Galaxy has no content to associate with this request."
-   },
-   {
-      "name": "USER_CANNOT_RUN_AS",
-      "code": 400001,
-      "message": "User does not have permissions to run jobs as another user."
-   },
-   {
-      "name": "USER_INVALID_RUN_AS",
-      "code": 400002,
-      "message": "Invalid run_as request - run_as user does not exist."
-   },
-   {
-      "name": "USER_INVALID_JSON",
-      "code": 400003,
-      "message": "Your request did not appear to be valid JSON, please consult the API documentation."
-   },
-   {
-      "name": "USER_OBJECT_ATTRIBUTE_INVALID",
-      "code": 400004,
-      "message": "Attempted to create or update object with invalid attribute value."
-   },
-   {
-      "name": "USER_OBJECT_ATTRIBUTE_MISSING",
-      "code": 400005,
-      "message": "Attempted to create or update object without required attribute."
-   },
-   {
-      "name": "USER_SLUG_DUPLICATE",
-      "code": 400006,
-      "message": "Slug must be unique per user."
-   },
-   {
-      "name": "USER_REQUEST_MISSING_PARAMETER",
-      "code": 400007,
-      "message": "Request is missing parameter required to complete desired action."
-   },
-   {
-      "name": "USER_REQUEST_INVALID_PARAMETER",
-      "code": 400008,
-      "message": "Request contained invalid parameter, action could not be completed."
-   },
-   {
-      "name": "MALFORMED_ID",
-      "code": 400009,
-      "message": "The id of the resource is malformed."
-   },
-   {
-      "name": "UNKNOWN_CONTENTS_TYPE",
-      "code": 400010,
-      "message": "The request contains unknown type of contents."
-   },
-   {
-      "name": "USER_IDENTIFIER_DUPLICATE",
-      "code": 400011,
-      "message": "Request contained a duplicated identifier that must be unique."
-   },
-   {
-      "name": "USER_TOOL_META_PARAMETER_PROBLEM",
-      "code": 400012,
-      "message": "Supplied incorrect or incompatible tool meta parameters."
-   },
-   {
-      "name": "MALFORMED_CONTENTS",
-      "code": 400013,
-      "message": "The contents of the request are malformed."
-   },
-   {
-      "name": "USER_TOOL_MISSING_PROBLEM",
-      "code": 400014,
-      "message": "Tool could not be found."
-   },
-   {
-      "name": "TOOL_INPUTS_NOT_READY",
-      "code": 400015,
-      "message": "Tool inputs not yet ready, try again later."
-   },
-   {
-      "name": "REAL_USER_REQUIRED",
-      "code": 400016,
-      "message": "Only real users can make this request."
-   },
-   {
-      "name": "USER_AUTHENTICATION_FAILED",
-      "code": 401001,
-      "message": "Authentication failed, invalid credentials supplied."
-   },
-   {
-      "name": "USER_NO_API_KEY",
-      "code": 403001,
-      "message": "Authentication required for this request"
-   },
-   {
-      "name": "USER_CANNOT_ACCESS_ITEM",
-      "code": 403002,
-      "message": "User cannot access specified item."
-   },
-   {
-      "name": "USER_DOES_NOT_OWN_ITEM",
-      "code": 403003,
-      "message": "User does not own specified item."
-   },
-   {
-      "name": "CONFIG_DOES_NOT_ALLOW",
-      "code": 403004,
-      "message": "The configuration of this Galaxy instance does not allow that operation"
-   },
-   {
-      "name": "INSUFFICIENT_PERMISSIONS",
-      "code": 403005,
-      "message": "You don't have proper permissions to perform the requested operation"
-   },
-   {
-      "name": "ADMIN_REQUIRED",
-      "code": 403006,
-      "message": "Action requires admin account."
-   },
-   {
-      "name": "USER_ACTIVATION_REQUIRED",
-      "code": 403007,
-      "message": "Action requires account activation."
-   },
-   {
-      "name": "USER_OBJECT_NOT_FOUND",
-      "code": 404001,
-      "message": "No such object found."
-   },
-   {
-      "name": "CONFLICT",
-      "code": 409001,
-      "message": "Database conflict prevented fulfilling the request."
-   },
-   {
-      "name": "DEPRECATED_API_CALL",
-      "code": 410001,
-      "message": "This API method or call signature has been deprecated and is no longer available"
-   },
-   {
-      "name": "INTERNAL_SERVER_ERROR",
-      "code": 500001,
-      "message": "Internal server error."
-   },
-   {
-      "name": "INCONSISTENT_DATABASE",
-      "code": 500002,
-      "message": "Inconsistent database prevented fulfilling the request."
-   },
-   {
-      "name": "CONFIG_ERROR",
-      "code": 500003,
-      "message": "Error in a configuration file."
-   },
-   {
-      "name": "TOOL_EXECUTION_ERROR",
-      "code": 500004,
-      "message": "Tool execution failed due to an internal server error."
-   },
-   {
-      "name": "INVALID_FILE_FORMAT",
-      "code": 500005,
-      "message": "File format not supported for this operation."
-   },
-   {
-      "name": "REFERENCE_DATA_ERROR",
-      "code": 500006,
-      "message": "Reference data required for program execution failed to load."
-   },
-   {
-      "name": "NOT_IMPLEMENTED",
-      "code": 501001,
-      "message": "Method is not implemented."
-   },
-   {
-      "name": "SERVER_NOT_CONFIGURED_FOR_REQUEST",
-      "code": 501002,
-      "message": "Server not configured for the request. The Galaxy admin may be able to resolve the problem by installing additional dependencies or setting up new infrastructure."
-   }
+    {
+        "name": "UNKNOWN",
+        "code": 0,
+        "message": "Unknown error occurred while processing request."
+    },
+    {
+        "name": "ACCEPTED_RETRY_LATER",
+        "code": 202001,
+        "message": "Galaxy has accepted this request and is processing. A retry-after header indicates to the client when to retry."
+    },
+    {
+        "name": "NO_CONTENT_GENERIC",
+        "code": 204001,
+        "message": "Galaxy has no content to associate with this request."
+    },
+    {
+        "name": "USER_CANNOT_RUN_AS",
+        "code": 400001,
+        "message": "User does not have permissions to run jobs as another user."
+    },
+    {
+        "name": "USER_INVALID_RUN_AS",
+        "code": 400002,
+        "message": "Invalid run_as request - run_as user does not exist."
+    },
+    {
+        "name": "USER_INVALID_JSON",
+        "code": 400003,
+        "message": "Your request did not appear to be valid JSON, please consult the API documentation."
+    },
+    {
+        "name": "USER_OBJECT_ATTRIBUTE_INVALID",
+        "code": 400004,
+        "message": "Attempted to create or update object with invalid attribute value."
+    },
+    {
+        "name": "USER_OBJECT_ATTRIBUTE_MISSING",
+        "code": 400005,
+        "message": "Attempted to create or update object without required attribute."
+    },
+    {
+        "name": "USER_SLUG_DUPLICATE",
+        "code": 400006,
+        "message": "Slug must be unique per user."
+    },
+    {
+        "name": "USER_REQUEST_MISSING_PARAMETER",
+        "code": 400007,
+        "message": "Request is missing parameter required to complete desired action."
+    },
+    {
+        "name": "USER_REQUEST_INVALID_PARAMETER",
+        "code": 400008,
+        "message": "Request contained invalid parameter, action could not be completed."
+    },
+    {
+        "name": "MALFORMED_ID",
+        "code": 400009,
+        "message": "The id of the resource is malformed."
+    },
+    {
+        "name": "UNKNOWN_CONTENTS_TYPE",
+        "code": 400010,
+        "message": "The request contains unknown type of contents."
+    },
+    {
+        "name": "USER_IDENTIFIER_DUPLICATE",
+        "code": 400011,
+        "message": "Request contained a duplicated identifier that must be unique."
+    },
+    {
+        "name": "USER_TOOL_META_PARAMETER_PROBLEM",
+        "code": 400012,
+        "message": "Supplied incorrect or incompatible tool meta parameters."
+    },
+    {
+        "name": "MALFORMED_CONTENTS",
+        "code": 400013,
+        "message": "The contents of the request are malformed."
+    },
+    {
+        "name": "USER_TOOL_MISSING_PROBLEM",
+        "code": 400014,
+        "message": "Tool could not be found."
+    },
+    {
+        "name": "TOOL_INPUTS_NOT_READY",
+        "code": 400015,
+        "message": "Tool inputs not yet ready, try again later."
+    },
+    {
+        "name": "REAL_USER_REQUIRED",
+        "code": 400016,
+        "message": "Only real users can make this request."
+    },
+    {
+        "name": "USER_AUTHENTICATION_FAILED",
+        "code": 401001,
+        "message": "Authentication failed, invalid credentials supplied."
+    },
+    {
+        "name": "USER_NO_API_KEY",
+        "code": 403001,
+        "message": "Authentication required for this request"
+    },
+    {
+        "name": "USER_CANNOT_ACCESS_ITEM",
+        "code": 403002,
+        "message": "User cannot access specified item."
+    },
+    {
+        "name": "USER_DOES_NOT_OWN_ITEM",
+        "code": 403003,
+        "message": "User does not own specified item."
+    },
+    {
+        "name": "ITEM_IS_IMMUTABLE",
+        "code": 403003,
+        "message": "The specified item is immutable."
+    },
+    {
+        "name": "CONFIG_DOES_NOT_ALLOW",
+        "code": 403004,
+        "message": "The configuration of this Galaxy instance does not allow that operation"
+    },
+    {
+        "name": "INSUFFICIENT_PERMISSIONS",
+        "code": 403005,
+        "message": "You don't have proper permissions to perform the requested operation"
+    },
+    {
+        "name": "ADMIN_REQUIRED",
+        "code": 403006,
+        "message": "Action requires admin account."
+    },
+    {
+        "name": "USER_ACTIVATION_REQUIRED",
+        "code": 403007,
+        "message": "Action requires account activation."
+    },
+    {
+        "name": "USER_OBJECT_NOT_FOUND",
+        "code": 404001,
+        "message": "No such object found."
+    },
+    {
+        "name": "CONFLICT",
+        "code": 409001,
+        "message": "Database conflict prevented fulfilling the request."
+    },
+    {
+        "name": "DEPRECATED_API_CALL",
+        "code": 410001,
+        "message": "This API method or call signature has been deprecated and is no longer available"
+    },
+    {
+        "name": "INTERNAL_SERVER_ERROR",
+        "code": 500001,
+        "message": "Internal server error."
+    },
+    {
+        "name": "INCONSISTENT_DATABASE",
+        "code": 500002,
+        "message": "Inconsistent database prevented fulfilling the request."
+    },
+    {
+        "name": "CONFIG_ERROR",
+        "code": 500003,
+        "message": "Error in a configuration file."
+    },
+    {
+        "name": "TOOL_EXECUTION_ERROR",
+        "code": 500004,
+        "message": "Tool execution failed due to an internal server error."
+    },
+    {
+        "name": "INVALID_FILE_FORMAT",
+        "code": 500005,
+        "message": "File format not supported for this operation."
+    },
+    {
+        "name": "REFERENCE_DATA_ERROR",
+        "code": 500006,
+        "message": "Reference data required for program execution failed to load."
+    },
+    {
+        "name": "NOT_IMPLEMENTED",
+        "code": 501001,
+        "message": "Method is not implemented."
+    },
+    {
+        "name": "SERVER_NOT_CONFIGURED_FOR_REQUEST",
+        "code": 501002,
+        "message": "Server not configured for the request. The Galaxy admin may be able to resolve the problem by installing additional dependencies or setting up new infrastructure."
+    }
 ]

lib/galaxy/managers/histories.py

@@ -143,6 +143,7 @@ class HistoryManager(sharable.SharableModelManager, deletable.PurgableManagerMix
         """
         Purge this history and all HDAs, Collections, and Datasets inside this history.
         """
+        self.error_unless_mutable(history)
         self.hda_manager.dataset_manager.error_unless_dataset_purge_allowed()
         # First purge all the datasets
         for hda in history.datasets:

lib/galaxy/managers/secured.py

@@ -143,3 +143,25 @@ class OwnableManagerMixin:
         """
         # just alias to list_owned
         return self.list_owned(user, **kwargs)
+
+    def get_mutable(self, id: int, user: Optional[model.User], **kwargs: Any) -> Any:
+        """
+        Return the item with the given id if the user can mutate it,
+        otherwise raise an error. The user must be the owner of the item.
+
+        :raises exceptions.ItemOwnershipException:
+        """
+        item = self.get_owned(id, user, **kwargs)
+        self.error_unless_mutable(item)
+        return item
+
+    def error_unless_mutable(self, item):
+        """
+        Raise an error if the item is NOT mutable.
+
+        Items purged or archived are considered immutable.
+
+        :raises exceptions.ItemImmutableException:
+        """
+        if getattr(item, "purged", False) or getattr(item, "archived", False):
+            raise exceptions.ItemImmutableException(f"{self.model_class.__name__} is immutable")

lib/galaxy/webapps/galaxy/api/tools.py

@@ -498,6 +498,8 @@ class ToolsController(BaseGalaxyAPIController, UsesVisualizationMixin):
             else:
                 raise exceptions.RequestParameterInvalidException("Must run conversion on either hdca or hda.")
 
+        self.history_manager.error_unless_mutable(target_history)
+
         # Make the target datatype available to the converter
         params["__target_datatype__"] = target_type
         vars = converter.handle_input(trans, params, history=target_history)

lib/galaxy/webapps/galaxy/controllers/dataset.py

@@ -825,6 +825,7 @@ class DatasetInterface(BaseUIController, UsesAnnotations, UsesItemRatings, UsesE
         try:
             id = self.decode_id(dataset_id)
             hda = self.hda_manager.get_owned(id, trans.user, current_history=trans.history)
+            self.hda_manager.error_unless_mutable(hda.history)
             hda.mark_deleted()
             hda.clear_associated_files()
             trans.log_event(f"Dataset id {str(id)} marked as deleted")
@@ -844,6 +845,7 @@ class DatasetInterface(BaseUIController, UsesAnnotations, UsesItemRatings, UsesE
         try:
             id = self.decode_id(dataset_id)
             item = self.hda_manager.get_owned(id, trans.user, current_history=trans.history)
+            self.hda_manager.error_unless_mutable(item.history)
             self.hda_manager.undelete(item)
             trans.log_event(f"Dataset id {str(id)} has been undeleted")
         except Exception:
@@ -858,6 +860,7 @@ class DatasetInterface(BaseUIController, UsesAnnotations, UsesItemRatings, UsesE
         try:
             id = self.decode_id(dataset_id)
             item = self.hda_manager.get_owned(id, trans.user, current_history=trans.history)
+            self.hda_manager.error_unless_mutable(item.history)
             item.mark_unhidden()
             trans.sa_session.flush()
             trans.log_event(f"Dataset id {str(id)} has been unhidden")

lib/galaxy/webapps/galaxy/controllers/history.py

@@ -533,6 +533,7 @@ class HistoryController(BaseUIController, SharableMixin, UsesAnnotations, UsesIt
                 )
             return {"title": "Change default dataset permissions for history '%s'" % history.name, "inputs": inputs}
         else:
+            self.history_manager.error_unless_mutable(history)
             permissions = {}
             for action_key, action in trans.app.model.Dataset.permitted_actions.items():
                 in_roles = payload.get(action_key) or []
@@ -569,6 +570,7 @@ class HistoryController(BaseUIController, SharableMixin, UsesAnnotations, UsesIt
             trans.app.security_agent.permitted_actions.DATASET_ACCESS: [private_role],
         }
         for history in histories:
+            self.history_manager.error_unless_mutable(history)
             # Set default role for history to private
             trans.app.security_agent.history_set_default_permissions(history, private_permissions)
             # Set private role for all datasets
@@ -668,7 +670,7 @@ class HistoryController(BaseUIController, SharableMixin, UsesAnnotations, UsesIt
     @web.require_login("set history's accessible flag")
     def set_accessible_async(self, trans, id=None, accessible=False):
         """Set history's importable attribute and slug."""
-        history = self.history_manager.get_owned(self.decode_id(id), trans.user, current_history=trans.history)
+        history = self.history_manager.get_mutable(self.decode_id(id), trans.user, current_history=trans.history)
         # Only set if importable value would change; this prevents a change in the update_time unless attribute really changed.
         importable = accessible in ["True", "true", "t", "T"]
         if history and history.importable != importable:
@@ -690,7 +692,7 @@ class HistoryController(BaseUIController, SharableMixin, UsesAnnotations, UsesIt
         id = listify(id)
         histories = []
         for history_id in id:
-            history = self.history_manager.get_owned(
+            history = self.history_manager.get_mutable(
                 self.decode_id(history_id), trans.user, current_history=trans.history
             )
             if history and history.user_id == user.id:
@@ -748,7 +750,7 @@ class HistoryController(BaseUIController, SharableMixin, UsesAnnotations, UsesIt
     def set_as_current(self, trans, id):
         """Change the current user's current history to one with `id`."""
         try:
-            history = self.history_manager.get_owned(self.decode_id(id), trans.user, current_history=trans.history)
+            history = self.history_manager.get_mutable(self.decode_id(id), trans.user, current_history=trans.history)
             trans.set_history(history)
             return self.history_data(trans, history)
         except exceptions.MessageException as msg_exc:

lib/galaxy/webapps/galaxy/services/dataset_collections.py

@@ -124,7 +124,7 @@ class DatasetCollectionsService(ServiceBase, UsesLibraryMixinItems):
         if payload.instance_type == "history":
             if payload.history_id is None:
                 raise exceptions.RequestParameterInvalidException("Parameter history_id is required.")
-            history = self.history_manager.get_owned(payload.history_id, trans.user, current_history=trans.history)
+            history = self.history_manager.get_mutable(payload.history_id, trans.user, current_history=trans.history)
             create_params["parent"] = history
             create_params["history"] = history
         elif payload.instance_type == "library" and payload.folder_id:

lib/galaxy/webapps/galaxy/services/datasets.py

@@ -695,6 +695,7 @@ class DatasetsService(ServiceBase, UsesVisualizationMixin):
             try:
                 manager = self.dataset_manager_by_type[dataset.src]
                 dataset_instance = manager.get_owned(dataset.id, trans.user)
+                manager.error_unless_mutable(dataset_instance.history)
                 if dataset.src == DatasetSourceType.hda:
                     self.hda_manager.error_if_uploading(dataset_instance)
                 if payload.purge:

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

@@ -388,7 +388,7 @@ class HistoriesService(ServiceBase, ConsumesModelStores, ServesExportStores):
             any values that were different from the original and, therefore, updated
         """
         # TODO: PUT /api/histories/{encoded_history_id} payload = { rating: rating } (w/ no security checks)
-        history = self.manager.get_owned(history_id, trans.user, current_history=trans.history)
+        history = self.manager.get_mutable(history_id, trans.user, current_history=trans.history)
         self.deserializer.deserialize(history, payload, user=trans.user, trans=trans)
         return self._serialize_history(trans, history, serialization_params)
 
@@ -406,7 +406,7 @@ class HistoriesService(ServiceBase, ConsumesModelStores, ServesExportStores):
         You can purge a history, removing all it's datasets from disk (if unshared),
         by passing in ``purge=True`` in the url.
         """
-        history = self.manager.get_owned(history_id, trans.user, current_history=trans.history)
+        history = self.manager.get_mutable(history_id, trans.user, current_history=trans.history)
         if purge:
             self.manager.purge(history)
         else:
@@ -427,7 +427,7 @@ class HistoriesService(ServiceBase, ConsumesModelStores, ServesExportStores):
 
         :returns:   the undeleted history
         """
-        history = self.manager.get_owned(history_id, trans.user, current_history=trans.history)
+        history = self.manager.get_mutable(history_id, trans.user, current_history=trans.history)
         self.manager.undelete(history)
         return self._serialize_history(trans, history, serialization_params)
 

lib/galaxy/webapps/galaxy/services/history_contents.py

@@ -510,7 +510,7 @@ class HistoriesContentsService(ServiceBase, ServesExportStores, ConsumesModelSto
         ..note:
             Currently, a user can only copy an HDA from a history that the user owns.
         """
-        history = self.history_manager.get_owned(history_id, trans.user, current_history=trans.history)
+        history = self.history_manager.get_mutable(history_id, trans.user, current_history=trans.history)
 
         serialization_params.default_view = "detailed"
         history_content_type = payload.type
@@ -531,7 +531,7 @@ class HistoriesContentsService(ServiceBase, ServesExportStores, ConsumesModelSto
         payload: CreateHistoryContentFromStore,
         serialization_params: SerializationParams,
     ) -> List[AnyHistoryContentItem]:
-        history = self.history_manager.get_owned(history_id, trans.user, current_history=trans.history)
+        history = self.history_manager.get_mutable(history_id, trans.user, current_history=trans.history)
         object_tracker = self.create_objects_from_store(
             trans,
             payload,
@@ -558,7 +558,7 @@ class HistoriesContentsService(ServiceBase, ServesExportStores, ConsumesModelSto
         request: MaterializeDatasetInstanceRequest,
     ) -> AsyncTaskResultSummary:
         # DO THIS JUST TO MAKE SURE IT IS OWNED...
-        self.history_manager.get_owned(request.history_id, trans.user, current_history=trans.history)
+        self.history_manager.get_mutable(request.history_id, trans.user, current_history=trans.history)
         assert trans.app.config.enable_celery_tasks
         task_request = MaterializeDatasetInstanceTaskRequest(
             history_id=request.history_id,
@@ -593,6 +593,7 @@ class HistoriesContentsService(ServiceBase, ServesExportStores, ConsumesModelSto
         payload_dict = payload.dict(by_alias=True)
         hda = self.hda_manager.get_owned(history_content_id, trans.user, current_history=trans.history, trans=trans)
         assert hda is not None
+        self.history_manager.error_unless_mutable(hda.history)
         self.hda_manager.update_permissions(trans, hda, **payload_dict)
         roles = self.hda_manager.serialize_dataset_association_roles(trans, hda)
         return DatasetAssociationRoles.construct(**roles)
@@ -617,8 +618,9 @@ class HistoriesContentsService(ServiceBase, ServesExportStores, ConsumesModelSto
         :returns:   an error object if an error occurred or a dictionary containing
                     any values that were different from the original and, therefore, updated
         """
+        history = self.history_manager.get_mutable(history_id, trans.user, current_history=trans.history)
         if contents_type == HistoryContentType.dataset:
-            return self.__update_dataset(trans, history_id, id, payload, serialization_params)
+            return self.__update_dataset(trans, history, id, payload, serialization_params)
         elif contents_type == HistoryContentType.dataset_collection:
             return self.__update_dataset_collection(trans, id, payload)
         else:
@@ -645,7 +647,7 @@ class HistoriesContentsService(ServiceBase, ServesExportStores, ConsumesModelSto
         :returns:   an error object if an error occurred or a dictionary containing
                     any values that were different from the original and, therefore, updated
         """
-        history = self.history_manager.get_owned(history_id, trans.user, current_history=trans.history)
+        history = self.history_manager.get_mutable(history_id, trans.user, current_history=trans.history)
         items = payload.items
         hda_ids: List[DecodedDatabaseIdField] = []
         hdca_ids: List[DecodedDatabaseIdField] = []
@@ -677,7 +679,7 @@ class HistoriesContentsService(ServiceBase, ServesExportStores, ConsumesModelSto
         filter_query_params: ValueFilterQueryParams,
         payload: HistoryContentBulkOperationPayload,
     ) -> HistoryContentBulkOperationResult:
-        history = self.history_manager.get_owned(history_id, trans.user, current_history=trans.history)
+        history = self.history_manager.get_mutable(history_id, trans.user, current_history=trans.history)
         filters = self.history_contents_filters.parse_query_filters(filter_query_params)
         self._validate_bulk_operation_params(payload, trans.user, trans)
         contents: List[HistoryItemModel]
@@ -710,7 +712,7 @@ class HistoriesContentsService(ServiceBase, ServesExportStores, ConsumesModelSto
         :returns:   TODO
         """
         decoded_id = history_content_id
-        history = self.history_manager.get_owned(history_id, trans.user, current_history=trans.history)
+        history = self.history_manager.get_mutable(history_id, trans.user, current_history=trans.history)
         hda = self.hda_manager.get_owned_ids([decoded_id], history=history)[0]
         if hda:
             self.hda_manager.set_metadata(trans, hda, overwrite=True, validate=True)
@@ -851,6 +853,7 @@ class HistoriesContentsService(ServiceBase, ServesExportStores, ConsumesModelSto
         self, trans, id: DecodedDatabaseIdField, purge: bool, stop_job: bool, serialization_params: SerializationParams
     ):
         hda = self.hda_manager.get_owned(id, trans.user, current_history=trans.history)
+        self.history_manager.error_unless_mutable(hda.history)
         self.hda_manager.error_if_uploading(hda)
 
         async_result = None
@@ -869,14 +872,13 @@ class HistoriesContentsService(ServiceBase, ServesExportStores, ConsumesModelSto
     def __update_dataset(
         self,
         trans,
-        history_id: DecodedDatabaseIdField,
+        history: History,
         id: DecodedDatabaseIdField,
         payload: Dict[str, Any],
         serialization_params: SerializationParams,
     ):
         # anon user: ensure that history ids match up and the history is the current,
         #   check for uploading, and use only the subset of attribute keys manipulatable by anon users
-        history = self.history_manager.get_owned(history_id, trans.user, current_history=trans.history)
         hda = self.__datasets_for_update(trans, history, [id], payload)[0]
         if hda:
             self.__deserialize_dataset(trans, hda, payload)

lib/galaxy/webapps/galaxy/services/tools.py

@@ -133,7 +133,7 @@ class ToolsService(ServiceBase):
         history_id = payload.get("history_id")
         if history_id:
             history_id = trans.security.decode_id(history_id) if isinstance(history_id, str) else history_id
-            target_history = self.history_manager.get_owned(history_id, trans.user, current_history=trans.history)
+            target_history = self.history_manager.get_mutable(history_id, trans.user, current_history=trans.history)
         else:
             target_history = None
 

lib/galaxy/workflow/run_request.py

@@ -253,7 +253,7 @@ def _get_target_history(
             history_name = history_param
     if history_id:
         history_manager = trans.app.history_manager
-        target_history = history_manager.get_owned(
+        target_history = history_manager.get_mutable(
             trans.security.decode_id(history_id), trans.user, current_history=trans.history
         )
     else:

lib/galaxy_test/api/test_datasets.py

@@ -788,3 +788,15 @@ class TestDatasetsApi(ApiTestCase):
         output = self.dataset_populator.fetch_hda(history_id, item)
         dataset_details = self.dataset_populator.get_history_dataset_details(history_id, dataset=output, assert_ok=True)
         assert "display_application/" in dataset_details["display_apps"][0]["links"][0]["href"]
+
+    def test_cannot_update_datatype_on_immutable_history(self, history_id):
+        hda_id = self.dataset_populator.new_dataset(history_id)["id"]
+        self.dataset_populator.wait_for_history_jobs(history_id)
+
+        # once we purge the history, it becomes immutable
+        self._delete(f"histories/{history_id}", data={"purge": True}, json=True)
+
+        # now we can't update the datatype
+        response = self._put(f"histories/{history_id}/contents/{hda_id}", data={"datatype": "tabular"}, json=True)
+        self._assert_status_code_is(response, 403)
+        assert response.json()["err_msg"] == "History is immutable"

lib/galaxy_test/api/test_histories.py

@@ -352,6 +352,61 @@ class TestHistoriesApi(ApiTestCase, BaseHistories):
             }
             self.dataset_populator.import_history(import_data)
 
+    def test_immutable_history_update_fails(self):
+        history_id = self._create_history("TestHistoryForImmutability")["id"]
+
+        # we can update the name as usual
+        self._update(history_id, {"name": "Immutable Name"})
+        show_response = self._show(history_id)
+        assert show_response["name"] == "Immutable Name"
+
+        # once we purge the history, it becomes immutable
+        self._delete(f"histories/{history_id}", data={"purge": True}, json=True)
+
+        # we cannot update the name anymore
+        response = self._update(history_id, {"name": "New Name"})
+        self._assert_status_code_is(response, 403)
+        assert response.json()["err_msg"] == "History is immutable"
+        show_response = self._show(history_id)
+        assert show_response["name"] == "Immutable Name"
+
+    def test_immutable_history_cannot_add_datasets(self):
+        history_id = self._create_history("TestHistoryForAddImmutability")["id"]
+
+        # we add a dataset
+        self.dataset_populator.new_dataset(history_id, content="TestContents")
+
+        # once we purge the history, it becomes immutable
+        self._delete(f"histories/{history_id}", data={"purge": True}, json=True)
+
+        # we cannot add another dataset
+        with self.assertRaisesRegex(AssertionError, "History is immutable"):
+            self.dataset_populator.new_dataset(history_id, content="TestContents")
+
+    def test_cannot_modify_tags_on_immutable_history(self):
+        history_id = self._create_history("TestHistoryForTagImmutability")["id"]
+        hda = self.dataset_populator.new_dataset(history_id, content="TestContents")
+
+        # we add a tag
+        self._update(history_id, {"tags": ["FirstTag"]})
+
+        # once we purge the history, it becomes immutable
+        self._delete(f"histories/{history_id}", data={"purge": True}, json=True)
+
+        # we cannot add another tag
+        response = self._update(history_id, {"tags": ["SecondTag"]})
+        self._assert_status_code_is(response, 403)
+        assert response.json()["err_msg"] == "History is immutable"
+
+        # we cannot remove the tag
+        response = self._update(history_id, {"tags": []})
+        self._assert_status_code_is(response, 403)
+        assert response.json()["err_msg"] == "History is immutable"
+
+        # we cannot add a tag to the dataset
+        response = self.dataset_populator.tag_dataset(history_id, hda["id"], ["DatasetTag"], raise_on_error=False)
+        assert response["err_msg"] == "History is immutable"
+
 
 class ImportExportTests(BaseHistories):
     task_based: ClassVar[bool]

lib/galaxy_test/api/test_history_contents.py

@@ -802,6 +802,45 @@ class TestHistoryContentsApi(ApiTestCase):
         collection = contents_response.json()[0]
         assert sorted(collection["elements_datatypes"]) == sorted(expected_datatypes)
 
+    @skip_without_tool("cat1")
+    def test_cannot_run_tools_on_immutable_histories(self, history_id):
+        create_response = self.dataset_collection_populator.create_pair_in_history(
+            history_id, contents=["123", "456"], wait=True
+        )
+        hdca_id = create_response.json()["outputs"][0]["id"]
+        inputs = {
+            "input1": {"batch": True, "values": [{"src": "hdca", "id": hdca_id}]},
+        }
+
+        # once we purge the history, it becomes immutable
+        self._delete(f"histories/{history_id}", data={"purge": True}, json=True)
+
+        with self.assertRaisesRegex(AssertionError, "History is immutable"):
+            self.dataset_populator.run_tool("cat1", inputs=inputs, history_id=history_id)
+
+    def test_cannot_update_dataset_collection_on_immutable_history(self, history_id):
+        hdca = self._create_pair_collection(history_id)
+
+        # once we purge the history, it becomes immutable
+        self._delete(f"histories/{history_id}", data={"purge": True}, json=True)
+
+        body = dict(name="newnameforpair")
+        update_response = self._put(
+            f"histories/{history_id}/contents/dataset_collections/{hdca['id']}", data=body, json=True
+        )
+        self._assert_status_code_is(update_response, 403)
+        assert update_response.json()["err_msg"] == "History is immutable"
+
+    def test_cannot_update_dataset_on_immutable_history(self, history_id):
+        hda1 = self._wait_for_new_hda(history_id)
+
+        # once we purge the history, it becomes immutable
+        self._delete(f"histories/{history_id}", data={"purge": True}, json=True)
+
+        update_response = self._update(history_id, hda1["id"], dict(name="Updated Name"))
+        self._assert_status_code_is(update_response, 403)
+        assert update_response.json()["err_msg"] == "History is immutable"
+
 
 class TestHistoryContentsApiBulkOperation(ApiTestCase):
     """

lib/galaxy_test/api/test_workflows.py

@@ -6484,6 +6484,18 @@ input_c:
         finally:
             shutil.rmtree(workflow_directory)
 
+    def test_cannot_run_workflow_on_immutable_history(self) -> None:
+        with self.dataset_populator.test_history() as history_id:
+            # once we purge the history, it becomes immutable
+            self._delete(f"histories/{history_id}", data={"purge": True}, json=True)
+
+            with self.assertRaisesRegex(AssertionError, "History is immutable"):
+                self.workflow_populator.run_workflow(
+                    WORKFLOW_INPUTS_AS_OUTPUTS,
+                    test_data={"input1": "hello world", "text_input": {"value": "A text variable", "type": "raw"}},
+                    history_id=history_id,
+                )
+
     def _invoke_paused_workflow(self, history_id):
         workflow = self.workflow_populator.load_workflow_from_resource("test_workflow_pause")
         workflow_id = self.workflow_populator.create_workflow(workflow)

lib/galaxy_test/base/populators.py

@@ -476,10 +476,11 @@ class BaseDatasetPopulator(BasePopulator):
         details = self.get_history_dataset_details(history_id, dataset=output)
         return details
 
-    def tag_dataset(self, history_id, hda_id, tags):
+    def tag_dataset(self, history_id, hda_id, tags, raise_on_error=True):
         url = f"histories/{history_id}/contents/{hda_id}"
         response = self._put(url, {"tags": tags}, json=True)
-        response.raise_for_status()
+        if raise_on_error:
+            response.raise_for_status()
         return response.json()
 
     def create_from_store_raw(self, payload: Dict[str, Any]) -> Response: