Nested interpreter JSON grows quadratically in DiskCache #204

#!/usr/bin/env python3
import hashlib
import json
import shutil
import subprocess
import tempfile
from pathlib import Path

VENV_PYTHON = "/opt/venv/bin/python"  # has python-discovery installed from /src
COLLECTOR = "/src/src/python_discovery/_py_info_collect.py"
work = Path(tempfile.mkdtemp(prefix="indent-amplification-"))


def install_candidate(name):
    candidate = work / name / "bin" / "python3.99"
    candidate.parent.mkdir(parents=True)
    shutil.copy(f"/tmp/repro.d/candidate_{name}.py", candidate)
    candidate.chmod(0o755)
    return candidate


def framed_answer(candidate):
    return subprocess.run([candidate, COLLECTOR, "s" * 32, "e" * 32], capture_output=True, check=True).stdout


def lookup(candidate, cache_root):
    command = [VENV_PYTHON, "/tmp/repro.d/find_python.py", str(candidate.parent), str(cache_root)]
    result = json.loads(subprocess.run(command, capture_output=True, text=True, check=True).stdout)
    for argv in result["spawned"]:
        print("  spawned ", *argv)
    print(f"  returned {result['returned']}, caller peak RSS {result['peak_rss_kib'] // 1024} MiB")
    return result


def cache_entry_size(candidate, cache_root):
    # DiskCache keeps it at <root>/py_info/4/<sha256 of the interpreter path>.json
    key = hashlib.sha256(str(candidate).encode()).hexdigest()
    return (cache_root / "py_info" / "4" / f"{key}.json").stat().st_size


# set up: each candidate alone on its own PATH directory, each with its own empty cache
# ("flat" and "deep" have the same length: the candidate's sys.path is part of its answer)
flat, deep = install_candidate("flat"), install_candidate("deep")
flat_cache, deep_cache = work / "flat" / "cache", work / "deep" / "cache"
flat_answer, deep_answer = framed_answer(flat), framed_answer(deep)
print(f"flat candidate answers {flat_answer[:50]} ... {flat_answer[-50:]}")
print(f"deep candidate answers {deep_answer[:50]} ... {deep_answer[-50:]}")

# control: padding is one 32000-byte string
print("lookup, flat candidate, empty cache")
flat_lookup = lookup(flat, flat_cache)
flat_entry = cache_entry_size(flat, flat_cache)

# attack: padding is 16000 nested empty lists, also 32000 bytes
print("lookup, deep candidate, empty cache")
deep_lookup = lookup(deep, deep_cache)
deep_entry = cache_entry_size(deep, deep_cache)

# the entry stays: the next lookup reads it instead of running the candidate
print("lookup again, deep candidate, the cache the first lookup left")
second_lookup = lookup(deep, deep_cache)
deep_spawned_again = any(argv[0] == str(deep) for argv in second_lookup["spawned"])
shutil.rmtree(work)

# check: a correct cache entry is about as big as the answer it stores
flat_rss, deep_rss, second_rss = (r["peak_rss_kib"] // 1024 for r in (flat_lookup, deep_lookup, second_lookup))
print(f"CONTROL  flat: answer {len(flat_answer)} B -> cache entry {flat_entry} B, caller peak RSS {flat_rss} MiB")
print(f"OBSERVED deep: answer {len(deep_answer)} B -> cache entry {deep_entry} B, caller peak RSS {deep_rss} MiB")
print(f"OBSERVED deep, second lookup: candidate spawned {deep_spawned_again}, caller peak RSS {second_rss} MiB")

same_size_answers = len(flat_answer) == len(deep_answer)
control_entry_ok = flat_entry < 2 * len(flat_answer)
deep_entry_amplified = deep_entry > 1000 * len(deep_answer)
deep_memory_amplified = deep_rss > 10 * flat_rss
paid_again_from_cache = not deep_spawned_again and second_rss > 10 * flat_rss
print(
    "BUG"
    if same_size_answers
    and control_entry_ok
    and deep_entry_amplified
    and deep_memory_amplified
    and paid_again_from_cache
    else "OK"
)

#!/usr/local/bin/python3.15
import json
import os
import sys

collector, start_cookie, end_cookie = sys.argv[1:4]
sys.path.insert(0, os.path.dirname(collector))
from _py_info_collect import PythonInfoCollector

padding = json.loads("[" * 16000 + "]" * 16000)  # 16000 nested empty lists, 32000 bytes as JSON

answer = PythonInfoCollector().to_dict()
answer["padding"] = padding
sys.stdout.write(start_cookie[::-1] + json.dumps(answer) + end_cookie[::-1])

#!/usr/local/bin/python3.15
import json
import os
import sys

collector, start_cookie, end_cookie = sys.argv[1:4]
sys.path.insert(0, os.path.dirname(collector))
from _py_info_collect import PythonInfoCollector

padding = "x" * 31998  # 32000 bytes as JSON, with the quotes

answer = PythonInfoCollector().to_dict()
answer["padding"] = padding
sys.stdout.write(start_cookie[::-1] + json.dumps(answer) + end_cookie[::-1])

import json
import resource
import sys
from pathlib import Path

spawned = []
sys.addaudithook(lambda event, args: spawned.append([str(a) for a in args[1]]) if event == "subprocess.Popen" else None)

from python_discovery import DiskCache, get_interpreter

path_dir, cache_root = sys.argv[1:3]
found = get_interpreter("python3.99", env={"PATH": path_dir}, cache=DiskCache(Path(cache_root)))

print(
    json.dumps({
        "spawned": spawned,
        "returned": found and str(found.executable),
        "peak_rss_kib": resource.getrusage(resource.RUSAGE_SELF).ru_maxrss,
    })
)