Changing prefix spellings can loop during base resolution #190

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

PYTHON = "/opt/venv/bin/python"  # the image's install of python-discovery
PROBE_TIMEOUT = 2  # PY_DISCOVERY_TIMEOUT; a candidate that never answers is probed twice, so costs at most 4 s
PATIENCE = 10 * PROBE_TIMEOUT
REPRO_D = Path("/tmp/repro.d")
work = Path(tempfile.mkdtemp())


def find_python_with_only(candidate_source, name):
    """Install candidate_source as <work>/<name>/bin/python3.99, the one file on PATH, and run the caller on it."""
    candidate = work / name / "bin" / "python3.99"
    candidate.parent.mkdir(parents=True)
    shutil.copy(candidate_source, candidate)
    candidate.chmod(0o755)
    cache = work / name / "cache"
    try:
        caller = subprocess.run(
            [PYTHON, REPRO_D / "trace_spawns.py", REPRO_D / "find_python.py", cache],
            env={"PATH": str(candidate.parent), "PY_DISCOVERY_TIMEOUT": str(PROBE_TIMEOUT)},
            stdout=subprocess.PIPE,
            timeout=PATIENCE,
            check=True,
        )
        output, still_running = caller.stdout, False
    except subprocess.TimeoutExpired as expired:
        output, still_running = expired.stdout, True
    return candidate, output.decode().splitlines(), still_running, cache


# control: the candidate spells its real_prefix the same way in every answer
candidate, output, _, _ = find_python_with_only(REPRO_D / "candidate_same_prefix_spelling.py", "control")
for line in output:
    print("CONTROL ", line[:140])
control_returned = f"RETURNED {candidate}" in output

# attack: the candidate spells its real_prefix differently in every answer
candidate, output, still_running, cache = find_python_with_only(
    REPRO_D / "candidate_new_prefix_spelling_each_time.py", "attack"
)
for line in output[:4]:
    print("OBSERVED", line[:140])
print("OBSERVED …")

# check: the attack call must end like the control, after a couple of probes of the one file
attack_returned = any(line.startswith("RETURNED") for line in output)
spawned = [line.split()[1] for line in output if line.startswith("SPAWNED")]
paths_to_candidate = {exe for exe in spawned if os.path.samefile(exe, candidate)}
cache_entries = list(cache.rglob("*.json"))
print(f"CONTROL  get_interpreter() returned the candidate: {control_returned}")
print(f"OBSERVED get_interpreter() returned: {attack_returned}; still running after {PATIENCE} s: {still_running}")
print(
    f"OBSERVED {len(spawned)} processes spawned, {len(paths_to_candidate)} distinct paths that are the candidate file"
)
print(f"OBSERVED {len(cache_entries)} cache entries, {sum(entry.stat().st_size for entry in cache_entries)} bytes")
shutil.rmtree(work)
# 100: far more than any real chain of virtual environments
print("BUG" if control_returned and still_running and not attack_returned and len(paths_to_candidate) > 100 else "OK")

#!/usr/bin/python3
"""The attacker's candidate, installed as python3.99 on PATH.

It answers every interrogation at once, like a legacy virtualenv: no system_executable, and a real_prefix to look in.
The real_prefix is its own directory, spelled differently in every answer.
"""

import json
import os
import random
import sys

start_cookie, end_cookie = sys.argv[2:4]
my_dir = os.path.dirname(os.path.realpath(sys.argv[0]))
parent, leaf = os.path.split(my_dir)
# both lead back to my_dir; "..", because pathlib (which keys the in-memory cache) folds "/./" and "//" but keeps ".."
detours = [os.path.join("..", leaf), os.path.join("..", "..", os.path.basename(parent), leaf)]

answer = {
    "version_info": {"major": 3, "minor": 99, "micro": 0, "releaselevel": "final", "serial": 0},
    "implementation": "CPython",
    "architecture": 64,
    "sysconfig_platform": "linux-x86_64",
    "free_threaded": False,
    "debug_build": False,
    "executable": sys.argv[0],
    "original_executable": sys.argv[0],
    "system_executable": None,
    "prefix": my_dir,
    "base_prefix": None,
    "real_prefix": os.path.join(my_dir, *random.choices(detours, k=64)),  # 2**64 spellings
}
# the framing _py_info_collect.py uses
sys.stdout.write(start_cookie[::-1] + json.dumps(answer) + end_cookie[::-1])

#!/usr/bin/python3
"""The control candidate, installed as python3.99 on PATH.

Identical to candidate_new_prefix_spelling_each_time.py, except that real_prefix is spelled the same in every answer.
"""

import json
import os
import sys

start_cookie, end_cookie = sys.argv[2:4]
my_dir = os.path.dirname(os.path.realpath(sys.argv[0]))
parent, leaf = os.path.split(my_dir)
detours = [os.path.join("..", leaf), os.path.join("..", "..", os.path.basename(parent), leaf)]

answer = {
    "version_info": {"major": 3, "minor": 99, "micro": 0, "releaselevel": "final", "serial": 0},
    "implementation": "CPython",
    "architecture": 64,
    "sysconfig_platform": "linux-x86_64",
    "free_threaded": False,
    "debug_build": False,
    "executable": sys.argv[0],
    "original_executable": sys.argv[0],
    "system_executable": None,
    "prefix": my_dir,
    "base_prefix": None,
    "real_prefix": os.path.join(my_dir, *detours * 32),  # the same 64 detours every time
}
sys.stdout.write(start_cookie[::-1] + json.dumps(answer) + end_cookie[::-1])

"""The caller, as virtualenv or tox run it for `-p 3.99`: a version spec, the process environment, a disk cache."""

import os
import sys
from pathlib import Path

from python_discovery import DiskCache, get_interpreter

info = get_interpreter("3.99", env=os.environ, cache=DiskCache(Path(sys.argv[1])))
print("RETURNED", info and info.executable, flush=True)

"""Run the script in argv[1] with the remaining arguments, printing the argv of every process it starts."""

import runpy
import sys


def print_spawn(event, args):
    if event == "subprocess.Popen":
        print("SPAWNED", *args[1], flush=True)


sys.addaudithook(print_spawn)
sys.argv = sys.argv[1:]
runpy.run_path(sys.argv[0], run_name="__main__")