Add advanced secrets backend configurations

airflow/config_templates/config.yml

@@ -782,6 +782,17 @@
       type: string
       example: ~
       default: ""
+    - name: backends_config
+      description: |
+        Advanced secrets backend configuration, allow to user configure more than one secret backend,
+        order of secrets backend and turn off built-in backends.
+        Expected JSON list of objects. See
+        https://airflow.apache.org/docs/apache-airflow/stable/security/secrets/secrets-backend/index.html#advanced-configuration
+        for more details.
+      version_added: 2.4.0
+      example: ~
+      type: string
+      default: ""
 - name: cli
   description: ~
   options:

airflow/config_templates/default_airflow.cfg

@@ -422,6 +422,13 @@ backend =
 # ``{{"connections_prefix": "/airflow/connections", "profile_name": "default"}}``
 backend_kwargs =
 
+# Advanced secrets backend configuration, allow to user configure more than one secret backend,
+# order of secrets backend and turn off built-in backends.
+# Expected JSON list of objects. See
+# https://airflow.apache.org/docs/apache-airflow/stable/security/secrets/secrets-backend/index.html#advanced-configuration
+# for more details.
+backends_config =
+
 [cli]
 # In what way should the cli access the API. The LocalClient will use the
 # database directly, while the json_client will use the api running on the

airflow/configuration.py

@@ -27,20 +27,21 @@
 import sys
 import warnings
 from base64 import b64encode
-from collections import OrderedDict
+from collections import OrderedDict, UserList
 
 # Ignored Mypy on configparser because it thinks the configparser module has no _UNSET attribute
 from configparser import _UNSET, ConfigParser, NoOptionError, NoSectionError  # type: ignore
 from contextlib import suppress
+from dataclasses import dataclass, field
 from json.decoder import JSONDecodeError
 from re import Pattern
-from typing import IO, Any, Dict, Iterable, List, Optional, Set, Tuple, Union
+from typing import IO, Any, Dict, Iterable, List, Optional, Sequence, Set, Tuple, Union
 from urllib.parse import urlparse
 
 from typing_extensions import overload
 
 from airflow.exceptions import AirflowConfigException
-from airflow.secrets import DEFAULT_SECRETS_SEARCH_PATH, BaseSecretsBackend
+from airflow.secrets import BaseSecretsBackend
 from airflow.utils import yaml
 from airflow.utils.module_loading import import_string
 from airflow.utils.weight_rule import WeightRule
@@ -62,6 +63,16 @@
 ENV_VAR_PREFIX = 'AIRFLOW__'
 
 
+DEFAULT_SECRETS_SEARCH_PATH: List[Dict[str, Any]] = [
+    {
+        "backend": "airflow.secrets.environment_variables.EnvironmentVariablesBackend",
+    },
+    {
+        "backend": "airflow.secrets.metastore.MetastoreBackend",
+    },
+]
+
+
 def _parse_sqlite_version(s: str) -> Tuple[int, ...]:
     match = _SQLITE3_VERSION_PATTERN.match(s)
     if match is None:
@@ -113,18 +124,20 @@ def run_command(command: str) -> str:
 
 def _get_config_value_from_secret_backend(config_key: str) -> Optional[str]:
     """Get Config option values from Secret Backend"""
-    try:
-        secrets_client = get_custom_secret_backend()
-        if not secrets_client:
-            return None
-        return secrets_client.get_config(config_key)
-    except Exception as e:
-        raise AirflowConfigException(
-            'Cannot retrieve config from alternative secrets backend. '
-            'Make sure it is configured properly and that the Backend '
-            'is accessible.\n'
-            f'{e}'
-        )
+    config_value = None
+    for secrets_client in ensure_secrets_loaded():
+        try:
+            config_value = secrets_client.get_config(config_key)
+            if config_value:
+                break
+        except Exception as e:
+            raise AirflowConfigException(
+                f'Cannot retrieve config from secrets backend `{secrets_client.__class__.__name__}`. '
+                f'Make sure it is configured properly and that the Backend is accessible.\n'
+                f'{e}'
+            )
+
+    return config_value
 
 
 def _default_config_file_path(file_name: str) -> str:
@@ -1487,47 +1500,118 @@ def set(*args, **kwargs) -> None:
     conf.set(*args, **kwargs)
 
 
