Do not allow Metastore as secret backend in Supervisor

Author: kaxil

.pre-commit-config.yaml

@@ -1613,6 +1613,7 @@ repos:
           ^airflow-core/src/airflow/operators/subdag\.py$|
           ^airflow-core/src/airflow/plugins_manager\.py$|
           ^airflow-core/src/airflow/providers_manager\.py$|
+          ^airflow-core/src/airflow/secrets/__init__.py$|
           ^airflow-core/src/airflow/serialization/definitions/[_a-z]+\.py$|
           ^airflow-core/src/airflow/serialization/enums\.py$|
           ^airflow-core/src/airflow/serialization/helpers\.py$|

airflow-core/newsfragments/56583.significant.rst

@@ -1,33 +1,41 @@
-Fix connection access in API server contexts (plugins, log handlers)
+Fix Connection & Variable access in API server contexts (plugins, log handlers)
 
 Previously, hooks used in API server contexts (plugins, middlewares, log handlers) would fail with an ``ImportError``
-for ``SUPERVISOR_COMMS``, because ``SUPERVISOR_COMMS`` only exists in worker execution contexts.
+for ``SUPERVISOR_COMMS``, because ``SUPERVISOR_COMMS`` only exists in task runner child processes.
 
-This has been fixed by implementing automatic context detection with separate secrets backend chains:
+This has been fixed by implementing automatic context detection with three separate secrets backend chains:
 
 **Context Detection:**
 
-- **Client contexts** (workers, DAG processors, triggerers): Automatically detected via ``SUPERVISOR_COMMS`` presence
-- **Server contexts** (API server, scheduler, plugins): Automatically detected when ``SUPERVISOR_COMMS`` is not available
-- No configuration required - works regardless of import order or plugin loading timing
+1. **Client contexts** (task runner in worker): Detected via ``SUPERVISOR_COMMS`` presence
+2. **Server contexts** (API server, scheduler): Explicitly marked with ``_AIRFLOW_PROCESS_CONTEXT=server`` environment variable
+3. **Fallback contexts** (supervisor, unknown contexts): Neither marker present, uses minimal safe chain
 
 **Backend Chains:**
 
-- **Client**: ``EnvironmentVariablesBackend`` → ``ExecutionAPISecretsBackend`` (routes to Execution API)
-- **Server**: ``EnvironmentVariablesBackend`` → ``MetastoreBackend`` (direct DB access)
+- **Client**: ``EnvironmentVariablesBackend`` → ``ExecutionAPISecretsBackend`` (routes to Execution API via SUPERVISOR_COMMS)
+- **Server**: ``EnvironmentVariablesBackend`` → ``MetastoreBackend`` (direct database access)
+- **Fallback**: ``EnvironmentVariablesBackend`` only (+ external backends from config like AWS Secrets Manager, Vault)
 
-This maintains the architectural separation where workers access resources only through the Execution API,
-while API server components have direct database access.
+The fallback chain is crucial for supervisor processes (worker-side, before task runner starts) which need to access
+external secrets for remote logging setup but should not use ``MetastoreBackend`` (to maintain worker isolation).
+
+**Architecture Benefits:**
+
+- Workers (supervisor + task runner) never use ``MetastoreBackend``, maintaining strict isolation
+- External secrets backends (AWS Secrets Manager, Vault, etc.) work in all three contexts
+- Supervisor falls back to Execution API client for connections not found in external backends
+- API server and scheduler have direct database access for optimal performance
 
 **Impact:**
 
 - Hooks like ``GCSHook``, ``S3Hook`` now work correctly in log handlers and plugins
 - No code changes required for existing plugins or hooks
-- Workers remain isolated from direct database access (network-level blocking still possible)
-- External secrets backends (AWS Secrets Manager, Vault, etc.) continue to work in all contexts
-- Automatic detection works regardless of initialization order
+- Workers remain isolated from direct database access (network-level DB blocking fully supported)
+- External secrets work everywhere (workers, supervisor, API server)
+- Robust handling of unknown contexts with safe minimal chain
 
-See: `#56120 <https://github.com/apache/airflow/issues/56120>`__, `#56583 <https://github.com/apache/airflow/issues/56583>`__
+See: `#56120 <https://github.com/apache/airflow/issues/56120>`__, `#56583 <https://github.com/apache/airflow/issues/56583>`__, `#51816 <https://github.com/apache/airflow/issues/51816>`__
 
 * Types of change
 

