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
@@ -923,8 +923,10 @@ def test_initialize_secrets_backends_on_workers(self):
         backends = initialize_secrets_backends(DEFAULT_SECRETS_SEARCH_PATH_WORKERS)
         backend_classes = [backend.__class__.__name__ for backend in backends]
 
-        assert len(backends) == 2
+        assert len(backends) == 3
         assert "SystemsManagerParameterStoreBackend" in backend_classes
+        assert "EnvironmentVariablesBackend" in backend_classes
+        assert "ExecutionAPISecretsBackend" in backend_classes
 
     @skip_if_force_lowest_dependencies_marker
     @conf_vars(

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

@@ -190,12 +190,16 @@ async def _async_get_connection(conn_id: str) -> Connection:
 
     from airflow.sdk.execution_time.supervisor import ensure_secrets_backend_loaded
 
-    # Try secrets backends using async wrapper (which may include SupervisorCommsSecretsBackend
-    # in worker contexts or MetastoreBackend in API server contexts)
+    # Try secrets backends
     backends = ensure_secrets_backend_loaded()
     for secrets_backend in backends:
         try:
-            conn = await sync_to_async(secrets_backend.get_connection)(conn_id)  # type: ignore[assignment]
+            # Use async method if available, otherwise wrap sync method
+            if hasattr(secrets_backend, "aget_connection"):
+                conn = await secrets_backend.aget_connection(conn_id)  # type: ignore[assignment]
+            else:
+                conn = await sync_to_async(secrets_backend.get_connection)(conn_id)  # type: ignore[assignment]
+
             if conn:
                 SecretCache.save_connection_uri(conn_id, conn.get_uri())
                 _mask_connection_secrets(conn)
@@ -233,7 +237,8 @@ def _get_variable(key: str, deserialize_json: bool) -> Any:
         pass  # Continue to check backends
 
     backends = ensure_secrets_backend_loaded()
-    # iterate over backends if not in cache (or expired)
+
+    # Iterate over backends if not in cache (or expired)
     for secrets_backend in backends:
         try:
             var_val = secrets_backend.get_variable(key=key)
@@ -253,31 +258,13 @@ def _get_variable(key: str, deserialize_json: bool) -> Any:
                 type(secrets_backend).__name__,
             )
 
-    if backends:
-        log.debug(
-            "Variable not found in any of the configured Secrets Backends. Trying to retrieve from API server",
-            key=key,
-        )
-
-    # TODO: This should probably be moved to a separate module like `airflow.sdk.execution_time.comms`
-    #   or `airflow.sdk.execution_time.variable`
-    #   A reason to not move it to `airflow.sdk.execution_time.comms` is that it
-    #   will make that module depend on Task SDK, which is not ideal because we intend to
-    #   keep Task SDK as a separate package than execution time mods.
-    from airflow.sdk.execution_time.comms import ErrorResponse, GetVariable
-    from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
-
-    msg = SUPERVISOR_COMMS.send(GetVariable(key=key))
-
-    if isinstance(msg, ErrorResponse):
-        raise AirflowRuntimeError(msg)
+    # If no backend found the variable, raise a not found error (mirrors _get_connection)
+    from airflow.sdk.exceptions import AirflowRuntimeError, ErrorType
+    from airflow.sdk.execution_time.comms import ErrorResponse
 
-    if TYPE_CHECKING:
-        assert isinstance(msg, VariableResult)
-    variable = _convert_variable_result_to_variable(msg, deserialize_json)
-    # Save raw value to ensure cache consistency regardless of deserialize_json parameter
-    SecretCache.save_variable(key, msg.value)
-    return variable.value
+    raise AirflowRuntimeError(
+        ErrorResponse(error=ErrorType.VARIABLE_NOT_FOUND, detail={"message": f"Variable {key} not found"})
+    )
 
 
 def _set_variable(key: str, value: Any, description: str | None = None, serialize_json: bool = False) -> None:
