fix(pm): forward index and transport settings into uv; bridge pip mirrors
pm.environment strips every ambient UV_* so a caller's uv settings cannot select PM's project, interpreter or cache. That also dropped the knobs mirrored and air-gapped networks depend on (UV_INDEX_URL/UV_DEFAULT_INDEX, UV_NATIVE_TLS, UV_INSECURE_HOST, UV_HTTP_TIMEOUT, index credentials), and uv never reads pip's configuration, so PIP_INDEX_URL / pip.conf mirrors stalled against pypi.org (#88453, #94613, #95608 on main). pm.index_config owns the allowlist and the pip→uv bridge with pip's own precedence (PIP_INDEX_URL over pip.conf; any explicit uv index wins). Only transport/index config crosses; the lockfile stays authoritative. A uv timeout now raises InstallError naming those knobs instead of a raw TimeoutExpired, so `hermes pm install` and the venv_sync status report tell the user what to configure.
This commit is contained in:
@@ -173,15 +173,20 @@ def _run_streaming(command: list[str], *, cwd: Path, env: dict[str, str],
|
||||
|
||||
|
||||
def _base_environment(env: Mapping[str, str] | None = None) -> dict[str, str]:
|
||||
"""Ambient UV settings are never policy; explicit build index settings are."""
|
||||
index_settings = {
|
||||
"UV_DEFAULT_INDEX", "UV_EXTRA_INDEX_URL", "UV_NO_INDEX", "UV_FIND_LINKS",
|
||||
"UV_INSECURE_HOST", "UV_KEYRING_PROVIDER", "UV_NATIVE_TLS",
|
||||
}
|
||||
return {key: value for key, value in (os.environ if env is None else env).items()
|
||||
"""Ambient UV settings never select the project, interpreter or cache.
|
||||
|
||||
Index and transport settings are the exception (pm.index_config): without
|
||||
them mirrored and air-gapped networks cannot resolve anything.
|
||||
"""
|
||||
from pm.index_config import bridged_index_settings, is_forwarded
|
||||
|
||||
source = os.environ if env is None else env
|
||||
base = {key: value for key, value in source.items()
|
||||
if not key.startswith("PYTHON") and key != "VIRTUAL_ENV"
|
||||
and (not key.startswith("UV_") or
|
||||
(env is not None and (key.startswith("UV_INDEX") or key in index_settings)))}
|
||||
and (not key.startswith("UV_") or is_forwarded(key))}
|
||||
if env is None:
|
||||
base.update(bridged_index_settings(os.environ))
|
||||
return base
|
||||
|
||||
|
||||
def managed_environment(destination: Path, *, python: Path | None = None,
|
||||
@@ -255,12 +260,19 @@ class PythonEnvironment:
|
||||
command.append("--no-config")
|
||||
if self.offline:
|
||||
command.append("--offline")
|
||||
if self.output is not None:
|
||||
# uv hides build-backend output until failure without verbose mode.
|
||||
command.append("--verbose")
|
||||
return _run_streaming(command, cwd=cwd, env=env, timeout=timeout, output=self.output)
|
||||
return subprocess.run(command, cwd=str(cwd), env=env, capture_output=True,
|
||||
text=True, encoding="utf-8", errors="replace", timeout=timeout)
|
||||
try:
|
||||
if self.output is not None:
|
||||
# uv hides build-backend output until failure without verbose mode.
|
||||
command.append("--verbose")
|
||||
return _run_streaming(command, cwd=cwd, env=env, timeout=timeout, output=self.output)
|
||||
return subprocess.run(command, cwd=str(cwd), env=env, capture_output=True,
|
||||
text=True, encoding="utf-8", errors="replace", timeout=timeout)
|
||||
except subprocess.TimeoutExpired as exc:
|
||||
from pm.index_config import TIMEOUT_HINT
|
||||
|
||||
# A silent stall against an unreachable index is the #95608 shape;
|
||||
# name the mirror knobs instead of surfacing a raw TimeoutExpired.
|
||||
raise InstallError("venv", f"uv {args[0]} timed out after {timeout}s", TIMEOUT_HINT) from exc
|
||||
|
||||
def create(self) -> None:
|
||||
"""Create at the final destination; callers must not move a live venv."""
|
||||
|
||||
107
pm/index_config.py
Normal file
107
pm/index_config.py
Normal file
@@ -0,0 +1,107 @@
|
||||
"""Forward the user's package-index and transport configuration into uv.
|
||||
|
||||
PM strips ambient ``UV_*`` so a caller's uv settings cannot steer which
|
||||
project, interpreter or cache an operation uses (pm.environment). Index and
|
||||
transport knobs are different: on mirrored or air-gapped networks they are the
|
||||
only way any dependency resolves at all (#88453, #94613, #95608). Only those
|
||||
cross the boundary; the lockfile stays authoritative for what gets installed.
|
||||
|
||||
uv never reads pip's configuration, so a pip-only mirror (``PIP_INDEX_URL`` or
|
||||
``index-url`` in pip.conf) is bridged to ``UV_INDEX_URL`` unless uv already has
|
||||
an index of its own. Stdlib only: the bootstrap runner imports this before any
|
||||
dependency exists.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
import configparser
|
||||
import os
|
||||
from pathlib import Path
|
||||
import sys
|
||||
|
||||
# Explicit uv index / transport settings that survive into uv. UV_INDEX_<NAME>_
|
||||
# {USERNAME,PASSWORD} credentials match by prefix in is_forwarded().
|
||||
FORWARDED_UV_SETTINGS = frozenset({
|
||||
"UV_INDEX_URL", "UV_EXTRA_INDEX_URL", "UV_DEFAULT_INDEX", "UV_INDEX", "UV_NO_INDEX",
|
||||
"UV_FIND_LINKS", "UV_INDEX_STRATEGY", "UV_KEYRING_PROVIDER",
|
||||
"UV_NATIVE_TLS", "UV_INSECURE_HOST", "UV_HTTP_TIMEOUT",
|
||||
})
|
||||
|
||||
_UV_INDEX_KNOBS = ("UV_INDEX_URL", "UV_DEFAULT_INDEX", "UV_INDEX")
|
||||
|
||||
# pip knob → uv knob, applied only when uv has no value of its own.
|
||||
_PIP_TO_UV = (
|
||||
("PIP_EXTRA_INDEX_URL", "UV_EXTRA_INDEX_URL"),
|
||||
("PIP_TRUSTED_HOST", "UV_INSECURE_HOST"),
|
||||
)
|
||||
|
||||
TIMEOUT_HINT = ("uv timed out. If your network needs a package mirror, set index-url in "
|
||||
"pip.conf (bridged to uv automatically) or UV_INDEX_URL; raise UV_HTTP_TIMEOUT "
|
||||
"for slow links.")
|
||||
|
||||
|
||||
def is_forwarded(key: str) -> bool:
|
||||
return key in FORWARDED_UV_SETTINGS or key.startswith("UV_INDEX_")
|
||||
|
||||
|
||||
def pip_config_candidates(env: Mapping[str, str]) -> list[Path]:
|
||||
"""pip's config files, lowest precedence first, as ``pip._internal.configuration`` ranks them.
|
||||
|
||||
Global, then user (skipped entirely when ``PIP_CONFIG_FILE`` names an existing file), then
|
||||
the interpreter's ``sys.prefix`` site file, then ``PIP_CONFIG_FILE`` itself on top.
|
||||
``RawConfigParser.read`` applies them in order, so the last file wins.
|
||||
``PIP_CONFIG_FILE=os.devnull`` disables all of them.
|
||||
"""
|
||||
explicit = env.get("PIP_CONFIG_FILE", "")
|
||||
if explicit == os.devnull:
|
||||
return []
|
||||
home = Path.home()
|
||||
if sys.platform == "win32":
|
||||
name = "pip.ini"
|
||||
global_files = [Path(env.get("ProgramData") or r"C:\ProgramData") / "pip" / name]
|
||||
user_files = [home / "pip" / name,
|
||||
Path(env.get("APPDATA") or home / "AppData" / "Roaming") / "pip" / name]
|
||||
elif sys.platform == "darwin":
|
||||
name = "pip.conf"
|
||||
global_files = [Path("/Library/Application Support/pip") / name]
|
||||
app_support = home / "Library" / "Application Support" / "pip"
|
||||
user_files = [home / ".pip" / name,
|
||||
(app_support if app_support.is_dir() else home / ".config" / "pip") / name]
|
||||
else:
|
||||
name = "pip.conf"
|
||||
xdg_dirs = (env.get("XDG_CONFIG_DIRS") or "/etc/xdg").split(os.pathsep)
|
||||
global_files = [Path(d) / "pip" / name for d in xdg_dirs if d] + [Path("/etc") / name]
|
||||
user_files = [home / ".pip" / name, Path(env.get("XDG_CONFIG_HOME") or home / ".config") / "pip" / name]
|
||||
explicit_files = [Path(explicit)] if explicit else []
|
||||
if explicit_files and explicit_files[0].is_file():
|
||||
user_files = []
|
||||
return global_files + user_files + [Path(sys.prefix) / name] + explicit_files
|
||||
|
||||
|
||||
def pip_conf_index_url(env: Mapping[str, str]) -> str | None:
|
||||
# Raw: pip does not interpolate, and mirror URLs carry percent-encoded credentials.
|
||||
parser = configparser.RawConfigParser()
|
||||
try:
|
||||
parser.read(str(path) for path in pip_config_candidates(env))
|
||||
if not parser.has_section("global"):
|
||||
return None
|
||||
return parser.get("global", "index-url", fallback="").strip() or None
|
||||
except configparser.Error:
|
||||
return None
|
||||
|
||||
|
||||
def bridged_index_settings(ambient: Mapping[str, str]) -> dict[str, str]:
|
||||
"""The uv index/transport settings *ambient* asks for, pip knobs translated.
|
||||
|
||||
``PIP_INDEX_URL`` beats pip.conf, as in pip; any explicit uv index knob beats both.
|
||||
"""
|
||||
settings = {key: value for key, value in ambient.items() if is_forwarded(key)}
|
||||
if not any(settings.get(key) for key in _UV_INDEX_KNOBS):
|
||||
index_url = (ambient.get("PIP_INDEX_URL") or "").strip() or pip_conf_index_url(ambient)
|
||||
if index_url:
|
||||
settings["UV_INDEX_URL"] = index_url
|
||||
for pip_key, uv_key in _PIP_TO_UV:
|
||||
value = (ambient.get(pip_key) or "").strip()
|
||||
if value and not settings.get(uv_key):
|
||||
settings[uv_key] = value
|
||||
return settings
|
||||
@@ -461,7 +461,9 @@ def streaming_runner(request, tmp_path):
|
||||
env=dict(os.environ), timeout=timeout)
|
||||
return environment._run(["-c", script], cwd=tmp_path, timeout=timeout)
|
||||
|
||||
return run, output, RuntimeError if request.param == "cli" else subprocess.TimeoutExpired
|
||||
from pm.package import InstallError
|
||||
|
||||
return run, output, RuntimeError if request.param == "cli" else InstallError
|
||||
|
||||
|
||||
@pytest.mark.parametrize("parent_exits", [True, False])
|
||||
|
||||
64
tests/pm/test_index_bridging.py
Normal file
64
tests/pm/test_index_bridging.py
Normal file
@@ -0,0 +1,64 @@
|
||||
"""Mirrored and air-gapped networks configure indexes through pip or uv; PM forwards
|
||||
exactly that into uv while still refusing every other ambient uv setting."""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
|
||||
import pytest
|
||||
|
||||
from pm.environment import PythonEnvironment, _base_environment
|
||||
from pm.package import InstallError
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def clean_index_env(monkeypatch, tmp_path):
|
||||
for key in list(os.environ):
|
||||
if key.startswith(("UV_", "PIP_")):
|
||||
monkeypatch.delenv(key)
|
||||
monkeypatch.setenv("PIP_CONFIG_FILE", os.devnull)
|
||||
return tmp_path
|
||||
|
||||
|
||||
def test_pip_index_reaches_uv_but_ambient_uv_selection_does_not(clean_index_env, monkeypatch):
|
||||
monkeypatch.setenv("PIP_INDEX_URL", "https://mirror.example/simple")
|
||||
monkeypatch.setenv("PIP_TRUSTED_HOST", "mirror.example")
|
||||
monkeypatch.setenv("UV_HTTP_TIMEOUT", "300")
|
||||
monkeypatch.setenv("UV_INDEX_CORP_PASSWORD", "s3cret")
|
||||
monkeypatch.setenv("UV_PYTHON", "/poison/python")
|
||||
monkeypatch.setenv("UV_CACHE_DIR", "/poison/cache")
|
||||
monkeypatch.setenv("UV_PROJECT_ENVIRONMENT", "/poison/venv")
|
||||
|
||||
env = _base_environment()
|
||||
|
||||
assert env["UV_INDEX_URL"] == "https://mirror.example/simple"
|
||||
assert env["UV_INSECURE_HOST"] == "mirror.example"
|
||||
assert env["UV_HTTP_TIMEOUT"] == "300"
|
||||
assert env["UV_INDEX_CORP_PASSWORD"] == "s3cret"
|
||||
assert not {"UV_PYTHON", "UV_CACHE_DIR", "UV_PROJECT_ENVIRONMENT"} & env.keys()
|
||||
|
||||
|
||||
def test_pip_conf_is_bridged_only_when_uv_has_no_index(clean_index_env, monkeypatch):
|
||||
pip_conf = clean_index_env / "pip.conf"
|
||||
# Percent-encoded credentials: pip reads its config raw, so must the bridge.
|
||||
pip_conf.write_text("[global]\nindex-url = https://user:p%40ss@mirror.example/simple\n", encoding="utf-8")
|
||||
monkeypatch.setenv("PIP_CONFIG_FILE", str(pip_conf))
|
||||
|
||||
assert _base_environment()["UV_INDEX_URL"] == "https://user:p%40ss@mirror.example/simple"
|
||||
|
||||
monkeypatch.setenv("UV_DEFAULT_INDEX", "https://explicit.example/simple")
|
||||
env = _base_environment()
|
||||
assert env["UV_DEFAULT_INDEX"] == "https://explicit.example/simple"
|
||||
assert "UV_INDEX_URL" not in env
|
||||
|
||||
|
||||
def test_uv_timeout_names_the_mirror_knobs(tmp_path, monkeypatch):
|
||||
def stall(*args, **kwargs):
|
||||
raise subprocess.TimeoutExpired(args[0], kwargs["timeout"])
|
||||
|
||||
monkeypatch.setattr(subprocess, "run", stall)
|
||||
environment = PythonEnvironment(uv=tmp_path / "uv", python=tmp_path / "python",
|
||||
destination=tmp_path / "venv", cache=tmp_path / "cache", env={})
|
||||
with pytest.raises(InstallError, match="UV_INDEX_URL") as info:
|
||||
environment._run(["sync"], cwd=tmp_path, timeout=7)
|
||||
assert "timed out after 7s" in str(info.value)
|
||||
@@ -31,7 +31,7 @@ def test_ambient_uv_config_does_not_affect_pm_venv_sync(tmp_path, monkeypatch):
|
||||
(config / "uv" / "uv.toml").write_text('required-version="<0.0.1"\n')
|
||||
for key, value in {
|
||||
"UV_NO_CONFIG": "1", "UV_CONFIG_FILE": "/poison/uv.toml",
|
||||
"UV_DEFAULT_INDEX": "https://poison.invalid/simple", "UV_PYTHON": "/poison/python",
|
||||
"UV_PYTHON": "/poison/python",
|
||||
"UV_PROJECT_ENVIRONMENT": str(tmp_path / "unrelated-environment"),
|
||||
"UV_CACHE_DIR": str(tmp_path / "hostile-cache"),
|
||||
"UV_PROJECT": "/poison/project", "VIRTUAL_ENV": "/poison/venv",
|
||||
|
||||
Reference in New Issue
Block a user