optimize_cfg() can use freed constants during list mutation

[Nievesjyl]

Description

_testinternalcapi.optimize_cfg() accepts a caller-owned constants list and passes it through the flowgraph optimizer. Multiple optimizer paths use PyList_GET_ITEM() without the list critical section and only later promote or dereference the borrowed result. Concurrent clear, resize, or replacement can destroy an entry before _PyCfg_OptimizeCodeUnit(), get_const_value(), or remove_unused_consts() finishes using it.

Observed Behavior

One thread repeatedly cleared and repopulated the constants list while another invoked optimize_cfg() on a stable instruction sequence. The free-threaded ASan build crashed in under one second in _PyCfg_OptimizeCodeUnit() at Python/flowgraph.c:3681. With the GIL enabled, the same binary completed 96,316 optimizations in four seconds without a sanitizer failure.

Affected Version

CPython 3.14.7 at commit 823f0323ee6ec1402088b73bce1a38473cac36dc, tested with a free-threaded ASan build.

Reproduction

ASAN_OPTIONS=abort_on_error=1:detect_leaks=0 PYTHON_GIL=0 python3.14 poc/reproduce.py --seconds 5

Setting PYTHON_GIL=1 provides the GIL-enabled control.

PoC Source Code

poc/reproduce.py

import _testinternalcapi
import argparse
import ast
import threading
import time


parser = argparse.ArgumentParser()
parser.add_argument("--seconds", type=float, default=5.0)
args = parser.parse_args()

sequence, metadata = _testinternalcapi.compiler_codegen(
    ast.parse("x = 1", mode="exec"), "<concurdep-probe>", 0
)
consts = metadata["consts"]
stop = threading.Event()
optimizations = 0


class Payload:
    __slots__ = ("data",)

    def __init__(self):
        self.data = bytearray(4096)


def mutate():
    while not stop.is_set():
        consts.clear()
        consts.extend(Payload() for _ in range(64))


def optimize():
    global optimizations
    while not stop.is_set():
        try:
            _testinternalcapi.optimize_cfg(sequence, consts, 0)
        except (IndexError, SystemError, ValueError):
            pass
        optimizations += 1


threads = [threading.Thread(target=mutate), threading.Thread(target=optimize)]
for thread in threads:
    thread.start()
time.sleep(args.seconds)
stop.set()
for thread in threads:
    thread.join()
print("optimizations:", optimizations)

[ZeroIntensity]

Refer to #157124 (comment)