-
Notifications
You must be signed in to change notification settings - Fork 5
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge branch 'main' into feature/TED-87
- Loading branch information
Showing
7 changed files
with
202 additions
and
10 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,104 @@ | ||
| #!/usr/bin/python3 | ||
|
|
||
| # config_resolver.py | ||
| # Date: 01/07/2021 | ||
| # Author: Stratulat Ștefan | ||
|
|
||
| """ | ||
| This module aims to provide a simple method of resolving configurations, | ||
| through the process of searching for them in different sources. | ||
| """ | ||
| import inspect | ||
| import logging | ||
| import os | ||
| from abc import ABC | ||
|
|
||
| from ted_sws.adapters.vault_secrets_store import VaultSecretsStore | ||
|
|
||
| logger = logging.getLogger(__name__) | ||
|
|
||
|
|
||
| class abstractstatic(staticmethod): | ||
| """ | ||
| This class serves to create decorators | ||
| with the property of a static method and an abstract method. | ||
| """ | ||
|
|
||
| __slots__ = () | ||
|
|
||
| def __init__(self, function): | ||
| super(abstractstatic, self).__init__(function) | ||
| function.__isabstractmethod__ = True | ||
|
|
||
| __isabstractmethod__ = True | ||
|
|
||
|
|
||
| class ConfigResolverABC(ABC): | ||
| """ | ||
| This class defines a configuration resolution abstraction. | ||
| """ | ||
|
|
||
| @classmethod | ||
| def config_resolve(cls, default_value: str = None) -> str: | ||
| """ | ||
| This method aims to search for a configuration and return its value. | ||
| :param default_value: the default return value, if the configuration is not found. | ||
| :return: the value of the search configuration if found, otherwise default_value returns | ||
| """ | ||
| config_name = inspect.stack()[1][3] | ||
| return cls._config_resolve(config_name, default_value) | ||
|
|
||
| @abstractstatic | ||
| def _config_resolve(config_name: str, default_value: str = None): | ||
| """ | ||
| This abstract method is used to be able to define the configuration search in different environments. | ||
| :param config_name: the name of the configuration you are looking for | ||
| :param default_value: the default return value, if the configuration is not found. | ||
| :return: the value of the search configuration if found, otherwise default_value returns | ||
| """ | ||
| raise NotImplementedError | ||
|
|
||
|
|
||
| class EnvConfigResolver(ConfigResolverABC): | ||
| """ | ||
| This class aims to search for configurations in environment variables. | ||
| """ | ||
|
|
||
| def _config_resolve(config_name: str, default_value: str = None): | ||
| value = os.environ.get(config_name, default=default_value) | ||
| logger.debug("[ENV] Value of '" + str(config_name) + "' is " + str(value) + "(supplied default is '" + str( | ||
| default_value) + "')") | ||
| return value | ||
|
|
||
|
|
||
| class VaultConfigResolver(ConfigResolverABC): | ||
| """ | ||
| This class aims to search for configurations in Vault secrets. | ||
| """ | ||
|
|
||
| def _config_resolve(config_name: str, default_value: str = None): | ||
| value = VaultSecretsStore().get_secret(config_name, default_value) | ||
| logger.debug("[VAULT] Value of '" + str(config_name) + "' is " + str(value) + "(supplied default is '" + str( | ||
| default_value) + "')") | ||
| return value | ||
|
|
||
|
|
||
| class VaultAndEnvConfigResolver(ConfigResolverABC): | ||
| """ | ||
| This class aims to combine the search for configurations in Vault secrets and environmental variables. | ||
| """ | ||
|
|
||
| def _config_resolve(config_name: str, default_value: str = None): | ||
| value = VaultSecretsStore().get_secret(config_name, default_value) | ||
| logger.debug( | ||
| "[VAULT&ENV] Value of '" + str(config_name) + "' is " + str(value) + "(supplied default is '" + str( | ||
| default_value) + "')") | ||
| if value is not None: | ||
| os.environ[config_name] = str(value) | ||
| return value | ||
| else: | ||
| value = EnvConfigResolver._config_resolve(config_name, default_value) | ||
| logger.debug( | ||
| "[VAULT&ENV] Value of '" + str(config_name) + "' is " + str(value) + "(supplied default is '" + str( | ||
| default_value) + "')") | ||
| return value |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,68 @@ | ||
| from abc import ABC, abstractmethod | ||
| from typing import List | ||
| import hvac | ||
| import json | ||
| import os | ||
| import dotenv | ||
|
|
||
| dotenv.load_dotenv(verbose=True, override=True) | ||
|
|
||
| class SecretsStoreABC(ABC): | ||
| """ | ||
| This class aims to define an interface for obtaining secrets from similar resources as Vault. | ||
| """ | ||
|
|
||
| @abstractmethod | ||
| def get_secrets(self, path: str) -> dict: | ||
| """ | ||
| This method defines abstraction to obtain a dictionary of secrets based on a direction to them. | ||
| :param path: the direction of secrets | ||
| :return: returns a dictionary of secrets | ||
| """ | ||
| raise NotImplementedError | ||
|
|
||
|
|
||
| class VaultSecretsStore(SecretsStoreABC): | ||
| """ | ||
| This class is an adapter for the Vault, which allows you to extract secrets from the Vault. | ||
| """ | ||
| default_vault_addr: str = os.environ.get('VAULT_ADDR') | ||
| default_vault_token: str = os.environ.get('VAULT_TOKEN') | ||
| default_secret_mount: str = None | ||
| default_secret_paths: List[str] = None | ||
|
|
||
| def __init__(self, | ||
| vault_addr: str = None, | ||
| vault_token: str = None, | ||
| secret_mount: str = None, | ||
| secret_paths: List[str] = None | ||
| ): | ||
| self._vault_addr = vault_addr if vault_addr else self.default_vault_addr | ||
| self._vault_token = vault_token if vault_token else self.default_vault_token | ||
| self._secret_mount = secret_mount if secret_mount else self.default_secret_mount | ||
| self._secret_paths = secret_paths if secret_paths else self.default_secret_paths | ||
| self._client = hvac.Client(url=self._vault_addr, token=self._vault_token) | ||
|
|
||
| def get_secrets(self, path: str) -> dict: | ||
| secret_response = self._client.secrets.kv.v2.read_secret_version( | ||
| path=path, mount_point=self._secret_mount) | ||
| result_data_str = str(secret_response['data']['data']) | ||
| result_data_json = result_data_str.replace("'", "\"") | ||
| result_data = json.loads(result_data_json) | ||
| return result_data | ||
|
|
||
| def get_secret(self, secret_key: str, default_value: str = None): | ||
| """ | ||
| This method extracts from the vault of a secret based on the name of the secret. | ||
| :param secret_key: the name of the secret sought. | ||
| :param default_value: the default return value in case the secret is not found. | ||
| :return: | ||
| """ | ||
|
|
||
| secrets_dict = {} | ||
| for path in self._secret_paths: | ||
| secrets_dict.update(self.get_secrets(path)) | ||
| if secret_key in secrets_dict.keys(): | ||
| return secrets_dict[secret_key] | ||
| else: | ||
| return default_value |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,6 +1,22 @@ | ||
| from ted_sws import config | ||
| from ted_sws.adapters.config_resolver import EnvConfigResolver, VaultConfigResolver, VaultAndEnvConfigResolver | ||
|
|
||
|
|
||
| def test_config_resolver(): | ||
| mongo_db_url = config.MONGO_DB_AUTH_URL | ||
| assert mongo_db_url | ||
|
|
||
|
|
||
| def test_env_config_resolver(): | ||
| config_value = EnvConfigResolver().config_resolve() | ||
| assert config_value is None | ||
|
|
||
|
|
||
| def test_vault_config_resolver(): | ||
| config_value = VaultConfigResolver().config_resolve() | ||
| assert config_value is None | ||
|
|
||
|
|
||
| def test_vault_and_env_config_resolver(): | ||
| config_value = VaultAndEnvConfigResolver().config_resolve() | ||
| assert config_value is None |