diff --git a/providers/hashicorp/docs/secrets-backends/hashicorp-vault.rst b/providers/hashicorp/docs/secrets-backends/hashicorp-vault.rst index faf4db70f015a..5ac1e7bdfe4bb 100644 --- a/providers/hashicorp/docs/secrets-backends/hashicorp-vault.rst +++ b/providers/hashicorp/docs/secrets-backends/hashicorp-vault.rst @@ -69,6 +69,24 @@ For example, if you want to set parameter ``connections_path`` to ``"airflow-con backend = airflow.providers.hashicorp.secrets.vault.VaultBackend backend_kwargs = {"connections_path": "airflow-connections", "variables_path": null, "mount_point": "airflow", "url": "http://127.0.0.1:8200"} +Falling back across multiple secret paths +""""""""""""""""""""""""""""""""""""""""" + +``connections_path``, ``variables_path``, and ``config_path`` can take a list of paths. Each path is tried in order, and +the first path where a secret is found returns. This can be used as a fallback mechanism for e.g. teams, environments, +etc.: + +.. code-block:: ini + + [secrets] + backend = airflow.providers.hashicorp.secrets.vault.VaultBackend + backend_kwargs = {"connections_path": ["airflow-prod/connections", "airflow-common/connections"], "mount_point": "airflow", "url": "http://127.0.0.1:8200"} + +.. note:: + Historically, the ``global_secrets_path`` parameter partially supported the above. The ``global_secrets_path`` + parameter was deprecated in favor of a more generic approach. It is appended to the list of paths until it is + completely removed. + Storing and Retrieving Connections using connection URI representation """""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""" diff --git a/providers/hashicorp/src/airflow/providers/hashicorp/secrets/vault.py b/providers/hashicorp/src/airflow/providers/hashicorp/secrets/vault.py index 8200c23c506ea..ca106e3c74ab5 100644 --- a/providers/hashicorp/src/airflow/providers/hashicorp/secrets/vault.py +++ b/providers/hashicorp/src/airflow/providers/hashicorp/secrets/vault.py @@ -20,7 +20,10 @@ from __future__ import annotations import json +import warnings +from collections.abc import Sequence +from airflow.exceptions import AirflowProviderDeprecationWarning from airflow.providers.common.compat.sdk import conf from airflow.providers.hashicorp._internal_client.vault_client import _VaultClient from airflow.secrets import BaseSecretsBackend @@ -47,16 +50,20 @@ class VaultBackend(BaseSecretsBackend, LoggingMixin): would be accessible if you provide ``{"connections_path": "connections"}`` and request conn_id ``smtp_default``. - :param connections_path: Specifies the path of the secret to read to get Connections. + :param connections_path: Specifies the path(s) of the secret to read to get Connections. Accepts a + single path or a list of paths, tried in order until a secret is found. (default: 'connections'). If set to None (null), requests for connections will not be sent to Vault. - :param variables_path: Specifies the path of the secret to read to get Variable. + :param variables_path: Specifies the path(s) of the secret to read to get Variable. Accepts a + single path or a list of paths, tried in order until a secret is found. (default: 'variables'). If set to None (null), requests for variables will not be sent to Vault. - :param config_path: Specifies the path of the secret to read Airflow Configurations + :param config_path: Specifies the path(s) of the secret to read Airflow Configurations. Accepts a + single path or a list of paths, tried in order until a secret is found. (default: 'config'). If set to None (null), requests for configurations will not be sent to Vault. :param use_team_secrets_path: Flag to enable team scoped secret retrieval from {base_path}/{team_name}/{key} in multi team deployments. (default: true) - :param global_secrets_path: Path prefix to add to global scoped connections and variables in multi team deployments. - (default: No prefix) + :param global_secrets_path: (Deprecated) Path appended as an extra fallback entry to + ``connections_path``, ``variables_path``, and ``config_path``. (default: No prefix) Use a list of + paths in those parameters instead, e.g. ``["connections/team1", "connections/common"]``. :param url: Base URL for the Vault instance being addressed. :param auth_type: Authentication Type for Vault. Default is ``token``. Available values are: ('approle', 'aws_iam', 'azure', 'github', 'gcp', 'jwt', 'kubernetes', 'ldap', 'radius', 'token', 'userpass') @@ -101,9 +108,9 @@ class VaultBackend(BaseSecretsBackend, LoggingMixin): def __init__( self, - connections_path: str | None = "connections", - variables_path: str | None = "variables", - config_path: str | None = "config", + connections_path: str | Sequence[str] | None = "connections", + variables_path: str | Sequence[str] | None = "variables", + config_path: str | Sequence[str] | None = "config", use_team_secrets_path: bool = True, global_secrets_path: str | None = None, url: str | None = None, @@ -136,13 +143,23 @@ def __init__( **kwargs, ): super().__init__() - self.connections_path = connections_path.rstrip("/") if connections_path is not None else None - self.variables_path = variables_path.rstrip("/") if variables_path is not None else None - self.config_path = config_path.rstrip("/") if config_path is not None else None + self.connections_path = self._normalize_paths(connections_path) + self.variables_path = self._normalize_paths(variables_path) + self.config_path = self._normalize_paths(config_path) self.use_team_secrets_path = use_team_secrets_path - self.global_secrets_path = ( - global_secrets_path.rstrip("/") if global_secrets_path is not None else None - ) + if global_secrets_path is not None: + warnings.warn( + "The `global_secrets_path` parameter is deprecated and will be removed in a future " + "release. Provide a list of fallback paths to `connections_path`, `variables_path`, " + 'and `config_path` instead, e.g. `connections_path=["connections/foobar", "connections/common"]`.', + AirflowProviderDeprecationWarning, + stacklevel=2, + ) + global_secrets_path = global_secrets_path.rstrip("/") + self.connections_path = self._append_path(self.connections_path, global_secrets_path) + self.variables_path = self._append_path(self.variables_path, global_secrets_path) + self.config_path = self._append_path(self.config_path, global_secrets_path) + self.mount_point = mount_point self.kv_engine_version = kv_engine_version self.vault_client = _VaultClient( @@ -176,6 +193,31 @@ def __init__( **kwargs, ) + @staticmethod + def _normalize_paths(paths: str | Sequence[str] | None) -> list[str] | None: + """ + Normalize a single path or a sequence of paths into a list, stripping trailing slashes. + + :param paths: Paths to normalize. + :return: Normalized paths. + """ + if paths is None: + return None + if isinstance(paths, str): + paths = [paths] + return [path.rstrip("/") for path in paths] + + @staticmethod + def _append_path(paths: list[str] | None, path: str) -> list[str]: + """ + Append a path into a list (which could be None). + + :param paths: List of paths to append to. + :param path: Path to append. + :return: Resulting list of paths. + """ + return [*(paths or []), path] + def _parse_path(self, secret_path: str) -> tuple[str | None, str | None]: if not self.mount_point: split_secret_path = secret_path.split("/", 1) @@ -200,29 +242,31 @@ def _get_secret_with_base(self, base_path: str | None, key: str) -> dict | None: secret_path=(mount_point + "/" if mount_point else "") + secret_path ) - def _get_team_or_global_secret(self, base_path: str | None, team_name: str | None, key: str): + def _get_secret(self, base_paths: list[str] | None, team_name: str | None, key: str): """ - Get a secret from a team specific path or the global path. + Get a secret, trying each of ``base_paths`` in order until one yields a value. + + If multi team is enabled and a team name is given, check {base_path}/{team_name}/{key}. + Otherwise, check {base_path}/{key}. - If multi team is enabled, check {base_path}/{team_name}/{key}, then fallback to {base_path}/{global_path}/{key} or {base_path}/{key}. + :param base_paths: Base paths to try, in order. + :param team_name: Team name associated to the task trying to access the secret (if any). + :param key: Secret key. """ - if base_path is None: + if not base_paths: return None - if ( - conf.getboolean("core", "multi_team", fallback=False) - and self.use_team_secrets_path - and team_name is not None - ): - response = self._get_secret_with_base(self.build_path(base_path, team_name), key) + multi_team_enabled = conf.getboolean("core", "multi_team", fallback=False) + for base_path in base_paths: + if multi_team_enabled and self.use_team_secrets_path and team_name is not None: + path_ = self.build_path(base_path, team_name) + else: + path_ = base_path + + response = self._get_secret_with_base(path_, key) if response is not None: return response - # Fallback to global secret - if conf.getboolean("core", "multi_team", fallback=False) and self.global_secrets_path is not None: - path = self.build_path(base_path, self.global_secrets_path) - else: - path = base_path - return self._get_secret_with_base(path, key) + return None def get_conn_value(self, conn_id: str, team_name: str | None = None) -> str | None: """ @@ -242,7 +286,7 @@ def get_conn_value(self, conn_id: str, team_name: str | None = None) -> str | No :param team_name: Team name associated to the task trying to access the connection (if any) :return: Serialized connection string or None """ - response = self._get_team_or_global_secret(self.connections_path, team_name, conn_id) + response = self._get_secret(self.connections_path, team_name, conn_id) if response is None: return None @@ -260,7 +304,7 @@ def get_variable(self, key: str, team_name: str | None = None) -> str | None: :param team_name: Team name associated to the task trying to access the variable (if any) :return: Variable Value retrieved from the vault """ - response = self._get_team_or_global_secret(self.variables_path, team_name, key) + response = self._get_secret(self.variables_path, team_name, key) if not response: return None @@ -274,10 +318,16 @@ def get_config(self, key: str) -> str | None: """ Get Airflow Configuration. + Tries each path in ``config_path`` in order and returns the value from the first one found. + :param key: Configuration Option Key :return: Configuration Option Value retrieved from the vault """ - response = self._get_secret_with_base(self.config_path, key) + response = None + for base_path in self.config_path or []: + response = self._get_secret_with_base(base_path, key) + if response is not None: + break if not response: return None try: diff --git a/providers/hashicorp/tests/unit/hashicorp/secrets/test_vault.py b/providers/hashicorp/tests/unit/hashicorp/secrets/test_vault.py index 53dc52a3f39e7..3b45df0e23e71 100644 --- a/providers/hashicorp/tests/unit/hashicorp/secrets/test_vault.py +++ b/providers/hashicorp/tests/unit/hashicorp/secrets/test_vault.py @@ -22,6 +22,7 @@ import pytest from hvac.exceptions import InvalidPath, VaultError +from airflow.exceptions import AirflowProviderDeprecationWarning from airflow.providers.hashicorp.secrets.vault import VaultBackend from tests_common.test_utils.config import conf_vars @@ -143,20 +144,6 @@ def test_get_conn_value(self, mock_hvac): "foo", id="team_conn_no_separation", ), - pytest.param( - ["secret_not_found", "connection_result"], - {"global_secrets_path": "global"}, - ["/foo/test_postgres", "/global/test_postgres"], - "foo", - id="fallback_global_conn", - ), - pytest.param( - ["connection_result"], - {"global_secrets_path": "global"}, - ["/global/test_postgres"], - None, - id="global_conn_no_team", - ), ], ) @conf_vars({("core", "multi_team"): "True"}) @@ -189,7 +176,7 @@ def test_get_connection_value_multi_team( mock_client.secrets.kv.v2.read_secret_version.assert_has_calls( [ mock.call( - path=test_client.connections_path + path, + path=test_client.connections_path[0] + path, mount_point="airflow", version=None, raise_on_deleted_version=True, @@ -200,6 +187,54 @@ def test_get_connection_value_multi_team( assert value is not None assert json.loads(value)["conn_type"] == "postgresql" + @mock.patch("airflow.providers.hashicorp._internal_client.vault_client.hvac") + def test_get_connection_value_multipath(self, mock_hvac, secret_not_found, connection_result): + """Test fetching a connection when two paths are supplied and a secret is found in the 2nd path.""" + mock_client = mock.MagicMock() + mock_hvac.Client.return_value = mock_client + mock_client.secrets.kv.v2.read_secret_version.side_effect = [secret_not_found, connection_result] + + kwargs = { + "connections_path": ["connections/team1", "connections/common"], + "mount_point": "airflow", + "auth_type": "token", + "url": "http://127.0.0.1:8200", + "token": "s.7AU0I51yv1Q1lxOIg1F3ZRAS", + } + + test_client = VaultBackend(**kwargs) + connection = test_client.get_connection(conn_id="test_postgres") + mock_client.secrets.kv.v2.read_secret_version.assert_has_calls( + [ + mock.call( + path=path, + mount_point="airflow", + version=None, + raise_on_deleted_version=True, + ) + for path in ("connections/team1/test_postgres", "connections/common/test_postgres") + ] + ) + assert connection.get_uri() == "postgres://airflow:airflow@host:5432/airflow?foo=bar&baz=taz" + + @mock.patch("airflow.providers.hashicorp._internal_client.vault_client.hvac") + def test_get_connection_value_with_list_of_paths_not_found(self, mock_hvac, secret_not_found): + """Test fetching a connection when two paths are supplied and a secret is not found in either path.""" + mock_client = mock.MagicMock() + mock_hvac.Client.return_value = mock_client + mock_client.secrets.kv.v2.read_secret_version.side_effect = [secret_not_found, secret_not_found] + + kwargs = { + "connections_path": ["connections/team1", "connections/common"], + "mount_point": "airflow", + "auth_type": "token", + "url": "http://127.0.0.1:8200", + "token": "s.7AU0I51yv1Q1lxOIg1F3ZRAS", + } + + test_client = VaultBackend(**kwargs) + assert test_client.get_connection(conn_id="test_postgres") is None + @mock.patch("airflow.providers.hashicorp._internal_client.vault_client.hvac") def test_get_connection_without_predefined_mount_point(self, mock_hvac): mock_client = mock.MagicMock() @@ -301,20 +336,6 @@ def test_get_variable_value(self, mock_hvac): "foo", id="team_var_no_separation", ), - pytest.param( - ["secret_not_found", "variable_result"], - {"global_secrets_path": "global"}, - ["/foo/hello", "/global/hello"], - "foo", - id="fallback_global_var", - ), - pytest.param( - ["variable_result"], - {"global_secrets_path": "global"}, - ["/global/hello"], - None, - id="global_var_no_team", - ), ], ) @conf_vars({("core", "multi_team"): "True"}) @@ -347,7 +368,7 @@ def test_get_variable_value_multi_team( mock_client.secrets.kv.v2.read_secret_version.assert_has_calls( [ mock.call( - path=test_client.variables_path + path, + path=test_client.variables_path[0] + path, mount_point="airflow", version=None, raise_on_deleted_version=True, @@ -357,6 +378,36 @@ def test_get_variable_value_multi_team( ) assert returned_uri == "world" + @mock.patch("airflow.providers.hashicorp._internal_client.vault_client.hvac") + def test_get_variable_value_multipath(self, mock_hvac, secret_not_found, variable_result): + """Verify fetching a variable and falling back with multiple paths configured.""" + mock_client = mock.MagicMock() + mock_hvac.Client.return_value = mock_client + mock_client.secrets.kv.v2.read_secret_version.side_effect = [secret_not_found, variable_result] + + kwargs = { + "variables_path": ["variables/team1", "variables/common"], + "mount_point": "airflow", + "auth_type": "token", + "url": "http://127.0.0.1:8200", + "token": "s.7AU0I51yv1Q1lxOIg1F3ZRAS", + } + + test_client = VaultBackend(**kwargs) + returned_uri = test_client.get_variable("hello") + mock_client.secrets.kv.v2.read_secret_version.assert_has_calls( + [ + mock.call( + path=path, + mount_point="airflow", + version=None, + raise_on_deleted_version=True, + ) + for path in ("variables/team1/hello", "variables/common/hello") + ] + ) + assert returned_uri == "world" + @mock.patch("airflow.providers.hashicorp._internal_client.vault_client.hvac") def test_get_variable_value_without_predefined_mount_point(self, mock_hvac): mock_client = mock.MagicMock() @@ -694,6 +745,42 @@ def test_get_config_value(self, mock_hvac): returned_uri = test_client.get_config("sql_alchemy_conn") assert returned_uri == "sqlite:////Users/airflow/airflow/airflow.db" + @mock.patch("airflow.providers.hashicorp._internal_client.vault_client.hvac") + def test_get_config_value_multipath(self, mock_hvac, secret_not_found): + """Verify fetching a config value and falling back with multiple paths configured.""" + mock_client = mock.MagicMock() + mock_hvac.Client.return_value = mock_client + found_config = { + "data": { + "data": {"value": "sqlite:////Users/airflow/airflow/airflow.db"}, + "metadata": {"version": 1}, + }, + } + mock_client.secrets.kv.v2.read_secret_version.side_effect = [secret_not_found, found_config] + + kwargs = { + "config_path": ["config/team1", "config/common"], + "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") + mock_client.secrets.kv.v2.read_secret_version.assert_has_calls( + [ + mock.call( + path=path, + mount_point="secret", + version=None, + raise_on_deleted_version=True, + ) + for path in ("config/team1/sql_alchemy_conn", "config/common/sql_alchemy_conn") + ] + ) + assert returned_uri == "sqlite:////Users/airflow/airflow/airflow.db" + @mock.patch("airflow.providers.hashicorp._internal_client.vault_client.hvac") def test_get_config_value_without_predefined_mount_point(self, mock_hvac): mock_client = mock.MagicMock() @@ -824,3 +911,38 @@ def test_config_path_none_value(self, mock_hvac): test_client = VaultBackend(**kwargs) assert test_client.get_config("test") is None mock_hvac.Client.assert_not_called() + + @mock.patch("airflow.providers.hashicorp._internal_client.vault_client.hvac") + def test_global_secrets_path(self, mock_hvac, secret_not_found, connection_result): + """ + Verify if the (deprecated) global_secrets_path is appended to connections path. + """ + mock_client = mock.MagicMock() + mock_hvac.Client.return_value = mock_client + mock_client.secrets.kv.v2.read_secret_version.side_effect = [secret_not_found, connection_result] + + kwargs = { + "connections_path": "connections", + "global_secrets_path": "connections/common", + "mount_point": "airflow", + "auth_type": "token", + "url": "http://127.0.0.1:8200", + "token": "s.7AU0I51yv1Q1lxOIg1F3ZRAS", + } + + # Ensure global_secrets_path raises a deprecation warning + with pytest.warns(AirflowProviderDeprecationWarning): + test_client = VaultBackend(**kwargs) + connection = test_client.get_connection(conn_id="test_postgres") + mock_client.secrets.kv.v2.read_secret_version.assert_has_calls( + [ + mock.call( + path=path, + mount_point="airflow", + version=None, + raise_on_deleted_version=True, + ) + for path in ("connections/test_postgres", "connections/common/test_postgres") + ] + ) + assert connection.get_uri() == "postgres://airflow:airflow@host:5432/airflow?foo=bar&baz=taz"