@@ -290,18 +277,21 @@ def _set_variable(key: str, value: Any, description: str | None = None, serializ
 
     from airflow.sdk.execution_time.cache import SecretCache
     from airflow.sdk.execution_time.comms import PutVariable
+    from airflow.sdk.execution_time.secrets.execution_api import ExecutionAPISecretsBackend
     from airflow.sdk.execution_time.supervisor import ensure_secrets_backend_loaded
     from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
 
     # check for write conflicts on the worker
     for secrets_backend in ensure_secrets_backend_loaded():
+        if isinstance(secrets_backend, ExecutionAPISecretsBackend):
+            continue
         try:
             var_val = secrets_backend.get_variable(key=key)
             if var_val is not None:
                 _backend_name = type(secrets_backend).__name__
                 log.warning(
                     "The variable %s is defined in the %s secrets backend, which takes "
-                    "precedence over reading from the database. The value in the database will be "
+                    "precedence over reading from the API Server. The value from the API Server will be "
                     "updated, but to read it you have to delete the conflicting variable "
                     "from %s",
                     key,
@@ -362,12 +352,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/secrets/execution_api.py

@@ -43,7 +43,7 @@ def get_conn_value(self, conn_id: str) -> str | None:
         """
         raise NotImplementedError("Use get_connection instead")
 
-    def get_connection(self, conn_id: str) -> Connection | None:
+    def get_connection(self, conn_id: str) -> Connection | None:  # type: ignore[override]
         """
         Return connection object by routing through SUPERVISOR_COMMS.
 
@@ -75,7 +75,7 @@ def get_variable(self, key: str) -> str | None:
         :param key: Variable key
         :return: Variable value or None if not found
         """
-        from airflow.sdk.execution_time.comms import ErrorResponse, GetVariable
+        from airflow.sdk.execution_time.comms import ErrorResponse, GetVariable, VariableResult
         from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
 
         try:
@@ -86,8 +86,59 @@ def get_variable(self, key: str) -> str | None:
                 return None
 
             # Extract value from VariableResult
-            if hasattr(msg, "value"):
-                return msg.value
+            if isinstance(msg, VariableResult):
+                return msg.value  # Already a string | None
+            return None
+        except Exception:
+            # If SUPERVISOR_COMMS fails for any reason, return None
+            # to allow fallback to other backends
+            return None
+
+    async def aget_connection(self, conn_id: str) -> Connection | None:  # type: ignore[override]
+        """
+        Return connection object asynchronously via SUPERVISOR_COMMS.
+
+        :param conn_id: connection id
+        :return: Connection object or None if not found
+        """
+        from airflow.sdk.execution_time.comms import ErrorResponse, GetConnection
+        from airflow.sdk.execution_time.context import _process_connection_result_conn
+        from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+
+        try:
+            msg = await SUPERVISOR_COMMS.asend(GetConnection(conn_id=conn_id))
+
+            if isinstance(msg, ErrorResponse):
+                # Connection not found or error occurred
+                return None
+
+            # Convert ExecutionAPI response to SDK Connection
+            return _process_connection_result_conn(msg)
+        except Exception:
+            # If SUPERVISOR_COMMS fails for any reason, return None
+            # to allow fallback to other backends
+            return None
+
+    async def aget_variable(self, key: str) -> str | None:
+        """
+        Return variable value asynchronously via SUPERVISOR_COMMS.
+
+        :param key: Variable key
+        :return: Variable value or None if not found
+        """
+        from airflow.sdk.execution_time.comms import ErrorResponse, GetVariable, VariableResult
+        from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+
+        try:
+            msg = await SUPERVISOR_COMMS.asend(GetVariable(key=key))
+
+            if isinstance(msg, ErrorResponse):
+                # Variable not found or error occurred
+                return None
+
+            # Extract value from VariableResult
+            if isinstance(msg, VariableResult):
+                return msg.value  # Already a string | None
             return None
         except Exception:
             # If SUPERVISOR_COMMS fails for any reason, return None