gh-159041: Fix use-after-free in asyncio task tracking during thread state cleanup

Lib/test/test_asyncio/test_tasks.py

@@ -1,5 +1,6 @@
 """Tests for tasks.py."""
 
+import _thread
 import collections
 import contextlib
 import contextvars
@@ -20,6 +21,7 @@
 from asyncio import tasks
 from test.test_asyncio import utils as test_utils
 from test import support
+from test.support import threading_helper
 from test.support.script_helper import assert_python_ok
 
 
@@ -3095,6 +3097,39 @@ class CTask_CFuture_Tests(BaseTaskTests, SetMethodsTest,
     all_tasks = getattr(tasks, '_c_all_tasks', None)
     current_task = staticmethod(getattr(tasks, '_c_current_task', None))
 
+    @threading_helper.requires_working_threading()
+    def test_task_created_during_thread_state_clear(self):
+        loop = self.loop
+        task = None
+        var = contextvars.ContextVar('var')
+        # Avoid recreating the thread state's context after it is cleared.
+        task_context = contextvars.Context()
+
+        async def noop():
+            pass
+
+        class CreatesTaskOnClear:
+            def __del__(self):
+                nonlocal task
+                task = loop.create_task(noop(), context=task_context)
+
+        def create_finalizer():
+            # gh-159041: Run the finalizer when PyThreadState_Clear()
+            # clears the thread's context.
+            var.set(CreatesTaskOnClear())
+
+        # threading.Thread owns its Context, so use _thread to make the
+        # Context lifetime match the thread state's lifetime.
+        handle = _thread.start_joinable_thread(create_finalizer)
+        handle.join(support.SHORT_TIMEOUT)
+        self.assertTrue(handle.is_done(), 'thread failed to exit')
+        self.assertIsNotNone(task)
+
+        try:
+            self.assertEqual(self.all_tasks(loop), {task})
+        finally:
+            loop.run_until_complete(task)
+
     def test_del__log_destroy_pending_segfault(self):
         async def coro():
             pass

Lib/test/test_embed.py

@@ -440,6 +440,43 @@ def test_finalize_structseq(self):
         out, err = self.run_embedded_interpreter("test_repeated_init_exec", code)
         self.assertEqual(out, 'Tests passed\n' * INIT_LOOPS)
 
+    def test_asyncio_task_during_interpreter_clear(self):
+        import_helper.import_module('_asyncio')
+        code = textwrap.dedent("""
+            import asyncio
+            import contextvars
+            import gc
+            import os
+
+            gc.disable()
+
+            async def noop():
+                pass
+
+            class Loop:
+                def get_debug(self):
+                    return False
+
+                def call_soon(self, callback, *args, context=None):
+                    self.callback = callback
+
+            class CreatesTaskOnClear:
+                def __del__(self):
+                    asyncio.Task(noop(), loop=Loop(), context=contextvars.Context())
+                    os.write(1, b'task created\\n')
+
+            # Keep the finalizer alive until interpreter_clear()'s last GC.
+            finalizer = CreatesTaskOnClear()
+            finalizer.cycle = finalizer
+            del finalizer
+        """)
+        # gh-159041: Delete the cleared interpreter with no thread state attached.
+        out, err = self.run_embedded_interpreter(
+            "test_interpreter_clear_and_delete", code,
+            timeout=support.SHORT_TIMEOUT)
+        self.assertEqual(out, "task created\n")
+        self.assertEqual(err, "")
+
     def test_simple_initialization_api(self):
         # _testembed now uses Py_InitializeFromConfig by default
         # This case specifically checks Py_Initialize(Ex) still works

Misc/NEWS.d/next/Core_and_Builtins/2026-10-09-16-04-40.gh-issue-159041.PUP8bU.rst

@@ -0,0 +1,2 @@
+Fix a use-after-free in :mod:`asyncio` task bookkeeping when a finalizer
+creates a task during thread shutdown.

Programs/_testembed.c

@@ -268,6 +268,23 @@ static int test_repeated_init_exec(void)
     return 0;
 }
 
+
+static int test_interpreter_clear_and_delete(void)
+{
+    assert(main_argc == 3);
+    _testembed_initialize();
+    if (PyRun_SimpleString(main_argv[2]) < 0) {
+        return 1;
+    }
+
+    PyInterpreterState *interp = PyInterpreterState_Get();
+    PyInterpreterState_Clear(interp);
+    PyThreadState_Swap(NULL);
+    PyInterpreterState_Delete(interp);
+    // The interpreter has already been deleted, so don't call Py_FinalizeEx().
+    return 0;
+}
+
 /****************************************************************************
  * Test the Py_Initialize(Ex) convenience/compatibility wrappers
  ***************************************************************************/
