Add config option [secrets]backends_order

airflow/config_templates/config.yml

@@ -1296,6 +1296,20 @@ secrets:
       sensitive: true
       example: ~
       default: ""
+    backends_order:
+      description: |
+        Comma-separated list of secret backends. These backends will be used in the order they are specified.
+        Please note that the `environment_variable` and `metastore` are required values and cannot be removed
+        from the list. Supported values are:
+
+        * ``custom``: Custom secret backend specified in the ``secrets[backend]`` configuration option.
+        * ``environment_variable``: Standard environment variable backend
+        ``airflow.secrets.environment_variables.EnvironmentVariablesBackend``.
+        * ``metastore``: Standard metastore backend ``airflow.secrets.metastore.MetastoreBackend``.
+      version_added: 3.0.0
+      type: string
+      example: ~
+      default: "custom,environment_variable,metastore"
     use_cache:
       description: |
         .. note:: |experimental|

airflow/configuration.py

@@ -46,7 +46,6 @@
 from typing_extensions import overload
 
 from airflow.exceptions import AirflowConfigException
-from airflow.secrets import DEFAULT_SECRETS_SEARCH_PATH
 from airflow.utils import yaml
 from airflow.utils.module_loading import import_string
 from airflow.utils.providers_configuration_loader import providers_configuration_loaded
@@ -2089,23 +2088,58 @@ def get_custom_secret_backend() -> BaseSecretsBackend | None:
     return secrets_backend_cls(**backend_kwargs)
 
 
-def initialize_secrets_backends() -> list[BaseSecretsBackend]:
+def get_importable_secret_backend(class_name: str) -> BaseSecretsBackend:
+    """Get secret backend defined in the given class name."""
+    secrets_backend_cls = import_string(class_name)
+    return secrets_backend_cls()
+
+
+def initialize_secrets_backends() -> list[BaseSecretsBackend]:  # here
     """
     Initialize secrets backend.
 
     * import secrets backend classes
     * instantiate them and return them in a list
     """
-    backend_list = []
+    backends_map: dict[str, dict[str, Any]] = {
+        "environment_variable": {
+            "callback": get_importable_secret_backend,
+            "args": ("airflow.secrets.environment_variables.EnvironmentVariablesBackend",),
+        },
+        "metastore": {
+            "callback": get_importable_secret_backend,
+            "args": ("airflow.secrets.metastore.MetastoreBackend",),
+        },
+        "custom": {
+            "callback": get_custom_secret_backend,
+            "args": None,
+        },
+    }
 
-    custom_secret_backend = get_custom_secret_backend()
+    backends_order = conf.getlist("secrets", "backends_order", delimiter=",")
 
-    if custom_secret_backend is not None:
-        backend_list.append(custom_secret_backend)
+    required_backends = ["metastore", "environment_variable"]
+    if missing_backends := [b for b in required_backends if b not in backends_order]:
+        raise AirflowConfigException(
+            "The configuration option [secrets]backends_order is misconfigured. "
+            "The following backend types are missing: %s",
+            missing_backends,
+        )
+
+    if unsupported_backends := [b for b in backends_order if b not in backends_map.keys()]:
+        raise AirflowConfigException(
+            "The configuration option [secrets]backends_order is misconfigured. "
+            "The following backend types are unsupported: %s",
+            unsupported_backends,
+        )
 
-    for class_name in DEFAULT_SECRETS_SEARCH_PATH:
-        secrets_backend_cls = import_string(class_name)
-        backend_list.append(secrets_backend_cls())
+    backend_list = []
+    for backend_type in backends_order:
+        backend_item = backends_map[backend_type]
+        callback, args = backend_item["callback"], backend_item["args"]
+        backend = callback(*args) if args else callback()
+        if backend:
+            backend_list.append(backend)
 
     return backend_list
 

airflow/secrets/__init__.py

@@ -27,11 +27,6 @@
 
 from __future__ import annotations
 
-__all__ = ["BaseSecretsBackend", "DEFAULT_SECRETS_SEARCH_PATH"]
+__all__ = ["BaseSecretsBackend"]
 
 from airflow.secrets.base_secrets import BaseSecretsBackend
-
-DEFAULT_SECRETS_SEARCH_PATH = [
-    "airflow.secrets.environment_variables.EnvironmentVariablesBackend",
-    "airflow.secrets.metastore.MetastoreBackend",
-]

docs/apache-airflow/security/secrets/secrets-backend/index.rst

@@ -39,10 +39,12 @@ When looking up a connection/variable, by default Airflow will search environmen
 database second.
 
 If you enable an alternative secrets backend, it will be searched first, followed by environment variables,
-then metastore.  This search ordering is not configurable. Though, in some alternative secrets backend you might have
+then metastore. Though, in some alternative secrets backend you might have
 the option to filter which connection/variable/config is searched in the secret backend. Please look at the
 documentation of the secret backend you are using to see if such option is available.
 
+The secrets backends search ordering is also configurable via the configuration option ``[secrets]backends_order``.
+
 .. warning::
 
     When using environment variables or an alternative secrets backend to store secrets or variables, it is possible to create key collisions.

tests/always/test_secrets.py

@@ -22,6 +22,7 @@
 import pytest
 
 from airflow.configuration import ensure_secrets_loaded, initialize_secrets_backends
+from airflow.exceptions import AirflowConfigException
 from airflow.models import Connection, Variable
 from airflow.secrets.cache import SecretCache
 
@@ -117,6 +118,40 @@ def test_backend_fallback_to_env_var(self, mock_get_connection):
 
         assert conn.get_uri() == "mysql://airflow:airflow@host:5432/airflow"
 
+    @conf_vars(
+        {
+            (
+                "secrets",
+                "backend",
+            ): "airflow.providers.amazon.aws.secrets.systems_manager.SystemsManagerParameterStoreBackend",
+            ("secrets", "backend_kwargs"): '{"connections_prefix": "/airflow", "profile_name": null}',
+            ("secrets", "backends_order"): "custom,environment_variable,metastore",
+        }
+    )
+    def test_backends_order(self):
+        backends = ensure_secrets_loaded()
+        backend_classes = [backend.__class__.__name__ for backend in backends]
+        assert backend_classes == [
+            "SystemsManagerParameterStoreBackend",
+            "EnvironmentVariablesBackend",
+            "MetastoreBackend",
+        ]
+
+    @conf_vars({("secrets", "backends_order"): "custom,metastore"})
+    def test_backends_order_no_environment_variable_backend(self):
+        with pytest.raises(AirflowConfigException):
+            ensure_secrets_loaded()
+
+    @conf_vars({("secrets", "backends_order"): "environment_variable"})
+    def test_backends_order_no_metastore_backend(self):
+        with pytest.raises(AirflowConfigException):
+            ensure_secrets_loaded()
+
+    @conf_vars({("secrets", "backends_order"): "metastore,environment_variable,unsupported"})
+    def test_backends_order_unsupported(self):
+        with pytest.raises(AirflowConfigException):
+            ensure_secrets_loaded()
+
 
 @pytest.mark.db_test
 class TestVariableFromSecrets: