Support secrets in tool requirements

lib/galaxy/tool_util/cwl/parser.py

@@ -219,6 +219,9 @@ def software_requirements(self) -> List:
     def resource_requirements(self) -> List:
         return self.hints_or_requirements_of_class("ResourceRequirement")
 
+    def credentials(self) -> List:
+        return self.hints_or_requirements_of_class("Credentials")
+
 
 class CommandLineToolProxy(ToolProxy):
     _class = "CommandLineTool"

lib/galaxy/tool_util/deps/requirements.py

@@ -20,6 +20,7 @@
 
 from galaxy.util import (
     asbool,
+    string_as_bool,
     xml_text,
 )
 from galaxy.util.oset import OrderedSet
@@ -305,27 +306,125 @@ def resource_requirements_from_list(requirements: Iterable[Dict[str, Any]]) -> L
     return rr
 
 
+class SecretOrVariable:
+    def __init__(
+        self,
+        type: str,
+        name: str,
+        inject_as_env: str,
+        label: str = "",
+        description: str = "",
+    ) -> None:
+        self.type = type
+        self.name = name
+        self.inject_as_env = inject_as_env
+        self.label = label
+        self.description = description
+        if self.type not in {"secret", "variable"}:
+            raise ValueError(f"Invalid credential type '{self.type}'")
+        if not self.inject_as_env:
+            raise ValueError("Missing inject_as_env")
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "type": self.type,
+            "name": self.name,
+            "inject_as_env": self.inject_as_env,
+            "label": self.label,
+            "description": self.description,
+        }
+
+    @classmethod
+    def from_element(cls, elem) -> "SecretOrVariable":
+        return cls(
+            type=elem.tag,
+            name=elem.get("name"),
+            inject_as_env=elem.get("inject_as_env"),
+            label=elem.get("label", ""),
+            description=elem.get("description", ""),
+        )
+
+    @classmethod
+    def from_dict(cls, dict: Dict[str, Any]) -> "SecretOrVariable":
+        type = dict["type"]
+        name = dict["name"]
+        inject_as_env = dict["inject_as_env"]
+        label = dict.get("label", "")
+        description = dict.get("description", "")
+        return cls(type=type, name=name, inject_as_env=inject_as_env, label=label, description=description)
+
+
+class CredentialsRequirement:
+    def __init__(
+        self,
+        name: str,
+        reference: str,
+        required: bool = False,
+        label: str = "",
+        description: str = "",
+        secrets_and_variables: Optional[List[SecretOrVariable]] = None,
+    ) -> None:
+        self.name = name
+        self.reference = reference
+        self.required = required
+        self.label = label
+        self.description = description
+        self.secrets_and_variables = secrets_and_variables if secrets_and_variables is not None else []
+
+        if not self.reference:
+            raise ValueError("Missing reference")
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "name": self.name,
+            "reference": self.reference,
+            "required": self.required,
+            "label": self.label,
+            "description": self.description,
+            "secrets_and_variables": [s.to_dict() for s in self.secrets_and_variables],
+        }
+
+    @classmethod
+    def from_dict(cls, dict: Dict[str, Any]) -> "CredentialsRequirement":
+        name = dict["name"]
+        reference = dict["reference"]
+        required = dict.get("required", False)
+        label = dict.get("label", "")
+        description = dict.get("description", "")
+        secrets_and_variables = [SecretOrVariable.from_dict(s) for s in dict.get("secrets_and_variables", [])]
+        return cls(
+            name=name,
+            reference=reference,
+            required=required,
+            label=label,
+            description=description,
+            secrets_and_variables=secrets_and_variables,
+        )
+
+
 def parse_requirements_from_lists(
     software_requirements: List[Union[ToolRequirement, Dict[str, Any]]],
     containers: Iterable[Dict[str, Any]],
     resource_requirements: Iterable[Dict[str, Any]],
-) -> Tuple[ToolRequirements, List[ContainerDescription], List[ResourceRequirement]]:
+    credentials: Iterable[Dict[str, Any]],
+) -> Tuple[ToolRequirements, List[ContainerDescription], List[ResourceRequirement], List[CredentialsRequirement]]:
     return (
         ToolRequirements.from_list(software_requirements),
         [ContainerDescription.from_dict(c) for c in containers],
         resource_requirements_from_list(resource_requirements),
+        [CredentialsRequirement.from_dict(s) for s in credentials],
     )
 
 
