Harden standalone updates with required checksums and private staging

This commit is contained in:
utku 2026-10-24 21:55:13 -07:00
parent 200a501f15
commit 2bd118280e
4 changed files with 469 additions and 49 deletions

View file

@ -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

View file

@ -11,6 +11,21 @@ strix (--target <target> | --target-list <path>) [options]
## Options
<ParamField path="--update" type="boolean">
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.
</ParamField>
<ParamField path="--target, -t" type="string">
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://<collection-uuid>`). Can be specified multiple times. Fresh runs require at least one target source: `--target` or `--target-list`.

View file

@ -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())

368
tests/test_self_update.py Normal file
View file

@ -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)