Merge ce580989ef into cb581bf93d

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

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

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"

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)