diff --git a/pyproject.toml b/pyproject.toml index dd957766b9..bf4b118d68 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -46,6 +46,7 @@ studio = [ "*.sh", "*.ps1", "*.bat", + "node_prebuilt_pins.json", "frontend/dist/**/*", "frontend/*.json", "frontend/*.ts", diff --git a/studio/install_node_prebuilt.py b/studio/install_node_prebuilt.py index bc67189e2b..4038216eef 100644 --- a/studio/install_node_prebuilt.py +++ b/studio/install_node_prebuilt.py @@ -9,6 +9,9 @@ Downloads an official Node.js archive from nodejs.org into an isolated LTS clears the Studio frontend build floor (Vite 8: Node ^20.19 || >=22.12, npm >= 11) with the npm it bundles. +Archives are verified against sha256 digests pinned in ``node_prebuilt_pins.json`` +(committed in-tree), not a checksum re-fetched from the same origin as the archive. + Mirrors ``install_llama_prebuilt.py`` so the setup scripts drive it the same way. Exit codes: 0 success, 1 error, 2 fallback, 3 busy. A re-run that already matches logs "already matches" and returns 0 without downloading (the scripts grep it). @@ -65,6 +68,14 @@ INSTALL_STAGING_ROOT_NAME = ".staging" METADATA_FILENAME = "UNSLOTH_NODE_PREBUILT_INFO.json" METADATA_SCHEMA_VERSION = 1 +# Trust anchor: verify archives against sha256 pins committed in +# node_prebuilt_pins.json (in-tree, code-reviewed), not a same-origin checksum. +PINS_FILENAME = "node_prebuilt_pins.json" +PINS_SCHEMA_VERSION = 1 +DEFAULT_NODE_CHANNEL = "pinned" +# Opt-in to install an unpinned version, trusting the same-origin SHASUMS256.txt. +ALLOW_UNVERIFIED_ENV = "UNSLOTH_NODE_ALLOW_UNVERIFIED" + # PowerShell renders stderr as NativeCommandError noise; main() flips logs to stdout. _LOG_TO_STDOUT = False @@ -73,6 +84,11 @@ class PrebuiltFallback(RuntimeError): """Recoverable failure -- caller should fall back (exit code 2).""" +class UnpinnedNodeRefused(PrebuiltFallback): + """Unpinned Node requested without opt-in. Distinct type so the orchestrator + re-raises it instead of letting the keep-existing path swallow the refusal.""" + + class BusyInstallConflict(RuntimeError): """Another process holds the install lock (exit code 3).""" @@ -311,6 +327,81 @@ def download_file_verified( raise PrebuiltFallback(f"{label} checksum mismatch after retry") +# ── Pinned digest manifest (trust anchor) ── +def pins_path() -> Path: + return Path(__file__).resolve().parent / PINS_FILENAME + + +def load_pins() -> dict: + path = pins_path() + try: + data = json.loads(path.read_text(encoding = "utf-8")) + except FileNotFoundError as exc: + raise PrebuiltFallback(f"pinned Node manifest missing: {path}") from exc + except (json.JSONDecodeError, OSError) as exc: + raise PrebuiltFallback(f"pinned Node manifest unreadable ({path}): {exc}") from exc + if not isinstance(data, dict) or data.get("schema_version") != PINS_SCHEMA_VERSION: + raise PrebuiltFallback(f"pinned Node manifest has an unexpected schema: {path}") + return data + + +def pinned_default_version(pins: dict) -> str: + version = str(pins.get("default_version", "")).lstrip("v") + if not _version_tuple(version): + raise PrebuiltFallback("pinned Node manifest is missing a valid 'default_version'") + return version + + +def pinned_sha256(pins: dict, version: str, asset_name: str) -> str | None: + versions = pins.get("versions") + if not isinstance(versions, dict): + return None + entry = versions.get(version.lstrip("v")) + if not isinstance(entry, dict): + return None + digest = entry.get(asset_name) + if not isinstance(digest, str): + return None + digest = digest.strip().lower() + if len(digest) == 64 and all(c in "0123456789abcdef" for c in digest): + return digest + return None + + +def allow_unverified_node() -> bool: + return os.environ.get(ALLOW_UNVERIFIED_ENV, "").strip().lower() in {"1", "true", "yes", "on"} + + +def resolve_expected_sha256(pins: dict, version: str, asset: str, *, allow_unverified: bool) -> str: + """Sha256 to verify the archive against: the committed pin, or (only with explicit + opt-in) the same-origin SHASUMS256.txt. Unpinned without opt-in is refused.""" + pinned = pinned_sha256(pins, version, asset) + if pinned is not None: + log(f"verifying {asset} against pinned sha256 from {PINS_FILENAME}") + return pinned + + if not allow_unverified: + raise UnpinnedNodeRefused( + f"refusing to install Node v{version}: {asset} is not in the pinned manifest " + f"({PINS_FILENAME}); its only checksum would arrive over the same channel as the " + f"archive. Use the pinned default version (unset UNSLOTH_NODE_VERSION), add a pin " + f"for {asset}, or set {ALLOW_UNVERIFIED_ENV}=1 to trust the upstream SHASUMS256.txt " + f"at your own risk." + ) + + log( + f"WARNING: {asset} is not pinned; trusting upstream SHASUMS256.txt because " + f"{ALLOW_UNVERIFIED_ENV} is set. This checksum shares the archive's origin and is " + f"not an independent integrity guarantee." + ) + # A non-UTF8 body just yields no hex match below -> clean PrebuiltFallback. + shasums = download_bytes(node_shasums_url(version), timeout = 30).decode("utf-8", "replace") + expected = expected_sha256_for(shasums, asset) + if not expected: + raise PrebuiltFallback(f"no sha256 for {asset} in SHASUMS256.txt (v{version})") + return expected + + # ── Safe archive extraction (zip + tar.gz, traversal/symlink guarded) ── def _safe_extract_path(base: Path, member_name: str) -> Path: member_path = Path(member_name.replace("\\", "/")) @@ -583,11 +674,21 @@ def load_metadata(install_dir: Path) -> dict | None: return data if isinstance(data, dict) else None -def existing_install_matches(install_dir: Path, host: HostInfo, *, version: str) -> bool: - """True iff the on-disk install is exactly this version and runs.""" +def existing_install_matches( + install_dir: Path, + host: HostInfo, + *, + version: str, + expected_sha: str | None = None, +) -> bool: + """True iff the on-disk install is exactly this version, runs, and (when + expected_sha is given) was recorded with that digest, so a non-pinned or + tampered artifact is not kept just because its version string matches.""" meta = load_metadata(install_dir) if not meta or meta.get("version") != version: return False + if expected_sha is not None and meta.get("sha256") != expected_sha: + return False if installed_node_version(install_dir, host) != version: return False npm_major = installed_npm_major(install_dir, host) @@ -634,8 +735,12 @@ def _ensure_npm_floor(install_dir: Path, host: HostInfo) -> None: # ── Orchestration ── def install_prebuilt(install_dir: Path, *, channel: str, min_major: int, force: bool) -> int: host = detect_host() + pins = load_pins() - if channel in {"lts", "latest"}: + if channel in {"", "pinned", "default"}: + # Default path: the committed pin, no index.json round-trip. + version = pinned_default_version(pins) + elif channel in {"lts", "latest"}: try: index = fetch_json(NODE_DIST_INDEX) except Exception as exc: # noqa: BLE001 @@ -658,21 +763,35 @@ def install_prebuilt(install_dir: Path, *, channel: str, min_major: int, force: asset = node_asset_name(version, host) log(f"target Node v{version} ({asset})") - if not force and existing_install_matches(install_dir, host, version = version): + # Only keep an existing install if it matches the committed pin (or the caller + # opted out of pinning); an unpinned target without opt-in falls through to the + # refusal in resolve_expected_sha256 rather than short-circuiting on it. + pin = pinned_sha256(pins, version, asset) + allow_unverified = allow_unverified_node() + may_keep = pin is not None or allow_unverified + + if ( + not force + and may_keep + and existing_install_matches(install_dir, host, version = version, expected_sha = pin) + ): log(f"existing Node install already matches v{version}; nothing to do") return EXIT_SUCCESS with install_lock(install_lock_path(install_dir)): # Re-check under the lock: a concurrent run may have just finished. - if not force and existing_install_matches(install_dir, host, version = version): + if ( + not force + and may_keep + and existing_install_matches(install_dir, host, version = version, expected_sha = pin) + ): log(f"existing Node install already matches v{version}; nothing to do") return EXIT_SUCCESS try: - shasums = download_bytes(node_shasums_url(version), timeout = 30).decode("utf-8") - expected_sha = expected_sha256_for(shasums, asset) - if not expected_sha: - raise PrebuiltFallback(f"no sha256 for {asset} in SHASUMS256.txt (v{version})") + expected_sha = resolve_expected_sha256( + pins, version, asset, allow_unverified = allow_unverified + ) staging_root = install_dir.parent / INSTALL_STAGING_ROOT_NAME staging_root.mkdir(parents = True, exist_ok = True) @@ -705,10 +824,22 @@ def install_prebuilt(install_dir: Path, *, channel: str, min_major: int, force: staging_root.rmdir() except OSError: pass + except UnpinnedNodeRefused: + # A policy refusal, not a transient failure: fail closed, never keep-existing. + raise except Exception as exc: # noqa: BLE001 - # A newer Node exists upstream but the shasums/archive fetch failed; - # keep an existing usable Node rather than aborting the update. - if not force and existing_install_usable(install_dir, host): + # Transient download/verify failure: keep an existing usable Node, but + # never keep a same-version install whose recorded digest is not the pin + # (the artifact the short-circuit above just rejected). A different usable + # version is still kept for offline resilience. + meta = load_metadata(install_dir) + pin_mismatch = ( + pin is not None + and bool(meta) + and meta.get("version") == version + and meta.get("sha256") != pin + ) + if not force and not pin_mismatch and existing_install_usable(install_dir, host): log(f"Node download failed ({exc}); keeping existing isolated Node") return EXIT_SUCCESS raise @@ -733,8 +864,12 @@ def main(argv: list[str] | None = None) -> int: ) parser.add_argument( "--node-version", - default = os.environ.get("UNSLOTH_NODE_VERSION", "lts"), - help = "'lts' (default), 'latest', or an explicit version like 24.4.1", + default = os.environ.get("UNSLOTH_NODE_VERSION", DEFAULT_NODE_CHANNEL), + help = ( + f"'pinned' (default; installs the digest-pinned version from {PINS_FILENAME}), " + f"'lts', 'latest', or an explicit version like 24.4.1. Non-pinned versions require " + f"{ALLOW_UNVERIFIED_ENV}=1." + ), ) parser.add_argument("--min-major", type = int, default = NODE_MIN_LTS_MAJOR) parser.add_argument( @@ -753,6 +888,10 @@ def main(argv: list[str] | None = None) -> int: except BusyInstallConflict as exc: log(str(exc)) return EXIT_BUSY + except UnpinnedNodeRefused as exc: + # Catch before PrebuiltFallback so the refusal logs its own message. + log(str(exc)) + return EXIT_FALLBACK except PrebuiltFallback as exc: log(f"prebuilt unavailable: {exc}") return EXIT_FALLBACK diff --git a/studio/node_prebuilt_pins.json b/studio/node_prebuilt_pins.json new file mode 100644 index 0000000000..d279c04316 --- /dev/null +++ b/studio/node_prebuilt_pins.json @@ -0,0 +1,15 @@ +{ + "schema_version": 1, + "comment": "Trust anchor for install_node_prebuilt.py: sha256 frozen from the official nodejs.org SHASUMS256.txt. To bump: update default_version and every per-asset digest from the new SHASUMS256.txt.", + "default_version": "24.18.0", + "versions": { + "24.18.0": { + "node-v24.18.0-linux-x64.tar.gz": "783130984963db7ba9cbd01089eaf2c2efb055c7c1693c943174b967b3050cb8", + "node-v24.18.0-linux-arm64.tar.gz": "6b4484c2190274175df9aa8f28e2d758a819cb1c1fe6ab481e2f95b463ab8508", + "node-v24.18.0-darwin-x64.tar.gz": "dfd0dbd3e721503434df7b7205e719f61b3a3a31b2bcf9729b8b91fea240f080", + "node-v24.18.0-darwin-arm64.tar.gz": "e1a97e14c99c803e96c7339403282ea05a499c32f8d83defe9ef5ec66f979ed1", + "node-v24.18.0-win-x64.zip": "0ae68406b42d7725661da979b1403ec9926da205c6770827f33aac9d8f26e821", + "node-v24.18.0-win-arm64.zip": "f274669adb93b1fd0fbf8f21fd078609e9dcc84333d4f2718d2dde3f9a161a01" + } + } +} diff --git a/tests/studio/install/test_install_node_prebuilt_logic.py b/tests/studio/install/test_install_node_prebuilt_logic.py index 0c6da87655..5476702d65 100644 --- a/tests/studio/install/test_install_node_prebuilt_logic.py +++ b/tests/studio/install/test_install_node_prebuilt_logic.py @@ -120,7 +120,7 @@ def test_expected_sha256_rejects_malformed(): # ── Version selection from index.json ── INDEX = [ {"version": "v26.3.1", "lts": False}, - {"version": "v24.17.0", "lts": "Krypton"}, + {"version": "v24.18.0", "lts": "Krypton"}, {"version": "v24.9.0", "lts": "Krypton"}, {"version": "v22.20.0", "lts": "Jod"}, {"version": "v20.19.0", "lts": "Iron"}, @@ -128,8 +128,8 @@ INDEX = [ def test_select_lts_respects_min_major(): - # Newest LTS at/above 24 -> 24.17.0 (22.x LTS is below the floor). - assert M.select_node_version(INDEX, channel = "lts", min_major = 24) == "24.17.0" + # Newest LTS at/above 24 -> 24.18.0 (22.x LTS is below the floor). + assert M.select_node_version(INDEX, channel = "lts", min_major = 24) == "24.18.0" def test_select_latest_overall(): @@ -301,10 +301,13 @@ def test_existing_install_matches_true_when_version_and_runtime_ok(tmp_path: Pat def test_install_prebuilt_short_circuits_when_version_matches(tmp_path: Path, monkeypatch): install_dir = tmp_path / "node" install_dir.mkdir() - M.write_metadata(install_dir, version = "24.17.0", asset = "x", sha256 = "y") + version = M.pinned_default_version(M.load_pins()) # == INDEX's newest LTS + asset = M.node_asset_name(version, _host("linux", "x64")) + pin = M.pinned_sha256(M.load_pins(), version, asset) # short-circuit now needs the pin + M.write_metadata(install_dir, version = version, asset = asset, sha256 = pin) monkeypatch.setattr(M, "detect_host", lambda: _host("linux", "x64")) monkeypatch.setattr(M, "fetch_json", lambda url: INDEX) - monkeypatch.setattr(M, "installed_node_version", lambda d, h: "24.17.0") + monkeypatch.setattr(M, "installed_node_version", lambda d, h: version) monkeypatch.setattr(M, "installed_npm_major", lambda d, h: 11) def boom(*a, **k): @@ -407,26 +410,25 @@ def test_install_prebuilt_rejects_explicit_below_floor(tmp_path: Path, monkeypat M.install_prebuilt(install_dir, channel = "20.18.0", min_major = 24, force = False) -def test_install_prebuilt_keeps_existing_when_shasums_fetch_fails(tmp_path: Path, monkeypatch): - # index.json resolves a newer version, but the later SHASUMS fetch fails and a - # usable older isolated Node is on disk -> keep it instead of aborting. +def test_install_prebuilt_keeps_existing_when_download_fails(tmp_path: Path, monkeypatch): + # Archive download fails but a usable older Node is on disk -> keep it. install_dir = tmp_path / "node" install_dir.mkdir() M.write_metadata(install_dir, version = "24.9.0", asset = "x", sha256 = "y") monkeypatch.setattr(M, "detect_host", lambda: _host("linux", "x64")) - monkeypatch.setattr(M, "fetch_json", lambda url: INDEX) # newest LTS = 24.17.0 + monkeypatch.setattr(M, "fetch_json", lambda url: INDEX) # newest LTS = 24.18.0 (pinned) monkeypatch.setattr(M, "installed_node_version", lambda d, h: "24.9.0") monkeypatch.setattr(M, "installed_npm_major", lambda d, h: 11) - monkeypatch.setattr(M, "download_bytes", _offline) # SHASUMS fetch fails + monkeypatch.setattr(M, "download_file_verified", _offline) # archive download fails rc = M.install_prebuilt(install_dir, channel = "lts", min_major = 24, force = False) assert rc == M.EXIT_SUCCESS -def test_install_prebuilt_reraises_shasums_failure_without_existing(tmp_path: Path, monkeypatch): +def test_install_prebuilt_reraises_download_failure_without_existing(tmp_path: Path, monkeypatch): install_dir = tmp_path / "node" # nothing usable on disk monkeypatch.setattr(M, "detect_host", lambda: _host("linux", "x64")) monkeypatch.setattr(M, "fetch_json", lambda url: INDEX) - monkeypatch.setattr(M, "download_bytes", _offline) + monkeypatch.setattr(M, "download_file_verified", _offline) with pytest.raises(OSError): M.install_prebuilt(install_dir, channel = "lts", min_major = 24, force = False) @@ -475,3 +477,288 @@ def test_ensure_npm_floor_noop_when_npm_meets_bar(tmp_path: Path, monkeypatch): monkeypatch.setattr(M, "_run_node", boom) M._ensure_npm_floor(tmp_path / "node", _host("linux", "x64")) + + +# ── Pinned digest manifest (trust anchor) ── +# Archives must be verified against committed pins, never a same-origin re-fetch. +ALL_SUPPORTED_HOSTS = [ + ("linux", "x64"), + ("linux", "arm64"), + ("darwin", "x64"), + ("darwin", "arm64"), + ("win", "x64"), + ("win", "arm64"), +] + + +def test_load_pins_exposes_valid_default_version(): + pins = M.load_pins() + version = M.pinned_default_version(pins) + assert M._version_tuple(version) # parses as a real version + assert M._meets_node_floor(version) # the pinned default clears the build floor + + +def test_pinned_manifest_covers_every_supported_asset(): + # A future Node bump must pin all six os/arch archives or a host loses its anchor. + pins = M.load_pins() + version = M.pinned_default_version(pins) + for node_os, node_arch in ALL_SUPPORTED_HOSTS: + host = _host(node_os, node_arch) + asset = M.node_asset_name(version, host) + digest = M.pinned_sha256(pins, version, asset) + assert digest is not None and len(digest) == 64, f"missing pin for {asset}" + assert all(c in "0123456789abcdef" for c in digest) + + +def test_pinned_sha256_unknown_pair_returns_none(): + pins = M.load_pins() + assert M.pinned_sha256(pins, "99.0.0", "node-v99.0.0-linux-x64.tar.gz") is None + version = M.pinned_default_version(pins) + assert M.pinned_sha256(pins, version, "node-vX-bogus-arch.tar.gz") is None + + +def test_load_pins_rejects_bad_schema(tmp_path: Path, monkeypatch): + bad = tmp_path / "node_prebuilt_pins.json" + bad.write_text(json.dumps({"schema_version": 999})) + monkeypatch.setattr(M, "pins_path", lambda: bad) + with pytest.raises(PrebuiltFallback): + M.load_pins() + + +def test_load_pins_raises_when_missing(tmp_path: Path, monkeypatch): + monkeypatch.setattr(M, "pins_path", lambda: tmp_path / "does_not_exist.json") + with pytest.raises(PrebuiltFallback): + M.load_pins() + + +def test_resolve_expected_sha256_uses_pin_without_touching_network(monkeypatch): + pins = M.load_pins() + version = M.pinned_default_version(pins) + asset = M.node_asset_name(version, _host("linux", "x64")) + + def boom(*a, **k): + raise AssertionError("pinned path must not fetch the remote SHASUMS256.txt") + + monkeypatch.setattr(M, "download_bytes", boom) + sha = M.resolve_expected_sha256(pins, version, asset, allow_unverified = False) + assert sha == M.pinned_sha256(pins, version, asset) + + +def test_resolve_expected_sha256_failcloses_on_unpinned(monkeypatch): + pins = M.load_pins() + + def boom(*a, **k): + raise AssertionError("must not reach the network when refusing an unpinned version") + + monkeypatch.setattr(M, "download_bytes", boom) + with pytest.raises(PrebuiltFallback): + M.resolve_expected_sha256( + pins, "26.3.1", "node-v26.3.1-linux-x64.tar.gz", allow_unverified = False + ) + + +def test_resolve_expected_sha256_optin_falls_back_to_remote_shasums(monkeypatch): + pins = M.load_pins() + asset = "node-v26.3.1-linux-x64.tar.gz" + remote_sha = "d" * 64 + monkeypatch.setattr(M, "download_bytes", lambda url, **k: f"{remote_sha} {asset}\n".encode()) + # Only with the explicit opt-in does the legacy remote-checksum path run. + sha = M.resolve_expected_sha256(pins, "26.3.1", asset, allow_unverified = True) + assert sha == remote_sha + + +@pytest.mark.parametrize( + "value,expected", + [("1", True), ("true", True), ("YES", True), ("on", True), ("0", False), ("", False)], +) +def test_allow_unverified_node_reads_env(monkeypatch, value, expected): + monkeypatch.setenv(M.ALLOW_UNVERIFIED_ENV, value) + assert M.allow_unverified_node() is expected + + +def test_install_prebuilt_default_channel_resolves_pinned_version(tmp_path: Path, monkeypatch): + # Default channel installs the pinned version with no index.json round-trip. + pins = M.load_pins() + version = M.pinned_default_version(pins) + install_dir = tmp_path / "node" + install_dir.mkdir() + asset = M.node_asset_name(version, _host("linux", "x64")) + pin = M.pinned_sha256(pins, version, asset) # kept only if the recorded digest is the pin + M.write_metadata(install_dir, version = version, asset = asset, sha256 = pin) + monkeypatch.setattr(M, "detect_host", lambda: _host("linux", "x64")) + monkeypatch.setattr(M, "installed_node_version", lambda d, h: version) + monkeypatch.setattr(M, "installed_npm_major", lambda d, h: 11) + + def boom(*a, **k): + raise AssertionError("default channel must not hit nodejs.org when the install matches") + + monkeypatch.setattr(M, "fetch_json", boom) # no index.json + monkeypatch.setattr(M, "download_file", boom) + monkeypatch.setattr(M, "download_bytes", boom) + rc = M.install_prebuilt(install_dir, channel = "pinned", min_major = 24, force = False) + assert rc == M.EXIT_SUCCESS + + +def test_install_prebuilt_failcloses_on_unpinned_latest(tmp_path: Path, monkeypatch): + # Unpinned `latest`, no opt-in, nothing on disk to keep: refuse. + install_dir = tmp_path / "node" + monkeypatch.setattr(M, "detect_host", lambda: _host("linux", "x64")) + monkeypatch.setattr(M, "fetch_json", lambda url: INDEX) # latest overall = 26.3.1 (unpinned) + monkeypatch.delenv(M.ALLOW_UNVERIFIED_ENV, raising = False) + + def boom(*a, **k): + raise AssertionError("must not download an unpinned archive") + + monkeypatch.setattr(M, "download_file", boom) + with pytest.raises(M.UnpinnedNodeRefused): + M.install_prebuilt(install_dir, channel = "latest", min_major = 24, force = False) + + +@pytest.mark.parametrize("channel", ["latest", "26.3.1"]) +def test_install_prebuilt_unpinned_refusal_does_not_keep_existing( + tmp_path: Path, monkeypatch, channel +): + # Regression: an unpinned refusal must fail closed even with a usable install on + # disk; the keep-existing fallback is for transient failures only. + install_dir = tmp_path / "node" + install_dir.mkdir() + M.write_metadata(install_dir, version = "24.9.0", asset = "old", sha256 = "old") + monkeypatch.setattr(M, "detect_host", lambda: _host("linux", "x64")) + monkeypatch.setattr(M, "fetch_json", lambda url: INDEX) # latest overall = 26.3.1 (unpinned) + monkeypatch.setattr(M, "installed_node_version", lambda d, h: "24.9.0") + monkeypatch.setattr(M, "installed_npm_major", lambda d, h: 11) # existing install is usable + monkeypatch.delenv(M.ALLOW_UNVERIFIED_ENV, raising = False) + + def boom(*a, **k): + raise AssertionError("must not download an unpinned archive") + + monkeypatch.setattr(M, "download_file", boom) + with pytest.raises(M.UnpinnedNodeRefused): + M.install_prebuilt(install_dir, channel = channel, min_major = 24, force = False) + + +def test_unpinned_refusal_maps_to_fallback_exit_code(tmp_path: Path, monkeypatch, capsys): + # main() must surface the refusal as EXIT_FALLBACK (setup treats it as a failed + # install with guidance), not as a success masked by the keep-existing path. + install_dir = tmp_path / "node" + install_dir.mkdir() + M.write_metadata(install_dir, version = "24.9.0", asset = "old", sha256 = "old") + monkeypatch.setattr(M, "detect_host", lambda: _host("linux", "x64")) + monkeypatch.setattr(M, "fetch_json", lambda url: INDEX) + monkeypatch.setattr(M, "installed_node_version", lambda d, h: "24.9.0") + monkeypatch.setattr(M, "installed_npm_major", lambda d, h: 11) + monkeypatch.delenv(M.ALLOW_UNVERIFIED_ENV, raising = False) + rc = M.main(["--install-dir", str(install_dir), "--node-version", "latest"]) + assert rc == M.EXIT_FALLBACK + # Guard the main() catch order: UnpinnedNodeRefused must be caught before the + # generic PrebuiltFallback, so assert the message, not just the exit code. + out = capsys.readouterr().out + assert "refusing to install Node" in out + assert "prebuilt unavailable" not in out + + +def test_resolve_expected_sha256_rejects_malformed_pins(): + # A structurally-broken manifest is "not pinned" (None), never a bogus digest. + asset = "node-v24.17.0-linux-x64.tar.gz" + assert M.pinned_sha256({"versions": "nope"}, "24.17.0", asset) is None + assert M.pinned_sha256({"versions": {"24.17.0": "nope"}}, "24.17.0", asset) is None + assert M.pinned_sha256({"versions": {"24.17.0": {asset: "x" * 63}}}, "24.17.0", asset) is None + assert M.pinned_sha256({"versions": {"24.17.0": {asset: "z" * 64}}}, "24.17.0", asset) is None + # an uppercase but otherwise valid digest is normalized to lowercase + up = "A" * 64 + assert M.pinned_sha256({"versions": {"24.17.0": {asset: up}}}, "24.17.0", asset) == up.lower() + for bad in [{}, {"default_version": ""}, {"default_version": "not-a-version"}]: + with pytest.raises(PrebuiltFallback): + M.pinned_default_version(bad) + + +def test_install_prebuilt_optin_takes_remote_shasums_path(tmp_path: Path, monkeypatch): + # With the opt-in set, an unpinned version drives the remote-SHASUMS path end to + # end: fetch SHASUMS256.txt, then the verified archive download (no refusal). + install_dir = tmp_path / "node" # nothing on disk -> errors re-raise, not keep-existing + asset = "node-v26.3.1-linux-x64.tar.gz" + monkeypatch.setattr(M, "detect_host", lambda: _host("linux", "x64")) + monkeypatch.setattr(M, "fetch_json", lambda url: INDEX) # latest = 26.3.1 (unpinned) + monkeypatch.setenv(M.ALLOW_UNVERIFIED_ENV, "1") + shasums_fetched = {"n": 0} + + def fake_shasums(url, **k): + shasums_fetched["n"] += 1 + return f"{'d' * 64} {asset}\n".encode() + + class _ReachedDownload(Exception): + pass + + def reached(*a, **k): + raise _ReachedDownload + + monkeypatch.setattr(M, "download_bytes", fake_shasums) + monkeypatch.setattr(M, "download_file_verified", reached) + with pytest.raises(_ReachedDownload): + M.install_prebuilt(install_dir, channel = "latest", min_major = 24, force = False) + assert shasums_fetched["n"] == 1 # the opt-in path fetched the remote SHASUMS + + +def test_pins_manifest_ships_next_to_installer(): + # The committed manifest must sit beside the installer so __file__ resolution finds it. + assert M.pins_path() == MODULE_PATH.parent / M.PINS_FILENAME + assert M.pins_path().is_file() + + +def test_pins_manifest_is_declared_in_package_data(): + # An unpackaged trust anchor is no trust anchor: a pip install must ship it. + # tomllib is stdlib only on 3.11+; fall back to tomli, else skip on 3.9/3.10. + tomllib = pytest.importorskip("tomllib" if sys.version_info >= (3, 11) else "tomli") + + data = tomllib.loads((PACKAGE_ROOT / "pyproject.toml").read_text(encoding = "utf-8")) + studio_globs = data["tool"]["setuptools"]["package-data"]["studio"] + assert M.PINS_FILENAME in studio_globs + + +def test_existing_install_matches_enforces_expected_sha(tmp_path: Path, monkeypatch): + # The short-circuit must not keep a version-matching install whose recorded digest + # is not the pin (old remote-SHASUMS install or a tampered artifact). + host = _host("linux", "x64") + M.write_metadata(tmp_path, version = "24.17.0", asset = "x", sha256 = "aa") + monkeypatch.setattr(M, "installed_node_version", lambda d, h: "24.17.0") + monkeypatch.setattr(M, "installed_npm_major", lambda d, h: 11) + assert M.existing_install_matches(tmp_path, host, version = "24.17.0") is True # back-compat + assert M.existing_install_matches(tmp_path, host, version = "24.17.0", expected_sha = "aa") is True + assert M.existing_install_matches(tmp_path, host, version = "24.17.0", expected_sha = "bb") is False + + +def test_install_prebuilt_refuses_existing_unpinned_install(tmp_path: Path, monkeypatch): + # Codex P2: an unpinned version already on disk must still fail closed without the + # opt-in, not be kept by the version-only short-circuit. + install_dir = tmp_path / "node" + install_dir.mkdir() + M.write_metadata(install_dir, version = "26.3.1", asset = "a", sha256 = "s") + monkeypatch.setattr(M, "detect_host", lambda: _host("linux", "x64")) + monkeypatch.setattr(M, "installed_node_version", lambda d, h: "26.3.1") + monkeypatch.setattr(M, "installed_npm_major", lambda d, h: 11) + monkeypatch.delenv(M.ALLOW_UNVERIFIED_ENV, raising = False) + + def boom(*a, **k): + raise AssertionError("must not keep or download an unpinned install") + + monkeypatch.setattr(M, "download_file", boom) + with pytest.raises(M.UnpinnedNodeRefused): + M.install_prebuilt(install_dir, channel = "26.3.1", min_major = 24, force = False) + + +def test_pinned_target_wrong_sha_not_kept_when_download_fails(tmp_path: Path, monkeypatch): + # Symmetry with the short-circuit guard: the transient-failure fallback must not + # keep a same-version install whose recorded digest is not the pin. (A different + # usable version is still kept for offline resilience -- covered above.) + host = _host("linux", "x64") + version = M.pinned_default_version(M.load_pins()) + asset = M.node_asset_name(version, host) + install_dir = tmp_path / "node" + install_dir.mkdir() + M.write_metadata(install_dir, version = version, asset = asset, sha256 = "0" * 64) # not the pin + monkeypatch.setattr(M, "detect_host", lambda: host) + monkeypatch.setattr(M, "installed_node_version", lambda d, h: version) + monkeypatch.setattr(M, "installed_npm_major", lambda d, h: 11) + monkeypatch.setattr(M, "download_file_verified", _offline) # transient download failure + with pytest.raises(OSError): + M.install_prebuilt(install_dir, channel = "pinned", min_major = 24, force = False)