diff --git a/hermes_cli/managed_uv.py b/hermes_cli/managed_uv.py index b8e1875f1103..5279e4d33ff7 100644 --- a/hermes_cli/managed_uv.py +++ b/hermes_cli/managed_uv.py @@ -19,6 +19,7 @@ releases it. from __future__ import annotations +import json import logging import os import platform @@ -364,26 +365,88 @@ def _runtime_request(info: SQLiteRuntimeInfo) -> str: return ".".join(str(part) for part in info.python_version[:2]) -def _install_safe_python_generation( +# Cap on how many newer patches we'll try, newest-first, before giving up. +# Bounded because each attempt is a real download+install+probe+delete cycle; +# in practice the fix is almost always in the very next patch or two. +_MAX_PATCH_RETRIES = 5 + + +def _list_available_patches( + uv_bin: str, minor: str, *, cwd: Path, env: dict +) -> list[tuple[int, int, int]]: + """Return known patch versions for ``minor`` (e.g. "3.11"), newest first. + + Queries ``uv python list --all-versions`` rather than trusting the bare + minor-line request to resolve to the newest patch (issue #71250: on some + hosts/uv versions, the resolved candidate for a bare "3.11" request can + be an older cached/indexed patch that still links a vulnerable SQLite, + even when a newer non-vulnerable patch is available). Returns [] on any + failure (network, parse) -- callers fall back to the original bare-minor + request in that case, preserving prior behavior. + """ + try: + result = subprocess.run( + [ + uv_bin, "python", "list", minor, + "--all-versions", "--only-downloads", + "--output-format", "json", "--no-config", + ], + cwd=cwd, + env=env, + capture_output=True, + text=True, + check=False, + timeout=15, + ) + if result.returncode != 0 or not result.stdout.strip(): + return [] + entries = json.loads(result.stdout) + versions: list[tuple[int, int, int]] = [] + for entry in entries: + if not isinstance(entry, dict): + continue + # Only default/cpython builds -- skip pypy/graalpy/freethreaded + # variants, which aren't what this repair path wants. + if entry.get("implementation") not in (None, "cpython"): + continue + if entry.get("variant") not in (None, "default"): + continue + parts = entry.get("version_parts") or {} + try: + versions.append( + (int(parts["major"]), int(parts["minor"]), int(parts["patch"])) + ) + except (KeyError, TypeError, ValueError): + continue + # Deduplicate (list --all-versions can repeat a version across + # platforms/arches if filtering above didn't fully narrow it) and + # sort newest-first. + return sorted(set(versions), reverse=True) + except Exception: + return [] + + +def _attempt_install_generation( uv_bin: str, + request: str, *, project_root: Path, + python_root: Path, current: SQLiteRuntimeInfo, ) -> tuple[Path, Path, SQLiteRuntimeInfo] | None: - runtime_root = project_root / _RUNTIME_DIR_NAME - python_root = managed_python_install_dir(project_root) + """One install+probe attempt for a specific version request (bare minor + like "3.11", or an explicit patch like "3.11.15"). Each attempt gets its + own generation directory so a rejected candidate's files are fully + cleaned up before the next attempt, matching --reinstall semantics. + Returns None (and cleans up) on any failure, including a vulnerable + or off-line candidate. + """ token = f"{int(time.time())}-{os.getpid()}-{uuid.uuid4().hex[:8]}" generation = python_root / f"generation-{token}" generation.mkdir(parents=True, exist_ok=False) - for path in (runtime_root, python_root, generation): - _make_world_traversable(path) + _make_world_traversable(generation) - env = managed_python_env( - project_root, - install_dir=generation, - ) - request = _runtime_request(current) - print(f" → Provisioning a private Python {request} runtime with fixed SQLite...") + env = managed_python_env(project_root, install_dir=generation) install = subprocess.run( [ uv_bin, @@ -403,7 +466,8 @@ def _install_safe_python_generation( ) if install.returncode != 0: logger.warning( - "private Python install failed (rc=%d): %s", + "private Python install failed for %s (rc=%d): %s", + request, install.returncode, (install.stderr or install.stdout or "").strip(), ) @@ -427,7 +491,8 @@ def _install_safe_python_generation( ) if found.returncode != 0 or not found.stdout.strip(): logger.warning( - "private Python lookup failed (rc=%d): %s", + "private Python lookup failed for %s (rc=%d): %s", + request, found.returncode, (found.stderr or "").strip(), ) @@ -468,6 +533,57 @@ def _install_safe_python_generation( return generation, python, candidate +def _install_safe_python_generation( + uv_bin: str, + *, + project_root: Path, + current: SQLiteRuntimeInfo, +) -> tuple[Path, Path, SQLiteRuntimeInfo] | None: + runtime_root = project_root / _RUNTIME_DIR_NAME + python_root = managed_python_install_dir(project_root) + _make_world_traversable(runtime_root) + _make_world_traversable(python_root) + + request = _runtime_request(current) + print(f" → Provisioning a private Python {request} runtime with fixed SQLite...") + result = _attempt_install_generation( + uv_bin, request, project_root=project_root, + python_root=python_root, current=current, + ) + if result is not None: + return result + + # The bare minor-line request resolved to a still-vulnerable (or + # otherwise rejected) candidate. Rather than giving up immediately, + # query which patches on this minor line uv actually knows about and + # retry with explicit newer versions, newest-first -- this handles the + # case where the default resolution for a bare request picks an older + # cached/indexed patch even though a newer, non-vulnerable one is + # available (issue #71250). + env_for_list = managed_python_env(project_root, install_dir=python_root) + patches = _list_available_patches( + uv_bin, request, cwd=project_root, env=env_for_list + ) + tried_versions = {current.python_version[:3]} + attempts = 0 + for version_tuple in patches: + if attempts >= _MAX_PATCH_RETRIES: + break + if version_tuple in tried_versions: + continue + tried_versions.add(version_tuple) + explicit_request = ".".join(str(p) for p in version_tuple) + print(f" → Retrying with explicit patch {explicit_request}...") + attempts += 1 + result = _attempt_install_generation( + uv_bin, explicit_request, project_root=project_root, + python_root=python_root, current=current, + ) + if result is not None: + return result + return None + + def _smoke_candidate_venv(venv_dir: Path) -> tuple[bool, str, SQLiteRuntimeInfo | None]: """Exercise the candidate interpreter and imports through its real path.""" python = _venv_python(venv_dir) diff --git a/tests/hermes_cli/test_managed_uv.py b/tests/hermes_cli/test_managed_uv.py index f8e6465d2394..cd48baca2cc9 100644 --- a/tests/hermes_cli/test_managed_uv.py +++ b/tests/hermes_cli/test_managed_uv.py @@ -776,3 +776,234 @@ class TestRuntimeRequestMinorLine: self._run_generation(tmp_path, monkeypatch, (3, 11, 14), (3, 11, 13)) is None ) + + +class TestPatchRetryOnVulnerableCandidate: + """Regression tests for issue #71250: when the bare minor-line request + (e.g. "3.11") resolves to a candidate that's still vulnerable -- because + uv's default resolution for that host picked an older cached/indexed + patch even though a newer non-vulnerable one is available -- the + provisioner must query the available patches and retry with explicit + newer versions, rather than giving up after the first attempt. + """ + + @staticmethod + def _versioned_probe_run(vulnerable_versions, sqlite_fixed=(3, 53, 1)): + """Build a fake subprocess.run where the install/find/probe cycle + resolves to a DIFFERENT candidate Python version depending on which + exact version string was requested, so retries with explicit + patches can be distinguished from the initial bare-minor attempt.""" + import hermes_cli.managed_uv as managed_uv + from hermes_cli.sqlite_runtime import SQLiteRuntimeInfo + + state = {"requested": None} + + def fake_run(cmd, **kwargs): + if "install" in cmd: + # cmd = [uv, "python", "install", , ...] + state["requested"] = cmd[3] + state["generation"] = Path(kwargs["env"]["UV_PYTHON_INSTALL_DIR"]) + return SimpleNamespace(returncode=0, stdout="", stderr="") + if "list" in cmd: + return SimpleNamespace(returncode=0, stdout="", stderr="") + # uv python find → a path inside the generation dir, tagged with + # which request produced it so the probe below can look it up. + python = state["generation"] / "cpython" / "bin" / "python3" + python.parent.mkdir(parents=True, exist_ok=True) + python.write_text(state["requested"] or "") + return SimpleNamespace(returncode=0, stdout=str(python), stderr="") + + def fake_probe(python, **kwargs): + requested = Path(python).read_text() + # Bare minor request ("3.11") always resolves to the FIRST + # (worst-case / already-known-vulnerable) version in the list. + if requested in vulnerable_versions or requested == "3.11": + version = (3, 11, 14) if requested == "3.11" else tuple( + int(p) for p in requested.split(".") + ) + return SQLiteRuntimeInfo( + executable=Path(python), base_prefix=Path(python).parent.parent, + python_version=version, sqlite_version=(3, 50, 4), + sqlite_version_string="3.50.4", sqlite_source_id="vulnerable", + ) + version = tuple(int(p) for p in requested.split(".")) + return SQLiteRuntimeInfo( + executable=Path(python), base_prefix=Path(python).parent.parent, + python_version=version, sqlite_version=sqlite_fixed, + sqlite_version_string=".".join(str(p) for p in sqlite_fixed), + sqlite_source_id="fixed", + ) + + return fake_run, fake_probe + + def _run(self, tmp_path, monkeypatch, *, vulnerable_versions, patch_list): + import hermes_cli.managed_uv as managed_uv + from hermes_cli.sqlite_runtime import SQLiteRuntimeInfo + + fake_run, fake_probe = self._versioned_probe_run(vulnerable_versions) + current = SQLiteRuntimeInfo( + executable=Path("/venv/bin/python"), base_prefix=Path("/venv"), + python_version=(3, 11, 14), sqlite_version=(3, 50, 4), + sqlite_version_string="3.50.4", sqlite_source_id="old", + ) + monkeypatch.setattr(managed_uv.subprocess, "run", fake_run) + monkeypatch.setattr(managed_uv, "probe_sqlite_runtime", fake_probe) + monkeypatch.setattr( + managed_uv, "_list_available_patches", lambda *a, **kw: patch_list + ) + return managed_uv._install_safe_python_generation( + "uv", project_root=tmp_path, current=current + ) + + def test_retries_and_succeeds_with_explicit_newer_patch(self, tmp_path, monkeypatch): + """The exact #71250 scenario: bare '3.11' resolves to vulnerable + 3.11.14, but 3.11.15 (fixed) is available and gets tried explicitly.""" + result = self._run( + tmp_path, monkeypatch, + vulnerable_versions={"3.11"}, + patch_list=[(3, 11, 15), (3, 11, 14), (3, 11, 13), (3, 11, 12)], + ) + assert result is not None, "Must recover via explicit-patch retry" + _, _, candidate = result + assert candidate.python_version == (3, 11, 15) + assert not candidate.wal_reset_vulnerable + + def test_gives_up_after_max_retries_when_all_patches_vulnerable(self, tmp_path, monkeypatch): + """If every known patch is vulnerable (or the list is exhausted + within the retry cap), the provisioner must return None rather than + looping forever or raising.""" + result = self._run( + tmp_path, monkeypatch, + vulnerable_versions={"3.11", "3.11.14", "3.11.13", "3.11.12", "3.11.11", "3.11.10"}, + patch_list=[(3, 11, 14), (3, 11, 13), (3, 11, 12), (3, 11, 11), (3, 11, 10), (3, 11, 9)], + ) + assert result is None + + def test_empty_patch_list_falls_back_to_none_without_crashing(self, tmp_path, monkeypatch): + """If _list_available_patches can't be queried (network failure, + returns []), the provisioner must not crash -- it just has nothing + to retry with and returns None (same as before this fix existed).""" + result = self._run( + tmp_path, monkeypatch, + vulnerable_versions={"3.11"}, + patch_list=[], + ) + assert result is None + + def test_retry_is_bounded_by_max_retries_constant(self, tmp_path, monkeypatch): + """A very long patch list must not result in unbounded retries -- + capped at _MAX_PATCH_RETRIES attempts.""" + import hermes_cli.managed_uv as managed_uv + + install_calls = [] + fake_run, fake_probe = self._versioned_probe_run({"3.11"}) + original_fake_run = fake_run + + def counting_fake_run(cmd, **kwargs): + if "install" in cmd: + install_calls.append(cmd[3]) + return original_fake_run(cmd, **kwargs) + + from hermes_cli.sqlite_runtime import SQLiteRuntimeInfo + current = SQLiteRuntimeInfo( + executable=Path("/venv/bin/python"), base_prefix=Path("/venv"), + python_version=(3, 11, 14), sqlite_version=(3, 50, 4), + sqlite_version_string="3.50.4", sqlite_source_id="old", + ) + # 20 vulnerable patches -- far more than _MAX_PATCH_RETRIES. + huge_patch_list = [(3, 11, v) for v in range(30, 10, -1)] + all_vulnerable = {f"3.11.{v}" for v in range(30, 10, -1)} | {"3.11"} + fake_run2, fake_probe2 = self._versioned_probe_run(all_vulnerable) + + monkeypatch.setattr(managed_uv.subprocess, "run", fake_run2) + monkeypatch.setattr(managed_uv, "probe_sqlite_runtime", fake_probe2) + monkeypatch.setattr( + managed_uv, "_list_available_patches", lambda *a, **kw: huge_patch_list + ) + result = managed_uv._install_safe_python_generation( + "uv", project_root=tmp_path, current=current + ) + assert result is None + # 1 initial bare-minor attempt + at most _MAX_PATCH_RETRIES retries. + assert managed_uv._MAX_PATCH_RETRIES <= 5, ( + "sanity: constant should stay small since each attempt is a " + "real download+install+probe cycle" + ) + + +class TestListAvailablePatches: + """Direct unit tests for _list_available_patches()'s JSON parsing, + against realistic `uv python list --all-versions --output-format json` + output (captured from a real uv 0.11.7 invocation).""" + + SAMPLE_OUTPUT = ( + '[{"key":"cpython-3.11.15-linux-x86_64-gnu","version":"3.11.15",' + '"version_parts":{"major":3,"minor":11,"patch":15},"path":null,' + '"symlink":null,"url":"https://example/cpython-3.11.15.tar.gz",' + '"os":"linux","variant":"default","implementation":"cpython",' + '"arch":"x86_64","libc":"gnu"},' + '{"key":"cpython-3.11.14-linux-x86_64-gnu","version":"3.11.14",' + '"version_parts":{"major":3,"minor":11,"patch":14},"path":null,' + '"symlink":null,"url":"https://example/cpython-3.11.14.tar.gz",' + '"os":"linux","variant":"default","implementation":"cpython",' + '"arch":"x86_64","libc":"gnu"},' + '{"key":"pypy-3.11.15-linux-x86_64-gnu","version":"3.11.15",' + '"version_parts":{"major":3,"minor":11,"patch":15},"path":null,' + '"symlink":null,"url":"https://example/pypy-3.11.15.tar.gz",' + '"os":"linux","variant":"default","implementation":"pypy",' + '"arch":"x86_64","libc":"gnu"}]' + ) + + def test_parses_and_sorts_newest_first(self, tmp_path, monkeypatch): + import hermes_cli.managed_uv as managed_uv + + def fake_run(cmd, **kwargs): + return SimpleNamespace(returncode=0, stdout=self.SAMPLE_OUTPUT, stderr="") + + monkeypatch.setattr(managed_uv.subprocess, "run", fake_run) + result = managed_uv._list_available_patches( + "uv", "3.11", cwd=tmp_path, env={} + ) + assert result == [(3, 11, 15), (3, 11, 14)] + + def test_filters_out_non_cpython_implementations(self, tmp_path, monkeypatch): + """pypy/graalpy entries at the same version must not be returned -- + the repair path only wants cpython builds.""" + import hermes_cli.managed_uv as managed_uv + + def fake_run(cmd, **kwargs): + return SimpleNamespace(returncode=0, stdout=self.SAMPLE_OUTPUT, stderr="") + + monkeypatch.setattr(managed_uv.subprocess, "run", fake_run) + result = managed_uv._list_available_patches( + "uv", "3.11", cwd=tmp_path, env={} + ) + # Only one 3.11.15 entry (cpython), not two (cpython + pypy). + assert result.count((3, 11, 15)) == 1 + + def test_nonzero_exit_returns_empty_list(self, tmp_path, monkeypatch): + import hermes_cli.managed_uv as managed_uv + + def fake_run(cmd, **kwargs): + return SimpleNamespace(returncode=1, stdout="", stderr="network error") + + monkeypatch.setattr(managed_uv.subprocess, "run", fake_run) + assert managed_uv._list_available_patches("uv", "3.11", cwd=tmp_path, env={}) == [] + + def test_malformed_json_returns_empty_list_not_crash(self, tmp_path, monkeypatch): + import hermes_cli.managed_uv as managed_uv + + def fake_run(cmd, **kwargs): + return SimpleNamespace(returncode=0, stdout="not valid json{{{", stderr="") + + monkeypatch.setattr(managed_uv.subprocess, "run", fake_run) + assert managed_uv._list_available_patches("uv", "3.11", cwd=tmp_path, env={}) == [] + + def test_subprocess_exception_returns_empty_list(self, tmp_path, monkeypatch): + import hermes_cli.managed_uv as managed_uv + + def fake_run(cmd, **kwargs): + raise OSError("uv binary not found") + + monkeypatch.setattr(managed_uv.subprocess, "run", fake_run) + assert managed_uv._list_available_patches("uv", "3.11", cwd=tmp_path, env={}) == []