From 2bd118280ec05831beaf87cc6eee433772b144f3 Mon Sep 17 00:00:00 2001 From: utku Date: Sat, 24 Oct 2026 21:55:13 -0700 Subject: [PATCH] Harden standalone updates with required checksums and private staging --- README.md | 2 + docs/usage/cli.mdx | 15 ++ strix/interface/update_check.py | 133 +++++++----- tests/test_self_update.py | 368 ++++++++++++++++++++++++++++++++ 4 files changed, 469 insertions(+), 49 deletions(-) create mode 100644 tests/test_self_update.py diff --git a/README.md b/README.md index 5134c8bd1..f2ea7283c 100644 --- a/README.md +++ b/README.md @@ -106,6 +106,8 @@ curl -sSL https://strix.ai/install | bash Installed with pipx, uv, or pip? `strix --update` prints the matching upgrade command instead (`pipx upgrade strix-agent`, `uv tool upgrade strix-agent`, `pip install --upgrade strix-agent`). Strix also shows a notice with the right command when a new version is available. +Standalone self-updates verify the download against the SHA-256 digest in GitHub's release metadata before replacing the installed binary. If the digest is unavailable or does not match, the update stops and keeps the current installation. Resolve the error and retry `strix --update`. + --- ## Ways to Run Strix diff --git a/docs/usage/cli.mdx b/docs/usage/cli.mdx index 2fc492e9b..efa8e7de8 100644 --- a/docs/usage/cli.mdx +++ b/docs/usage/cli.mdx @@ -11,6 +11,21 @@ strix (--target | --target-list ) [options] ## Options + + Update the standalone binary to the latest stable GitHub release and exit. No + target, Docker daemon, or LLM credentials are required. For pip, pipx, and uv + installs, print the corresponding package-manager upgrade command instead. + + The standalone updater requires a valid SHA-256 digest from GitHub's release + metadata and verifies the archive before installing it. Unavailable metadata, + checksum mismatches, download failures, and invalid archives leave the current + binary in place and exit with code `1`. Retry after resolving the error. + + Updates are staged in a private directory next to the installed binary, so that + directory must be writable. Windows keeps the previous executable as + `strix.exe.old` until the next update and restores it if replacement fails. + + Target to test. Accepts URLs, repositories, local directories, domains, IP addresses, API spec files (OpenAPI/Swagger `.json`/`.yaml`, a Postman collection export), or a live Postman collection by id (`postman://`). Can be specified multiple times. Fresh runs require at least one target source: `--target` or `--target-list`. diff --git a/strix/interface/update_check.py b/strix/interface/update_check.py index 9e98b38fd..7ccecc998 100644 --- a/strix/interface/update_check.py +++ b/strix/interface/update_check.py @@ -14,6 +14,7 @@ import json import logging import os import platform +import re import shutil import stat import subprocess @@ -117,7 +118,7 @@ def _fetch_latest_version() -> str | None: def _fetch_asset_digest(version: str, filename: str) -> str | None: - """Return the expected sha256 (hex) for a release asset, if the API provides one.""" + """Return a valid SHA-256 for the exact release asset, or None on failure.""" try: with requests.get( f"https://api.github.com/repos/{GITHUB_REPO}/releases/tags/v{version}", @@ -128,8 +129,8 @@ def _fetch_asset_digest(version: str, filename: str) -> str | None: for asset in assets: if asset.get("name") == filename: digest = asset.get("digest") or "" - if digest.startswith("sha256:"): - return digest.removeprefix("sha256:") + if isinstance(digest, str) and re.fullmatch(r"sha256:[0-9a-fA-F]{64}", digest): + return digest.removeprefix("sha256:").lower() except Exception: # noqa: BLE001 logger.debug("release asset digest lookup failed", exc_info=True) return None @@ -312,6 +313,60 @@ def _release_target() -> str | None: return target if target in supported else None +def _unpack_binary(archive_path: Path, binary_name: str, staged: Path, *, is_windows: bool) -> None: + """Copy only the expected regular file, without extracting archive paths or links.""" + if is_windows: + with zipfile.ZipFile(archive_path) as archive: + members = [member for member in archive.infolist() if member.filename == binary_name] + if len(members) != 1: + raise RuntimeError(f"Expected exactly one {binary_name} in the release archive") + member = members[0] + file_type = stat.S_IFMT(member.external_attr >> 16) + if member.is_dir() or file_type not in (0, stat.S_IFREG) or member.file_size <= 0: + raise RuntimeError(f"Release binary {binary_name} must be a nonempty regular file") + with archive.open(member) as source, staged.open("xb") as destination: + shutil.copyfileobj(source, destination) + destination.flush() + os.fsync(destination.fileno()) + else: + with tarfile.open(archive_path, "r:gz") as archive: + tar_members = [member for member in archive.getmembers() if member.name == binary_name] + if len(tar_members) != 1: + raise RuntimeError(f"Expected exactly one {binary_name} in the release archive") + tar_member = tar_members[0] + if not tar_member.isfile() or tar_member.size <= 0: + raise RuntimeError(f"Release binary {binary_name} must be a nonempty regular file") + tar_source = archive.extractfile(tar_member) + if tar_source is None: + raise RuntimeError(f"Could not read release binary {binary_name}") + with tar_source, staged.open("xb") as destination: + shutil.copyfileobj(tar_source, destination) + destination.flush() + os.fsync(destination.fileno()) + + +def _replace_binary(staged: Path, current_exe: Path, *, is_windows: bool) -> None: + if not is_windows: + staged.replace(current_exe) + return + + # Windows cannot replace a running executable. Retain it for rollback; + # the next update can remove it once the old process has exited. + old = current_exe.with_name(current_exe.name + ".old") + old.unlink(missing_ok=True) + try: + current_exe.rename(old) + staged.replace(current_exe) + except BaseException: + # Restore the installed path on interruption as well as filesystem errors. + try: + if not current_exe.exists(): + old.rename(current_exe) + except OSError as exc: + raise RuntimeError(f"Update failed; restore the previous binary from {old}") from exc + raise + + def _download_and_replace(version: str, target: str, console: Console) -> bool: is_windows = target.startswith("windows") archive_ext = ".zip" if is_windows else ".tar.gz" @@ -320,7 +375,16 @@ def _download_and_replace(version: str, target: str, console: Console) -> bool: binary_name = f"strix-{version}-{target}" + (".exe" if is_windows else "") current_exe = Path(sys.executable).resolve() - with tempfile.TemporaryDirectory() as tmp: + expected_digest = _fetch_asset_digest(version, filename) + if expected_digest is None: + raise RuntimeError( + f"Cannot verify {filename}: release metadata is unavailable or has no valid " + "SHA-256 checksum. Try again when GitHub release metadata is available." + ) + + # A private directory avoids clobbering a shared .new file, and keeps the + # replacement on the installation's filesystem for an atomic POSIX rename. + with tempfile.TemporaryDirectory(prefix=".strix-update-", dir=current_exe.parent) as tmp: tmp_dir = Path(tmp) archive_path = tmp_dir / filename console.print(f"[dim]Downloading[/] {url}") @@ -334,45 +398,19 @@ def _download_and_replace(version: str, target: str, console: Console) -> bool: for chunk in response.iter_content(chunk_size=1 << 20): f.write(chunk) - expected_digest = _fetch_asset_digest(version, filename) - if expected_digest: - actual_digest = _sha256_file(archive_path) - if actual_digest != expected_digest: - raise RuntimeError( - f"checksum mismatch for {filename}: " - f"expected sha256 {expected_digest}, got {actual_digest}" - ) - else: - console.print("[dim yellow]No published checksum available; skipping verification[/]") + actual_digest = _sha256_file(archive_path) + if actual_digest != expected_digest: + raise RuntimeError( + f"checksum mismatch for {filename}: " + f"expected sha256 {expected_digest}, got {actual_digest}" + ) - if is_windows: - with zipfile.ZipFile(archive_path) as zf: - zf.extract(binary_name, tmp_dir) - else: - with tarfile.open(archive_path, "r:gz") as tf: - tf.extract(binary_name, tmp_dir, filter="data") - - new_binary = tmp_dir / binary_name - new_binary.chmod(new_binary.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH) - - staged = current_exe.with_name(current_exe.name + ".new") - try: - shutil.copy2(new_binary, staged) - if is_windows: - # Windows can't replace a running executable in place; move it aside first. - old = current_exe.with_name(current_exe.name + ".old") - old.unlink(missing_ok=True) - current_exe.rename(old) - try: - staged.replace(current_exe) - except Exception: - old.rename(current_exe) - raise - else: - staged.replace(current_exe) - except Exception: - staged.unlink(missing_ok=True) - raise + staged = tmp_dir / binary_name + _unpack_binary(archive_path, binary_name, staged, is_windows=is_windows) + # Preserve local access restrictions, but never carry setuid/setgid or + # archive-supplied permissions into a newly downloaded executable. + staged.chmod(stat.S_IMODE(current_exe.stat().st_mode) & 0o777) + _replace_binary(staged, current_exe, is_windows=is_windows) return True @@ -392,9 +430,9 @@ def self_update(console: Console | None = None, version: str | None = None) -> b ) return False - latest = version or _fetch_latest_version() - if not latest: - console.print("[bold red]Could not determine the latest strix version.[/]") + latest = (version or _fetch_latest_version() or "").removeprefix("v") + if re.fullmatch(r"[0-9]+\.[0-9]+\.[0-9]+", latest) is None: + console.print("[bold red]Could not determine a valid stable strix release version.[/]") return False current = get_version() @@ -415,10 +453,7 @@ def self_update(console: Console | None = None, version: str | None = None) -> b except Exception as e: # noqa: BLE001 logger.debug("self-update failed", exc_info=True) console.print(f"[bold red]Update failed:[/] {e}") - console.print( - "[dim]You can reinstall manually with:[/] " - "[#60a5fa]curl -sSL https://strix.ai/install | bash[/]" - ) + console.print("[dim]Resolve the error, then retry with:[/] [#60a5fa]strix --update[/]") return False _write_cache(latest_version=latest, checked_at=time.time()) diff --git a/tests/test_self_update.py b/tests/test_self_update.py new file mode 100644 index 000000000..ae5c471cc --- /dev/null +++ b/tests/test_self_update.py @@ -0,0 +1,368 @@ +"""Exercise binary updates with real archives and an isolated installation.""" + +from __future__ import annotations + +import errno +import hashlib +import io +import json +import stat +import sys +import tarfile +import zipfile +from dataclasses import dataclass, field +from pathlib import Path +from typing import TYPE_CHECKING, Self + +import pytest +import requests +from rich.console import Console + +from strix.interface import cli_args, update_check + + +if TYPE_CHECKING: + from collections.abc import Iterator + + +VERSION = "1.8.0" +OLD_BINARY = b"previous executable" +NEW_BINARY = b"new executable" + + +@dataclass +class _Response: + payload: object = None + content: bytes = b"" + stream_error: Exception | None = None + + def __enter__(self) -> Self: + return self + + def __exit__(self, *_args: object) -> None: + pass + + def raise_for_status(self) -> None: + pass + + def json(self) -> object: + return self.payload + + def iter_content(self, chunk_size: int) -> Iterator[bytes]: + for offset in range(0, len(self.content), chunk_size): + yield self.content[offset : offset + chunk_size] + if self.stream_error: + raise self.stream_error + + +@dataclass +class _Release: + executable: Path + target: str + archive: bytes = b"" + digest: object = None + metadata_error: Exception | None = None + stream_error: Exception | None = None + urls: list[str] = field(default_factory=list) + output: io.StringIO = field(default_factory=io.StringIO) + + @property + def binary_name(self) -> str: + suffix = ".exe" if self.target.startswith("windows") else "" + return f"strix-{VERSION}-{self.target}{suffix}" + + @property + def filename(self) -> str: + suffix = ".zip" if self.target.startswith("windows") else ".tar.gz" + return f"strix-{VERSION}-{self.target}{suffix}" + + def set_archive(self, kind: str = "regular") -> None: + buffer = io.BytesIO() + name = "unexpected" if kind == "missing" else self.binary_name + content = b"" if kind == "empty" else NEW_BINARY + if self.target.startswith("windows"): + with zipfile.ZipFile(buffer, "w") as archive: + member = zipfile.ZipInfo(name) + member.create_system = 3 + member.external_attr = (stat.S_IFREG | 0o755) << 16 + if kind == "symlink": + member.external_attr = (stat.S_IFLNK | 0o777) << 16 + elif kind == "directory": + member.external_attr = (stat.S_IFDIR | 0o755) << 16 + archive.writestr(member, content) + if kind == "duplicate": + with pytest.warns(UserWarning, match="Duplicate name"): + archive.writestr(member, content) + else: + with tarfile.open(fileobj=buffer, mode="w:gz") as archive: + member = tarfile.TarInfo(name) + member.mode = 0o7777 + member.size = len(content) + if kind in {"symlink", "hardlink"}: + member.type = tarfile.SYMTYPE if kind == "symlink" else tarfile.LNKTYPE + member.linkname = "payload" + member.size = 0 + payload = tarfile.TarInfo("payload") + payload.size = len(content) + archive.addfile(payload, io.BytesIO(content)) + elif kind == "directory": + member.type = tarfile.DIRTYPE + member.size = 0 + archive.addfile(member, io.BytesIO(content)) + if kind == "duplicate": + archive.addfile(member, io.BytesIO(content)) + self.archive = buffer.getvalue() + self.digest = "sha256:" + hashlib.sha256(self.archive).hexdigest() + + def get(self, url: str, **_kwargs: object) -> _Response: + self.urls.append(url) + if url == f"https://api.github.com/repos/usestrix/strix/releases/tags/v{VERSION}": + if self.metadata_error: + raise self.metadata_error + return _Response(payload={"assets": [{"name": self.filename, "digest": self.digest}]}) + assert url == ( + f"https://github.com/usestrix/strix/releases/download/v{VERSION}/{self.filename}" + ) + return _Response(content=self.archive, stream_error=self.stream_error) + + def update(self, version: str = VERSION) -> bool: + return update_check.self_update(Console(file=self.output), version=version) + + def assert_preserved(self) -> None: + assert self.executable.read_bytes() == OLD_BINARY + assert list(self.executable.parent.iterdir()) == [self.executable] + assert not update_check._CACHE_PATH.exists() + + +@pytest.fixture(params=["linux-x86_64", "macos-arm64", "windows-x86_64"]) +def release( + request: pytest.FixtureRequest, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> _Release: + install = tmp_path / "install with spaces" + install.mkdir() + executable = install / ("strix.exe" if request.param.startswith("windows") else "strix") + executable.write_bytes(OLD_BINARY) + executable.chmod(0o700) + result = _Release(executable=executable, target=request.param) + result.set_archive() + monkeypatch.setattr(sys, "executable", str(executable)) + monkeypatch.setattr(update_check, "is_binary_install", lambda: True) + monkeypatch.setattr(update_check, "get_version", lambda: "1.7.0") + monkeypatch.setattr(update_check, "_release_target", lambda: result.target) + monkeypatch.setattr(update_check, "_CACHE_PATH", tmp_path / "cache.json") + monkeypatch.setattr(update_check.requests, "get", result.get) + return result + + +def test_verified_update_installs_only_expected_binary(release: _Release) -> None: + assert release.update() is True + assert release.executable.read_bytes() == NEW_BINARY + assert release.urls[0].startswith("https://api.github.com/") + if sys.platform != "win32": + assert stat.S_IMODE(release.executable.stat().st_mode) == 0o700 + assert json.loads(update_check._CACHE_PATH.read_text())["latest_version"] == VERSION + expected_paths = {release.executable} + if release.target.startswith("windows"): + backup = release.executable.with_name(release.executable.name + ".old") + assert backup.read_bytes() == OLD_BINARY + expected_paths.add(backup) + assert set(release.executable.parent.iterdir()) == expected_paths + + +@pytest.mark.parametrize("digest", [None, "", "sha256:", "sha256:xyz", "md5:" + "a" * 32, 123]) +def test_missing_or_invalid_checksum_never_downloads(release: _Release, digest: object) -> None: + release.digest = digest + assert release.update() is False + assert len(release.urls) == 1 + assert "checksum" in release.output.getvalue().lower() + release.assert_preserved() + + +@pytest.mark.parametrize("error", [requests.Timeout("offline"), requests.HTTPError("403")]) +def test_metadata_failure_never_downloads(release: _Release, error: Exception) -> None: + release.metadata_error = error + assert release.update() is False + assert len(release.urls) == 1 + release.assert_preserved() + + +def test_checksum_mismatch_preserves_installation(release: _Release) -> None: + release.digest = "sha256:" + "0" * 64 + assert release.update() is False + assert "checksum mismatch" in release.output.getvalue() + release.assert_preserved() + + +def test_interrupted_download_preserves_installation(release: _Release) -> None: + release.stream_error = requests.ConnectionError("download interrupted") + assert release.update() is False + release.assert_preserved() + + +@pytest.mark.parametrize("kind", ["missing", "empty", "directory", "symlink", "duplicate"]) +def test_invalid_archive_member_preserves_installation(release: _Release, kind: str) -> None: + release.set_archive(kind) + assert release.update() is False + release.assert_preserved() + + +def test_tar_hardlink_is_not_an_executable(release: _Release) -> None: + if release.target.startswith("windows"): + pytest.skip("ZIP does not have tar hardlinks") + release.set_archive("hardlink") + assert release.update() is False + release.assert_preserved() + + +def test_corrupt_archive_preserves_installation(release: _Release) -> None: + release.archive = b"not an archive" + release.digest = "sha256:" + hashlib.sha256(release.archive).hexdigest() + assert release.update() is False + release.assert_preserved() + + +def test_update_does_not_overwrite_existing_staging_file(release: _Release) -> None: + unrelated = release.executable.with_name(release.executable.name + ".new") + unrelated.write_bytes(b"unrelated file") + assert release.update() is True + assert release.executable.read_bytes() == NEW_BINARY + assert unrelated.read_bytes() == b"unrelated file" + + +def test_staging_write_failure_preserves_installation( + release: _Release, monkeypatch: pytest.MonkeyPatch +) -> None: + def fail_copy(*_args: object, **_kwargs: object) -> None: + raise OSError(errno.ENOSPC, "disk full") + + monkeypatch.setattr(update_check.shutil, "copyfileobj", fail_copy) + assert release.update() is False + release.assert_preserved() + + +@pytest.mark.parametrize("interrupted", [False, True]) +def test_failed_replacement_restores_previous_binary( + release: _Release, monkeypatch: pytest.MonkeyPatch, interrupted: bool +) -> None: + original_replace = Path.replace + + def fail_install(path: Path, target: str | Path) -> Path: + if Path(target) == release.executable and path.name != release.executable.name + ".old": + assert path.parent.parent == release.executable.parent + if interrupted: + raise KeyboardInterrupt + raise OSError(errno.EACCES, "replacement denied") + return original_replace(path, target) + + monkeypatch.setattr(Path, "replace", fail_install) + if interrupted: + with pytest.raises(KeyboardInterrupt): + release.update() + else: + assert release.update() is False + release.assert_preserved() + + +@pytest.mark.parametrize("version", ["../../other", "1.8.0/other", "1.8.0?x=y", "garbage"]) +def test_invalid_release_version_never_contacts_network(release: _Release, version: str) -> None: + assert release.update(version) is False + assert release.urls == [] + release.assert_preserved() + + +def test_uppercase_digest_and_prefixed_version_are_accepted(release: _Release) -> None: + release.digest = "sha256:" + hashlib.sha256(release.archive).hexdigest().upper() + assert release.update("v" + VERSION) is True + assert release.executable.read_bytes() == NEW_BINARY + + +@pytest.mark.parametrize("metadata", [{"assets": []}, {"assets": None}, [], None]) +def test_unusable_release_metadata_preserves_installation( + release: _Release, monkeypatch: pytest.MonkeyPatch, metadata: object +) -> None: + def get_metadata(url: str, **_kwargs: object) -> _Response: + assert url.startswith("https://api.github.com/") + return _Response(payload=metadata) + + monkeypatch.setattr(update_check.requests, "get", get_metadata) + assert release.update() is False + release.assert_preserved() + + +def test_failed_update_preserves_existing_cache(release: _Release) -> None: + original_cache = json.dumps({"latest_version": VERSION, "skipped_version": "1.7.1"}) + update_check._CACHE_PATH.write_text(original_cache) + release.digest = None + assert release.update() is False + assert update_check._CACHE_PATH.read_text() == original_cache + assert release.executable.read_bytes() == OLD_BINARY + + +def test_existing_staging_symlink_is_not_followed(release: _Release, tmp_path: Path) -> None: + unrelated = tmp_path / "unrelated" + unrelated.write_bytes(b"keep this file") + link = release.executable.with_name(release.executable.name + ".new") + try: + link.symlink_to(unrelated) + except OSError: + pytest.skip("symlink creation is unavailable") + assert release.update() is True + assert unrelated.read_bytes() == b"keep this file" + assert link.is_symlink() + + +@pytest.mark.parametrize("release", ["windows-x86_64"], indirect=True) +def test_windows_interruption_after_backup_move_restores_binary( + release: _Release, monkeypatch: pytest.MonkeyPatch +) -> None: + original_rename = Path.rename + + def interrupt_after_move(path: Path, target: str | Path) -> Path: + result = original_rename(path, target) + if path == release.executable: + raise KeyboardInterrupt + return result + + monkeypatch.setattr(Path, "rename", interrupt_after_move) + with pytest.raises(KeyboardInterrupt): + release.update() + release.assert_preserved() + + +@pytest.mark.parametrize("release", ["windows-x86_64"], indirect=True) +def test_windows_failed_rollback_retains_backup_and_reports_location( + release: _Release, monkeypatch: pytest.MonkeyPatch +) -> None: + original_rename = Path.rename + + def deny_replace(_path: Path, _target: str | Path) -> Path: + raise PermissionError("replacement denied") + + def deny_restore(path: Path, target: str | Path) -> Path: + if Path(target) == release.executable: + raise PermissionError("restore denied") + return original_rename(path, target) + + monkeypatch.setattr(Path, "replace", deny_replace) + monkeypatch.setattr(Path, "rename", deny_restore) + assert release.update() is False + backup = release.executable.with_name(release.executable.name + ".old") + assert backup.read_bytes() == OLD_BINARY + assert not release.executable.exists() + assert "restore the previous binary" in release.output.getvalue() + assert backup.name in release.output.getvalue() + assert set(release.executable.parent.iterdir()) == {backup} + + +@pytest.mark.parametrize("verified", [True, False]) +def test_update_cli_exits_without_requiring_a_scan_target( + release: _Release, monkeypatch: pytest.MonkeyPatch, verified: bool +) -> None: + monkeypatch.setattr(sys, "argv", ["strix", "--update"]) + monkeypatch.setattr(update_check, "_fetch_latest_version", lambda: VERSION) + if not verified: + release.digest = None + with pytest.raises(SystemExit) as exc: + cli_args.parse_arguments() + assert exc.value.code == (0 if verified else 1) + assert release.executable.read_bytes() == (NEW_BINARY if verified else OLD_BINARY)