Validate size of workflow metadata.

lib/galaxy/managers/workflows.py

@@ -819,10 +819,13 @@ def _workflow_from_raw_description(
 
         if "logo_url" in data:
             workflow.logo_url = data["logo_url"]
-        if "help" in data:
-            workflow.help = data["help"]
-        if "readme" in data:
-            workflow.readme = data["readme"]
+        try:
+            if "help" in data:
+                workflow.help = data["help"]
+            if "readme" in data:
+                workflow.readme = data["readme"]
+        except ValueError as e:
+            raise exceptions.RequestParameterInvalidException(str(e))
 
         if getattr(workflow_state_resolution_options, "archive_source", None):
             source_metadata = {}

lib/galaxy/model/__init__.py

@@ -124,6 +124,7 @@
     reconstructor,
     registry,
     relationship,
+    validates,
 )
 from sqlalchemy.orm.attributes import flag_modified
 from sqlalchemy.orm.collections import attribute_keyed_dict
@@ -235,6 +236,8 @@
 
 _datatypes_registry = None
 
+MAX_WORKFLOW_README_SIZE = 20000
+MAX_WORKFLOW_HELP_SIZE = 40000
 STR_TO_STR_DICT = Dict[str, str]
 
 
@@ -7920,6 +7923,20 @@ def __init__(self, uuid=None):
         self.user = None
         self.uuid = get_uuid(uuid)
 
+    @validates("readme")
+    def validates_readme(self, key, readme):
+        size = len(readme)
+        if size > MAX_WORKFLOW_README_SIZE:
+            raise ValueError(f"Workflow readme too large ({size}), maximum allowed length ({MAX_WORKFLOW_README_SIZE}).")
+        return readme
+
+    @validates("help")
+    def validates_help(self, key, help):
+        size = len(help)
+        if size > MAX_WORKFLOW_HELP_SIZE:
+            raise ValueError(f"Workflow help too large ({size}), maximum allowed length ({MAX_WORKFLOW_HELP_SIZE}).")
+        return help
+
     def has_outputs_defined(self):
         """
         Returns true or false indicating whether or not a workflow has outputs defined.

lib/galaxy_test/api/test_workflows.py

@@ -1213,6 +1213,30 @@ def test_readme_metadata(self):
         assert workflow["help"] == help
         assert workflow["logo_url"] == logo_url
 
+    def test_readme_too_large(self):
+        big_but_valid_readme = "READ_" * 3999
+        too_big_readme = "READ_" * 4001
+        base_workflow_json = json.loads(workflow_str)
+        logo_url = "https://galaxyproject.org/images/galaxy_logo_hub_white.svg"
+        help = "This is my instruction for the workflow!"
+        base_workflow_json["readme"] = too_big_readme
+        base_workflow_json["logo_url"] = logo_url
+        base_workflow_json["help"] = help
+        workflow_with_metadata_str = json.dumps(base_workflow_json)
+        base64_url = "base64://" + base64.b64encode(workflow_with_metadata_str.encode("utf-8")).decode("utf-8")
+        response = self._post("workflows", data={"archive_source": base64_url})
+        self._assert_status_code_is(response, 400)
+        self._assert_error_code_is(response, error_codes.error_codes_by_name["USER_REQUEST_INVALID_PARAMETER"])
+
+        base_workflow_json["readme"] = big_but_valid_readme
+        workflow_with_metadata_str = json.dumps(base_workflow_json)
+        base64_url = "base64://" + base64.b64encode(workflow_with_metadata_str.encode("utf-8")).decode("utf-8")
+        response = self._post("workflows", data={"archive_source": base64_url})
+        response.raise_for_status()
+        workflow_id = response.json()["id"]
+        workflow = self._download_workflow(workflow_id)
+        assert workflow["readme"] == big_but_valid_readme
+
     def test_trs_import(self):
         trs_payload = {
             "archive_source": "trs_tool",