gh-126080: fix UAF on task->task_context in task_call_step_soon due to an evil loop.__getattribute__

Lib/test/support/__init__.py

@@ -23,7 +23,7 @@
     # globals
     "PIPE_MAX_SIZE", "verbose", "max_memuse", "use_resources", "failfast",
     # exceptions
-    "Error", "TestFailed", "TestDidNotRun", "ResourceDenied",
+    "ReachableCode", "Error", "TestFailed", "TestDidNotRun", "ResourceDenied",
     # io
     "record_original_stdout", "get_original_stdout", "captured_stdout",
     "captured_stdin", "captured_stderr", "captured_output",
@@ -107,6 +107,11 @@
 STDLIB_DIR = os.path.dirname(TEST_HOME_DIR)
 REPO_ROOT = os.path.dirname(STDLIB_DIR)
 
+class ReachableCode(Exception):
+    """Exception to raise to indicate that some code was reached.
+
+    Use this exception if using mocks is not a good alternative.
+    """
 
 class Error(Exception):
     """Base class for regression test exceptions."""

Lib/test/test_asyncio/test_futures.py

@@ -14,6 +14,7 @@
 import warnings
 from test.test_asyncio import utils as test_utils
 from test import support
+from test.support import ReachableCode
 
 
 def tearDownModule():
@@ -32,13 +33,6 @@ def last_cb():
     pass
 
 
-class ReachableCode(Exception):
-    """Exception to raise to indicate that some code was reached.
-
-    Use this exception if using mocks is not a good alternative.
-    """
-
-
 class SimpleEvilEventLoop(asyncio.base_events.BaseEventLoop):
     """Base class for UAF and other evil stuff requiring an evil event loop."""
 

Lib/test/test_asyncio/test_tasks.py

@@ -19,6 +19,7 @@
 from asyncio import tasks
 from test.test_asyncio import utils as test_utils
 from test import support
+from test.support import ReachableCode
 from test.support.script_helper import assert_python_ok
 from test.support.warnings_helper import ignore_warnings
 
@@ -89,8 +90,8 @@ class BaseTaskTests:
     Future = None
     all_tasks = None
 
-    def new_task(self, loop, coro, name='TestTask', context=None):
-        return self.__class__.Task(coro, loop=loop, name=name, context=context)
+    def new_task(self, loop, coro, name='TestTask', context=None, **kwargs):
+        return self.__class__.Task(coro, loop=loop, name=name, context=context, **kwargs)
 
     def new_future(self, loop):
         return self.__class__.Future(loop=loop)
@@ -2688,6 +2689,76 @@ def test_get_context(self):
         finally:
             loop.close()
 
+    def test_use_after_free_on_task_call_step_soon_with_ridiculous_setup(self):
+        # Special thanks to Nico-Posada for the original PoC.
+        # see: https://github.com/python/cpython/issues/126080
+        asserter = self
+
+        class Break:
+            def __str__(self):
+                # to break recursion errors in Task.__init__
+                raise ReachableCode(type(self))
+
+        class EvilEventLoop:
+            get_debug = staticmethod(lambda: False)
+            is_running = staticmethod(lambda: True)
+
+            def call_soon(self, *args, **kwargs):
+                # raise an exception just to make sure this was called
+                raise ReachableCode(type(self))
+
+            def __getattribute__(self, name):
+                if name == "call_soon":
+                    with asserter.assertRaises(ReachableCode) as cm:
+                        # The context must be set to `None` for it to use
+                        # Py_XSETREF instead of a plain regular assignment.
+                        evil_task.__init__(evil_coro, loop=self, name=Break())
+                    asserter.assertEqual(len(cm.exception.args), 1)
+                    asserter.assertIs(cm.exception.args[0], Break)
+                return object.__getattribute__(self, name)
+
+        class TaskWakeupCatch:
+            _asyncio_future_blocking = True
+            get_loop = staticmethod(lambda: evil_loop)
+            task_wakeup_method = None
+
+            def add_done_callback(self, callback, *args, **kwargs):
+                # Retrieve the 'task_wakeup' method of the Task object
+                # which is not accessible from pure Python code.
+                if self.task_wakeup_method is None:
+                    self.task_wakeup_method = callback
+
+        catcher = TaskWakeupCatch()
+
+        # We want a synchronous generator wrapped in a coroutine function
+        # and not an asynchronous generator defined via 'async def'.
+        async def evil_coroutine():
+            @types.coroutine
+            def sync_generator():
+                # ensure to keep catcher alive after the first send() call
+                nonlocal catcher
+                while 1:
+                    yield catcher
+            await sync_generator()
+
+        evil_coro = evil_coroutine()
+        evil_loop = EvilEventLoop()
+
+        self.assertIsNone(catcher.task_wakeup_method)
+        evil_task = self.new_task(evil_loop, evil_coro, eager_start=True)
+        self.assertIsInstance(catcher.task_wakeup_method, types.BuiltinMethodType)
+
+        with asserter.assertRaises(ReachableCode) as cm:
+            evil_task.__init__(evil_coro, loop=evil_loop, name=Break())
+        self.assertEqual(len(cm.exception.args), 1)
+        self.assertIs(cm.exception.args[0], Break)
+
+        self.assertIsInstance(catcher.task_wakeup_method, types.BuiltinMethodType)
+        with self.assertRaises(ReachableCode) as cm:
+            catcher.task_wakeup_method(mock.Mock())
+        self.assertEqual(len(cm.exception.args), 1)
+        self.assertIs(cm.exception.args[0], EvilEventLoop)
+
 
 def add_subclass_tests(cls):
     BaseTask = cls.Task
