diff env/lib/python3.9/site-packages/cwltool/secrets.py @ 0:4f3585e2f14b draft default tip

"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/env/lib/python3.9/site-packages/cwltool/secrets.py	Mon Mar 22 18:12:50 2021 +0000
@@ -0,0 +1,62 @@
+"""Minimal in memory storage of secrets."""
+import uuid
+from typing import Dict, List, MutableMapping, MutableSequence, Optional, cast
+
+from .utils import CWLObjectType, CWLOutputType
+
+
+class SecretStore:
+    """Minimal implementation of a secret storage."""
+
+    def __init__(self) -> None:
+        """Initialize the secret store."""
+        self.secrets = {}  # type: Dict[str, str]
+
+    def add(self, value: Optional[CWLOutputType]) -> Optional[CWLOutputType]:
+        """
+        Add the given value to the store.
+
+        Returns a placeholder value to use until the real value is needed.
+        """
+        if not isinstance(value, str):
+            raise Exception("Secret store only accepts strings")
+
+        if value not in self.secrets:
+            placeholder = "(secret-%s)" % str(uuid.uuid4())
+            self.secrets[placeholder] = value
+            return placeholder
+        return value
+
+    def store(self, secrets: List[str], job: CWLObjectType) -> None:
+        """Sanitize the job object of any of the given secrets."""
+        for j in job:
+            if j in secrets:
+                job[j] = self.add(job[j])
+
+    def has_secret(self, value: CWLOutputType) -> bool:
+        """Test if the provided document has any of our secrets."""
+        if isinstance(value, str):
+            for k in self.secrets:
+                if k in value:
+                    return True
+        elif isinstance(value, MutableMapping):
+            for this_value in value.values():
+                if self.has_secret(cast(CWLOutputType, this_value)):
+                    return True
+        elif isinstance(value, MutableSequence):
+            for this_value in value:
+                if self.has_secret(cast(CWLOutputType, this_value)):
+                    return True
+        return False
+
+    def retrieve(self, value: CWLOutputType) -> CWLOutputType:
+        """Replace placeholders with their corresponding secrets."""
+        if isinstance(value, str):
+            for key, this_value in self.secrets.items():
+                value = value.replace(key, this_value)
+            return value
+        elif isinstance(value, MutableMapping):
+            return {k: self.retrieve(cast(CWLOutputType, v)) for k, v in value.items()}
+        elif isinstance(value, MutableSequence):
+            return [self.retrieve(cast(CWLOutputType, v)) for v in value]
+        return value