Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 15 additions & 11 deletions agents/hooks/cleanup_worktrees.py
Original file line number Diff line number Diff line change
Expand Up @@ -142,22 +142,26 @@ def sandbox_writable_paths() -> set[Path] | None:
paths = {
home / entry.removeprefix("~").lstrip("/") if entry.startswith("~") else home / entry for entry in entries
}
paths.update(
{
Path("/tmp"),
Path("/private/tmp"),
Path("/tmp/claude"),
Path("/private/tmp/claude"),
home / HERDR_WORKTREES,
}
)
if temporary := os.environ.get("TMPDIR"):
paths.add(Path(temporary))
paths.update(untrusted_locations())
return paths | {path.resolve() for path in paths}
except OSError, ValueError, KeyError, TypeError:
return None


def untrusted_locations() -> set[Path]:
"""Return locations writable by lanes independently of rendered allowWrite entries."""
paths = {
Path("/tmp"),
Path("/private/tmp"),
Path("/tmp/claude"),
Path("/private/tmp/claude"),
Path.home().resolve() / HERDR_WORKTREES,
}
if temporary := os.environ.get("TMPDIR"):
paths.add(Path(temporary))
return paths


def cleanup(primary: Path, current: Path | None, *, dry_run: bool) -> list[str]:
"""Remove what is merged in one repository and return one line per removal."""
merged = merged_heads(primary)
Expand Down
1 change: 0 additions & 1 deletion tests/agents/hooks/cleanup_worktrees_runtime.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,6 @@ def cleanup(cwd: Path, *arguments: str, payload: dict | None = None) -> subproce
stderr = io.StringIO()
with pytest.MonkeyPatch.context() as patch, redirect_stdout(stdout), redirect_stderr(stderr):
patch.chdir(cwd)
patch.setattr(cleanup_worktrees, "sandbox_writable_paths", set)
if payload is not None:
patch.setattr(sys, "stdin", io.StringIO(json.dumps(payload)))
returncode = cleanup_worktrees.main(list(arguments))
Expand Down
6 changes: 6 additions & 0 deletions tests/agents/hooks/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
sys.path.insert(0, str(REPO_ROOT / "agents/hooks/lib"))
from tests.agents.hooks.cleanup_worktrees_runtime import git

import cleanup_worktrees
import session_store


