From 46010c535761ab83579027e32e8c6a033c272f8d Mon Sep 17 00:00:00 2001
From: Kaxil Naik <kaxilnaik@gmail.com>
Date: Fri, 3 Jul 2020 16:39:32 +0100
Subject: [PATCH 1/9] Get Airflow configs with sensitive data from Secret
 Backends

---
 airflow/configuration.py                      | 76 +++++++++++++++++--
 .../amazon/aws/secrets/secrets_manager.py     | 20 ++++-
 airflow/providers/hashicorp/secrets/vault.py  | 22 +++++-
 airflow/secrets/base_secrets.py               |  9 +++
 .../providers/hashicorp/secrets/test_vault.py | 31 ++++++++
 tests/test_configuration.py                   | 50 ++++++++++--
 6 files changed, 194 insertions(+), 14 deletions(-)

diff --git a/airflow/configuration.py b/airflow/configuration.py
index 8e00011f11e20..8c7518f35967a 100644
--- a/airflow/configuration.py
+++ b/airflow/configuration.py
@@ -17,6 +17,7 @@
 # under the License.
 
 import copy
+import json
 import logging
 import multiprocessing
 import os
@@ -30,9 +31,11 @@
 from collections import OrderedDict
 # Ignored Mypy on configparser because it thinks the configparser module has no _UNSET attribute
 from configparser import _UNSET, ConfigParser, NoOptionError, NoSectionError  # type: ignore
+from json import JSONDecodeError
 from typing import Dict, Optional, Tuple, Union
 
 import yaml
+from cached_property import cached_property
 from cryptography.fernet import Fernet
 
 from airflow.exceptions import AirflowConfigException
