gh-142615: disallow multiple initializations of asyncio tasks and futures

Lib/asyncio/futures.py

@@ -79,6 +79,10 @@ def __init__(self, *, loop=None):
         loop object used by the future. If it's not provided, the future uses
         the default event loop.
         """
+        if self._loop is not None:
+            raise RuntimeError(f"{self.__class__.__name__} object is already "
+                                "initialized")
+
         if loop is None:
             self._loop = events.get_event_loop()
         else:

Lib/test/test_asyncio/test_futures.py

@@ -750,6 +750,10 @@ def test_future_cancelled_exception_refcycles(self):
         self.assertIsNotNone(exc)
         self.assertListEqual(gc.get_referrers(exc), [])
 
+    def test_future_disallow_multiple_initialization(self):
+        f = self._new_future(loop=self.loop)
+        with self.assertRaises(RuntimeError, msg="is already initialized"):
+            f.__init__(loop=self.loop)
 
 @unittest.skipUnless(hasattr(futures, '_CFuture'),
                      'requires the C _asyncio module')
@@ -1091,33 +1095,6 @@ def __getattribute__(self, name):
             fut.add_done_callback(fut_callback_0)
             self.assertRaises(ReachableCode, fut.set_result, "boom")
 
-    def test_use_after_free_on_fut_context_0_with_evil__getattribute__(self):
-        # see: https://github.com/python/cpython/issues/125984
-
-        class EvilEventLoop(SimpleEvilEventLoop):
-            def call_soon(self, *args, **kwargs):
-                super().call_soon(*args, **kwargs)
-                raise ReachableCode
-
-            def __getattribute__(self, name):
-                if name == 'call_soon':
-                    # resets the future's event loop
-                    fut.__init__(loop=SimpleEvilEventLoop())
-                return object.__getattribute__(self, name)
-
-        evil_loop = EvilEventLoop()
-        with mock.patch.object(self, 'loop', evil_loop):
-            fut = self._new_future()
-            self.assertIs(fut.get_loop(), evil_loop)
-
-            fut_callback_0 = mock.Mock()
-            fut_context_0 = mock.Mock()
-            fut.add_done_callback(fut_callback_0, context=fut_context_0)
-            del fut_context_0
-            del fut_callback_0
-            self.assertRaises(ReachableCode, fut.set_result, "boom")
-
-
 @unittest.skipUnless(hasattr(futures, '_CFuture'),
                      'requires the C _asyncio module')
 class CFutureDoneCallbackTests(BaseFutureDoneCallbackTests,

Lib/test/test_asyncio/test_tasks.py

@@ -2776,28 +2776,17 @@ def test_get_context(self):
         finally:
             loop.close()
 
-    def test_proper_refcounts(self):
-        # see: https://github.com/python/cpython/issues/126083
-        class Break:
-            def __str__(self):
-                raise RuntimeError("break")
-
-        obj = object()
-        initial_refcount = sys.getrefcount(obj)
-
-        coro = coroutine_function()
-        with contextlib.closing(asyncio.EventLoop()) as loop:
-            task = asyncio.Task.__new__(asyncio.Task)
-            for _ in range(5):
-                with self.assertRaisesRegex(RuntimeError, 'break'):
-                    task.__init__(coro, loop=loop, context=obj, name=Break())
-
-            coro.close()
-            task._log_destroy_pending = False
-            del task
+    def test_task_disallow_multiple_initialization(self):
+        async def foo():
+            pass
 
-            self.assertEqual(sys.getrefcount(obj), initial_refcount)
+        coro = foo()
+        self.addCleanup(coro.close)
+        task = self.new_task(self.loop, coro)
+        task._log_destroy_pending = False
 
+        with self.assertRaises(RuntimeError, msg="is already initialized"):
+            task.__init__(coro, loop=self.loop)
 
 def add_subclass_tests(cls):
     BaseTask = cls.Task
@@ -2921,19 +2910,6 @@ class CTask_CFuture_Tests(BaseTaskTests, SetMethodsTest,
     all_tasks = getattr(tasks, '_c_all_tasks', None)
     current_task = staticmethod(getattr(tasks, '_c_current_task', None))
 
-    @support.refcount_test
-    def test_refleaks_in_task___init__(self):
-        gettotalrefcount = support.get_attribute(sys, 'gettotalrefcount')
-        async def coro():
-            pass
-        task = self.new_task(self.loop, coro())
-        self.loop.run_until_complete(task)
-        refs_before = gettotalrefcount()
-        for i in range(100):
-            task.__init__(coro(), loop=self.loop)
-            self.loop.run_until_complete(task)
-        self.assertAlmostEqual(gettotalrefcount() - refs_before, 0, delta=10)
-
     def test_del__log_destroy_pending_segfault(self):
         async def coro():
             pass

Misc/NEWS.d/next/Library/2025-12-12-08-51-29.gh-issue-142615.GoJ6el.rst

@@ -0,0 +1,3 @@
+Fix possible crashes when initializing :class:`asyncio.Task` or :class:`asyncio.Future` multiple times.
+These classes can now be initialized only once and any subsequent initialization attempt will raise a RuntimeError.
+Patch by Kumar Aditya.

Modules/_asynciomodule.c

@@ -498,21 +498,13 @@ future_schedule_callbacks(asyncio_state *state, FutureObj *fut)
 static int
 future_init(FutureObj *fut, PyObject *loop)
 {
+    if (fut->fut_loop != NULL) {
+        PyErr_Format(PyExc_RuntimeError, "%T object is already initialized", fut);
+        return -1;
+    }
+
     PyObject *res;
     int is_true;
-
-    Py_CLEAR(fut->fut_loop);
-    Py_CLEAR(fut->fut_callback0);
-    Py_CLEAR(fut->fut_context0);
-    Py_CLEAR(fut->fut_callbacks);
-    Py_CLEAR(fut->fut_result);
-    Py_CLEAR(fut->fut_exception);
-    Py_CLEAR(fut->fut_exception_tb);
-    Py_CLEAR(fut->fut_source_tb);
-    Py_CLEAR(fut->fut_cancel_msg);
-    Py_CLEAR(fut->fut_cancelled_exc);
-    Py_CLEAR(fut->fut_awaited_by);
-
     fut->fut_state = STATE_PENDING;
     fut->fut_log_tb = 0;
     fut->fut_blocking = 0;