Mercurial > repos > guerler > springsuite
diff planemo/lib/python3.7/site-packages/cwltool/secrets.py @ 0:d30785e31577 draft
"planemo upload commit 6eee67778febed82ddd413c3ca40b3183a3898f1"
author | guerler |
---|---|
date | Fri, 31 Jul 2020 00:18:57 -0400 |
parents | |
children |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/planemo/lib/python3.7/site-packages/cwltool/secrets.py Fri Jul 31 00:18:57 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