@@ -2863,6 +2934,9 @@ class PyTask_CFutureSubclass_Tests(BaseTaskTests, test_utils.TestCase):
     Task = tasks._PyTask
     all_tasks = tasks._py_all_tasks
 
+    def test_use_after_free_on_task_call_step_soon_with_ridiculous_setup(self):
+        self.skipTest("Python implementation is safe")
+
 
 @unittest.skipUnless(hasattr(tasks, '_CTask'),
                      'requires the C _asyncio module')
@@ -2881,6 +2955,9 @@ class PyTask_CFuture_Tests(BaseTaskTests, test_utils.TestCase):
     Future = getattr(futures, '_CFuture', None)
     all_tasks = staticmethod(tasks._py_all_tasks)
 
+    def test_use_after_free_on_task_call_step_soon_with_ridiculous_setup(self):
+        self.skipTest("Python implementation is safe")
+
 
 class PyTask_PyFuture_Tests(BaseTaskTests, SetMethodsTest,
                             test_utils.TestCase):
@@ -2889,12 +2966,18 @@ class PyTask_PyFuture_Tests(BaseTaskTests, SetMethodsTest,
     Future = futures._PyFuture
     all_tasks = staticmethod(tasks._py_all_tasks)
 
+    def test_use_after_free_on_task_call_step_soon_with_ridiculous_setup(self):
+        self.skipTest("Python implementation is safe")
+
 
 @add_subclass_tests
 class PyTask_PyFuture_SubclassTests(BaseTaskTests, test_utils.TestCase):
     Task = tasks._PyTask
     Future = futures._PyFuture
 
+    def test_use_after_free_on_task_call_step_soon_with_ridiculous_setup(self):
+        self.skipTest("Python implementation is safe")
+
 
 @unittest.skipUnless(hasattr(tasks, '_CTask'),
                      'requires the C _asyncio module')

Misc/NEWS.d/next/Library/2024-10-29-10-38-28.gh-issue-126080.qKRBuo.rst

@@ -0,0 +1,3 @@
+Fix a use-after-free crash on :class:`asyncio.Task` objects for which the
+underlying event loop implements an evil :meth:`~object.__getattribute__`.
+Reported by Nico-Posada. Patch by Bénédikt Tran.

Modules/_asynciomodule.c

@@ -2738,7 +2738,11 @@ task_call_step_soon(asyncio_state *state, TaskObj *task, PyObject *arg)
         return -1;
     }
 
+    // Beware: An evil call_soon could alter task_context.
+    // See: https://github.com/python/cpython/issues/126080.
+    PyObject *task_context = Py_NewRef(task->task_context);
     int ret = call_soon(state, task->task_loop, cb, NULL, task->task_context);
+    Py_DECREF(task_context);
     Py_DECREF(cb);
     return ret;
 }