19
19
from asyncio import tasks
20
20
from test .test_asyncio import utils as test_utils
21
21
from test import support
22
- from test .support import ReachableCode
23
22
from test .support .script_helper import assert_python_ok
24
23
from test .support .warnings_helper import ignore_warnings
25
24
@@ -2711,76 +2710,6 @@ def __str__(self):
2711
2710
2712
2711
self .assertEqual (sys .getrefcount (obj ), initial_refcount )
2713
2712
2714
- def test_use_after_free_on_task_call_step_soon_with_ridiculous_setup (self ):
2715
- # Special thanks to Nico-Posada for the original PoC.
2716
- # see: https://github.com/python/cpython/issues/126080
2717
- asserter = self
2718
-
2719
- class Break :
2720
- def __str__ (self ):
2721
- # to break recursion errors in Task.__init__
2722
- raise ReachableCode (type (self ))
2723
-
2724
- class EvilEventLoop :
2725
- get_debug = staticmethod (lambda : False )
2726
- is_running = staticmethod (lambda : True )
2727
-
2728
- def call_soon (self , * args , ** kwargs ):
2729
- # raise an exception just to make sure this was called
2730
- raise ReachableCode (type (self ))
2731
-
2732
- def __getattribute__ (self , name ):
2733
- if name == "call_soon" :
2734
- with asserter .assertRaises (ReachableCode ) as cm :
2735
- # The context must be set to `None` for it to use
2736
- # Py_XSETREF instead of a plain regular assignment.
2737
- evil_task .__init__ (evil_coro , loop = self , name = Break ())
2738
- asserter .assertEqual (len (cm .exception .args ), 1 )
2739
- asserter .assertIs (cm .exception .args [0 ], Break )
2740
- return object .__getattribute__ (self , name )
2741
-
2742
- class TaskWakeupCatch :
2743
- _asyncio_future_blocking = True
2744
- get_loop = staticmethod (lambda : evil_loop )
2745
- task_wakeup_method = None
2746
-
2747
- def add_done_callback (self , callback , * args , ** kwargs ):
2748
- # Retrieve the 'task_wakeup' method of the Task object
2749
- # which is not accessible from pure Python code.
2750
- if self .task_wakeup_method is None :
2751
- self .task_wakeup_method = callback
2752
-
2753
- catcher = TaskWakeupCatch ()
2754
-
2755
- # We want a synchronous generator wrapped in a coroutine function
2756
- # and not an asynchronous generator defined via 'async def'.
2757
- async def evil_coroutine ():
2758
- @types .coroutine
2759
- def sync_generator ():
2760
- # ensure to keep catcher alive after the first send() call
2761
- nonlocal catcher
2762
- while 1 :
2763
- yield catcher
2764
- await sync_generator ()
2765
-
2766
- evil_coro = evil_coroutine ()
2767
- evil_loop = EvilEventLoop ()
2768
-
2769
- self .assertIsNone (catcher .task_wakeup_method )
2770
- evil_task = self .new_task (evil_loop , evil_coro , eager_start = True )
2771
- self .assertIsInstance (catcher .task_wakeup_method , types .BuiltinMethodType )
2772
-
2773
- with asserter .assertRaises (ReachableCode ) as cm :
2774
- evil_task .__init__ (evil_coro , loop = evil_loop , name = Break ())
2775
- self .assertEqual (len (cm .exception .args ), 1 )
2776
- self .assertIs (cm .exception .args [0 ], Break )
2777
-
2778
- self .assertIsInstance (catcher .task_wakeup_method , types .BuiltinMethodType )
2779
- with self .assertRaises (ReachableCode ) as cm :
2780
- catcher .task_wakeup_method (mock .Mock ())
2781
- self .assertEqual (len (cm .exception .args ), 1 )
2782
- self .assertIs (cm .exception .args [0 ], EvilEventLoop )
2783
-
2784
2713
2785
2714
def add_subclass_tests (cls ):
2786
2715
BaseTask = cls .Task
@@ -2956,9 +2885,6 @@ class PyTask_CFutureSubclass_Tests(BaseTaskTests, test_utils.TestCase):
2956
2885
Task = tasks ._PyTask
2957
2886
all_tasks = tasks ._py_all_tasks
2958
2887
2959
- def test_use_after_free_on_task_call_step_soon_with_ridiculous_setup (self ):
2960
- self .skipTest ("Python implementation is safe" )
2961
-
2962
2888
2963
2889
@unittest .skipUnless (hasattr (tasks , '_CTask' ),
2964
2890
'requires the C _asyncio module' )
@@ -2977,9 +2903,6 @@ class PyTask_CFuture_Tests(BaseTaskTests, test_utils.TestCase):
2977
2903
Future = getattr (futures , '_CFuture' , None )
2978
2904
all_tasks = staticmethod (tasks ._py_all_tasks )
2979
2905
2980
- def test_use_after_free_on_task_call_step_soon_with_ridiculous_setup (self ):
2981
- self .skipTest ("Python implementation is safe" )
2982
-
2983
2906
2984
2907
class PyTask_PyFuture_Tests (BaseTaskTests , SetMethodsTest ,
2985
2908
test_utils .TestCase ):
@@ -2988,18 +2911,12 @@ class PyTask_PyFuture_Tests(BaseTaskTests, SetMethodsTest,
2988
2911
Future = futures ._PyFuture
2989
2912
all_tasks = staticmethod (tasks ._py_all_tasks )
2990
2913
2991
- def test_use_after_free_on_task_call_step_soon_with_ridiculous_setup (self ):
2992
- self .skipTest ("Python implementation is safe" )
2993
-
2994
2914
2995
2915
@add_subclass_tests
2996
2916
class PyTask_PyFuture_SubclassTests (BaseTaskTests , test_utils .TestCase ):
2997
2917
Task = tasks ._PyTask
2998
2918
Future = futures ._PyFuture
2999
2919
3000
- def test_use_after_free_on_task_call_step_soon_with_ridiculous_setup (self ):
3001
- self .skipTest ("Python implementation is safe" )
3002
-
3003
2920
3004
2921
@unittest .skipUnless (hasattr (tasks , '_CTask' ),
3005
2922
'requires the C _asyncio module' )
0 commit comments