Add config option [secrets]backends_order

Author: moiseenkov

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

@@ -39,13 +39,15 @@ 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.
 
 On the other hand, if a workers secrets backend is defined, the order of lookup has higher priority for the workers secrets
 backend and then the secrets backend.
 
+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.
@@ -64,12 +66,21 @@ The ``[secrets]`` section has the following options:
     [secrets]
     backend =
     backend_kwargs =
+    backends_order =
 
 Set ``backend`` to the fully qualified class name of the backend you want to enable.
 
 You can provide ``backend_kwargs`` with json and it will be passed as kwargs to the ``__init__`` method of
 your secrets backend.
 
+``backends_order`` 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``.
+
 If you want to check which secret backend is currently set, you can use ``airflow config get-value secrets backend`` command as in
 the example below.
 

airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml

@@ -77,6 +77,52 @@ paths:
       security:
       - OAuth2PasswordBearer: []
       - HTTPBearer: []
+  /ui/backends_order:
+    get:
+      tags:
+      - Config
+      summary: Get Backends Order Value
+      operationId: get_backends_order_value
+      security:
+      - OAuth2PasswordBearer: []
+      - HTTPBearer: []
+      parameters:
+      - name: accept
+        in: header
+        required: false
+        schema:
+          type: string
+          enum:
+          - application/json
+          - text/plain
+          - '*/*'
+          default: '*/*'
+          title: Accept
+      responses:
+        '200':
+          description: Successful Response
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/Config'
+        '404':
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+          description: Not Found
+        '406':
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+          description: Not Acceptable
+        '422':
+          description: Validation Error
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPValidationError'
   /ui/connections/hook_meta:
     get:
       tags:
@@ -1206,6 +1252,41 @@ components:
       - count
       title: CalendarTimeRangeResponse
       description: Represents a summary of DAG runs for a specific calendar time range.
+    Config:
+      properties:
+        sections:
+          items:
+            $ref: '#/components/schemas/ConfigSection'
+          type: array
+          title: Sections
+      additionalProperties: false
+      type: object
+      required:
+      - sections
+      title: Config
+      description: List of config sections with their options.
+    ConfigOption:
+      properties:
+        key:
+          type: string
+          title: Key
+        value:
+          anyOf:
+          - type: string
+          - prefixItems:
+            - type: string
+            - type: string
+            type: array
+            maxItems: 2
+            minItems: 2
+          title: Value
+      additionalProperties: false
+      type: object
+      required:
+      - key
+      - value
+      title: ConfigOption
+      description: Config option.
     ConfigResponse:
       properties:
         page_size:
@@ -1259,6 +1340,23 @@ components:
       - show_external_log_redirect
       title: ConfigResponse
       description: configuration serializer.
+    ConfigSection:
+      properties:
+        name:
+          type: string
+          title: Name
+        options:
+          items:
+            $ref: '#/components/schemas/ConfigOption'
+          type: array
+          title: Options
+      additionalProperties: false
+      type: object
+      required:
+      - name
+      - options
+      title: ConfigSection
+      description: Config Section Schema.
     ConnectionHookFieldBehavior:
       properties:
         hidden:

airflow-core/src/airflow/api_fastapi/core_api/routes/ui/config.py

@@ -18,13 +18,20 @@
 
 from typing import Any
 
-from fastapi import Depends, status
+from fastapi import Depends, HTTPException, status
 
+from airflow.api_fastapi.common.headers import HeaderAcceptJsonOrText
 from airflow.api_fastapi.common.router import AirflowRouter
 from airflow.api_fastapi.common.types import UIAlert
+from airflow.api_fastapi.core_api.datamodels.config import (
+    Config,
+    ConfigOption,
+    ConfigSection,
+)
 from airflow.api_fastapi.core_api.datamodels.ui.config import ConfigResponse
 from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
 from airflow.api_fastapi.core_api.security import requires_authenticated
+from airflow.api_fastapi.core_api.services.public.config import _response_based_on_accept
 from airflow.configuration import conf
 from airflow.settings import DASHBOARD_UIALERTS
 from airflow.utils.log.log_reader import TaskLogReader