airflow-core/src/airflow/api_fastapi/main.py

@@ -19,6 +19,10 @@
 
 import os
 
+# Mark this as a server context before any airflow imports
+# This ensures plugins loaded at import time get the correct secrets backend chain
+os.environ["_AIRFLOW_PROCESS_CONTEXT"] = "server"
+
 from airflow.api_fastapi.app import cached_app
 
 # There is no way to pass the apps to this file from Airflow CLI

airflow-core/src/airflow/jobs/scheduler_job_runner.py

@@ -1020,6 +1020,11 @@ def set_ti_span_attrs(cls, span, state, ti):
             span.add_event(name="airflow.task.ended", timestamp=datetime_to_nano(ti.end_date))
 
     def _execute(self) -> int | None:
+        import os
+
+        # Mark this as a server context for secrets backend detection
+        os.environ["_AIRFLOW_PROCESS_CONTEXT"] = "server"
+
         self.log.info("Starting the scheduler")
 
         reset_signals = self.register_signals()

airflow-core/src/airflow/secrets/__init__.py

@@ -29,7 +29,7 @@
 
 from airflow.utils.deprecation_tools import add_deprecated_classes
 
-__all__ = ["BaseSecretsBackend", "DEFAULT_SECRETS_SEARCH_PATH", "DEFAULT_SECRETS_SEARCH_PATH_WORKERS"]
+__all__ = ["BaseSecretsBackend", "DEFAULT_SECRETS_SEARCH_PATH"]
 
 from airflow.secrets.base_secrets import BaseSecretsBackend
 
@@ -38,15 +38,33 @@
     "airflow.secrets.metastore.MetastoreBackend",
 ]
 
-DEFAULT_SECRETS_SEARCH_PATH_WORKERS = [
-    "airflow.secrets.environment_variables.EnvironmentVariablesBackend",
-    "airflow.sdk.execution_time.secrets.execution_api.ExecutionAPISecretsBackend",
-]
-
 
 __deprecated_classes = {
     "cache": {
         "SecretCache": "airflow.sdk.execution_time.cache.SecretCache",
     },
 }
 add_deprecated_classes(__deprecated_classes, __name__)
+
+
+def __getattr__(name):
+    if name == "DEFAULT_SECRETS_SEARCH_PATH_WORKERS":
+        import warnings
+
+        warnings.warn(
+            "airflow.secrets.DEFAULT_SECRETS_SEARCH_PATH_WORKERS is moved to the Task SDK. "
+            "Use airflow.sdk.execution_time.secrets.DEFAULT_SECRETS_SEARCH_PATH_WORKERS instead.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        try:
+            from airflow.sdk.execution_time.secrets import DEFAULT_SECRETS_SEARCH_PATH_WORKERS
+
+            return DEFAULT_SECRETS_SEARCH_PATH_WORKERS
+        except (ImportError, AttributeError):
+            # Back-compat for older Task SDK clients
+            return [
+                "airflow.secrets.environment_variables.EnvironmentVariablesBackend",
+            ]
+
+    raise AttributeError(f"module '{__name__}' has no attribute '{name}'")

airflow-core/tests/unit/core/test_configuration.py

@@ -43,7 +43,7 @@
     write_default_airflow_configuration_if_needed,
 )
 from airflow.providers_manager import ProvidersManager
-from airflow.secrets import DEFAULT_SECRETS_SEARCH_PATH_WORKERS
+from airflow.sdk.execution_time.secrets import DEFAULT_SECRETS_SEARCH_PATH_WORKERS
 
 from tests_common.test_utils.config import conf_vars
 from tests_common.test_utils.markers import skip_if_force_lowest_dependencies_marker

task-sdk/src/airflow/sdk/execution_time/context.py

@@ -362,12 +362,16 @@ def __eq__(self, other):
         return True
 
     def get(self, conn_id: str, default_conn: Any = None) -> Any:
+        from airflow.exceptions import AirflowNotFoundException
+
         try:
             return _get_connection(conn_id)
         except AirflowRuntimeError as e:
             if e.error.error == ErrorType.CONNECTION_NOT_FOUND:
                 return default_conn
             raise
+        except AirflowNotFoundException:
+            return default_conn
 
 
 class VariableAccessor:

task-sdk/src/airflow/sdk/execution_time/secrets/__init__.py

@@ -21,4 +21,9 @@
 
 from airflow.sdk.execution_time.secrets.execution_api import ExecutionAPISecretsBackend
 
