Support multiple paths fallback in VaultBackend

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

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: