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.
This commit is contained in:
LuckVd 2026-09-01 14:07:17 +08:00
parent 14cc948c36
commit ce580989ef
4 changed files with 276 additions and 22 deletions

View file

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

View file

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

133
tests/test_cli_resume.py Normal file
View file

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

View file

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