Allow opting into fork+exec task processes via execute_tasks_new_python_interpreter

airflow-core/docs/security/security_model.rst

@@ -591,8 +591,11 @@ model — Airflow does not enforce these natively.
    For higher security, pass sensitive configuration values via environment variables rather than
    configuration files. Environment variables are inherently safer than configuration files in
    Airflow's worker processes because of a built-in protection: on Linux, the supervisor process
-   calls ``prctl(PR_SET_DUMPABLE, 0)`` before forking the task process, and this flag is inherited
-   by the forked child. This marks both processes as non-dumpable, which prevents same-UID sibling
+   calls ``prctl(PR_SET_DUMPABLE, 0)`` before forking the task process; a bare-forked child inherits
+   the flag, and a child started through ``exec`` restores it in its bootstrap, before importing
+   Airflow, because ``execve`` resets it (``kernel.yama.ptrace_scope >= 1`` covers ``/proc/<pid>/mem``
+   and ``ptrace`` attach for the remaining interpreter-start window). This marks both processes as
+   non-dumpable, which prevents same-UID sibling
    processes from reading ``/proc/<pid>/environ``, ``/proc/<pid>/mem``, or attaching via
    ``ptrace``. In contrast, configuration files on disk are readable by any process running as
    the same Unix user. Environment variables can also be scoped to individual processes or

airflow-core/docs/security/workload.rst