Expand Down Expand Up @@ -104,6 +105,11 @@ def merged(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict[str, str]:
(home / ".claude/settings.json").write_text('{"sandbox":{"filesystem":{"allowWrite":[]}}}')
monkeypatch.setenv("HOME", str(home))
monkeypatch.setenv("TMPDIR", str(tmp_path / "sandbox-tmp"))
monkeypatch.setattr(
cleanup_worktrees,
"untrusted_locations",
lambda: {tmp_path / "writable", Path(os.environ["TMPDIR"]), Path.home() / cleanup_worktrees.HERDR_WORKTREES},
)
directory = tmp_path / "bin"
directory.mkdir()
state = tmp_path / "merged.json"
Expand Down
8 changes: 4 additions & 4 deletions tests/agents/hooks/test_cleanup_worktrees_isolation.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ def test_lane_status_never_runs_a_filter_from_raced_admin_configuration(
worktree = next(tree for tree in cleanup_worktrees.linked_worktrees(primary) if tree.path == path)
admin = worktree.admin_dir
assert admin is not None
fake = tmp_path / "fake.git"
fake = tmp_path / "writable/fake.git"
shutil.copytree(primary / ".git", fake)
git("config", "--file", str(fake / "config"), "extensions.worktreeConfig", "true", cwd=primary)
marker = tmp_path / "clean-filter-ran"
Expand Down Expand Up @@ -173,7 +173,8 @@ def test_a_gitlink_worktree_is_skipped_without_running_submodule_git(
def test_a_worktree_with_a_replaced_git_file_is_skipped(primary: Path, merged: dict[str, str], tmp_path: Path) -> None:
path = task_worktree(primary, "feat/planted")
merged["feat/planted"] = git("rev-parse", "HEAD", cwd=path)
fake = tmp_path / "fake"
fake = tmp_path / "writable/fake"
fake.parent.mkdir()
git("clone", "--quiet", "--no-checkout", str(primary), str(fake), cwd=primary)
git(f"--git-dir={fake / '.git'}", "read-tree", merged["feat/planted"], cwd=primary)
(path / ".git").write_text(f"gitdir: {fake / '.git'}\n")
Expand All @@ -194,7 +195,7 @@ def test_a_worktree_with_untrusted_admin_configuration_is_skipped(
merged["feat/planted"] = git("rev-parse", "HEAD", cwd=path)
admin = Path(git("rev-parse", "--absolute-git-dir", cwd=path))
if planted == "commondir":
fake = tmp_path / "fake.git"
fake = tmp_path / "writable/fake.git"
shutil.copytree(primary / ".git", fake)
(admin / "commondir").write_text(str(fake))
else:
Expand Down Expand Up @@ -313,5 +314,4 @@ def test_a_planted_git_entry_inside_a_lane_is_neither_current_nor_primary(
assert path.exists()
assert not other.exists()
assert branches(primary) == {"main", "feat/here"}
monkeypatch.setattr(cleanup_worktrees, "sandbox_writable_paths", set)
assert cleanup_worktrees.toplevel(nested) == path
62 changes: 54 additions & 8 deletions tests/agents/hooks/test_cleanup_worktrees_trust.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,13 +31,9 @@ def temporary(tmp_path: Path, merged: dict[str, str], location: str) -> Iterator
sandbox_tmp = tmp_path / "sandbox-tmp"
sandbox_tmp.mkdir()
candidates = [Path("/tmp"), Path("/private/tmp"), Path("/tmp/claude"), Path("/private/tmp/claude")]
candidates.extend(Path(value) for name in ("TMPDIR", "TMP", "TEMP") if (value := os.environ.get(name)))
writable = cleanup_worktrees.sandbox_writable_paths()
assert writable is not None
candidates.append(Path(os.environ["TMPDIR"]))
failures = []
for candidate in dict.fromkeys(candidates):
if not any(candidate.resolve().is_relative_to(root.resolve()) for root in writable):
continue
try:
directory = tempfile.TemporaryDirectory(dir=candidate)
except OSError as error:
Expand All @@ -46,7 +42,7 @@ def temporary(tmp_path: Path, merged: dict[str, str], location: str) -> Iterator
with directory:
yield Path(directory.name)
return
pytest.skip("No writable temporary directory inside the checked writable set: " + "; ".join(failures))
pytest.skip("No writable temporary directory inside the production writable locations: " + "; ".join(failures))


def herdr_lane(primary: Path) -> Path:
Expand Down Expand Up @@ -161,9 +157,9 @@ def test_unknown_writable_paths_skip_lane_primaries_but_clean_the_callers_reposi
path.write_text("not json" if settings == "invalid" else '{"sandbox":null}')
marker = record_commands(tmp_path, other)

assert cleanup_with_real_policy(lane).returncode == 0
assert cleanup(lane).returncode == 0
assert not marker.exists()
result = cleanup_with_real_policy(primary)
result = cleanup(primary)

assert result.returncode == 0
assert not marker.exists()
Expand Down Expand Up @@ -234,3 +230,53 @@ def test_git_removal_is_isolated_and_herdr_only_closes_the_workspace(

assert not lane.exists()
assert (tmp_path / "herdr_removed.txt").read_text().splitlines() == ["workspace close w9"]


def test_production_writable_paths_include_all_untrusted_locations(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
home = tmp_path / "home"
(home / ".claude").mkdir(parents=True)
alias = tmp_path / "alias"
target = tmp_path / "target"
target.mkdir()
alias.symlink_to(target, target_is_directory=True)
entries = [str(alias), "relative", "~/tilde"]
(home / ".claude/settings.json").write_text(json.dumps({"sandbox": {"filesystem": {"allowWrite": entries}}}))
monkeypatch.setenv("HOME", str(home))
monkeypatch.setenv("TMPDIR", str(tmp_path / "temporary"))

writable = cleanup_worktrees.sandbox_writable_paths()

expected = {
Path("/tmp"),
Path("/private/tmp"),
Path("/tmp/claude"),
Path("/private/tmp/claude"),
home / ".herdr/worktrees",
tmp_path / "temporary",
alias,
home / "relative",
home / "tilde",
}
assert writable is not None
assert expected | {path.resolve() for path in expected} <= writable


@pytest.mark.parametrize("entry", ["sweep", "caller"])
def test_fixture_policy_still_rejects_a_planted_writable_primary(
primary: Path, merged: dict[str, str], tmp_path: Path, entry: str
) -> None:
fake = tmp_path / "writable/fake"
fake.parent.mkdir()
git("clone", "--quiet", str(primary), str(fake), cwd=primary)
lane = herdr_lane(fake)
merged["feat/done"] = git("rev-parse", "HEAD", cwd=lane)
marker = record_commands(tmp_path, fake)

result = cleanup(tmp_path if entry == "sweep" else fake)

assert result.returncode == 0
assert not marker.exists()
assert not (tmp_path / "commands").exists()
assert lane.exists()
Loading