-def ensure_secrets_loaded() -> List[BaseSecretsBackend]:
-    """
-    Ensure that all secrets backends are loaded.
-    If the secrets_backend_list contains only 2 default backends, reload it.
-    """
-    # Check if the secrets_backend_list contains only 2 default backends
-    if len(secrets_backend_list) == 2:
-        return initialize_secrets_backends()
-    return secrets_backend_list
+class DefaultSecretsBackend(UserList):
+    """List Container which use for store default secrets backends."""
+
+
+@dataclass(frozen=True)
+class SecretsBackendConfig:
+    """Secrets Backend Config dataclass helper."""
+
+    backend: str
+    backend_kwargs: Dict[str, Any] = field(default_factory=dict)
+
+    @classmethod
+    def from_dict(cls, d: Dict[str, Any]) -> 'SecretsBackendConfig':
+        """
+        Read Secret Backend Config from dictionary
+
+        Ignores all unexpected keywords
+        """
+        return cls(d['backend'], d.get('backend_kwargs', {}))
 
+    @classmethod
+    def from_config(cls) -> Optional['SecretsBackendConfig']:
+        """Try to get ``SecretsBackendConfig`` from airflow config [secrets] section"""
+        secrets_backend = conf.get(section='secrets', key='backend', fallback=None)
 
-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 not secrets_backend:
+            return None
 
-    if secrets_backend_cls:
         try:
-            backends: Any = conf.get(section='secrets', key='backend_kwargs', fallback='{}')
-            alternative_secrets_config_dict = json.loads(backends)
+            secrets_backend_kwargs: Any = conf.get(section='secrets', key='backend_kwargs', fallback='{}')
+            secrets_backend_kwargs = json.loads(secrets_backend_kwargs)
         except JSONDecodeError:
-            alternative_secrets_config_dict = {}
+            secrets_backend_kwargs = {}
+
+        return cls(secrets_backend, secrets_backend_kwargs)
+
+    def initialize(self) -> BaseSecretsBackend:
+        """Initialize Secrets Backend."""
+        return import_string(self.backend)(**self.backend_kwargs)
+
+
+class UniqueSecretsBackendsConfigs(UserList):
+    """List Container which use for store unique secrets backends configs."""
 
-        return secrets_backend_cls(**alternative_secrets_config_dict)
-    return None
+    def append(self, config) -> None:
+        """Append item to a list if it not exists yet"""
+        if config in self.data:
+            log.warning("%r already exists.", config)
+            return
+        self.data.append(config)
+
+    def extend(self, configs) -> None:
+        """Extends item if it not exists yet"""
+        for config in configs:
+            self.append(config)
+
+
+def ensure_secrets_loaded() -> Sequence[BaseSecretsBackend]:
+    """Ensure that all secrets backends are loaded."""
+    if isinstance(secrets_backend_list, DefaultSecretsBackend):
+        return initialize_secrets_backends()
+    return secrets_backend_list
 
 
-def initialize_secrets_backends() -> List[BaseSecretsBackend]:
+def initialize_secrets_backends() -> Sequence[BaseSecretsBackend]:
     """
+    * read secrets backend configurations
+    * keep only unique secret backend configurations
     * import secrets backend classes
     * instantiate them and return them in a list
     """
-    backend_list = []
+    backend_configs = UniqueSecretsBackendsConfigs()
+    default_configs = [SecretsBackendConfig.from_dict(config) for config in DEFAULT_SECRETS_SEARCH_PATH]
+
+    secrets_backend_config = conf.getjson(section='secrets', key='backends_config', fallback=None)
+    if secrets_backend_config:
+        # If [secrets] 'backend_config' exists and defined than use this configuration
+        # instead of [secrets] 'backend' and [secrets] 'backend_kwargs'.
+        if not isinstance(secrets_backend_config, list):
+            raise AirflowConfigException(
+                f"[secrets] 'backends_config' expected list of backends configurations but got"
+                f" {secrets_backend_config!r}."
+            )
+
+        for config in secrets_backend_config:
+            try:
+                backend_configs.append(SecretsBackendConfig.from_dict(config))
+            except Exception as e:
+                raise AirflowConfigException(
+                    f"Cannot read config: {config!r} from [secrets] 'backends_config'.\n{e}"
+                ) from e
 
-    custom_secret_backend = get_custom_secret_backend()
+    else:
+        custom_secrets_backend_config = SecretsBackendConfig.from_config()
+        if not custom_secrets_backend_config:
+            # Returns default secrets backend list for further checks in `ensure_secrets_loaded()`.
+            return DefaultSecretsBackend(config.initialize() for config in default_configs)
 
