diff --git a/airflow/configuration.py b/airflow/configuration.py index 8e00011f11e20..798852a4c4c2a 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) @@ -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): diff --git a/airflow/providers/amazon/aws/secrets/secrets_manager.py b/airflow/providers/amazon/aws/secrets/secrets_manager.py index 6932d8e5f32b7..39dd8a70e3c2e 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 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,6 +67,7 @@ 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 @@ -69,6 +75,7 @@ def __init__( 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 diff --git a/airflow/providers/hashicorp/secrets/vault.py b/airflow/providers/hashicorp/secrets/vault.py index 482632311ca02..635971666ce57 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 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 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 100bd16666693..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 [] @@ -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 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/docs/howto/set-config.rst b/docs/howto/set-config.rst index 241720fe325f1..9fc12b66af773 100644 --- a/docs/howto/set-config.rst +++ b/docs/howto/set-config.rst @@ -46,7 +46,21 @@ 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 + # 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` 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,14 +79,23 @@ 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: -#. 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`` #. Airflow's built in defaults .. note:: diff --git a/tests/providers/hashicorp/secrets/test_vault.py b/tests/providers/hashicorp/secrets/test_vault.py index fbb3f5ad22cd4..5a4449b357352 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 = { + "configs_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_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 09ab57adc5b9a..3a374345031a4 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'), } @@ -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") + @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() + 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 +''' + 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.sensitive_config_values = test_conf.sensitive_config_values | { + ('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.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'}): @@ -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.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')