@@ -64,3 +71,32 @@ def get_configs() -> ConfigResponse:
     config.update({key: value for key, value in additional_config.items()})
 
     return ConfigResponse.model_validate(config)
+
+
+@config_router.get(
+    "/backends_order",
+    responses={
+        **create_openapi_http_exception_doc(
+            [
+                status.HTTP_404_NOT_FOUND,
+                status.HTTP_406_NOT_ACCEPTABLE,
+            ]
+        ),
+    },
+    response_model=Config,
+    dependencies=[Depends(requires_authenticated())],
+)
+def get_backends_order_value(
+    accept: HeaderAcceptJsonOrText,
+):
+    section, option = "secrets", "backends_order"
+    if not conf.has_option(section, option):
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Option [{section}/{option}] not found.",
+        )
+
+    value = conf.get(section, option)
+
+    config = Config(sections=[ConfigSection(name=section, options=[ConfigOption(key=option, value=value)])])
+    return _response_based_on_accept(accept, config)

airflow-core/src/airflow/config_templates/config.yml

@@ -1323,6 +1323,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-core/src/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.task.weight_rule import WeightRule
 from airflow.utils import yaml
 from airflow.utils.module_loading import import_string
@@ -2154,7 +2153,7 @@ def make_group_other_inaccessible(file_path: str):
 
 
 def ensure_secrets_loaded(
-    default_backends: list[str] = DEFAULT_SECRETS_SEARCH_PATH,
+    default_backends: list[str] | None = None,
 ) -> list[BaseSecretsBackend]:
     """
     Ensure that all secrets backends are loaded.
@@ -2163,9 +2162,8 @@ def ensure_secrets_loaded(
     """
     # Check if the secrets_backend_list contains only 2 default backends.
 
-    # Check if we are loading the backends for worker too by checking if the default_backends is equal
-    # to DEFAULT_SECRETS_SEARCH_PATH.
-    if len(secrets_backend_list) == 2 or default_backends != DEFAULT_SECRETS_SEARCH_PATH:
+    # Check if we are loading the backends for worker too by checking if the default_backends is not None
+    if len(secrets_backend_list) == 2 or default_backends is not None:
         return initialize_secrets_backends(default_backends=default_backends)
     return secrets_backend_list
 
@@ -2210,28 +2208,73 @@ def get_custom_secret_backend(worker_mode: bool = False) -> BaseSecretsBackend |
     return secrets_backend_cls(**backend_kwargs)
 
 
+def get_importable_secret_backend(class_name: str | None) -> BaseSecretsBackend | None:
+    """Get secret backend defined in the given class name."""
+    if class_name is not None:
+        secrets_backend_cls = import_string(class_name)
+        return secrets_backend_cls()
+    return None
+
+
 def initialize_secrets_backends(
-    default_backends: list[str] = DEFAULT_SECRETS_SEARCH_PATH,
+    default_backends: list[str] | None = None,
 ) -> list[BaseSecretsBackend]:
     """
     Initialize secrets backend.
 
     * import secrets backend classes
     * instantiate them and return them in a list
     """
-    backend_list = []
     worker_mode = False
-    if default_backends != DEFAULT_SECRETS_SEARCH_PATH:
+    environment_variable_args: str | None = (
+        "airflow.secrets.environment_variables.EnvironmentVariablesBackend"
+    )
+    metastore_args: str | None = "airflow.secrets.metastore.MetastoreBackend"
+    if default_backends is not None:
         worker_mode = True
+        environment_variable_args = (
+            environment_variable_args if environment_variable_args in default_backends else None
+        )
+        metastore_args = metastore_args if metastore_args in default_backends else None
+    backends_map: dict[str, dict[str, Any]] = {
+        "environment_variable": {
+            "callback": get_importable_secret_backend,
+            "args": (environment_variable_args,),
+        },
+        "metastore": {
+            "callback": get_importable_secret_backend,
+            "args": (metastore_args,),
+        },
+        "custom": {
+            "callback": get_custom_secret_backend,
+            "args": (worker_mode,),
+        },
+    }
 
