pickle.dumps() can copy resized bytearray storage under free threading

[Nievesjyl]

Description

For protocol 5, _pickle passes separately loaded bytearray address and length values to its payload writer without holding the bytearray lock or acquiring a buffer export. A concurrent resize can replace the storage or make the pointer and length describe different generations during the copy.

Observed Behavior

On a free-threaded build, four workers fail on their first iteration with nominal 32 MiB results containing the original A prefix and zero-filled suffix, although the source only holds complete A, complete B, or empty states. An earlier run also ended with SIGSEGV. 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 15 33554432

PoC Source Code

poc/reproduce.py:

#!/usr/bin/env python3
"""Detect torn pickle payloads while an exact bytearray is resized."""

import pickle
import sys
import threading
import time


SIZE = int(sys.argv[2]) if len(sys.argv) > 2 else 32 * 1024 * 1024
SMALL = 16
A_BYTES = b"A" * SIZE
B_BYTES = b"B" * SMALL
DATA = bytearray(A_BYTES)
STOP = threading.Event()
FAILURES = []


def mutate():
    while not STOP.is_set():
        DATA.clear()
        DATA.extend(B_BYTES)
        DATA.clear()
        DATA.extend(A_BYTES)


def pickle_worker(deadline):
    count = 0
    try:
        while time.monotonic() < deadline and not STOP.is_set():
            payload = pickle.dumps(DATA, protocol=5)
            restored = pickle.loads(payload)
            if len(restored) not in {0, SMALL, SIZE}:
                raise AssertionError(("torn-length", len(restored), SMALL, SIZE))
            if len(restored) == SMALL and restored != B_BYTES:
                raise AssertionError(("torn-small", restored))
            if len(restored) == SIZE and restored != A_BYTES:
                raise AssertionError(("torn-large", restored[:32], restored[-32:]))
            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=pickle_worker, args=(deadline,)) for _ in range(4)]
    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)