add Vault Secrets Store & Config Resolver

requirements.txt

@@ -2,7 +2,7 @@
 pydantic~=1.9.0
 requests~=2.27.1
 deepdiff~=5.7.0
-git+https://github.com/meaningfy-ws/mfy-data-core
 python-dotenv~=0.19.2
 pymongo~=4.0.1
-apache-airflow==2.2.4
+apache-airflow==2.2.4
+hvac==0.11.2

ted_sws/__init__.py

@@ -12,8 +12,8 @@
 import os
 
 import dotenv
-from mfy_data_core.adapters.config_resolver import VaultAndEnvConfigResolver
-from mfy_data_core.adapters.vault_secrets_store import VaultSecretsStore
+from ted_sws.adapters.config_resolver import VaultAndEnvConfigResolver
+from ted_sws.adapters.vault_secrets_store import VaultSecretsStore
 
 dotenv.load_dotenv(verbose=True, override=True)
 

ted_sws/adapters/config_resolver.py

@@ -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(EnvConfigResolver):
+    """
+        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 = super()._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

ted_sws/adapters/vault_secrets_store.py

@@ -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

tests/unit/test_config_resolver.py

@@ -1,6 +1,16 @@
 from ted_sws import config
+from ted_sws.adapters.config_resolver import EnvConfigResolver, VaultConfigResolver
 
 
 def test_config_resolver():
     mongo_db_url = config.MONGO_DB_AUTH_URL
     assert mongo_db_url
+
+
+def test_env_config_resolver():
+    config_resolver = EnvConfigResolver().config_resolve()
+    assert config_resolver is None
+
+def test_vault_config_resolver():
+    config_resolver = VaultConfigResolver().config_resolve()
+    assert config_resolver is None