@@ -115,7 +118,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
+    configs_with_secret = {
         ('core', 'sql_alchemy_conn'),
         ('core', 'fernet_key'),
         ('celery', 'broker_url'),
@@ -267,17 +272,49 @@ 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.configs_with_secret:
                 return run_command(os.environ[env_var_cmd])
 
     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.configs_with_secret:
             if super().has_option(section, fallback_key):
                 command = super().get(section, fallback_key)
                 return run_command(command)
 
+    @cached_property
+    def _secrets_backend_client(self):
+        if super().has_option('secrets', 'backend') is False:
+            return None
+
+        secrets_backend_cls = super().get('secrets', 'backend')
+        if not secrets_backend_cls:
+            return None
+        print("secrets_backend_cls", secrets_backend_cls)
+        secrets_backend = import_string(secrets_backend_cls)
+        if secrets_backend:
+            try:
+                alternative_secrets_config_dict = json.loads(
+                    super().get('secrets', 'backend_kwargs', fallback='{}')
+                )
+            except JSONDecodeError:
+                alternative_secrets_config_dict = {}
+            return secrets_backend(**alternative_secrets_config_dict)
+
+    def _get_secret_option(self, section, key):
+        """Get Config option values from Secret Backend"""
+        secrets_client = self._secrets_backend_client
+        print("S clinet", secrets_client)
+        if not secrets_client:
+            return None
+        fallback_key = key + '_secret'
+        # if this is a valid secret key...
+        if (section, key) in self.configs_with_secret:
+            if super().has_option(section, fallback_key):
+                secrets_path = super().get(section, fallback_key)
+                return secrets_client.get_configuration(secrets_path)
+
     def get(self, section, key, **kwargs):
         section = str(section).lower()
         key = str(key).lower()
@@ -319,6 +356,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 +520,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 +543,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 +591,7 @@ def as_dict(
 
         # add bash commands
         if include_cmds:
-            for (section, key) in self.as_command_stdout:
+            for (section, key) in self.configs_with_secret:
                 opt = self._get_cmd_option(section, key)
                 if opt:
                     if not display_sensitive:
@@ -551,6 +603,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.configs_with_secret:
+                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):
diff --git a/airflow/providers/amazon/aws/secrets/secrets_manager.py b/airflow/providers/amazon/aws/secrets/secrets_manager.py
index 6932d8e5f32b7..ce8647885544c 100644
--- a/airflow/providers/amazon/aws/secrets/secrets_manager.py
+++ b/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 configurations_prefix is ``airflow/configurations/sql_alchemy_conn``, this would be accessible
+    if you provide ``{"configurations_prefix": "airflow/configurations"}`` and request variable
+    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 configurations_prefix: Specifies the prefix of the secret to read to get Variables.
+    :type configurations_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,6 +67,7 @@ def __init__(
         self,
         connections_prefix: str = 'airflow/connections',
         variables_prefix: str = 'airflow/variables',
+        configurations_prefix: str = 'airflow/configuration',
         profile_name: Optional[str] = None,
         sep: str = "/",
         **kwargs
@@ -69,6 +75,7 @@ def __init__(
         super().__init__()
         self.connections_prefix = connections_prefix.rstrip("/")
         self.variables_prefix = variables_prefix.rstrip('/')
+        self.configurations_prefix = configurations_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_configuration(self, key: str) -> Optional[str]:
+        """
+        Get Airflow Configuration
+
+        :param key: Configuration Option Key
+        :return: Configuration Option Value
+        """
+        return self._get_secret(self.configurations_prefix, key)
+
     def _get_secret(self, path_prefix: str, secret_id: str) -> Optional[str]:
         """
         Get secret value from Secrets Manager
diff --git a/airflow/providers/hashicorp/secrets/vault.py b/airflow/providers/hashicorp/secrets/vault.py
index 482632311ca02..4a31484f52161 100644
--- a/airflow/providers/hashicorp/secrets/vault.py
+++ b/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 configurations_path: Specifies the path of the secret to read Airflow Configurations
+        (default: 'configurations').
+    :type configurations_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',
+        configurations_path: str = 'configurations',
         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.configurations_path = configurations_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_configuration(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.configurations_path, key)
+        response = self.vault_client.get_secret(secret_path=secret_path)
+        return response.get("value") if response else None
diff --git a/airflow/secrets/base_secrets.py b/airflow/secrets/base_secrets.py
index 100bd16666693..d1864490e2f4d 100644
--- a/airflow/secrets/base_secrets.py
+++ b/airflow/secrets/base_secrets.py
@@ -73,3 +73,12 @@ def get_variable(self, key: str) -> Optional[str]:
         :return: Variable Value
         """
         raise NotImplementedError()
+
+    def get_configuration(self, key: str) -> Optional[str]:
+        """
+        Return value for Airflow Config Key
+
+        :param key: Config Key
+        :return: Config Value
+        """
+        raise NotImplementedError()
diff --git a/tests/providers/hashicorp/secrets/test_vault.py b/tests/providers/hashicorp/secrets/test_vault.py
index fbb3f5ad22cd4..5c2c65012b28a 100644
--- a/tests/providers/hashicorp/secrets/test_vault.py
+++ b/tests/providers/hashicorp/secrets/test_vault.py
@@ -260,3 +260,34 @@ def test_auth_type_kubernetes_with_unreadable_jwt_raises_error(self):
 
         with self.assertRaisesRegex(FileNotFoundError, path):
             VaultBackend(**kwargs).get_connections(conn_id='test')
+
+    @mock.patch("airflow.providers.hashicorp._internal_client.vault_client.hvac")
+    def test_get_config_value(self, mock_hvac):
+        mock_client = mock.MagicMock()
+        mock_hvac.Client.return_value = mock_client
+        mock_client.secrets.kv.v2.read_secret_version.return_value = {
+            'request_id': '2d48a2ad-6bcb-e5b6-429d-da35fdf31f56',
+            'lease_id': '',
+            'renewable': False,
+            'lease_duration': 0,
+            'data': {'data': {'value': 'sqlite:////Users/airflow/airflow/airflow.db'},
+                     'metadata': {'created_time': '2020-03-28T02:10:54.301784Z',
+                                  'deletion_time': '',
+                                  'destroyed': False,
+                                  'version': 1}},
+            'wrap_info': None,
+            'warnings': None,
+            'auth': None
+        }
+
+        kwargs = {
+            "configurations_path": "configurations",
+            "mount_point": "secret",
+            "auth_type": "token",
+            "url": "http://127.0.0.1:8200",
+            "token": "s.FnL7qg0YnHZDpf4zKKuFy0UK"
+        }
+
+        test_client = VaultBackend(**kwargs)
+        returned_uri = test_client.get_configuration("sql_alchemy_conn")
+        self.assertEqual('sqlite:////Users/airflow/airflow/airflow.db', returned_uri)
diff --git a/tests/test_configuration.py b/tests/test_configuration.py
index 09ab57adc5b9a..5b9179068f438 100644
--- a/tests/test_configuration.py
+++ b/tests/test_configuration.py
@@ -169,7 +169,7 @@ def test_command_precedence(self):
         test_conf = AirflowConfigParser(
             default_config=parameterized_config(test_config_default))
         test_conf.read_string(test_config)
-        test_conf.as_command_stdout = test_conf.as_command_stdout | {
+        test_conf.configs_with_secret = test_conf.configs_with_secret | {
             ('test', 'key2'),
             ('test', 'key4'),
         }
@@ -203,6 +203,46 @@ def test_command_precedence(self):
         self.assertNotIn('key4', cfg_dict['test'])
         self.assertEqual('printf key4_result', cfg_dict['test']['key4_cmd'])
 
+    @mock.patch("airflow.providers.hashicorp._internal_client.vault_client.hvac")
+    def test_config_from_secret_backend(self, mock_hvac):
+        """Get Config Value from a Secret Backend"""
+        mock_client = mock.MagicMock()
+        mock_hvac.Client.return_value = mock_client
+        mock_client.secrets.kv.v2.read_secret_version.return_value = {
+            'request_id': '2d48a2ad-6bcb-e5b6-429d-da35fdf31f56',
+            'lease_id': '',
+            'renewable': False,
+            'lease_duration': 0,
+            'data': {'data': {'value': 'sqlite:////Users/airflow/airflow/airflow.db'},
+                     'metadata': {'created_time': '2020-03-28T02:10:54.301784Z',
+                                  'deletion_time': '',
+                                  'destroyed': False,
+                                  'version': 1}},
+            'wrap_info': None,
+            'warnings': None,
+            'auth': None
+        }
+
+        test_config = '''[test]
+sql_alchemy_conn_secret = sql_alchemy_conn
+
+[secrets]
+backend = airflow.providers.hashicorp.secrets.vault.VaultBackend
+backend_kwargs = {"configurations_path": "configurations","url": "http://127.0.0.1:8200", "token": "token"}
+'''
+        test_config_default = '''[test]
+sql_alchemy_conn = airflow
+'''
+
+        test_conf = AirflowConfigParser(default_config=parameterized_config(test_config_default))
+        test_conf.read_string(test_config)
+        test_conf.configs_with_secret = test_conf.configs_with_secret | {
+            ('test', 'sql_alchemy_conn'),
+        }
+
+        self.assertEqual(
+            'sqlite:////Users/airflow/airflow/airflow.db', test_conf.get('test', 'sql_alchemy_conn'))
+
     def test_getboolean(self):
         """Test AirflowConfigParser.getboolean"""
         test_config = """
@@ -434,7 +474,7 @@ def test_deprecated_options_cmd(self):
         # Guarantee we have a deprecated setting, so we test the deprecation
         # lookup even if we remove this explicit fallback
         conf.deprecated_options[('celery', "result_backend")] = ('celery', 'celery_result_backend')
-        conf.as_command_stdout.add(('celery', 'celery_result_backend'))
+        conf.configs_with_secret.add(('celery', 'celery_result_backend'))
 
         conf.remove_option('celery', 'result_backend')
         with conf_vars({('celery', 'celery_result_backend_cmd'): '/bin/echo 99'}):
@@ -511,13 +551,13 @@ def test_command_from_env(self):
 '''
         test_cmdenv_conf = AirflowConfigParser()
         test_cmdenv_conf.read_string(test_cmdenv_config)
-        test_cmdenv_conf.as_command_stdout.add(('testcmdenv', 'itsacommand'))
+        test_cmdenv_conf.configs_with_secret.add(('testcmdenv', 'itsacommand'))
         with unittest.mock.patch.dict('os.environ'):
             # AIRFLOW__TESTCMDENV__ITSACOMMAND_CMD maps to ('testcmdenv', 'itsacommand') in
-            # as_command_stdout and therefore should return 'OK' from the environment variable's
+            # configs_with_secret and therefore should return 'OK' from the environment variable's
             # echo command, and must not return 'NOT OK' from the configuration
             self.assertEqual(test_cmdenv_conf.get('testcmdenv', 'itsacommand'), 'OK')
-            # AIRFLOW__TESTCMDENV__NOTACOMMAND_CMD maps to no entry in as_command_stdout and therefore
+            # AIRFLOW__TESTCMDENV__NOTACOMMAND_CMD maps to no entry in configs_with_secret and therefore
             # the option should return 'OK' from the configuration, and must not return 'NOT OK' from
             # the environement variable's echo command
             self.assertEqual(test_cmdenv_conf.get('testcmdenv', 'notacommand'), 'OK')

From 6b1b4c357c63e6433f805148ec12a9ec195ca061 Mon Sep 17 00:00:00 2001
From: Kaxil Naik <kaxilnaik@gmail.com>
Date: Fri, 3 Jul 2020 16:50:54 +0100
Subject: [PATCH 2/9] fixup! Get Airflow configs with sensitive data from
 Secret Backends

---
 airflow/configuration.py | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/airflow/configuration.py b/airflow/configuration.py
index 8c7518f35967a..f937acf62c88c 100644
--- a/airflow/configuration.py
+++ b/airflow/configuration.py
@@ -31,7 +31,6 @@
 from collections import OrderedDict
 # Ignored Mypy on configparser because it thinks the configparser module has no _UNSET attribute
 from configparser import _UNSET, ConfigParser, NoOptionError, NoSectionError  # type: ignore
-from json import JSONDecodeError
 from typing import Dict, Optional, Tuple, Union
 
 import yaml
@@ -298,7 +297,7 @@ def _secrets_backend_client(self):
                 alternative_secrets_config_dict = json.loads(
                     super().get('secrets', 'backend_kwargs', fallback='{}')
                 )
-            except JSONDecodeError:
+            except json.JSONDecodeError:
                 alternative_secrets_config_dict = {}
             return secrets_backend(**alternative_secrets_config_dict)
 

From e0f1b253ded6033ec0ad96010a3116e29a99f884 Mon Sep 17 00:00:00 2001
From: Kaxil Naik <kaxilnaik@gmail.com>
Date: Fri, 3 Jul 2020 17:40:03 +0100
Subject: [PATCH 3/9] fixup! fixup! Get Airflow configs with sensitive data
 from Secret Backends

---
 airflow/configuration.py                         | 14 +++++++-------
 .../amazon/aws/secrets/secrets_manager.py        | 16 ++++++++--------
 airflow/providers/hashicorp/secrets/vault.py     | 14 +++++++-------
 airflow/secrets/base_secrets.py                  |  4 ++--
 tests/providers/hashicorp/secrets/test_vault.py  |  4 ++--
 tests/test_configuration.py                      | 14 +++++++-------
 6 files changed, 33 insertions(+), 33 deletions(-)

diff --git a/airflow/configuration.py b/airflow/configuration.py
index f937acf62c88c..80ee5cd7243bc 100644
--- a/airflow/configuration.py
+++ b/airflow/configuration.py
@@ -119,7 +119,7 @@ class AirflowConfigParser(ConfigParser):
     # is to not store password on boxes in text files.
     # These configs can also be fetched from Secrets backend
     # following the "{section}__{name}__secret" pattern
-    configs_with_secret = {
+    senstive_config_values = {
         ('core', 'sql_alchemy_conn'),
         ('core', 'fernet_key'),
         ('celery', 'broker_url'),
@@ -271,13 +271,13 @@ 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.configs_with_secret:
+            if (section, key) in self.senstive_config_values:
                 return run_command(os.environ[env_var_cmd])
 
     def _get_cmd_option(self, section, key):
         fallback_key = key + '_cmd'
         # if this is a valid command key...
-        if (section, key) in self.configs_with_secret:
+        if (section, key) in self.senstive_config_values:
             if super().has_option(section, fallback_key):
                 command = super().get(section, fallback_key)
                 return run_command(command)
@@ -309,10 +309,10 @@ def _get_secret_option(self, section, key):
             return None
         fallback_key = key + '_secret'
         # if this is a valid secret key...
-        if (section, key) in self.configs_with_secret:
+        if (section, key) in self.senstive_config_values:
             if super().has_option(section, fallback_key):
                 secrets_path = super().get(section, fallback_key)
-                return secrets_client.get_configuration(secrets_path)
+                return secrets_client.get_config(secrets_path)
 
     def get(self, section, key, **kwargs):
         section = str(section).lower()
@@ -590,7 +590,7 @@ def as_dict(
 
         # add bash commands
         if include_cmds:
-            for (section, key) in self.configs_with_secret:
+            for (section, key) in self.senstive_config_values:
                 opt = self._get_cmd_option(section, key)
                 if opt:
                     if not display_sensitive:
@@ -604,7 +604,7 @@ def as_dict(
 
         # add config from secret backends
         if include_secret:
-            for (section, key) in self.configs_with_secret:
+            for (section, key) in self.senstive_config_values:
                 opt = self._get_secret_option(section, key)
                 if opt:
                     if not display_sensitive:
diff --git a/airflow/providers/amazon/aws/secrets/secrets_manager.py b/airflow/providers/amazon/aws/secrets/secrets_manager.py
index ce8647885544c..0a0a02bddddbd 100644
--- a/airflow/providers/amazon/aws/secrets/secrets_manager.py
+++ b/airflow/providers/amazon/aws/secrets/secrets_manager.py
@@ -44,8 +44,8 @@ class SecretsManagerBackend(BaseSecretsBackend, LoggingMixin):
     if you provide ``{"connections_prefix": "airflow/connections"}`` and request conn_id ``smtp_default``.
     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 configurations_prefix is ``airflow/configurations/sql_alchemy_conn``, this would be accessible
-    if you provide ``{"configurations_prefix": "airflow/configurations"}`` and request variable
+    And if configs_prefix is ``airflow/configs/sql_alchemy_conn``, this would be accessible
+    if you provide ``{"configs_prefix": "airflow/configs"}`` and request variable
     key ``sql_alchemy_conn``.
 
     You can also pass additional keyword arguments like ``aws_secret_access_key``, ``aws_access_key_id``
@@ -55,8 +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 configurations_prefix: Specifies the prefix of the secret to read to get Variables.
-    :type configurations_prefix: str
+    :param configs_prefix: Specifies the prefix of the secret to read to get Variables.
+    :type configs_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: "/"
@@ -67,7 +67,7 @@ def __init__(
         self,
         connections_prefix: str = 'airflow/connections',
         variables_prefix: str = 'airflow/variables',
-        configurations_prefix: str = 'airflow/configuration',
+        configs_prefix: str = 'airflow/configs',
         profile_name: Optional[str] = None,
         sep: str = "/",
         **kwargs
@@ -75,7 +75,7 @@ def __init__(
         super().__init__()
         self.connections_prefix = connections_prefix.rstrip("/")
         self.variables_prefix = variables_prefix.rstrip('/')
-        self.configurations_prefix = configurations_prefix.rstrip('/')
+        self.configs_prefix = configs_prefix.rstrip('/')
         self.profile_name = profile_name
         self.sep = sep
         self.kwargs = kwargs
@@ -108,14 +108,14 @@ def get_variable(self, key: str) -> Optional[str]:
         """
         return self._get_secret(self.variables_prefix, key)
 
-    def get_configuration(self, key: str) -> Optional[str]:
+    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.configurations_prefix, key)
+        return self._get_secret(self.configs_prefix, key)
 
     def _get_secret(self, path_prefix: str, secret_id: str) -> Optional[str]:
         """
diff --git a/airflow/providers/hashicorp/secrets/vault.py b/airflow/providers/hashicorp/secrets/vault.py
index 4a31484f52161..c1d99717815ab 100644
--- a/airflow/providers/hashicorp/secrets/vault.py
+++ b/airflow/providers/hashicorp/secrets/vault.py
@@ -52,9 +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 configurations_path: Specifies the path of the secret to read Airflow Configurations
-        (default: 'configurations').
-    :type configurations_path: str
+    :param configs_path: Specifies the path of the secret to read Airflow Configurations
+        (default: 'configs').
+    :type configs_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:
@@ -114,7 +114,7 @@ def __init__(  # pylint: disable=too-many-arguments
         self,
         connections_path: str = 'connections',
         variables_path: str = 'variables',
-        configurations_path: str = 'configurations',
+        configs_path: str = 'configs',
         url: Optional[str] = None,
         auth_type: str = 'token',
         auth_mount_point: Optional[str] = None,
@@ -142,7 +142,7 @@ def __init__(  # pylint: disable=too-many-arguments
         super().__init__()
         self.connections_path = connections_path.rstrip('/')
         self.variables_path = variables_path.rstrip('/')
-        self.configurations_path = configurations_path.rstrip('/')
+        self.configs_path = configs_path.rstrip('/')
         self.mount_point = mount_point
         self.kv_engine_version = kv_engine_version
         self.vault_client = _VaultClient(
@@ -197,7 +197,7 @@ def get_variable(self, key: str) -> Optional[str]:
         response = self.vault_client.get_secret(secret_path=secret_path)
         return response.get("value") if response else None
 
-    def get_configuration(self, key: str) -> Optional[str]:
+    def get_config(self, key: str) -> Optional[str]:
         """
         Get Airflow Configuration
 
@@ -206,6 +206,6 @@ def get_configuration(self, key: str) -> Optional[str]:
         :rtype: str
         :return: Configuration Option Value retrieved from the vault
         """
-        secret_path = self.build_path(self.configurations_path, key)
+        secret_path = self.build_path(self.configs_path, key)
         response = self.vault_client.get_secret(secret_path=secret_path)
         return response.get("value") if response else None
diff --git a/airflow/secrets/base_secrets.py b/airflow/secrets/base_secrets.py
index d1864490e2f4d..bd903f81aa608 100644
--- a/airflow/secrets/base_secrets.py
+++ b/airflow/secrets/base_secrets.py
@@ -74,11 +74,11 @@ def get_variable(self, key: str) -> Optional[str]:
         """
         raise NotImplementedError()
 
-    def get_configuration(self, key: str) -> Optional[str]:
+    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
         """
-        raise NotImplementedError()
+        return None
diff --git a/tests/providers/hashicorp/secrets/test_vault.py b/tests/providers/hashicorp/secrets/test_vault.py
index 5c2c65012b28a..5a4449b357352 100644
--- a/tests/providers/hashicorp/secrets/test_vault.py
+++ b/tests/providers/hashicorp/secrets/test_vault.py
@@ -281,7 +281,7 @@ def test_get_config_value(self, mock_hvac):
         }
 
         kwargs = {
-            "configurations_path": "configurations",
+            "configs_path": "configurations",
             "mount_point": "secret",
             "auth_type": "token",
             "url": "http://127.0.0.1:8200",
@@ -289,5 +289,5 @@ def test_get_config_value(self, mock_hvac):
         }
 
         test_client = VaultBackend(**kwargs)
-        returned_uri = test_client.get_configuration("sql_alchemy_conn")
+        returned_uri = test_client.get_config("sql_alchemy_conn")
         self.assertEqual('sqlite:////Users/airflow/airflow/airflow.db', returned_uri)
diff --git a/tests/test_configuration.py b/tests/test_configuration.py
index 5b9179068f438..f0b04c2d3970d 100644
--- a/tests/test_configuration.py
+++ b/tests/test_configuration.py
@@ -169,7 +169,7 @@ def test_command_precedence(self):
         test_conf = AirflowConfigParser(
             default_config=parameterized_config(test_config_default))
         test_conf.read_string(test_config)
-        test_conf.configs_with_secret = test_conf.configs_with_secret | {
+        test_conf.as_command_stdout = test_conf.as_command_stdout | {
             ('test', 'key2'),
             ('test', 'key4'),
         }
@@ -228,7 +228,7 @@ def test_config_from_secret_backend(self, mock_hvac):
 
 [secrets]
 backend = airflow.providers.hashicorp.secrets.vault.VaultBackend
-backend_kwargs = {"configurations_path": "configurations","url": "http://127.0.0.1:8200", "token": "token"}
+backend_kwargs = {"configs_path": "configurations","url": "http://127.0.0.1:8200", "token": "token"}
 '''
         test_config_default = '''[test]
 sql_alchemy_conn = airflow
@@ -236,7 +236,7 @@ def test_config_from_secret_backend(self, mock_hvac):
 
         test_conf = AirflowConfigParser(default_config=parameterized_config(test_config_default))
         test_conf.read_string(test_config)
-        test_conf.configs_with_secret = test_conf.configs_with_secret | {
+        test_conf.as_command_stdout = test_conf.as_command_stdout | {
             ('test', 'sql_alchemy_conn'),
         }
 
@@ -474,7 +474,7 @@ def test_deprecated_options_cmd(self):
         # Guarantee we have a deprecated setting, so we test the deprecation
         # lookup even if we remove this explicit fallback
         conf.deprecated_options[('celery', "result_backend")] = ('celery', 'celery_result_backend')
-        conf.configs_with_secret.add(('celery', 'celery_result_backend'))
+        conf.as_command_stdout.add(('celery', 'celery_result_backend'))
 
         conf.remove_option('celery', 'result_backend')
         with conf_vars({('celery', 'celery_result_backend_cmd'): '/bin/echo 99'}):
@@ -551,13 +551,13 @@ def test_command_from_env(self):
 '''
         test_cmdenv_conf = AirflowConfigParser()
         test_cmdenv_conf.read_string(test_cmdenv_config)
-        test_cmdenv_conf.configs_with_secret.add(('testcmdenv', 'itsacommand'))
+        test_cmdenv_conf.as_command_stdout.add(('testcmdenv', 'itsacommand'))
         with unittest.mock.patch.dict('os.environ'):
             # AIRFLOW__TESTCMDENV__ITSACOMMAND_CMD maps to ('testcmdenv', 'itsacommand') in
-            # configs_with_secret and therefore should return 'OK' from the environment variable's
+            # as_command_stdout and therefore should return 'OK' from the environment variable's
             # echo command, and must not return 'NOT OK' from the configuration
             self.assertEqual(test_cmdenv_conf.get('testcmdenv', 'itsacommand'), 'OK')
-            # AIRFLOW__TESTCMDENV__NOTACOMMAND_CMD maps to no entry in configs_with_secret and therefore
+            # AIRFLOW__TESTCMDENV__NOTACOMMAND_CMD maps to no entry in as_command_stdout and therefore
             # the option should return 'OK' from the configuration, and must not return 'NOT OK' from
             # the environement variable's echo command
             self.assertEqual(test_cmdenv_conf.get('testcmdenv', 'notacommand'), 'OK')

From 67bc20d91cd8954b924d80ca92287957a7ce8cc7 Mon Sep 17 00:00:00 2001
From: Kaxil Naik <kaxilnaik@gmail.com>
Date: Fri, 3 Jul 2020 17:45:56 +0100
Subject: [PATCH 4/9] fixup! fixup! fixup! Get Airflow configs with sensitive
 data from Secret Backends

---
 airflow/configuration.py    | 12 ++++++------
 tests/test_configuration.py | 14 +++++++-------
 2 files changed, 13 insertions(+), 13 deletions(-)

diff --git a/airflow/configuration.py b/airflow/configuration.py
index 80ee5cd7243bc..b744a4934c413 100644
--- a/airflow/configuration.py
+++ b/airflow/configuration.py
@@ -119,7 +119,7 @@ class AirflowConfigParser(ConfigParser):
     # is to not store password on boxes in text files.
     # These configs can also be fetched from Secrets backend
     # following the "{section}__{name}__secret" pattern
-    senstive_config_values = {
+    sensitive_config_values = {
         ('core', 'sql_alchemy_conn'),
         ('core', 'fernet_key'),
         ('celery', 'broker_url'),
@@ -271,13 +271,13 @@ 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.senstive_config_values:
+            if (section, key) in self.sensitive_config_values:
                 return run_command(os.environ[env_var_cmd])
 
     def _get_cmd_option(self, section, key):
         fallback_key = key + '_cmd'
         # if this is a valid command key...
-        if (section, key) in self.senstive_config_values:
+        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)
@@ -309,7 +309,7 @@ def _get_secret_option(self, section, key):
             return None
         fallback_key = key + '_secret'
         # if this is a valid secret key...
-        if (section, key) in self.senstive_config_values:
+        if (section, key) in self.sensitive_config_values:
             if super().has_option(section, fallback_key):
                 secrets_path = super().get(section, fallback_key)
                 return secrets_client.get_config(secrets_path)
@@ -590,7 +590,7 @@ def as_dict(
 
         # add bash commands
         if include_cmds:
-            for (section, key) in self.senstive_config_values:
+            for (section, key) in self.sensitive_config_values:
                 opt = self._get_cmd_option(section, key)
                 if opt:
                     if not display_sensitive:
@@ -604,7 +604,7 @@ def as_dict(
 
         # add config from secret backends
         if include_secret:
-            for (section, key) in self.senstive_config_values:
+            for (section, key) in self.sensitive_config_values:
                 opt = self._get_secret_option(section, key)
                 if opt:
                     if not display_sensitive:
diff --git a/tests/test_configuration.py b/tests/test_configuration.py
index f0b04c2d3970d..9c0284f5135dd 100644
--- a/tests/test_configuration.py
+++ b/tests/test_configuration.py
@@ -169,7 +169,7 @@ def test_command_precedence(self):
         test_conf = AirflowConfigParser(
             default_config=parameterized_config(test_config_default))
         test_conf.read_string(test_config)
-        test_conf.as_command_stdout = test_conf.as_command_stdout | {
+        test_conf.sensitive_config_values = test_conf.sensitive_config_values | {
             ('test', 'key2'),
             ('test', 'key4'),
         }
@@ -228,7 +228,7 @@ def test_config_from_secret_backend(self, mock_hvac):
 
 [secrets]
 backend = airflow.providers.hashicorp.secrets.vault.VaultBackend
-backend_kwargs = {"configs_path": "configurations","url": "http://127.0.0.1:8200", "token": "token"}
+backend_kwargs = {"configs_path": "configs","url": "http://127.0.0.1:8200", "token": "token"}
 '''
         test_config_default = '''[test]
 sql_alchemy_conn = airflow
@@ -236,7 +236,7 @@ def test_config_from_secret_backend(self, mock_hvac):
 
         test_conf = AirflowConfigParser(default_config=parameterized_config(test_config_default))
         test_conf.read_string(test_config)
-        test_conf.as_command_stdout = test_conf.as_command_stdout | {
+        test_conf.sensitive_config_values = test_conf.sensitive_config_values | {
             ('test', 'sql_alchemy_conn'),
         }
 
@@ -474,7 +474,7 @@ def test_deprecated_options_cmd(self):
         # Guarantee we have a deprecated setting, so we test the deprecation
         # lookup even if we remove this explicit fallback
         conf.deprecated_options[('celery', "result_backend")] = ('celery', 'celery_result_backend')
-        conf.as_command_stdout.add(('celery', 'celery_result_backend'))
+        conf.sensitive_config_values.add(('celery', 'celery_result_backend'))
 
         conf.remove_option('celery', 'result_backend')
         with conf_vars({('celery', 'celery_result_backend_cmd'): '/bin/echo 99'}):
@@ -551,13 +551,13 @@ def test_command_from_env(self):
 '''
         test_cmdenv_conf = AirflowConfigParser()
         test_cmdenv_conf.read_string(test_cmdenv_config)
-        test_cmdenv_conf.as_command_stdout.add(('testcmdenv', 'itsacommand'))
+        test_cmdenv_conf.sensitive_config_values.add(('testcmdenv', 'itsacommand'))
         with unittest.mock.patch.dict('os.environ'):
             # AIRFLOW__TESTCMDENV__ITSACOMMAND_CMD maps to ('testcmdenv', 'itsacommand') in
-            # as_command_stdout and therefore should return 'OK' from the environment variable's
+            # sensitive_config_values and therefore should return 'OK' from the environment variable's
             # echo command, and must not return 'NOT OK' from the configuration
             self.assertEqual(test_cmdenv_conf.get('testcmdenv', 'itsacommand'), 'OK')
-            # AIRFLOW__TESTCMDENV__NOTACOMMAND_CMD maps to no entry in as_command_stdout and therefore
+            # AIRFLOW__TESTCMDENV__NOTACOMMAND_CMD maps to no entry in sensitive_config_values and therefore
             # the option should return 'OK' from the configuration, and must not return 'NOT OK' from
             # the environement variable's echo command
             self.assertEqual(test_cmdenv_conf.get('testcmdenv', 'notacommand'), 'OK')

From c0717250929614cbceb632fea3dee28729ddcdee Mon Sep 17 00:00:00 2001
From: Kaxil Naik <kaxilnaik@gmail.com>
Date: Sat, 4 Jul 2020 02:11:52 +0100
Subject: [PATCH 5/9] fixup! fixup! fixup! fixup! Get Airflow configs with
 sensitive data from Secret Backends

---
 airflow/configuration.py        | 31 ++++++---------------------
 airflow/secrets/__init__.py     | 37 ++++++++++++++++++++++-----------
 airflow/secrets/base_secrets.py |  8 ++++---
 airflow/secrets/metastore.py    |  9 +++++---
 tests/test_configuration.py     |  8 +++----
 5 files changed, 46 insertions(+), 47 deletions(-)

diff --git a/airflow/configuration.py b/airflow/configuration.py
index b744a4934c413..7e1d615cefa98 100644
--- a/airflow/configuration.py
+++ b/airflow/configuration.py
@@ -17,7 +17,6 @@
 # under the License.
 
 import copy
-import json
 import logging
 import multiprocessing
 import os
@@ -34,7 +33,6 @@
 from typing import Dict, Optional, Tuple, Union
 
 import yaml
-from cached_property import cached_property
 from cryptography.fernet import Fernet
 
 from airflow.exceptions import AirflowConfigException
@@ -282,35 +280,18 @@ def _get_cmd_option(self, section, key):
                 command = super().get(section, fallback_key)
                 return run_command(command)
 
-    @cached_property
-    def _secrets_backend_client(self):
-        if super().has_option('secrets', 'backend') is False:
-            return None
-
-        secrets_backend_cls = super().get('secrets', 'backend')
-        if not secrets_backend_cls:
-            return None
-        print("secrets_backend_cls", secrets_backend_cls)
-        secrets_backend = import_string(secrets_backend_cls)
-        if secrets_backend:
-            try:
-                alternative_secrets_config_dict = json.loads(
-                    super().get('secrets', 'backend_kwargs', fallback='{}')
-                )
-            except json.JSONDecodeError:
-                alternative_secrets_config_dict = {}
-            return secrets_backend(**alternative_secrets_config_dict)
-
     def _get_secret_option(self, section, key):
         """Get Config option values from Secret Backend"""
-        secrets_client = self._secrets_backend_client
-        print("S clinet", secrets_client)
-        if not secrets_client:
-            return None
+        from airflow import secrets
+
         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_client = secrets.get_custom_secret_backend()
+                if not secrets_client:
+                    return None
+
                 secrets_path = super().get(section, fallback_key)
                 return secrets_client.get_config(secrets_path)
 
diff --git a/airflow/secrets/__init__.py b/airflow/secrets/__init__.py
index 2751ed5962dec..c62f8a5845736 100644
--- a/airflow/secrets/__init__.py
+++ b/airflow/secrets/__init__.py
@@ -22,18 +22,21 @@
     * 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",
@@ -41,7 +44,7 @@
 ]
 
 
-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)
diff --git a/airflow/secrets/base_secrets.py b/airflow/secrets/base_secrets.py
index bd903f81aa608..c5c8a4ae853b9 100644
--- a/airflow/secrets/base_secrets.py
+++ b/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 []
diff --git a/airflow/secrets/metastore.py b/airflow/secrets/metastore.py
index 9144c4ccab0e2..6d05005cfdda2 100644
--- a/airflow/secrets/metastore.py
+++ b/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
diff --git a/tests/test_configuration.py b/tests/test_configuration.py
index 9c0284f5135dd..3a374345031a4 100644
--- a/tests/test_configuration.py
+++ b/tests/test_configuration.py
@@ -204,6 +204,10 @@ def test_command_precedence(self):
         self.assertEqual('printf key4_result', cfg_dict['test']['key4_cmd'])
 
     @mock.patch("airflow.providers.hashicorp._internal_client.vault_client.hvac")
+    @conf_vars({
+        ("secrets", "backend"): "airflow.providers.hashicorp.secrets.vault.VaultBackend",
+        ("secrets", "backend_kwargs"): '{"url": "http://127.0.0.1:8200", "token": "token"}',
+    })
     def test_config_from_secret_backend(self, mock_hvac):
         """Get Config Value from a Secret Backend"""
         mock_client = mock.MagicMock()
@@ -225,10 +229,6 @@ def test_config_from_secret_backend(self, mock_hvac):
 
         test_config = '''[test]
 sql_alchemy_conn_secret = sql_alchemy_conn
-
-[secrets]
-backend = airflow.providers.hashicorp.secrets.vault.VaultBackend
-backend_kwargs = {"configs_path": "configs","url": "http://127.0.0.1:8200", "token": "token"}
 '''
         test_config_default = '''[test]
 sql_alchemy_conn = airflow

From 3b3810953a84f5d19f34245f1eed8a1f379104e4 Mon Sep 17 00:00:00 2001
From: Kaxil Naik <kaxilnaik@gmail.com>
Date: Sat, 4 Jul 2020 02:37:59 +0100
Subject: [PATCH 6/9] fixup! fixup! fixup! fixup! fixup! Get Airflow configs
 with sensitive data from Secret Backends

---
 airflow/configuration.py  | 23 ++++++++++++++++-------
 docs/howto/set-config.rst | 21 ++++++++++++++++++++-
 2 files changed, 36 insertions(+), 8 deletions(-)

diff --git a/airflow/configuration.py b/airflow/configuration.py
index 7e1d615cefa98..7a7a5aec3c1a2 100644
--- a/airflow/configuration.py
+++ b/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)
@@ -271,6 +280,12 @@ def _get_env_var_option(self, section, key):
             # if this is a valid command key...
             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_cmd = env_var + '_SECRET'
+        if env_var_cmd in os.environ:
+            # if this is a valid command key...
+            if (section, key) in self.sensitive_config_values:
+                return _get_config_value_from_secret_backend(os.environ[env_var_cmd])
 
     def _get_cmd_option(self, section, key):
         fallback_key = key + '_cmd'
@@ -282,18 +297,12 @@ def _get_cmd_option(self, section, key):
 
     def _get_secret_option(self, section, key):
         """Get Config option values from Secret Backend"""
-        from airflow import secrets
-
         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_client = secrets.get_custom_secret_backend()
-                if not secrets_client:
-                    return None
-
                 secrets_path = super().get(section, fallback_key)
-                return secrets_client.get_config(secrets_path)
+                return _get_config_value_from_secret_backend(secrets_path)
 
     def get(self, section, key, **kwargs):
         section = str(section).lower()
diff --git a/docs/howto/set-config.rst b/docs/howto/set-config.rst
index 241720fe325f1..bda105f9d2cb4 100644
--- a/docs/howto/set-config.rst
+++ b/docs/howto/set-config.rst
@@ -46,7 +46,18 @@ the key like this:
     [core]
     sql_alchemy_conn_cmd = bash_command_to_run
 
-The following config options support this ``_cmd`` version:
+You can also derive the connection string at run time by appending ``_secret`` to
+the key like this:
+
+.. code-block:: ini
+
+    [core]
+    sql_alchemy_conn_secret = sql_alchemy_conn
+
+This will retrieve config option from Secret Backends e.g Hashicorp Vault. See
+:ref:`Secrets Backends<secrets_backend_configuration>` for more details.
+
+The following config options support this ``_cmd`` and ``_secret`` version:
 
 * ``sql_alchemy_conn`` in ``[core]`` section
 * ``fernet_key`` in ``[core]`` section
@@ -65,6 +76,13 @@ the same way the usual config options can. For example:
 
     export AIRFLOW__CORE__SQL_ALCHEMY_CONN_CMD=bash_command_to_run
 
+Similarly, ``_secret`` config options can also be set using a corresponding environment variable.
+For example:
+
+.. code-block:: bash
+
+    export AIRFLOW__CORE__SQL_ALCHEMY_CONN_SECRET=sql_alchemy_conn
+
 The idea behind this is to not store passwords on boxes in plain text files.
 
 The universal order of precedence for all configuration options is as follows:
@@ -73,6 +91,7 @@ The universal order of precedence for all configuration options is as follows:
 #. set as a command environment variable
 #. set in ``airflow.cfg``
 #. command in ``airflow.cfg``
+#. secret key in ``airflow.cfg``
 #. Airflow's built in defaults
 
 .. note::

From 032d33f257681ac9d189357fe90fdba23a583da7 Mon Sep 17 00:00:00 2001
From: Kaxil Naik <kaxilnaik@gmail.com>
Date: Wed, 8 Jul 2020 00:25:45 +0100
Subject: [PATCH 7/9] fixup! fixup! fixup! fixup! fixup! fixup! Get Airflow
 configs with sensitive data from Secret Backends

---
 airflow/configuration.py  | 8 ++++----
 docs/howto/set-config.rst | 8 ++++++--
 2 files changed, 10 insertions(+), 6 deletions(-)

diff --git a/airflow/configuration.py b/airflow/configuration.py
index 7a7a5aec3c1a2..798852a4c4c2a 100644
--- a/airflow/configuration.py
+++ b/airflow/configuration.py
@@ -281,11 +281,11 @@ def _get_env_var_option(self, section, key):
             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_cmd = env_var + '_SECRET'
-        if env_var_cmd in os.environ:
-            # if this is a valid command key...
+        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_cmd])
+                return _get_config_value_from_secret_backend(os.environ[env_var_secret_path])
 
     def _get_cmd_option(self, section, key):
         fallback_key = key + '_cmd'
diff --git a/docs/howto/set-config.rst b/docs/howto/set-config.rst
index bda105f9d2cb4..9fc12b66af773 100644
--- a/docs/howto/set-config.rst
+++ b/docs/howto/set-config.rst
@@ -53,6 +53,9 @@ the key like this:
 
     [core]
     sql_alchemy_conn_secret = sql_alchemy_conn
+    # You can also add a nested path
+    # example:
+    # sql_alchemy_conn_secret = core/sql_alchemy_conn
 
 This will retrieve config option from Secret Backends e.g Hashicorp Vault. See
 :ref:`Secrets Backends<secrets_backend_configuration>` for more details.
@@ -87,8 +90,9 @@ The idea behind this is to not store passwords on boxes in plain text files.
 
 The universal order of precedence for all configuration options is as follows:
 
-#. set as an environment variable
-#. set as a command environment variable
+#. set as an environment variable (``AIRFLOW__CORE__SQL_ALCHEMY_CONN``)
+#. set as a command environment variable (``AIRFLOW__CORE__SQL_ALCHEMY_CONN_CMD``)
+#. set as a secret environment variable (``AIRFLOW__CORE__SQL_ALCHEMY_CONN_SECRET``)
 #. set in ``airflow.cfg``
 #. command in ``airflow.cfg``
 #. secret key in ``airflow.cfg``

From 35d75e5a52d413ecb97d935060011472625bac01 Mon Sep 17 00:00:00 2001
From: Kaxil Naik <kaxilnaik@gmail.com>
Date: Wed, 8 Jul 2020 12:26:10 +0100
Subject: [PATCH 8/9] fixup! fixup! fixup! fixup! fixup! fixup! fixup! Get
 Airflow configs with sensitive data from Secret Backends

---
 .../amazon/aws/secrets/secrets_manager.py          | 14 +++++++-------
 airflow/providers/hashicorp/secrets/vault.py       | 10 +++++-----
 2 files changed, 12 insertions(+), 12 deletions(-)

diff --git a/airflow/providers/amazon/aws/secrets/secrets_manager.py b/airflow/providers/amazon/aws/secrets/secrets_manager.py
index 0a0a02bddddbd..6882da7141230 100644
--- a/airflow/providers/amazon/aws/secrets/secrets_manager.py
+++ b/airflow/providers/amazon/aws/secrets/secrets_manager.py
@@ -44,8 +44,8 @@ class SecretsManagerBackend(BaseSecretsBackend, LoggingMixin):
     if you provide ``{"connections_prefix": "airflow/connections"}`` and request conn_id ``smtp_default``.
     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 configs_prefix is ``airflow/configs/sql_alchemy_conn``, this would be accessible
-    if you provide ``{"configs_prefix": "airflow/configs"}`` and request variable
+    And if config_prefix is ``airflow/config/sql_alchemy_conn``, this would be accessible
+    if you provide ``{"config_prefix": "airflow/config"}`` and request variable
     key ``sql_alchemy_conn``.
 
     You can also pass additional keyword arguments like ``aws_secret_access_key``, ``aws_access_key_id``
@@ -55,8 +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 configs_prefix: Specifies the prefix of the secret to read to get Variables.
-    :type configs_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: "/"
@@ -67,7 +67,7 @@ def __init__(
         self,
         connections_prefix: str = 'airflow/connections',
         variables_prefix: str = 'airflow/variables',
-        configs_prefix: str = 'airflow/configs',
+        config_prefix: str = 'airflow/config',
         profile_name: Optional[str] = None,
         sep: str = "/",
         **kwargs
@@ -75,7 +75,7 @@ def __init__(
         super().__init__()
         self.connections_prefix = connections_prefix.rstrip("/")
         self.variables_prefix = variables_prefix.rstrip('/')
-        self.configs_prefix = configs_prefix.rstrip('/')
+        self.config_prefix = config_prefix.rstrip('/')
         self.profile_name = profile_name
         self.sep = sep
         self.kwargs = kwargs
@@ -115,7 +115,7 @@ def get_config(self, key: str) -> Optional[str]:
         :param key: Configuration Option Key
         :return: Configuration Option Value
         """
-        return self._get_secret(self.configs_prefix, key)
+        return self._get_secret(self.config_prefix, key)
 
     def _get_secret(self, path_prefix: str, secret_id: str) -> Optional[str]:
         """
diff --git a/airflow/providers/hashicorp/secrets/vault.py b/airflow/providers/hashicorp/secrets/vault.py
index c1d99717815ab..635971666ce57 100644
--- a/airflow/providers/hashicorp/secrets/vault.py
+++ b/airflow/providers/hashicorp/secrets/vault.py
@@ -52,9 +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 configs_path: Specifies the path of the secret to read Airflow Configurations
+    :param config_path: Specifies the path of the secret to read Airflow Configurations
         (default: 'configs').
-    :type configs_path: str
+    :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:
@@ -114,7 +114,7 @@ def __init__(  # pylint: disable=too-many-arguments
         self,
         connections_path: str = 'connections',
         variables_path: str = 'variables',
-        configs_path: str = 'configs',
+        config_path: str = 'config',
         url: Optional[str] = None,
         auth_type: str = 'token',
         auth_mount_point: Optional[str] = None,
@@ -142,7 +142,7 @@ def __init__(  # pylint: disable=too-many-arguments
         super().__init__()
         self.connections_path = connections_path.rstrip('/')
         self.variables_path = variables_path.rstrip('/')
-        self.configs_path = configs_path.rstrip('/')
+        self.config_path = config_path.rstrip('/')
         self.mount_point = mount_point
         self.kv_engine_version = kv_engine_version
         self.vault_client = _VaultClient(
@@ -206,6 +206,6 @@ def get_config(self, key: str) -> Optional[str]:
         :rtype: str
         :return: Configuration Option Value retrieved from the vault
         """
-        secret_path = self.build_path(self.configs_path, key)
+        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

From d8d5b8fed26e6f7e04411a43115b3dc4b7c008c8 Mon Sep 17 00:00:00 2001
From: Kaxil Naik <kaxilnaik@gmail.com>
Date: Wed, 8 Jul 2020 12:28:18 +0100
Subject: [PATCH 9/9] fixup! fixup! fixup! fixup! fixup! fixup! fixup! fixup!
 Get Airflow configs with sensitive data from Secret Backends

---
 airflow/providers/amazon/aws/secrets/secrets_manager.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/airflow/providers/amazon/aws/secrets/secrets_manager.py b/airflow/providers/amazon/aws/secrets/secrets_manager.py
index 6882da7141230..39dd8a70e3c2e 100644
--- a/airflow/providers/amazon/aws/secrets/secrets_manager.py
+++ b/airflow/providers/amazon/aws/secrets/secrets_manager.py
@@ -45,7 +45,7 @@ class SecretsManagerBackend(BaseSecretsBackend, LoggingMixin):
     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 variable
+    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``