-    custom_secret_backend = get_custom_secret_backend(worker_mode)
+    backends_order = conf.getlist("secrets", "backends_order", delimiter=",")
 
-    if custom_secret_backend is not None:
-        backend_list.append(custom_secret_backend)
+    required_backends = ["environment_variable"] if worker_mode else ["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,
+        )
 
-    for class_name in default_backends:
-        secrets_backend_cls = import_string(class_name)
-        backend_list.append(secrets_backend_cls())
+    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,
+        )
+
+    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-core/src/airflow/secrets/__init__.py

@@ -29,15 +29,11 @@
 
 from airflow.utils.deprecation_tools import add_deprecated_classes
 
-__all__ = ["BaseSecretsBackend", "DEFAULT_SECRETS_SEARCH_PATH"]
-
-from airflow.secrets.base_secrets import BaseSecretsBackend
-
-DEFAULT_SECRETS_SEARCH_PATH = [
-    "airflow.secrets.environment_variables.EnvironmentVariablesBackend",
-    "airflow.secrets.metastore.MetastoreBackend",
+__all__ = [
+    "BaseSecretsBackend",
 ]
 
+from airflow.secrets.base_secrets import BaseSecretsBackend
 
 __deprecated_classes = {
     "cache": {

airflow-core/src/airflow/ui/openapi-gen/queries/common.ts

@@ -231,6 +231,12 @@ export type ConfigServiceGetConfigsDefaultResponse = Awaited<ReturnType<typeof C
 export type ConfigServiceGetConfigsQueryResult<TData = ConfigServiceGetConfigsDefaultResponse, TError = unknown> = UseQueryResult<TData, TError>;
 export const useConfigServiceGetConfigsKey = "ConfigServiceGetConfigs";
 export const UseConfigServiceGetConfigsKeyFn = (queryKey?: Array<unknown>) => [useConfigServiceGetConfigsKey, ...(queryKey ?? [])];
+export type ConfigServiceGetBackendsOrderValueDefaultResponse = Awaited<ReturnType<typeof ConfigService.getBackendsOrderValue>>;
+export type ConfigServiceGetBackendsOrderValueQueryResult<TData = ConfigServiceGetBackendsOrderValueDefaultResponse, TError = unknown> = UseQueryResult<TData, TError>;
+export const useConfigServiceGetBackendsOrderValueKey = "ConfigServiceGetBackendsOrderValue";
+export const UseConfigServiceGetBackendsOrderValueKeyFn = ({ accept }: {
+  accept?: "application/json" | "text/plain" | "*/*";
+} = {}, queryKey?: Array<unknown>) => [useConfigServiceGetBackendsOrderValueKey, ...(queryKey ?? [{ accept }])];
 export type DagWarningServiceListDagWarningsDefaultResponse = Awaited<ReturnType<typeof DagWarningService.listDagWarnings>>;
 export type DagWarningServiceListDagWarningsQueryResult<TData = DagWarningServiceListDagWarningsDefaultResponse, TError = unknown> = UseQueryResult<TData, TError>;
 export const useDagWarningServiceListDagWarningsKey = "DagWarningServiceListDagWarnings";

airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts

@@ -431,6 +431,16 @@ export const ensureUseConfigServiceGetConfigValueData = (queryClient: QueryClien
 */
 export const ensureUseConfigServiceGetConfigsData = (queryClient: QueryClient) => queryClient.ensureQueryData({ queryKey: Common.UseConfigServiceGetConfigsKeyFn(), queryFn: () => ConfigService.getConfigs() });
 /**
+* Get Backends Order Value
+* @param data The data for the request.
+* @param data.accept
+* @returns Config Successful Response
+* @throws ApiError
+*/
+export const ensureUseConfigServiceGetBackendsOrderValueData = (queryClient: QueryClient, { accept }: {
+  accept?: "application/json" | "text/plain" | "*/*";
+} = {}) => queryClient.ensureQueryData({ queryKey: Common.UseConfigServiceGetBackendsOrderValueKeyFn({ accept }), queryFn: () => ConfigService.getBackendsOrderValue({ accept }) });
+/**
 * List Dag Warnings
 * Get a list of DAG warnings.
 * @param data The data for the request.