🐛 fix(cache): keep shim results out of shared cache

docs/changelog/232.security.rst

@@ -0,0 +1,2 @@
+Interpreter discovery no longer reuses pyenv shim results across projects. A version choice in one checkout cannot
+leave an unsupported-version verdict that blocks discovery in another project.

docs/explanation.rst

@@ -250,22 +250,27 @@ links back to its own interpreter. Reading the raw field still tells the two sta
 How caching works
 -------------------
 
-Querying an interpreter requires a subprocess call, which is slow. The cache avoids repeating this
-work by storing the result as a JSON file keyed by the interpreter's path.
+Querying an interpreter requires a subprocess call, which is slow. The cache stores the result as a
+JSON file keyed by the queried path when that path identifies the interpreter that answered. Launchers
+such as pyenv shims can select different interpreters in different directories, so their results are
+not cached under the launcher path.
 
 .. mermaid::
 
     flowchart TD
         Lookup["py_info(path)"] --> Exists{"Cache hit?"}
         Exists -->|Yes| Read["Read JSON"]
         Exists -->|No| Run["Run subprocess"]
-        Run --> Write["Write JSON<br>(with filelock)"]
-        Write --> Return["Return PythonInfo"]
+        Run --> Same{"Queried path is<br>the interpreter?"}
+        Same -->|Yes| Write["Write JSON<br>(with filelock)"]
+        Same -->|No| Return["Return query result"]
+        Write --> Return
         Read --> Return
 
         style Lookup fill:#4a90d9,stroke:#2a5f8f,color:#fff
         style Return fill:#4a9f4a,stroke:#2a6f2a,color:#fff
         style Run fill:#d9904a,stroke:#8f5f2a,color:#fff
+        style Same fill:#d9904a,stroke:#8f5f2a,color:#fff
 
 The built-in :class:`~python_discovery.DiskCache` stores files under ``<root>/py_info/4/<sha256>.json``
 with `filelock <https://py-filelock.readthedocs.io/>`_-based locking for safe concurrent access. You
@@ -318,9 +323,10 @@ Supported Python versions
 Two floors apply: python-discovery itself runs on Python 3.9+ (``requires-python``), while discovery
 reaches down to Python 3.6 - old enough to cover RHEL 8's system Python.
 
-python-discovery skips candidates below 3.6 with one warning naming the executable, the version
-found, and the floor; the verdict is cached, so the interpreter is not queried again. Candidates too
-old to run the version check at all (before Python 2.7) fail as a generic query error. An
+python-discovery skips candidates below 3.6 with a warning naming the executable, the version
+found, and the floor. The verdict is cached when the candidate itself is the interpreter; launchers
+such as pyenv shims are queried again because their target can depend on the current directory.
+Candidates too old to run the version check at all (before Python 2.7) fail as a generic query error. An
 absolute-path spec for a below-floor interpreter raises :class:`RuntimeError` with the same message.
 The floor moves only in a major release.
 

src/python_discovery/_cached_py_info.py

@@ -53,6 +53,10 @@ class CacheEntryMeta(TypedDict):
 class _UnsupportedInterpreterError(RuntimeError):
     """The interpreter is older than the interrogation script supports; retrying cannot help."""
 