-def parse_requirements_from_xml(xml_root, parse_resources: bool = False):
+def parse_requirements_from_xml(xml_root, parse_resources_and_secrets: bool = False):
     """
     Parses requirements, containers and optionally resource requirements from Xml tree.
 
     >>> from galaxy.util import parse_xml_string
-    >>> def load_requirements(contents, parse_resources=False):
+    >>> def load_requirements(contents, parse_resources_and_secrets=False):
     ...     contents_document = '''<tool><requirements>%s</requirements></tool>'''
     ...     root = parse_xml_string(contents_document % contents)
-    ...     return parse_requirements_from_xml(root, parse_resources=parse_resources)
+    ...     return parse_requirements_from_xml(root, parse_resources_and_secrets=parse_resources_and_secrets)
     >>> reqs, containers = load_requirements('''<requirement>bwa</requirement>''')
     >>> reqs[0].name
     'bwa'
@@ -344,8 +443,10 @@ def parse_requirements_from_xml(xml_root, parse_resources: bool = False):
     requirements_elem = xml_root.find("requirements")
 
     requirement_elems = []
+    container_elems = []
     if requirements_elem is not None:
         requirement_elems = requirements_elem.findall("requirement")
+        container_elems = requirements_elem.findall("container")
 
     requirements = ToolRequirements()
     for requirement_elem in requirement_elems:
@@ -355,15 +456,13 @@ def parse_requirements_from_xml(xml_root, parse_resources: bool = False):
         requirement = ToolRequirement(name=name, type=type, version=version)
         requirements.append(requirement)
 
-    container_elems = []
-    if requirements_elem is not None:
-        container_elems = requirements_elem.findall("container")
-
     containers = [container_from_element(c) for c in container_elems]
-    if parse_resources:
+    if parse_resources_and_secrets:
         resource_elems = requirements_elem.findall("resource") if requirements_elem is not None else []
         resources = [resource_from_element(r) for r in resource_elems]
-        return requirements, containers, resources
+        credentials_elems = requirements_elem.findall("credentials") if requirements_elem is not None else []
+        credentials = [credentials_from_element(s) for s in credentials_elems]
+        return requirements, containers, resources, credentials
 
     return requirements, containers
 
@@ -386,3 +485,20 @@ def container_from_element(container_elem) -> ContainerDescription:
         shell=shell,
     )
     return container
+
+
+def credentials_from_element(credentials_elem) -> CredentialsRequirement:
+    name = credentials_elem.get("name")
+    reference = credentials_elem.get("reference")
+    required = string_as_bool(credentials_elem.get("required", "false"))
+    label = credentials_elem.get("label", "")
+    description = credentials_elem.get("description", "")
+    secrets_and_variables = [SecretOrVariable.from_element(elem) for elem in credentials_elem.findall("*")]
+    return CredentialsRequirement(
+        name=name,
+        reference=reference,
+        required=required,
+        label=label,
+        description=description,
+        secrets_and_variables=secrets_and_variables,
+    )

lib/galaxy/tool_util/linters/general.py

@@ -183,7 +183,7 @@ class RequirementNameMissing(Linter):
     @classmethod
     def lint(cls, tool_source: "ToolSource", lint_ctx: "LintContext"):
         _, tool_node = _tool_xml_and_root(tool_source)
-        requirements, containers, resource_requirements = tool_source.parse_requirements_and_containers()
+        requirements, *_ = tool_source.parse_requirements_and_containers()
         for r in requirements:
             if r.type != "package":
                 continue
@@ -195,7 +195,7 @@ class RequirementVersionMissing(Linter):
     @classmethod
     def lint(cls, tool_source: "ToolSource", lint_ctx: "LintContext"):
         _, tool_node = _tool_xml_and_root(tool_source)
-        requirements, containers, resource_requirements = tool_source.parse_requirements_and_containers()
+        requirements, *_ = tool_source.parse_requirements_and_containers()
         for r in requirements:
             if r.type != "package":
                 continue
@@ -207,7 +207,7 @@ class RequirementVersionWhitespace(Linter):
     @classmethod
     def lint(cls, tool_source: "ToolSource", lint_ctx: "LintContext"):
         _, tool_node = _tool_xml_and_root(tool_source)
-        requirements, containers, resource_requirements = tool_source.parse_requirements_and_containers()
+        requirements, *_ = tool_source.parse_requirements_and_containers()
         for r in requirements:
             if r.type != "package":
                 continue
@@ -223,7 +223,7 @@ class ResourceRequirementExpression(Linter):
     @classmethod
     def lint(cls, tool_source: "ToolSource", lint_ctx: "LintContext"):
         _, tool_node = _tool_xml_and_root(tool_source)
-        requirements, containers, resource_requirements = tool_source.parse_requirements_and_containers()
+        *_, resource_requirements, _ = tool_source.parse_requirements_and_containers()
         for rr in resource_requirements:
             if rr.runtime_required:
                 lint_ctx.warn(

lib/galaxy/tool_util/parser/cwl.py

@@ -162,10 +162,12 @@ def parse_requirements_and_containers(self):
 
         software_requirements = self.tool_proxy.software_requirements()
         resource_requirements = self.tool_proxy.resource_requirements()
+        credentials = self.tool_proxy.credentials()
         return requirements.parse_requirements_from_lists(
             software_requirements=[{"name": r[0], "version": r[1], "type": "package"} for r in software_requirements],
             containers=containers,
             resource_requirements=resource_requirements,
+            credentials=credentials,
         )
 
     def parse_profile(self):

lib/galaxy/tool_util/parser/interface.py

@@ -34,6 +34,7 @@
 if TYPE_CHECKING:
     from galaxy.tool_util.deps.requirements import (
         ContainerDescription,
+        CredentialsRequirement,
         ResourceRequirement,
         ToolRequirements,
     )
@@ -307,8 +308,10 @@ def parse_required_files(self) -> Optional["RequiredFiles"]:
     @abstractmethod
     def parse_requirements_and_containers(
         self,
-    ) -> Tuple["ToolRequirements", List["ContainerDescription"], List["ResourceRequirement"]]:
-        """Return triple of ToolRequirement, ContainerDescription and ResourceRequirement lists."""
+    ) -> Tuple[
+        "ToolRequirements", List["ContainerDescription"], List["ResourceRequirement"], List["CredentialsRequirement"]
+    ]:
+        """Return triple of ToolRequirement, ContainerDescription, ResourceRequirement, and SecretsRequirement objects."""
 
     @abstractmethod
     def parse_input_pages(self) -> "PagesSource":

lib/galaxy/tool_util/parser/xml.py

@@ -413,7 +413,7 @@ def parse_include_exclude_list(tag_name):
         return RequiredFiles.from_dict(as_dict)
 
     def parse_requirements_and_containers(self):
-        return requirements.parse_requirements_from_xml(self.root, parse_resources=True)
+        return requirements.parse_requirements_from_xml(self.root, parse_resources_and_secrets=True)
 
     def parse_input_pages(self) -> "XmlPagesSource":
         return XmlPagesSource(self.root)

lib/galaxy/tool_util/parser/yaml.py

@@ -115,6 +115,7 @@ def parse_requirements_and_containers(self):
             software_requirements=[r for r in mixed_requirements if r.get("type") != "resource"],
             containers=self.root_dict.get("containers", []),
             resource_requirements=[r for r in mixed_requirements if r.get("type") == "resource"],
+            credentials=self.root_dict.get("credentials", []),
         )
 
     def parse_input_pages(self) -> PagesSource: