diff env/lib/python3.7/site-packages/cwltool/secrets.py @ 0:26e78fe6e8c4 draft

"planemo upload commit c699937486c35866861690329de38ec1a5d9f783"
author shellac
date Sat, 02 May 2020 07:14:21 -0400
parents
children
line wrap: on
line diff
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/env/lib/python3.7/site-packages/cwltool/secrets.py	Sat May 02 07:14:21 2020 -0400
@@ -0,0 +1,64 @@
+"""Minimal in memory storage of secrets."""
+import uuid
+from typing import Any, Dict, List, MutableMapping, MutableSequence
+
+from six import string_types
+from typing_extensions import Text  # pylint: disable=unused-import
+# move to a regular typing import when Python 3.3-3.6 is no longer supported
+
+
+class SecretStore(object):
+    """Minimal implementation of a secret storage."""
+
+    def __init__(self):  # type: () -> None
+        """Initialize the secret store."""
+        self.secrets = {}  # type: Dict[Text, Text]
+
+    def add(self, value):  # type: (Text) -> Text
+        """
+        Add the given value to the store.
+
+        Returns a placeholder value to use until the real value is needed.
+        """
+        if not isinstance(value, string_types):
+            raise Exception("Secret store only accepts strings")
+
+        if value not in self.secrets:
+            placeholder = "(secret-%s)" % Text(uuid.uuid4())
+            self.secrets[placeholder] = value
+            return placeholder
+        return value
+
+    def store(self, secrets, job):
+        # type: (List[Text], MutableMapping[Text, Any]) -> 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):  # type: (Any) -> bool
+        """Test if the provided document has any of our secrets."""
+        if isinstance(value, string_types):
+            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(this_value):
+                    return True
+        elif isinstance(value, MutableSequence):
+            for this_value in value:
+                if self.has_secret(this_value):
+                    return True
+        return False
+
+    def retrieve(self, value):  # type: (Any) -> Any
+        """Replace placeholders with their corresponding secrets."""
+        if isinstance(value, string_types):
+            for key, this_value in self.secrets.items():
+                value = value.replace(key, this_value)
+        elif isinstance(value, MutableMapping):
+            return {k: self.retrieve(v) for k, v in value.items()}
+        elif isinstance(value, MutableSequence):
+            return [self.retrieve(v) for k, v in enumerate(value)]
+        return value