+    def __init__(self, message: str, executable: str | None = None) -> None:
+        super().__init__(message)
+        self.executable = executable
+
 
 def from_exe(  # ruff:ignore[too-many-arguments]
     cls: type[PythonInfo],
@@ -94,13 +98,32 @@ def _get_from_cache(  # ruff:ignore[too-many-arguments]
     if not ignore_cache and key in _CACHE:
         result = _CACHE[key]
     else:
-        result = _CACHE[key] = _get_via_file_cache(cls, cache, key[0], exe, env, executable_args=executable_args)
+        result = _get_via_file_cache(cls, cache, key[0], exe, env, executable_args=executable_args)
+        if _cacheable_result(key[0], result):
+            _CACHE[key] = result
+        else:
+            _CACHE.pop(key, None)
     # with launcher arguments the queried binary is not the interpreter, so keep the one the query reported
     if isinstance(result, PythonInfo) and executable_args is None:
         result.executable = exe
     return result
 
 
+def _cacheable_result(path: Path, result: PythonInfo | Exception) -> bool:
+    if isinstance(result, PythonInfo):
+        executable = result.original_executable
+    elif isinstance(result, _UnsupportedInterpreterError):
+        executable = result.executable
+    else:
+        return False
+    if executable is None:
+        return False
+    try:
+        return path.samefile(executable)
+    except (OSError, TypeError, ValueError):
+        return False
+
+
 def _get_via_file_cache(  # ruff:ignore[too-many-arguments]
     cls: type[PythonInfo],
     cache: PyInfoCache | None,
@@ -130,26 +153,25 @@ def _get_via_file_cache(  # ruff:ignore[too-many-arguments]
     }
     with py_info_store.locked():
         cached = _read_cache_entry(cls, py_info_store, entry_meta)
-        if isinstance(cached, _UnsupportedInterpreterError):
+        if cached is not None:
             return cached
-        py_info = cached
-        if py_info is None:
+        failure, py_info = _run_subprocess(cls, exe, env, executable_args)
+        if failure is not None and not isinstance(failure, _UnsupportedInterpreterError):
+            _LOGGER.debug("first subprocess attempt failed for %s (%s), retrying", exe, failure)
             failure, py_info = _run_subprocess(cls, exe, env, executable_args)
-            if failure is not None and not isinstance(failure, _UnsupportedInterpreterError):
-                _LOGGER.debug("first subprocess attempt failed for %s (%s), retrying", exe, failure)
-                failure, py_info = _run_subprocess(cls, exe, env, executable_args)
-            if isinstance(failure, _UnsupportedInterpreterError):
-                _LOGGER.warning("%s", failure)  # the verdict is permanent, warn once and remember it
+        if isinstance(failure, _UnsupportedInterpreterError):
+            _LOGGER.warning("%s", failure)
+            if _cacheable_result(path, failure):
                 py_info_store.write({**entry_meta, "unsupported": str(failure)})
-                return failure
-            if failure is not None:
-                return failure
-            if py_info is not None:
-                py_info_store.write({**entry_meta, "content": py_info.to_dict()})
-    if py_info is None:
-        msg = f"{exe} failed to produce interpreter info"
-        return RuntimeError(msg)
-    return py_info
+            return failure
+        if failure is not None:
+            return failure
+        if py_info is None:
+            msg = f"{exe} failed to produce interpreter info"
+            return RuntimeError(msg)
+        if _cacheable_result(path, py_info):
+            py_info_store.write({**entry_meta, "content": py_info.to_dict()})
+        return py_info
 
 
 def _read_cache_entry(
@@ -161,7 +183,7 @@ def _read_cache_entry(
         return None
     if isinstance(data, dict) and all(data.get(key) == value for key, value in entry_meta.items()):
         if isinstance(unsupported := data.get("unsupported"), str):
-            return _UnsupportedInterpreterError(unsupported)
+            return _UnsupportedInterpreterError(unsupported, entry_meta["path"])
         if isinstance(content := data.get("content"), dict):
             return _load_cached_py_info(cls, py_info_store, content)
     py_info_store.remove()
@@ -374,10 +396,12 @@ def _query_failure(exe: str, out: str, err: str | None, code: int | None) -> Run
     # the gate's exit code and stderr marker must both be present: either alone can collide, an errno of 79 from a
     # failed exec or a shim that echoes the marker phrase, but only the gate produces the combination
     if code == UNSUPPORTED_EXIT_CODE and err is not None and (at := err.find(UNSUPPORTED_MARKER)) != -1:
-        version = err[at + len(UNSUPPORTED_MARKER) :].split() or ["unknown"]
+        details = err[at + len(UNSUPPORTED_MARKER) :].splitlines()
+        version = details[0].split() if details else []
         floor = ".".join(str(i) for i in MIN_INTERROGATE_VERSION)
-        msg = f"{exe} is Python {version[0]}, older than the minimum {floor} python-discovery can query"
-        return _UnsupportedInterpreterError(msg)
+        found = version[0] if version else "unknown"
+        msg = f"{exe} is Python {found}, older than the minimum {floor} python-discovery can query"
+        return _UnsupportedInterpreterError(msg, details[1].strip() if len(details) > 1 else None)
     msg = f"{exe} with code {code}{f' out: {out!r}' if out else ''}{f' err: {err!r}' if err else ''}"
     return RuntimeError(f"failed to query {msg}")
 

src/python_discovery/_py_info_collect.py

@@ -14,7 +14,7 @@
 UNSUPPORTED_MARKER = "unsupported Python version "
 
 if __name__ == "__main__" and sys.version_info[:2] < MIN_INTERROGATE_VERSION:  # pragma: no cover # needs <3.6
-    sys.stderr.write(UNSUPPORTED_MARKER + ".".join(str(i) for i in sys.version_info[:3]))
+    sys.stderr.write(UNSUPPORTED_MARKER + ".".join(str(i) for i in sys.version_info[:3]) + "\n" + sys.executable)
     raise SystemExit(UNSUPPORTED_EXIT_CODE)
 
 import json

tests/py_info/test_py_info.py

@@ -141,7 +141,7 @@ def test_py_info_cached_error(mocker: MockerFixture, tmp_path: Path, session_cac
         PythonInfo.from_exe(str(tmp_path), session_cache)
     with pytest.raises(RuntimeError):
         PythonInfo.from_exe(str(tmp_path), session_cache)
-    assert spy.call_count == 2
+    assert spy.call_count == 4
 
 
 def test_py_info_cache_clear(mocker: MockerFixture, session_cache: DiskCache) -> None:

tests/test_cached_py_info.py

@@ -344,13 +344,54 @@ def test_from_exe_too_old_python_raises_by_default(tmp_path: Path) -> None:
 def test_from_exe_too_old_python_verdict_cached(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
     caplog.set_level(logging.INFO)
     cache = DiskCache(tmp_path / "cache")
-    exe = _write_fake_python(tmp_path, stderr="unsupported Python version 3.5.10")
+    exe = _write_fake_python(tmp_path, stderr="unsupported Python version 3.5.10\n$0")
     assert PythonInfo.from_exe(str(exe), cache, raise_on_error=False, ignore_cache=True) is None
     assert PythonInfo.from_exe(str(exe), cache, raise_on_error=False, ignore_cache=True) is None
     assert len(_interrogations(exe)) == 1
     assert len([record for record in caplog.records if record.levelno == logging.WARNING]) == 1
 
 
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX shell fake interpreter")
+def test_from_exe_too_old_python_shim_verdict_not_cached(tmp_path: Path) -> None:
+    cache = DiskCache(tmp_path / "cache")
+    exe = _write_fake_python(tmp_path, stderr=f"unsupported Python version 3.5.10\n{sys.executable}")
+
+    assert PythonInfo.from_exe(str(exe), cache, raise_on_error=False, ignore_cache=True) is None
+    assert PythonInfo.from_exe(str(exe), cache, raise_on_error=False, ignore_cache=True) is None
+    assert len(_interrogations(exe)) == 2
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX executable script")
+@pytest.mark.parametrize("ignore_cache", [pytest.param(False, id="memory"), pytest.param(True, id="disk")])
+def test_from_exe_shim_result_changes_with_directory(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ignore_cache: bool
+) -> None:
+    shim = tmp_path / "shim"
+    shim.write_text(
+        f"#!{sys.executable}\n"
+        "from runpy import run_path\nimport sys\nfrom pathlib import Path\n"
+        "sys.prefix = (Path.cwd() / '.python-version').read_text(encoding='utf-8').strip()\n"
+        "run_path(sys.argv.pop(1), run_name='__main__')\n",
+        encoding="utf-8",
+    )
+    shim.chmod(0o755)
+    first, second = tmp_path / "first", tmp_path / "second"
+    first.mkdir()
+    second.mkdir()
+    (first / ".python-version").write_text(str(first), encoding="utf-8")
+    (second / ".python-version").write_text(str(second), encoding="utf-8")
+    cache = DiskCache(tmp_path / "cache")
+
+    monkeypatch.chdir(first)
+    earlier = PythonInfo.from_exe(str(shim), cache, ignore_cache=ignore_cache, resolve_to_host=False)
+    monkeypatch.chdir(second)
+    later = PythonInfo.from_exe(str(shim), cache, ignore_cache=ignore_cache, resolve_to_host=False)
+
+    assert earlier is not None
+    assert later is not None
+    assert (earlier.prefix, later.prefix) == (str(first), str(second))
+
+
 @pytest.mark.skipif(sys.platform == "win32", reason="POSIX shell fake interpreter")
 def test_from_exe_marker_without_version(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
     caplog.set_level(logging.INFO)
@@ -475,7 +516,7 @@ def test_executable_args_memory_cache_keyed_by_args(tmp_path: Path, mocker: Mock
     cache, env = DiskCache(tmp_path), dict(os.environ)
 
     results = [
-        _get_from_cache(PythonInfo, cache, _LAUNCHER, env, ignore_cache=False, executable_args=args)
+        _get_from_cache(PythonInfo, cache, sys.executable, env, ignore_cache=False, executable_args=args)
         for args in ("-3.11", "-3.12", "-3.11")
     ]
 

tests/test_py_info_collect.py

@@ -3,11 +3,11 @@
 import ast
 import json
 import os
-import subprocess
 import sys
 import sysconfig
 from itertools import takewhile
 from pathlib import Path
+from subprocess import run
 from typing import TYPE_CHECKING
 
 import pytest
@@ -44,7 +44,26 @@ def test_script_version_gate_precedes_imports() -> None:
 
 
 def _run_script(*args: str) -> str:
-    return subprocess.run([sys.executable, str(SCRIPT), *args], capture_output=True, text=True, check=True).stdout
+    return run([sys.executable, str(SCRIPT), *args], capture_output=True, text=True, check=True).stdout
+
+
+def test_script_version_gate_reports_executable() -> None:
+    result = run(
+        [
+            sys.executable,
+            "-c",
+            (
+                "from runpy import run_path; import sys; sys.version_info = (3, 5, 10, 'final', 0); "
+                "run_path(sys.argv[1], run_name='__main__')"
+            ),
+            str(SCRIPT),
+        ],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert result.returncode == 79
+    assert result.stderr == f"unsupported Python version 3.5.10\n{sys.executable}"
 
 
 @pytest.mark.parametrize(