-__all__ = ["ExecutionAPISecretsBackend"]
+__all__ = ["ExecutionAPISecretsBackend", "DEFAULT_SECRETS_SEARCH_PATH_WORKERS"]
+
+DEFAULT_SECRETS_SEARCH_PATH_WORKERS = [
+    "airflow.secrets.environment_variables.EnvironmentVariablesBackend",
+    "airflow.sdk.execution_time.secrets.execution_api.ExecutionAPISecretsBackend",
+]

task-sdk/src/airflow/sdk/execution_time/supervisor.py

@@ -1769,30 +1769,44 @@ def ensure_secrets_backend_loaded() -> list[BaseSecretsBackend]:
     Initialize secrets backend with auto-detected context.
 
     Detection strategy:
-    - If SUPERVISOR_COMMS exists and is set → client chain (ExecutionAPISecretsBackend)
-    - Otherwise → server chain (MetastoreBackend)
+    1. SUPERVISOR_COMMS exists and is set → client chain (ExecutionAPISecretsBackend)
+    2. _AIRFLOW_PROCESS_CONTEXT=server env var → server chain (MetastoreBackend)
+    3. Neither → fallback chain (only env vars + external backends, no MetastoreBackend)
 
-    Client contexts: workers, DAG processor, triggerer (have SUPERVISOR_COMMS)
-    Server contexts: API server, scheduler (no SUPERVISOR_COMMS)
+    Client contexts: task runner in worker (has SUPERVISOR_COMMS)
+    Server contexts: API server, scheduler (set _AIRFLOW_PROCESS_CONTEXT=server)
+    Fallback contexts: supervisor, unknown contexts (no SUPERVISOR_COMMS, no env var)
 
-    This approach works regardless of import order or plugin loading timing.
+    The fallback chain ensures supervisor can use external secrets (AWS Secrets Manager,
+    Vault, etc.) while falling back to API client, without trying MetastoreBackend.
     """
+    import os
+
     from airflow.configuration import ensure_secrets_loaded
-    from airflow.secrets import DEFAULT_SECRETS_SEARCH_PATH, DEFAULT_SECRETS_SEARCH_PATH_WORKERS
+    from airflow.sdk.execution_time.secrets import DEFAULT_SECRETS_SEARCH_PATH_WORKERS
 
-    # Check for client context (SUPERVISOR_COMMS)
+    # 1. Check for client context (SUPERVISOR_COMMS)
     try:
         from airflow.sdk.execution_time import task_runner
 
         if hasattr(task_runner, "SUPERVISOR_COMMS") and task_runner.SUPERVISOR_COMMS is not None:
-            # Client context: worker, DAG processor, triggerer
+            # Client context: task runner with SUPERVISOR_COMMS
             return ensure_secrets_loaded(default_backends=DEFAULT_SECRETS_SEARCH_PATH_WORKERS)
     except (ImportError, AttributeError):
         pass
 
-    # Default to server context (no SUPERVISOR_COMMS = server)
-    # Server context: API server, scheduler, plugins
-    return ensure_secrets_loaded(default_backends=DEFAULT_SECRETS_SEARCH_PATH)
+    # 2. Check for explicit server context
+    if os.environ.get("_AIRFLOW_PROCESS_CONTEXT") == "server":
+        # Server context: API server, scheduler
+        # uses the default server list
+        return ensure_secrets_loaded()
+
+    # 3. Fallback for unknown contexts (supervisor, etc.)
+    # Only env vars + external backends from config, no MetastoreBackend, no ExecutionAPISecretsBackend
+    fallback_backends = [
+        "airflow.secrets.environment_variables.EnvironmentVariablesBackend",
+    ]
+    return ensure_secrets_loaded(default_backends=fallback_backends)
 
 
 def _configure_logging(log_path: str, client: Client) -> tuple[FilteringBoundLogger, BinaryIO | TextIO]:

task-sdk/tests/task_sdk/definitions/test_connection.py

@@ -28,7 +28,7 @@
 from airflow.sdk import Connection
 from airflow.sdk.exceptions import ErrorType
 from airflow.sdk.execution_time.comms import ConnectionResult, ErrorResponse
-from airflow.secrets import DEFAULT_SECRETS_SEARCH_PATH_WORKERS
+from airflow.sdk.execution_time.secrets import DEFAULT_SECRETS_SEARCH_PATH_WORKERS
 
 from tests_common.test_utils.config import conf_vars