@@ -3001,6 +3018,7 @@ struct TestCase
 static struct TestCase TestCases[] = {
     // Python initialization
     {"test_repeated_init_exec", test_repeated_init_exec},
+    {"test_interpreter_clear_and_delete", test_interpreter_clear_and_delete},
     {"test_repeated_simple_init", test_repeated_simple_init},
     {"test_forced_io_encoding", test_forced_io_encoding},
     {"test_import_in_subinterpreters", test_import_in_subinterpreters},

Python/pystate.c

@@ -1837,13 +1837,6 @@ PyThreadState_Clear(PyThreadState *tstate)
     Py_CLEAR(((_PyThreadStateImpl *)tstate)->asyncio_running_task);
 
 
-    PyMutex_Lock(&tstate->interp->asyncio_tasks_lock);
-    // merge any lingering tasks from thread state to interpreter's
-    // tasks list
-    llist_concat(&tstate->interp->asyncio_tasks_head,
-                 &((_PyThreadStateImpl *)tstate)->asyncio_tasks_head);
-    PyMutex_Unlock(&tstate->interp->asyncio_tasks_lock);
-
     Py_CLEAR(tstate->dict);
     Py_CLEAR(tstate->async_exc);
 
@@ -1919,11 +1912,30 @@ PyThreadState_Clear(PyThreadState *tstate)
 static void
 decrement_stoptheworld_countdown(struct _stoptheworld_state *stw);
 
-/* Common code for PyThreadState_Delete() and PyThreadState_DeleteCurrent() */
+// `release_gil` releases the current thread's GIL. For states already removed
+// by _PyThreadState_RemoveExcept(), `is_unlinked` limits cleanup to task transfer.
 static void
-tstate_delete_common(PyThreadState *tstate, int release_gil)
+tstate_delete_common(PyThreadState *tstate, int release_gil, int is_unlinked)
 {
     assert(tstate->_status.cleared && !tstate->_status.finalized);
+    assert(!is_unlinked || !release_gil);
+
+    _PyThreadStateImpl *tstate_impl = (_PyThreadStateImpl *)tstate;
+    // Finalizers may register tasks during or after PyThreadState_Clear().
+    // Transfer them before normal unlinking or GIL release: cross-thread task
+    // removal uses stop-the-world or the GIL, not asyncio_tasks_lock.
+    if (!llist_empty(&tstate_impl->asyncio_tasks_head)) {
+        // Serialize transfers from concurrently exiting threads.
+        PyMutex_Lock(&tstate->interp->asyncio_tasks_lock);
+        llist_concat(&tstate->interp->asyncio_tasks_head,
+                     &tstate_impl->asyncio_tasks_head);
+        PyMutex_Unlock(&tstate->interp->asyncio_tasks_lock);
+    }
+    assert(llist_empty(&tstate_impl->asyncio_tasks_head));
+    if (is_unlinked) {
+        return;
+    }
+
     tstate_verify_not_active(tstate);
     assert(!_PyThreadState_IsRunningMain(tstate));
 
@@ -1959,7 +1971,6 @@ tstate_delete_common(PyThreadState *tstate, int release_gil)
 
 #if defined(Py_REF_DEBUG) && defined(Py_GIL_DISABLED)
     // Add our portion of the total refcount to the interpreter's total.
-    _PyThreadStateImpl *tstate_impl = (_PyThreadStateImpl *)tstate;
     tstate->interp->object_state.reftotal += tstate_impl->reftotal;
     tstate_impl->reftotal = 0;
     assert(tstate_impl->refcounts.values == NULL);
@@ -2006,7 +2017,7 @@ zapthreads(PyInterpreterState *interp)
     */
     while ((tstate = interp->threads.head) != NULL) {
         tstate_verify_not_active(tstate);
-        tstate_delete_common(tstate, 0);
+        tstate_delete_common(tstate, /*release_gil=*/0, /*is_unlinked=*/0);
         free_threadstate((_PyThreadStateImpl *)tstate);
     }
 }
@@ -2017,7 +2028,7 @@ PyThreadState_Delete(PyThreadState *tstate)
 {
     _Py_EnsureTstateNotNULL(tstate);
     tstate_verify_not_active(tstate);
-    tstate_delete_common(tstate, 0);
+    tstate_delete_common(tstate, /*release_gil=*/0, /*is_unlinked=*/0);
     free_threadstate((_PyThreadStateImpl *)tstate);
 }
 
@@ -2033,7 +2044,7 @@ _PyThreadState_DeleteCurrent(PyThreadState *tstate)
     _PyStats_Detach((_PyThreadStateImpl *)tstate);
 #endif
     current_fast_clear(tstate->interp->runtime);
-    tstate_delete_common(tstate, 1);  // release GIL as part of call
+    tstate_delete_common(tstate, /*release_gil=*/1, /*is_unlinked=*/0);
     free_threadstate((_PyThreadStateImpl *)tstate);
 }
 
@@ -2103,6 +2114,7 @@ _PyThreadState_DeleteList(PyThreadState *list, int is_after_fork)
     for (p = list; p; p = next) {
         next = p->next;
         PyThreadState_Clear(p);
+        tstate_delete_common(p, /*release_gil=*/0, /*is_unlinked=*/1);
         if (is_after_fork) {
             free_threadstate((_PyThreadStateImpl *)p);
         }