-    if custom_secret_backend is not None:
-        backend_list.append(custom_secret_backend)
+        backend_configs.append(custom_secrets_backend_config)
+        backend_configs.extend(default_configs)
 
-    for class_name in DEFAULT_SECRETS_SEARCH_PATH:
-        secrets_backend_cls = import_string(class_name)
-        backend_list.append(secrets_backend_cls())
+    # Initialize secrets backends
+    backend_list = []
+    for config in backend_configs:
+        try:
+            backend_list.append(config.initialize())
+        except Exception as e:
+            raise AirflowConfigException(
+                f"Cannot initialize secrets backend {config.backend!r} "
+                f"with keyword arguments {config.backend_kwargs}.\n"
+                f"{e}"
+            ) from e
 
     return backend_list
 

airflow/secrets/__init__.py

@@ -22,11 +22,6 @@
     * Metastore database
     * AWS SSM Parameter store
 """
-__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

@@ -38,8 +38,8 @@ Search path
 When looking up a connection/variable, by default Airflow will search environment variables first and metastore
 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.
+If you enable an alternative secrets backend, it will be searched first, followed by environment
+variables, then metastore. This search ordering is not configurable in Airflow <2.4.0.
 
 .. warning::
 
@@ -59,6 +59,7 @@ The ``[secrets]`` section has the following options:
     [secrets]
     backend =
     backend_kwargs =
+    backends_config =
 
 Set ``backend`` to the fully qualified class name of the backend you want to enable.
 
@@ -73,6 +74,47 @@ the example below.
     $ airflow config get-value secrets backend
     airflow.providers.google.cloud.secrets.secret_manager.CloudSecretManagerBackend
 
+Advanced Configuration
+""""""""""""""""""""""
+
+.. versionadded:: 2.4.0
+
+You can provide ``backends_config`` with json-list which contains json-objects of required backends for:
+
+* Configure more than one alternative secrets backend.
+* Change search ordering of secrets backends.
+* Turn off built-in backends (environment variables or the metastore database).
+
+Object should contains mandatory **backend** attribute with fully qualified class name of the backend you want to enable
+and optionally contains **backend_kwargs** with json-object which will be passed as kwargs to the ``__init__``
+method of your secrets backend.
+
+Here's a basic example of how to configure to use Environment Variables first,
+next :ref:`SSM Parameter Store <ssm_parameter_store_secrets>`, and then metastore.:
+
+.. code-block:: json
+
+    [
+      {
+        "backend": "airflow.secrets.environment_variables.EnvironmentVariablesBackend"
+      },
+      {
+        "backend": "airflow.providers.amazon.aws.secrets.systems_manager.SystemsManagerParameterStoreBackend",
+        "backend_kwargs": {
+          "connections_prefix": "/airflow/connections",
+          "profile_name": "default"
+        }
+      },
+      {
+        "backend": "airflow.secrets.metastore.MetastoreBackend"
+      }
+    ]
+
+.. warning::
+
+    If you specify ``backends_config`` options than Airflow will ignore values
+    in ``backend`` and ``backend_kwargs`` options.
+
 Supported core backends
 ^^^^^^^^^^^^^^^^^^^^^^^
 
@@ -101,8 +143,8 @@ Roll your own secrets backend
 A secrets backend is a subclass of :py:class:`airflow.secrets.BaseSecretsBackend` and must implement either
 :py:meth:`~airflow.secrets.BaseSecretsBackend.get_connection` or :py:meth:`~airflow.secrets.BaseSecretsBackend.get_conn_value`.
 
-After writing your backend class, provide the fully qualified class name in the ``backend`` key in the ``[secrets]``
-section of ``airflow.cfg``.
+After writing your backend class, provide the fully qualified class name in the ``backend`` key or provide
+configurations in the ``backends_config`` in the ``[secrets]`` section of ``airflow.cfg``.
 
 Additional arguments to your SecretsBackend can be configured in ``airflow.cfg`` by supplying a JSON string to ``backend_kwargs``, which will be passed to the ``__init__`` of your SecretsBackend.
 See :ref:`Configuration <secrets_backend_configuration>` for more details, and :ref:`SSM Parameter Store <ssm_parameter_store_secrets>` for an example.