Get Airflow configs with sensitive data from Secret Backends

airflow/configuration.py

@@ -85,6 +85,15 @@ def run_command(command):
     return output
 
 
+def _get_config_value_from_secret_backend(config_key):
+    """Get Config option values from Secret Backend"""
+    from airflow import secrets
+    secrets_client = secrets.get_custom_secret_backend()
+    if not secrets_client:
+        return None
+    return secrets_client.get_config(config_key)
+
+
 def _read_default_config_file(file_name: str) -> Tuple[str, str]:
     templates_dir = os.path.join(os.path.dirname(__file__), 'config_templates')
     file_path = os.path.join(templates_dir, file_name)
@@ -115,7 +124,9 @@ class AirflowConfigParser(ConfigParser):
     # These configuration elements can be fetched as the stdout of commands
     # following the "{section}__{name}__cmd" pattern, the idea behind this
     # is to not store password on boxes in text files.
-    as_command_stdout = {
+    # These configs can also be fetched from Secrets backend
+    # following the "{section}__{name}__secret" pattern
+    sensitive_config_values = {
         ('core', 'sql_alchemy_conn'),
         ('core', 'fernet_key'),
         ('celery', 'broker_url'),
@@ -267,17 +278,32 @@ def _get_env_var_option(self, section, key):
         env_var_cmd = env_var + '_CMD'
         if env_var_cmd in os.environ:
             # if this is a valid command key...
-            if (section, key) in self.as_command_stdout:
+            if (section, key) in self.sensitive_config_values:
                 return run_command(os.environ[env_var_cmd])
+        # alternatively AIRFLOW__{SECTION}__{KEY}_SECRET (to get from Secrets Backend)
+        env_var_secret_path = env_var + '_SECRET'
+        if env_var_secret_path in os.environ:
+            # if this is a valid secret path...
+            if (section, key) in self.sensitive_config_values:
+                return _get_config_value_from_secret_backend(os.environ[env_var_secret_path])
 
     def _get_cmd_option(self, section, key):
         fallback_key = key + '_cmd'
         # if this is a valid command key...
-        if (section, key) in self.as_command_stdout:
+        if (section, key) in self.sensitive_config_values:
             if super().has_option(section, fallback_key):
                 command = super().get(section, fallback_key)
                 return run_command(command)
 
+    def _get_secret_option(self, section, key):
+        """Get Config option values from Secret Backend"""
+        fallback_key = key + '_secret'
+        # if this is a valid secret key...
+        if (section, key) in self.sensitive_config_values:
+            if super().has_option(section, fallback_key):
+                secrets_path = super().get(section, fallback_key)
+                return _get_config_value_from_secret_backend(secrets_path)
+
     def get(self, section, key, **kwargs):
         section = str(section).lower()
         key = str(key).lower()
@@ -319,6 +345,16 @@ def get(self, section, key, **kwargs):
                 self._warn_deprecate(section, key, deprecated_section, deprecated_key)
                 return option
 
+        # ...then from secret backends
+        option = self._get_secret_option(section, key)
+        if option:
+            return option
+        if deprecated_section:
+            option = self._get_secret_option(deprecated_section, deprecated_key)
+            if option:
+                self._warn_deprecate(section, key, deprecated_section, deprecated_key)
+                return option
+
         # ...then the default config
         if self.airflow_defaults.has_option(section, key) or 'fallback' in kwargs:
             return expand_env_var(
@@ -473,7 +509,8 @@ def write(self, fp, space_around_delimiters=True):
 
     def as_dict(
             self, display_source=False, display_sensitive=False, raw=False,
-            include_env=True, include_cmds=True) -> Dict[str, Dict[str, str]]:
+            include_env=True, include_cmds=True, include_secret=True
+    ) -> Dict[str, Dict[str, str]]:
         """
         Returns the current configuration as an OrderedDict of OrderedDicts.
 
@@ -495,6 +532,10 @@ def as_dict(
             set (True, default), or should the _cmd options be left as the
             command to run (False)
         :type include_cmds: bool
+        :param include_secret: Should the result of calling any *_secret config be
+            set (True, default), or should the _secret options be left as the
+            path to get the secret from (False)
+        :type include_secret: bool
         :rtype: Dict[str, Dict[str, str]]
         :return: Dictionary, where the key is the name of the section and the content is
             the dictionary with the name of the parameter and its value.
@@ -539,7 +580,7 @@ def as_dict(
 
         # add bash commands
         if include_cmds:
-            for (section, key) in self.as_command_stdout:
+            for (section, key) in self.sensitive_config_values:
                 opt = self._get_cmd_option(section, key)
                 if opt:
                     if not display_sensitive:
@@ -551,6 +592,20 @@ def as_dict(
                     cfg.setdefault(section, OrderedDict()).update({key: opt})
                     del cfg[section][key + '_cmd']
 
+        # add config from secret backends
+        if include_secret:
+            for (section, key) in self.sensitive_config_values:
+                opt = self._get_secret_option(section, key)
+                if opt:
+                    if not display_sensitive:
+                        opt = '< hidden >'
+                    if display_source:
+                        opt = (opt, 'secret')
+                    elif raw:
+                        opt = opt.replace('%', '%%')
+                    cfg.setdefault(section, OrderedDict()).update({key: opt})
+                    del cfg[section][key + '_secret']
+
         return cfg
 
     def load_test_config(self):

airflow/providers/amazon/aws/secrets/secrets_manager.py

@@ -42,8 +42,11 @@ class SecretsManagerBackend(BaseSecretsBackend, LoggingMixin):
 
     For example, if secrets prefix is ``airflow/connections/smtp_default``, this would be accessible
     if you provide ``{"connections_prefix": "airflow/connections"}`` and request conn_id ``smtp_default``.
-    And if variables prefix is ``airflow/variables/hello``, this would be accessible
+    If variables prefix is ``airflow/variables/hello``, this would be accessible
     if you provide ``{"variables_prefix": "airflow/variables"}`` and request variable key ``hello``.
+    And if config_prefix is ``airflow/config/sql_alchemy_conn``, this would be accessible
+    if you provide ``{"config_prefix": "airflow/config"}`` and request config
+    key ``sql_alchemy_conn``.
 
     You can also pass additional keyword arguments like ``aws_secret_access_key``, ``aws_access_key_id``
     or ``region_name`` to this class and they would be passed on to Boto3 client.
@@ -52,6 +55,8 @@ class SecretsManagerBackend(BaseSecretsBackend, LoggingMixin):
     :type connections_prefix: str
     :param variables_prefix: Specifies the prefix of the secret to read to get Variables.
     :type variables_prefix: str
+    :param config_prefix: Specifies the prefix of the secret to read to get Variables.
+    :type config_prefix: str
     :param profile_name: The name of a profile to use. If not given, then the default profile is used.
     :type profile_name: str
     :param sep: separator used to concatenate secret_prefix and secret_id. Default: "/"
@@ -62,13 +67,15 @@ def __init__(
         self,
         connections_prefix: str = 'airflow/connections',
         variables_prefix: str = 'airflow/variables',
+        config_prefix: str = 'airflow/config',
         profile_name: Optional[str] = None,
         sep: str = "/",
         **kwargs
     ):
         super().__init__()
         self.connections_prefix = connections_prefix.rstrip("/")
         self.variables_prefix = variables_prefix.rstrip('/')
+        self.config_prefix = config_prefix.rstrip('/')
         self.profile_name = profile_name
         self.sep = sep
         self.kwargs = kwargs
@@ -94,13 +101,22 @@ def get_conn_uri(self, conn_id: str) -> Optional[str]:
 
     def get_variable(self, key: str) -> Optional[str]:
         """
-        Get Airflow Variable from Environment Variable
+        Get Airflow Variable
 
         :param key: Variable Key
         :return: Variable Value
         """
         return self._get_secret(self.variables_prefix, key)
 
+    def get_config(self, key: str) -> Optional[str]:
+        """
+        Get Airflow Configuration
+
+        :param key: Configuration Option Key
+        :return: Configuration Option Value
+        """
+        return self._get_secret(self.config_prefix, key)
+
     def _get_secret(self, path_prefix: str, secret_id: str) -> Optional[str]:
         """
         Get secret value from Secrets Manager

airflow/providers/hashicorp/secrets/vault.py

@@ -52,6 +52,9 @@ class VaultBackend(BaseSecretsBackend, LoggingMixin):
     :param variables_path: Specifies the path of the secret to read to get Variables
         (default: 'variables').
     :type variables_path: str
+    :param config_path: Specifies the path of the secret to read Airflow Configurations
+        (default: 'configs').
+    :type config_path: str
     :param url: Base URL for the Vault instance being addressed.
     :type url: str
     :param auth_type: Authentication Type for Vault. Default is ``token``. Available values are:
@@ -111,6 +114,7 @@ def __init__(  # pylint: disable=too-many-arguments
         self,
         connections_path: str = 'connections',
         variables_path: str = 'variables',
+        config_path: str = 'config',
         url: Optional[str] = None,
         auth_type: str = 'token',
         auth_mount_point: Optional[str] = None,
@@ -135,9 +139,10 @@ def __init__(  # pylint: disable=too-many-arguments
         radius_port: Optional[int] = None,
         **kwargs
     ):
-        super().__init__(**kwargs)
+        super().__init__()
         self.connections_path = connections_path.rstrip('/')
         self.variables_path = variables_path.rstrip('/')
+        self.config_path = config_path.rstrip('/')
         self.mount_point = mount_point
         self.kv_engine_version = kv_engine_version
         self.vault_client = _VaultClient(
@@ -181,7 +186,7 @@ def get_conn_uri(self, conn_id: str) -> Optional[str]:
 
     def get_variable(self, key: str) -> Optional[str]:
         """
-        Get Airflow Variable from Environment Variable
+        Get Airflow Variable
 
         :param key: Variable Key
         :type key: str
@@ -191,3 +196,16 @@ def get_variable(self, key: str) -> Optional[str]:
         secret_path = self.build_path(self.variables_path, key)
         response = self.vault_client.get_secret(secret_path=secret_path)
         return response.get("value") if response else None
+
+    def get_config(self, key: str) -> Optional[str]:
+        """
+        Get Airflow Configuration
+
+        :param key: Configuration Option Key
+        :type key: str
+        :rtype: str
+        :return: Configuration Option Value retrieved from the vault
+        """
+        secret_path = self.build_path(self.config_path, key)
+        response = self.vault_client.get_secret(secret_path=secret_path)
+        return response.get("value") if response else None

airflow/secrets/__init__.py

@@ -22,26 +22,29 @@
     * Metatsore database
     * AWS SSM Parameter store
 """
-__all__ = ['BaseSecretsBackend', 'get_connections', 'get_variable']
+__all__ = ['BaseSecretsBackend', 'get_connections', 'get_variable', 'get_custom_secret_backend']
 
 import json
 from json import JSONDecodeError
-from typing import List, Optional
+from typing import TYPE_CHECKING, List, Optional
 
 from airflow.configuration import conf
 from airflow.exceptions import AirflowException
-from airflow.models.connection import Connection
 from airflow.secrets.base_secrets import BaseSecretsBackend
 from airflow.utils.module_loading import import_string
 
+if TYPE_CHECKING:
+    from airflow.models.connection import Connection
+
+
 CONFIG_SECTION = "secrets"
 DEFAULT_SECRETS_SEARCH_PATH = [
     "airflow.secrets.environment_variables.EnvironmentVariablesBackend",
     "airflow.secrets.metastore.MetastoreBackend",
 ]
 
 
-def get_connections(conn_id: str) -> List[Connection]:
+def get_connections(conn_id: str) -> List['Connection']:
     """
     Get all connections as an iterable.
 
@@ -71,13 +74,9 @@ def get_variable(key: str) -> Optional[str]:
     return None
 
 
-def initialize_secrets_backends() -> List[BaseSecretsBackend]:
-    """
-    * import secrets backend classes
-    * instantiate them and return them in a list
-    """
-    secrets_backend_cls = conf.getimport(section=CONFIG_SECTION, key='backend')
-    backend_list = []
+def get_custom_secret_backend() -> Optional[BaseSecretsBackend]:
+    """Get Secret Backend if defined in airflow.cfg"""
+    secrets_backend_cls = conf.getimport(section='secrets', key='backend')
 
     if secrets_backend_cls:
         try:
@@ -87,7 +86,21 @@ def initialize_secrets_backends() -> List[BaseSecretsBackend]:
         except JSONDecodeError:
             alternative_secrets_config_dict = {}
 
-        backend_list.append(secrets_backend_cls(**alternative_secrets_config_dict))
+        return secrets_backend_cls(**alternative_secrets_config_dict)
+    return None
+
+
+def initialize_secrets_backends() -> List[BaseSecretsBackend]:
+    """
+    * import secrets backend classes
+    * instantiate them and return them in a list
+    """
+    backend_list = []
+
+    custom_secret_backend = get_custom_secret_backend()
+
+    if custom_secret_backend is not None:
+        backend_list.append(custom_secret_backend)
 
     for class_name in DEFAULT_SECRETS_SEARCH_PATH:
         secrets_backend_cls = import_string(class_name)

airflow/secrets/base_secrets.py

@@ -16,9 +16,10 @@
 # under the License.
 
 from abc import ABC
-from typing import List, Optional
+from typing import TYPE_CHECKING, List, Optional
 
-from airflow.models.connection import Connection
+if TYPE_CHECKING:
+    from airflow.models.connection import Connection
 
 
 class BaseSecretsBackend(ABC):
@@ -52,13 +53,14 @@ def get_conn_uri(self, conn_id: str) -> Optional[str]:
         """
         raise NotImplementedError()
 
-    def get_connections(self, conn_id: str) -> List[Connection]:
+    def get_connections(self, conn_id: str) -> List['Connection']:
         """
         Return connection object with a given ``conn_id``.
 
         :param conn_id: connection id
         :type conn_id: str
         """
+        from airflow.models.connection import Connection
         conn_uri = self.get_conn_uri(conn_id=conn_id)
         if not conn_uri:
             return []
@@ -73,3 +75,12 @@ def get_variable(self, key: str) -> Optional[str]:
         :return: Variable Value
         """
         raise NotImplementedError()
+
+    def get_config(self, key: str) -> Optional[str]:    # pylint: disable=unused-argument
+        """
+        Return value for Airflow Config Key
+
+        :param key: Config Key
+        :return: Config Value
+        """
+        return None

airflow/secrets/metastore.py

@@ -19,12 +19,14 @@
 Objects relating to sourcing connections from metastore database
 """
 
-from typing import List
+from typing import TYPE_CHECKING, List
 
-from airflow.models.connection import Connection
 from airflow.secrets import BaseSecretsBackend
 from airflow.utils.session import provide_session
 
+if TYPE_CHECKING:
+    from airflow.models.connection import Connection
+
 
 class MetastoreBackend(BaseSecretsBackend):
     """
@@ -33,7 +35,8 @@ class MetastoreBackend(BaseSecretsBackend):
 
     # pylint: disable=missing-docstring
     @provide_session
-    def get_connections(self, conn_id, session=None) -> List[Connection]:
+    def get_connections(self, conn_id, session=None) -> List['Connection']:
+        from airflow.models.connection import Connection
         conn_list = session.query(Connection).filter(Connection.conn_id == conn_id).all()
         session.expunge_all()
         return conn_list