This is an automated email from the ASF dual-hosted git repository.
kaxil pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new f46a370a95b Stop the secrets backend lookup from importing the task
runner outside tasks (#74641)
f46a370a95b is described below
commit f46a370a95bcc2be98532fd820c0b4ca8943b336
Author: Kaxil Naik <[email protected]>
AuthorDate: Wed Oct 14 04:01:56 2026 +0100
Stop the secrets backend lookup from importing the task runner outside
tasks (#74641)
ensure_secrets_backend_loaded() imported task_runner to check whether
SUPERVISOR_COMMS was set. Outside a task, that loaded the task runtime
just to learn the answer was no. Read the module from sys.modules
instead, as mask_secret already does.
That import also loaded the task runner in the supervisor before it
forked the task process, and before callback processes when remote
logging has a connection, so those children inherited it. Import it
explicitly before a bare fork of a task or callback process instead, so
each child does not import it again. If the import fails there, the
child imports it again and reports the error in its own log, as before.
---
.../sdk/execution_time/callback_supervisor.py | 2 +
.../src/airflow/sdk/execution_time/supervisor.py | 27 +++++++----
.../execution_time/test_callback_supervisor.py | 14 ++++++
.../tests/task_sdk/execution_time/test_secrets.py | 20 ++++++++
.../task_sdk/execution_time/test_supervisor.py | 54 ++++++++++++++++++++++
5 files changed, 109 insertions(+), 8 deletions(-)
diff --git a/task-sdk/src/airflow/sdk/execution_time/callback_supervisor.py
b/task-sdk/src/airflow/sdk/execution_time/callback_supervisor.py
index 57f1cae312f..01d1ecf7f38 100644
--- a/task-sdk/src/airflow/sdk/execution_time/callback_supervisor.py
+++ b/task-sdk/src/airflow/sdk/execution_time/callback_supervisor.py
@@ -52,6 +52,7 @@ from airflow.sdk.execution_time.supervisor import (
SOCKET_CLEANUP_TIMEOUT,
WatchedSubprocess,
_ensure_client,
+ _import_task_runner_before_fork,
_make_process_nondumpable,
)
@@ -255,6 +256,7 @@ class CallbackSubprocess(WatchedSubprocess):
_log.error("Callback failed", error=error_msg)
sys.exit(1)
+ _import_task_runner_before_fork()
return super().start(
id=UUID(id) if not isinstance(id, UUID) else id,
client=client,
diff --git a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
index 9677c1e0607..d19ccd3ef5e 100644
--- a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
+++ b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
@@ -172,6 +172,7 @@ from airflow.sdk.execution_time.request_handlers import (
handle_update_dag_run_note,
)
from airflow.sdk.execution_time.schema import get_schema_version_migrator,
resolve_body_class
+from airflow.sdk.log import _supervisor_comms
try:
from socket import send_fds
@@ -364,6 +365,19 @@ def _subprocess_main():
main()
+def _import_task_runner_before_fork() -> None:
+ """
+ Import the task runner so that a child forked from this process inherits
it.
+
+ Task and callback processes both import it, so importing it once here
saves every forked child
+ from importing it again. It is not imported at module level because
secrets lookups outside a
+ task import this module, and they must not load the task runtime.
+ """
+ # If it fails to import, the child imports it again and reports the error
in its own log.
+ with suppress(ImportError):
+ import airflow.sdk.execution_time.task_runner # noqa: F401
+
+
def _reset_signals():
# Uninstall the rich etc. exception handler
sys.excepthook = sys.__excepthook__
@@ -1669,6 +1683,8 @@ class ActivitySubprocess(WatchedSubprocess):
# Tests override `target` with a local stub to exercise the base
# infrastructure; keep bare fork for those.
use_exec = target is _subprocess_main and _task_process_uses_exec()
+ if target is _subprocess_main and not use_exec:
+ _import_task_runner_before_fork()
proc: Self = super().start(
id=what.id,
client=client,
@@ -2919,14 +2935,9 @@ def ensure_secrets_backend_loaded() ->
list[BaseSecretsBackend]:
from airflow.sdk.execution_time.secrets import
DEFAULT_SECRETS_SEARCH_PATH_WORKERS
# 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: task runner with SUPERVISOR_COMMS
- return
ensure_secrets_loaded(default_backends=DEFAULT_SECRETS_SEARCH_PATH_WORKERS)
- except (ImportError, AttributeError):
- pass
+ if _supervisor_comms() is not None:
+ # Client context: task runner with SUPERVISOR_COMMS
+ return
ensure_secrets_loaded(default_backends=DEFAULT_SECRETS_SEARCH_PATH_WORKERS)
# 2. Check for explicit server context
if os.environ.get("_AIRFLOW_PROCESS_CONTEXT") == "server":
diff --git a/task-sdk/tests/task_sdk/execution_time/test_callback_supervisor.py
b/task-sdk/tests/task_sdk/execution_time/test_callback_supervisor.py
index 8acd3feeddf..fae31290be5 100644
--- a/task-sdk/tests/task_sdk/execution_time/test_callback_supervisor.py
+++ b/task-sdk/tests/task_sdk/execution_time/test_callback_supervisor.py
@@ -566,6 +566,20 @@ class TestCallbackSubprocessStart:
assert self.mock_super_start.call_args.kwargs.get("use_exec", False)
is False
+ def test_start_imports_task_runner_before_forking(self, base_start_kwargs):
+ """The callback child is forked from this process, so it inherits a
task runner imported here."""
+ calls = []
+ self.mock_super_start.side_effect = lambda **kwargs:
calls.append("fork")
+
+ with patch(
+
"airflow.sdk.execution_time.callback_supervisor._import_task_runner_before_fork",
+ autospec=True,
+ side_effect=lambda: calls.append("import task runner"),
+ ):
+ CallbackSubprocess.start(**base_start_kwargs)
+
+ assert calls == ["import task runner", "fork"]
+
class TestSuperviseCallbackExchangesTokenFirst:
"""No callback code may run until the single-use token has been
exchanged."""
diff --git a/task-sdk/tests/task_sdk/execution_time/test_secrets.py
b/task-sdk/tests/task_sdk/execution_time/test_secrets.py
index 2fe41e328ce..ffad8712ed0 100644
--- a/task-sdk/tests/task_sdk/execution_time/test_secrets.py
+++ b/task-sdk/tests/task_sdk/execution_time/test_secrets.py
@@ -17,6 +17,10 @@
from __future__ import annotations
+import subprocess
+import sys
+import textwrap
+
import pytest
from airflow.sdk.api.datamodels._generated import ConnectionResponse
@@ -308,6 +312,22 @@ class TestContextDetection:
assert "MetastoreBackend" not in backend_classes
assert "ExecutionAPISecretsBackend" not in backend_classes
+ def test_does_not_import_task_runner_outside_task_execution(self):
+ code = textwrap.dedent(
+ """
+ import sys
+ from airflow.sdk.execution_time.supervisor import
ensure_secrets_backend_loaded
+
+ ensure_secrets_backend_loaded()
+ print("airflow.sdk.execution_time.task_runner" in sys.modules)
+ """
+ )
+
+ result = subprocess.run([sys.executable, "-c", code],
capture_output=True, text=True, check=False)
+
+ assert result.returncode == 0, result.stderr
+ assert result.stdout.strip().splitlines()[-1] == "False"
+
class TestSecretsBackendMemoisation:
BACKEND =
"airflow.secrets.environment_variables.EnvironmentVariablesBackend"
diff --git a/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
b/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
index 14e285e3cb3..c237a9376bd 100644
--- a/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
+++ b/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
@@ -5818,6 +5818,60 @@ def test_activity_start_opts_into_fork_exec(monkeypatch,
mocker, task_uses_exec,
assert base_start.call_args.kwargs["target"] is target
[email protected](
+ ("task_uses_exec", "target", "expected_calls"),
+ [
+ pytest.param(False, supervisor._subprocess_main, ["import task
runner", "fork"], id="bare-fork"),
+ pytest.param(True, supervisor._subprocess_main, ["fork"], id="exec"),
+ pytest.param(False, lambda: None, ["fork"], id="test-target"),
+ ],
+)
+def test_activity_start_imports_task_runner_before_bare_fork(
+ monkeypatch, mocker, task_uses_exec, target, expected_calls
+):
+ """A forked task process inherits the task runner from the supervisor; a
spawned one imports its own."""
+ calls = []
+
+ def fork(*args, **kwargs):
+ calls.append("fork")
+ return MagicMock(spec=ActivitySubprocess)
+
+ monkeypatch.setattr(supervisor, "_task_process_uses_exec", lambda:
task_uses_exec)
+ mocker.patch.object(
+ supervisor,
+ "_import_task_runner_before_fork",
+ autospec=True,
+ side_effect=lambda: calls.append("import task runner"),
+ )
+ mocker.patch.object(supervisor.WatchedSubprocess, "start", autospec=True,
side_effect=fork)
+
+ ActivitySubprocess.start(
+ dag_rel_path=os.devnull,
+ bundle_info=FAKE_BUNDLE,
+ what=TaskInstance(
+ id="4d828a62-a417-4936-a7a6-2b3fabacecab",
+ task_id="b",
+ dag_id="c",
+ run_id="d",
+ try_number=1,
+ dag_version_id=uuid7(),
+ queue="default",
+ ),
+ client=MagicMock(spec=sdk_client.Client),
+ target=target,
+ logger=MagicMock(),
+ )
+
+ assert calls == expected_calls
+
+
+def
test_import_task_runner_before_fork_leaves_import_errors_to_the_child(monkeypatch):
+ """A task runner that fails to import is reported by the child, which
imports it again."""
+ monkeypatch.setitem(sys.modules, "airflow.sdk.execution_time.task_runner",
None)
+
+ supervisor._import_task_runner_before_fork()
+
+
class TestTaskProcessUsesExec:
"""The config opt-in for fork+exec of the task process where the platform
does not force it."""
