From ce580989ef87aacbf6784b5f9e577ec15ad5b888 Mon Sep 17 00:00:00 2001 From: LuckVd <37114923+LuckVd@users.noreply.github.com> Date: Tue, 1 Sep 2026 14:07:17 +0800 Subject: [PATCH] fix(resume): re-validate persisted mounts with the mount guard --resume rebuilt sandbox bind mounts from run.json but skipped check_mountable_dir for workspace_mount and repository cloned_repo_path (local_code paths were already re-checked), and workspace_files sources were not guarded at all. run.json lives on disk under the scanned tree and must be treated as untrusted: a tampered record could mount '/' or $HOME writable, or copy credential files into the sandbox. Apply the same guard on resume: - workspace_mount and cloned_repo_path now pass through check_mountable_dir - workspace_files sources pass through the new check_mountable_file, which applies the same system/home/credential checks to a single file - split the shared root/tree/credential sets into _mount_guard_sets used by both guards Adds unit coverage for check_mountable_file and resume integration tests for the three bypass paths. --- strix/interface/cli_args.py | 46 +++++++++---- strix/interface/utils.py | 60 +++++++++++++--- tests/test_cli_resume.py | 133 ++++++++++++++++++++++++++++++++++++ tests/test_local_sources.py | 59 ++++++++++++++++ 4 files changed, 276 insertions(+), 22 deletions(-) create mode 100644 tests/test_cli_resume.py diff --git a/strix/interface/cli_args.py b/strix/interface/cli_args.py index c9b6951be..75458891a 100644 --- a/strix/interface/cli_args.py +++ b/strix/interface/cli_args.py @@ -14,6 +14,7 @@ from strix.interface.scan_setup import attach_workspace_mount, build_targets_inf from strix.interface.update_check import self_update from strix.interface.utils import ( check_mountable_dir, + check_mountable_file, collect_local_sources, resolve_workspace_files, validate_config_file, @@ -487,12 +488,17 @@ def load_resume_state(args: argparse.Namespace) -> None: cloned = details.get("cloned_repo_path") if not cloned: continue - if not Path(cloned).expanduser().exists(): + cloned_path = Path(cloned).expanduser() + if not cloned_path.exists(): raise ResumeError( f"--resume {args.resume}: cloned repo at {cloned} is missing. " f"It was deleted between runs. Pick a fresh --run-name to " f"re-clone, or restore the directory before resuming." ) + try: + check_mountable_dir(cloned_path) + except ValueError as exc: + raise ResumeError(f"--resume {args.resume}: {exc}") from exc if args.instruction is None: args.instruction = state.get("instruction") @@ -500,32 +506,46 @@ def load_resume_state(args: argparse.Namespace) -> None: args.user_instruction = state.get("user_instruction") or None args.local_sources = collect_local_sources(args.targets_info) # Remount the workspace the run was started with. The user already confirmed - # this directory, so the target mount guard does not apply to it; it only has - # to still be there. + # this directory on the original run, but run.json is on disk under the + # scanned tree and may have been tampered with, so the target mount guard + # still applies on resume: it has to both still be there and be mountable. args.workspace_mount = workspace_mount # Replace the workspace files the run started with, unless this resume names # its own. The persisted record is revalidated like a fresh flag, so an - # edited run.json cannot widen what a resume places. A file deleted between - # runs is dropped rather than fatal: it is context for the agent, not scope. + # edited run.json cannot widen what a resume places. Each source is also + # passed through the mount guard, so a tampered run.json cannot smuggle + # credential or system files into the sandbox. A file deleted between runs + # is dropped rather than fatal: it is context for the agent, not scope. if not getattr(args, "workspace_files", None): - restored = [ - f"{source_path}:{workspace_path}" - for workspace_file in state.get("workspace_files") or [] - if isinstance(workspace_file, dict) - and (source_path := Path(str(workspace_file.get("source_path") or ""))).is_file() - and (workspace_path := str(workspace_file.get("workspace_path") or "")) - ] + restored: list[str] = [] + for workspace_file in state.get("workspace_files") or []: + if not isinstance(workspace_file, dict): + continue + source_path = Path(str(workspace_file.get("source_path") or "")).expanduser() + workspace_path = str(workspace_file.get("workspace_path") or "") + if not source_path.is_file() or not workspace_path: + continue + try: + check_mountable_file(source_path) + except ValueError as error: + raise ResumeError(f"--resume {args.resume}: {error}") from error + restored.append(f"{source_path}:{workspace_path}") try: args.workspace_files = resolve_workspace_files(restored) except ValueError as error: raise ResumeError(f"--resume {args.resume}: invalid workspace file: {error}") from error if workspace_mount: - if not Path(workspace_mount).expanduser().is_dir(): + mount_path = Path(workspace_mount).expanduser() + if not mount_path.is_dir(): raise ResumeError( f"--resume {args.resume}: the working directory {workspace_mount} " f"is missing. Restore it before resuming, or start a fresh run." ) + try: + check_mountable_dir(mount_path) + except ValueError as exc: + raise ResumeError(f"--resume {args.resume}: {exc}") from exc attach_workspace_mount(args) if state.get("diff_scope"): args.diff_scope = state.get("diff_scope") diff --git a/strix/interface/utils.py b/strix/interface/utils.py index 7ab55d8ee..8570cffef 100644 --- a/strix/interface/utils.py +++ b/strix/interface/utils.py @@ -1403,6 +1403,23 @@ def _is_within(path: Path, ancestor: Path) -> bool: return path_parts[: len(ancestor_parts)] == ancestor_parts +def _mount_guard_sets(anchor: Path) -> tuple[set[str], list[Path]]: + """Shared exact-match and tree sets used by both mount guards. + + ``anchor`` only locates the drive for the Windows-specific checks. + """ + exact = {str(Path(root)).casefold() for root in _FORBIDDEN_MOUNT_ROOTS} + exact |= {str(Path(root).resolve()).casefold() for root in _FORBIDDEN_MOUNT_ROOTS} + exact.add(str(Path.home().resolve()).casefold()) + tree_roots = set(_FORBIDDEN_MOUNT_TREES) + if os.name == "nt": + drive = Path(anchor.anchor) + tree_roots |= {str(drive / name) for name in _FORBIDDEN_WINDOWS_TREE_NAMES} + exact.add(str(drive / "Users").casefold()) + trees = [Path(root) for root in tree_roots] + [Path(root).resolve() for root in tree_roots] + return exact, trees + + def check_mountable_dir(path: Path) -> None: resolved = path.resolve() if not resolved.is_dir(): @@ -1410,15 +1427,7 @@ def check_mountable_dir(path: Path) -> None: # Both the literal and the resolved form: macOS reaches /etc through the # /private/etc symlink, and only the resolved path is compared below. - exact = {str(Path(root)).casefold() for root in _FORBIDDEN_MOUNT_ROOTS} - exact |= {str(Path(root).resolve()).casefold() for root in _FORBIDDEN_MOUNT_ROOTS} - exact.add(str(Path.home().resolve()).casefold()) - tree_roots = set(_FORBIDDEN_MOUNT_TREES) - if os.name == "nt": - drive = Path(resolved.anchor) - tree_roots |= {str(drive / name) for name in _FORBIDDEN_WINDOWS_TREE_NAMES} - exact.add(str(drive / "Users").casefold()) - trees = [Path(root) for root in tree_roots] + [Path(root).resolve() for root in tree_roots] + exact, trees = _mount_guard_sets(resolved) if ( str(resolved).casefold() in exact or resolved.parent == resolved @@ -1440,6 +1449,39 @@ def check_mountable_dir(path: Path) -> None: ) +def check_mountable_file(path: Path) -> None: + """Validate a single host file may be mounted into the sandbox as a + workspace file. + + Applies the same system/home/credential checks that ``check_mountable_dir`` + applies to a directory, against the file's parent directory and its own + path components, so credential files and system files cannot be smuggled + into a run through workspace files. + """ + resolved = path.expanduser().resolve() + if not resolved.is_file(): + raise ValueError(f"'{path}' is not an existing file.") + + parent = resolved.parent + # Both the literal and the resolved form: macOS reaches /etc through the + # /private/etc symlink, and only the resolved path is compared below. + exact, trees = _mount_guard_sets(parent) + if str(parent).casefold() in exact or any(_is_within(parent, tree) for tree in trees): + raise ValueError( + f"Refusing to mount '{resolved}' into the sandbox: it is a system " + "or home path, not a project file." + ) + + credential = next( + (part for part in resolved.parts if part.casefold() in _FORBIDDEN_MOUNT_DIR_NAMES), None + ) + if credential is not None: + raise ValueError( + f"Refusing to mount '{resolved}' into the sandbox: '{credential}' " + "holds credentials, not code." + ) + + def dedupe_local_targets(targets_info: list[dict[str, Any]]) -> list[dict[str, Any]]: result: list[dict[str, Any]] = [] seen_paths: set[str] = set() diff --git a/tests/test_cli_resume.py b/tests/test_cli_resume.py new file mode 100644 index 000000000..4ca207d8b --- /dev/null +++ b/tests/test_cli_resume.py @@ -0,0 +1,133 @@ +"""--resume re-validates persisted mounts against the mount guard. + +run.json lives on disk under the scanned tree and is treated as untrusted on +resume: workspace_mount, repository cloned_repo_path and workspace_files +sources must all pass the same checks a fresh invocation would apply. +""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path + +import pytest + +from strix.interface.cli_args import ResumeError, load_resume_state + + +def _resume_args(**overrides: object) -> argparse.Namespace: + defaults: dict[str, object] = { + "resume": "old-run", + "targets_info": [], + "instruction": None, + "user_instruction": None, + "local_sources": [], + "workspace_mount": None, + "workspace_files": None, + "diff_scope": None, + "scan_mode": "quick", + } + defaults.update(overrides) + return argparse.Namespace(**defaults) + + +def _write_run(tmp_path: Path, record: dict[str, object]) -> None: + run_dir = tmp_path / "strix_runs" / "old-run" + (run_dir / ".state").mkdir(parents=True) + (run_dir / "run.json").write_text(json.dumps(record), encoding="utf-8") + # A run that got far enough to resume always has a first agent snapshot. + (run_dir / ".state" / "agents.json").write_text("{}", encoding="utf-8") + + +def _patch_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + home = tmp_path / "home" + home.mkdir() + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setattr(Path, "home", classmethod(lambda _cls: home)) + return home + + +def test_resume_rejects_forbidden_workspace_mount( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + home = _patch_home(tmp_path, monkeypatch) + monkeypatch.chdir(tmp_path) + _write_run( + tmp_path, + {"targets_info": [], "workspace_mount": str(home), "workspace_files": []}, + ) + + with pytest.raises(ResumeError, match="Refusing to mount"): + load_resume_state(_resume_args()) + + +def test_resume_rejects_system_workspace_mount( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.chdir(tmp_path) + _write_run( + tmp_path, + {"targets_info": [], "workspace_mount": "/etc", "workspace_files": []}, + ) + + with pytest.raises(ResumeError, match="Refusing to mount"): + load_resume_state(_resume_args()) + + +def test_resume_rejects_forbidden_cloned_repo_path( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.chdir(tmp_path) + _write_run( + tmp_path, + { + "targets_info": [{"type": "repository", "details": {"cloned_repo_path": "/etc"}}], + "workspace_mount": None, + "workspace_files": [], + }, + ) + + with pytest.raises(ResumeError, match="Refusing to mount"): + load_resume_state(_resume_args()) + + +def test_resume_rejects_credential_workspace_file( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + home = _patch_home(tmp_path, monkeypatch) + ssh = home / ".ssh" + ssh.mkdir() + key = ssh / "id_rsa" + key.write_text("key", encoding="utf-8") + monkeypatch.chdir(tmp_path) + _write_run( + tmp_path, + { + "targets_info": [], + "user_instruction": "scan the target", + "workspace_mount": None, + "workspace_files": [{"source_path": str(key), "workspace_path": "id_rsa"}], + }, + ) + + with pytest.raises(ResumeError, match="holds credentials"): + load_resume_state(_resume_args()) + + +def test_resume_accepts_a_clean_record(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.chdir(tmp_path) + _write_run( + tmp_path, + { + "targets_info": [], + "user_instruction": "scan the target", + "workspace_mount": None, + "workspace_files": [], + }, + ) + + args = _resume_args() + load_resume_state(args) + + assert args.user_instruction == "scan the target" diff --git a/tests/test_local_sources.py b/tests/test_local_sources.py index a799192de..54bc450a8 100644 --- a/tests/test_local_sources.py +++ b/tests/test_local_sources.py @@ -11,6 +11,7 @@ import pytest from strix.interface.scan_setup import attach_workspace_mount from strix.interface.utils import ( check_mountable_dir, + check_mountable_file, collect_local_sources, dedupe_local_targets, infer_target_type, @@ -266,3 +267,61 @@ def test_dedupe_collapses_the_same_path() -> None: assert dedupe_local_targets([_local_target("/repo"), _local_target("/repo")]) == [ _local_target("/repo") ] + + +def test_check_mountable_file_accepts_a_project_file(tmp_path: Path) -> None: + file = tmp_path / "notes.txt" + file.write_text("context", encoding="utf-8") + + check_mountable_file(file) + + +def test_check_mountable_file_rejects_missing_file(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="not an existing file"): + check_mountable_file(tmp_path / "nope.txt") + + +def test_check_mountable_file_rejects_system_file() -> None: + system_file = next( + (p for p in (Path("/etc/passwd"), Path("/usr/bin/env")) if p.is_file()), None + ) + if system_file is None: + pytest.skip("no system file on this platform") + with pytest.raises(ValueError, match="Refusing to mount"): + check_mountable_file(system_file) + + +def test_check_mountable_file_rejects_home_file( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + home = tmp_path / "home" + home.mkdir() + file = home / "personal.txt" + file.write_text("secret", encoding="utf-8") + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setattr(Path, "home", classmethod(lambda _cls: home)) + + with pytest.raises(ValueError, match="Refusing to mount"): + check_mountable_file(file) + + +def test_check_mountable_file_rejects_credential_files(tmp_path: Path) -> None: + ssh_dir = tmp_path / ".ssh" + ssh_dir.mkdir() + key = ssh_dir / "id_rsa" + key.write_text("key", encoding="utf-8") + + with pytest.raises(ValueError, match="holds credentials"): + check_mountable_file(key) + + +def test_check_mountable_file_accepts_a_project_file_under_home_root( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + project = tmp_path / "home" / "dev" / "project" + project.mkdir(parents=True) + file = project / "data.txt" + file.write_text("x", encoding="utf-8") + monkeypatch.setattr(Path, "home", classmethod(lambda _cls: tmp_path / "home" / "dev")) + + check_mountable_file(file)