Mercurial > repos > guerler > springsuite
view 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 source
"""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