@@ -85,8 +85,11 @@ Worker process memory protection (Linux)
 ''''''''''''''''''''''''''''''''''''''''
 
 On Linux, the supervisor process calls ``prctl(PR_SET_DUMPABLE, 0)`` at the start of
-``supervise_task()`` before forking the task process. This flag is inherited by the forked
-child. Marking processes as non-dumpable prevents same-UID sibling processes from reading
+``supervise_task()`` before forking the task process. A bare-forked child inherits the flag;
+a child started through ``exec`` (macOS, or ``[core] execute_tasks_new_python_interpreter``)
+restores it in its bootstrap, before importing Airflow, because ``execve`` resets it; for the remaining
+interpreter-start window, ``kernel.yama.ptrace_scope >= 1`` covers ``/proc/<pid>/mem`` and ``ptrace``
+attach. Marking processes as non-dumpable prevents same-UID sibling processes from reading
 ``/proc/<pid>/mem``, ``/proc/<pid>/environ``, or ``/proc/<pid>/maps``, and blocks
 ``ptrace(PTRACE_ATTACH)``. This is critical because each supervisor holds a distinct JWT
 token in memory — without this protection, a malicious task process running as the same

airflow-core/newsfragments/72164.significant.rst

@@ -0,0 +1,23 @@
+``[core] execute_tasks_new_python_interpreter`` now applies to Airflow 3 task processes
+
+On Airflow 3 the option had no effect on task execution (only the Edge worker read it). When set to
+``True``, the task supervisor now ``exec``\ s a fresh interpreter right after forking the task process,
+which prevents the fork from inheriting a lock held by a supervisor thread (a permanent hang at the
+task's first TLS call). Deployments that kept the option ``True`` from Airflow 2 get this behaviour,
+and its per-task interpreter start-up cost, on upgrade without further action; set it to ``False`` to
+keep bare fork. The task process reads the global value, so a team-scoped Edge setting alone does not
+turn it on. Edge workers with the option ``True`` already start a fresh interpreter for the supervisor
+and will now start a second one for the task. The Dag processor, the triggerer and task
+callbacks (which the callback supervisor runs from a closure that cannot be named for an exec'd
+child) are not affected and keep bare fork.
+
+* Types of change
+
+  * [ ] Dag changes
+  * [x] Config changes
+  * [ ] API changes
+  * [ ] CLI changes
+  * [x] Behaviour changes
+  * [ ] Plugin changes
+  * [ ] Dependency changes
+  * [ ] Code interface changes

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

@@ -223,6 +223,14 @@ core:
         * ``False``: Execute via forking of the parent process
         * ``True``: Spawning a new python process, slower than fork, but means plugin changes picked
           up by tasks straight away
+
+        On Airflow 3 this applies to the task process only. When ``True`` the supervisor starts
+        the task in a fresh interpreter with ``exec`` right after forking it, so the task cannot
+        inherit a lock a supervisor thread held at fork time (which can otherwise hang it at its
+        first TLS call). The Dag processor, the triggerer and task callbacks are not affected and
+        keep bare fork. Each task pays an interpreter start plus re-import. The task process reads
+        the global value; Edge workers, which resolve it per team, already start a fresh interpreter
+        for the supervisor when it is ``True`` and start a second one for the task.
       default: "False"
       example: ~
       version_added: 2.0.0

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

@@ -517,6 +517,22 @@ def _should_use_exec() -> bool:
     return sys.platform in _FORK_EXEC_PLATFORMS
 
 
+def _task_process_uses_exec() -> bool:
+    """
+    Whether the task process should ``exec`` a fresh interpreter after the fork.
+
+    Forced where bare fork is unsafe (macOS); elsewhere a deployment opts in with
+    ``[core] execute_tasks_new_python_interpreter``. exec replaces the child's address
+    space, so it cannot inherit a lock a supervisor thread held at fork time (e.g.
+    OpenSSL's, which otherwise hangs the task at its first TLS call; #71707). Only the
+    task process reads the option -- it has always described task execution -- so the Dag
+    processor (one child per file per parse loop) and the triggerer keep the platform gate.
+    """
+    return _should_use_exec() or conf.getboolean(
+        "core", "execute_tasks_new_python_interpreter", fallback=False
+    )
+
+
 def _resolve_child_target(dotted: str) -> Callable[[], None]:
     """
     Resolve a ``module:qualname`` string to the callable the exec'd child runs.
@@ -531,15 +547,38 @@ def _resolve_child_target(dotted: str) -> Callable[[], None]:
     return pkgutil.resolve_name(dotted)
 
 
+# Runs in the exec'd child before anything else. execve reset PR_SET_DUMPABLE (4 in
+# <linux/prctl.h>); restore it before the Airflow import so the window in which a same-UID
+# sibling can open /proc/<pid>/mem or ptrace-attach is interpreter start only (a descriptor
+# or attach taken in that window survives a later prctl -- the kernel checks once, at open).
+# _child_exec_main() repeats the call as the logged fallback.
+_CHILD_EXEC_PRELUDE = """\
+import sys
+if sys.platform == "linux":
+    try:
+        import ctypes
+
+        ctypes.CDLL(None, use_errno=True).prctl(4, 0, 0, 0, 0)
+    except Exception:
+        pass
+"""
+_CHILD_EXEC_BOOTSTRAP = _CHILD_EXEC_PRELUDE + (
+    "from airflow.sdk.execution_time.supervisor import _child_exec_main\n_child_exec_main()\n"
+)
+
+
 def _child_exec_main():
     """
-    Entry point for the child process when using fork+exec (macOS).
+    Entry point for the child process when using fork+exec.
 
     After exec, FDs 0/1/2/3 are the requests/stdout/stderr/log sockets the parent
     placed there via dup2.  The target to run is named in ``_AIRFLOW_CHILD_TARGET``
     (``module:qualname``); it is rehydrated and handed to :func:`_fork_main`, which
     sets up the structured log channel from FD 3 exactly as the bare-fork path does.
     """
+    # The bootstrap already restored PR_SET_DUMPABLE before importing Airflow; this is the
+    # logged fallback (execve had reset what supervise_task() set before the fork).
+    _make_process_nondumpable()
     # FDs 0, 1, 2 were dup2'd onto the socketpairs before exec.
     child_requests = socket(fileno=0)
     child_stdout = socket(fileno=1)
@@ -690,9 +729,11 @@ def start(
         """
         Fork and start a new subprocess with the specified target function.
 
-        :param use_exec: If True, on platforms that need it (currently macOS),
-            immediately ``os.execv`` a fresh Python interpreter after ``os.fork``.
-            This avoids macOS fork-safety issues with Objective-C frameworks.
+        :param use_exec: If True, immediately ``os.execv`` a fresh Python interpreter
+            after ``os.fork``: forced on platforms that need it (macOS, whose Objective-C
+            frameworks are not fork-safe) and opted into for the task process elsewhere via
+            ``[core] execute_tasks_new_python_interpreter`` (a lock a supervisor thread
+            held at fork time cannot survive into a fresh address space).
             ``target`` is rehydrated in the exec'd child from its ``module:qualname``,
             so any importable entry point (task execution, DAG processor, triggerer)
             is supported.
@@ -742,8 +783,8 @@ def start(
 
             try:
                 if use_exec:
-                    # macOS: exec a fresh Python interpreter to drop the inherited
-                    # ObjC/CoreFoundation state that is not fork-safe. Redirect the
+                    # exec a fresh Python interpreter to drop inherited state that is not
+                    # fork-safe (ObjC/CoreFoundation on macOS; a held lock elsewhere). Redirect the
                     # socketpairs onto the fixed FDs the exec'd child reconstructs:
                     # 0 (requests/stdin), 1 (stdout), 2 (stderr), 3 (structured logs).
                     # The source fds are always >= 3 (0/1/2 stay open in every launch
@@ -760,12 +801,7 @@ def start(
                         os.set_inheritable(fd, True)
                     os.execv(
                         sys.executable,
-                        [
-                            sys.executable,
-                            "-c",
-                            "from airflow.sdk.execution_time.supervisor import _child_exec_main;"
-                            " _child_exec_main()",
-                        ],
+                        [sys.executable, "-c", _CHILD_EXEC_BOOTSTRAP],
                     )
                     # execv replaces the process -- unreachable on success
                 else:
@@ -1452,10 +1488,10 @@ def start(  # type: ignore[override]
         **kwargs,
     ) -> Self:
         """Fork and start a new subprocess to execute the given task."""
-        # Opt in to fork+exec on platforms that need it (currently macOS).
+        # fork+exec where the platform needs it (macOS) or the deployment opted in.
         # Tests override `target` with a local stub to exercise the base
         # infrastructure; keep bare fork for those.
-        use_exec = target is _subprocess_main and _should_use_exec()
+        use_exec = target is _subprocess_main and _task_process_uses_exec()
         proc: Self = super().start(
             id=what.id,
             client=client,

task-sdk/tests/task_sdk/execution_time/exec_probe_target.py

@@ -0,0 +1,30 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Importable entry point for the fork+exec end-to-end test; it runs inside the exec'd child."""
+
+from __future__ import annotations
+
+import sys
+
+from airflow.sdk.execution_time.comms import CommsDecoder
+
+
+def exec_probe_main() -> None:
+    """Stand-in for ``_subprocess_main``: consume the startup message, then report over stdout."""
+    CommsDecoder()._get_response()
+    print("exec-probe-ok")
+    sys.stdout.flush()