Unpickler can use freed extension-registry entries during concurrent replacement

[Nievesjyl]

Description

_pickle.load_extension() retrieves a borrowed (module_name, class_name) tuple from the public copyreg._inverted_registry and uses that tuple and its fields without locking or ownership promotion. Concurrent entry replacement can reclaim all three objects during validation or class resolution.

Observed Behavior

On a free-threaded build, repeatedly installing fresh but textually equal registry entries while workers decode the same EXT1 payload terminates with SIGSEGV in about 0.28 seconds. The compatibility-GIL control completes normally.

Affected Version

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

Reproduction

Run:

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

PoC Source Code

poc/reproduce.py:

#!/usr/bin/env python3
"""Race EXT1 loading with fresh-but-equal inverted-registry entries."""

import copyreg
import pickle
import sys
import threading
import time


CODE = 255
PAYLOAD = b"\x80\x02\x82\xff."
STOP = threading.Event()
FAILURES = []


class Target:
    pass


copyreg.add_extension(__name__, "Target", CODE)


def mutate():
    while not STOP.is_set():
        copyreg.clear_extension_cache()
        copyreg._inverted_registry[CODE] = (
            "".join(("__", "main__")),
            "".join(("Tar", "get")),
        )


def load_worker(deadline):
    count = 0
    try:
        while time.monotonic() < deadline and not STOP.is_set():
            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)]
    threads += [threading.Thread(target=load_worker, args=(deadline,)) for _ in range(8)]
    for thread in threads:
        thread.start()
    for thread in threads[1:]:
        thread.join()
    STOP.set()
    threads[0].join()
    print("gil=", sys._is_gil_enabled(), "failures=", FAILURES)
    return bool(FAILURES)


raise SystemExit(main())

[ZeroIntensity]

Refer to #157124 (comment)