feat(#630): drone rm safe-delete + rm_gate block-and-teach hook

Provider-agnostic temp/scratch cleanup that doesn't trip the rm -rf block.

- drone rm <path>: contained recursive delete (project root + /tmp + $TMPDIR),
  refuses outside roots and hard-carves .git/.trinity/.aipass/.codex/.agents +
  sibling-branch worktrees. Mirrors Codex's OS-sandbox boundary in software so
  behavior is consistent across CLIs. Pure-python, red-team tested.
- rm_gate (PreToolUse hook): blocks raw recursive rm, teaches 'drone rm',
  cross-provider, conservative-block on unparseable targets. Mirrors git_gate.
- Wired rm_gate into .aipass/hooks.json; CHANGELOG W23.

Tests: 56 drone rm + 56 rm_gate, seedgo 99% both. Phase 2 (remove the now-
redundant rm deny from setup.sh + aipass doctor migration) tracked separately.
This commit is contained in:
AIOSAI
2026-06-02 20:05:16 -07:00
parent 895b8f04fd
commit 2f5ce5ac87
11 changed files with 1290 additions and 3 deletions
+5
View File
@@ -41,6 +41,11 @@
"handler": "aipass.hooks.apps.handlers.security.git_gate.handle",
"matcher": "Bash|Edit|MultiEdit|Write|NotebookEdit"
},
"rm_gate": {
"enabled": true,
"handler": "aipass.hooks.apps.handlers.security.rm_gate.handle",
"matcher": "Bash"
},
"engine_test_sound": {
"enabled": false,
"command": "python3 $AIPASS_HOME/.claude/hooks/engine_test_sound.py",
+18
View File
@@ -10,6 +10,24 @@ and this project uses [Calendar Versioning](https://calver.org/) in the format
## [2026.W23] - 2026-06-02
### Added
- **`drone rm` — provider-agnostic safe delete** — a contained recursive delete
that lets agents clean up scratch dirs without tripping the `rm -rf` block.
Deletes are confined to the project root and the system temp dirs (`/tmp` and
`$TMPDIR`), refusing anything outside (home, `/etc`, `/`, etc.). Even inside
those roots it hard-refuses protected internals — `.git`, `.trinity/`,
`.aipass/`, `.codex/`, `.agents/`, and sibling-branch worktrees — mirroring the
filesystem boundary an OS-sandboxed agent (e.g. Codex) enforces, so behavior is
consistent across CLIs. Pure-Python (`shutil.rmtree`), with a red-team test
suite for containment escapes (symlinks, traversal, sibling branches). (#630)
- **`rm_gate` hook — block raw recursive `rm`, teach the safe path** — a
PreToolUse gate (mirroring `git_gate`) that blocks raw `rm -r`/`-rf`/`-fr`/
`--recursive` and redirects the agent to `drone rm`. Provider-agnostic (runs in
the hook engine, not tied to Claude Code permission rules), conservative
(unparseable targets are blocked, not allowed), and skips `drone rm` itself.
This makes the safe-delete path discoverable at the moment of friction. (#630)
### Fixed
- **A release merge can no longer destroy the `dev` branch** — `drone @git merge`
+25
View File
@@ -158,6 +158,31 @@
"standard": "architecture",
"reason": "Test file — intentionally outside 3-layer structure. tests/ is a peer of apps/, not part of it."
},
{
"file": "tests/test_rm.py",
"standard": "architecture",
"reason": "Test file — intentionally outside 3-layer structure. tests/ is a peer of apps/, not part of it."
},
{
"file": "tests/test_rm.py",
"standard": "encapsulation",
"reason": "Test file imports rm_handler directly to test its public interface. Unit tests require direct access to implementation components."
},
{
"file": "tests/test_rm.py",
"standard": "documentation",
"reason": "Test class docstrings describe intent; per-method docstrings would be noise for assertion-named test methods."
},
{
"file": "tests/test_rm.py",
"standard": "meta",
"reason": "Test file — META blocks are for production source files, not test suites."
},
{
"file": "tests/test_rm.py",
"standard": "trigger",
"reason": "Test file exercises .unlink() to verify deletion behavior — not a production file operation requiring trigger events."
},
{
"file": "CLAUDE.md",
"standard": "architecture",
+17
View File
@@ -85,6 +85,7 @@ def show_help() -> None:
table.add_row("activate @target", "Register all commands from a branch")
table.add_row("list", "List registered custom commands")
table.add_row("remove <name>", "Remove a custom command")
table.add_row("rm <path> [<path>...]", "Contained safe-delete (project + tmp)")
table.add_row("--help", "Show this help")
table.add_row("--version", "Show version")
@@ -310,6 +311,18 @@ def _handle_remove(name: str) -> int:
return 0 if success else 1
def _handle_rm(args: list[str]) -> int:
"""Handle ``drone rm <path> [<path>...]`` — contained safe-delete."""
from aipass.drone.apps.modules.rm import handle_command, print_help
if not args or args[0] in ("--help", "-h"):
print_help()
return 0
result = handle_command(args[0], args[1:] if len(args) > 1 else None)
return 0 if result else 1
def _handle_custom_command(args: list[str]) -> int:
"""Handle a custom command shortcut by matching and routing.
@@ -557,6 +570,10 @@ def main() -> int:
return 1
return _handle_remove(args[1])
# rm — contained safe-delete
if command == "rm":
return _handle_rm(args[1:])
# @target — route to branch or module
if command.startswith("@"):
return _handle_target(args)
@@ -0,0 +1,204 @@
# =================== AIPass ====================
# Name: rm_handler.py
# Description: Contained safe-delete handler
# Version: 1.1.0
# Created: 2026-06-02
# Modified: 2026-06-02
# =============================================
"""Contained safe-delete handler.
Deletes paths using shutil.rmtree (directories) or Path.unlink (files),
constrained to project root and system temp directories. Provider-agnostic
alternative to shell ``rm``.
Matches Codex sandbox boundaries: writable = {project, /tmp, $TMPDIR}.
Hard carve-outs protect .git, .trinity, .aipass, .codex, .agents, and
sibling branch worktrees even inside allowed roots.
"""
from __future__ import annotations
import os
import shutil
import tempfile
from pathlib import Path
from aipass.prax import logger
from aipass.drone.apps.handlers.json import json_handler
_CARVEOUT_DIRS = frozenset((".git", ".trinity", ".aipass", ".codex", ".agents"))
def _find_project_root() -> Path | None:
"""Walk up from CWD to find *_REGISTRY.json; return its parent as project root."""
cwd = Path.cwd()
for parent in [cwd, *cwd.parents]:
if list(parent.glob("*_REGISTRY.json")):
return parent.resolve()
aipass_home = os.environ.get("AIPASS_HOME")
if aipass_home:
home = Path(aipass_home)
if home.is_dir() and list(home.glob("*_REGISTRY.json")):
return home.resolve()
return None
def get_allowed_roots() -> list[Path]:
"""Return resolved roots under which deletion is permitted.
Union of: project root, ``/tmp``, and ``tempfile.gettempdir()`` (which
honors ``$TMPDIR``). Deduplicated by resolved path.
"""
seen: set[Path] = set()
roots: list[Path] = []
project_root = _find_project_root()
if project_root is not None and project_root not in seen:
seen.add(project_root)
roots.append(project_root)
for tmp_candidate in (Path("/tmp"), Path(tempfile.gettempdir())):
resolved = tmp_candidate.resolve()
if resolved not in seen:
seen.add(resolved)
roots.append(resolved)
return roots
def _detect_current_branch(project_root: Path | None) -> str | None:
"""Return the branch name the CWD lives in, or None."""
if project_root is None:
return None
cwd = Path.cwd().resolve()
aipass_src = project_root / "src" / "aipass"
if not cwd.is_relative_to(aipass_src):
return None
rel = cwd.relative_to(aipass_src)
return rel.parts[0] if rel.parts else None
def _resolve_git_dir(path: Path) -> Path | None:
"""If *path* is a ``.git`` file (worktree pointer), return the resolved gitdir."""
try:
if path.is_file():
text = path.read_text(encoding="utf-8", errors="replace").strip()
if text.startswith("gitdir:"):
return Path(text.split(":", 1)[1].strip()).resolve()
except OSError as exc:
logger.warning("Failed to read .git file at %s: %s", path, exc)
return None
def check_carveouts(resolved: Path, project_root: Path | None) -> tuple[bool, str]:
"""Refuse deletion of protected paths even inside allowed roots.
Returns ``(blocked, reason)``. ``blocked=True`` means the path must
NOT be deleted.
"""
parts = resolved.parts
for i, part in enumerate(parts):
if part in _CARVEOUT_DIRS:
return True, f"Protected directory: path is inside {part}/"
if part == ".git":
git_path = Path(*parts[: i + 1]) if i > 0 else Path(part)
real_gitdir = _resolve_git_dir(git_path)
if real_gitdir and resolved.is_relative_to(real_gitdir):
return True, "Protected: path resolves inside .git worktree gitdir"
if project_root is not None:
aipass_src = project_root / "src" / "aipass"
if resolved.is_relative_to(aipass_src):
rel = resolved.relative_to(aipass_src)
if rel.parts:
target_branch = rel.parts[0]
current_branch = _detect_current_branch(project_root)
if current_branch is None or target_branch != current_branch:
return True, (f"Protected: path is inside sibling branch src/aipass/{target_branch}/")
return False, ""
def check_containment(path: Path, roots: list[Path]) -> tuple[bool, str]:
"""Check if *path* (already resolved) is a strict child of any allowed root.
Returns ``(allowed, reason)``. Refuses the root directories themselves.
When multiple roots are nested (e.g. /tmp and /tmp/claude-1000), the path
must not equal ANY root — checked upfront before containment.
"""
root_set = frozenset(roots)
if path in root_set:
return False, f"Refusing to delete root directory itself: {path}"
for root in roots:
if path.is_relative_to(root):
return True, ""
allowed_str = ", ".join(str(r) for r in roots)
return False, (f"Path {path} is outside allowed roots.\n Allowed: {allowed_str}")
def safe_delete(paths: list[str]) -> list[tuple[str, bool, str]]:
"""Delete *paths* with containment checks.
Returns a list of ``(original_path, success, message)`` tuples.
Every path is checked independently; a refused path does not block others.
"""
roots = get_allowed_roots()
if not roots:
return [(p, False, "No allowed roots found (no project registry, no temp dir)") for p in paths]
project_root = _find_project_root()
json_handler.log_operation("rm", {"paths": paths, "roots": [str(r) for r in roots]})
results: list[tuple[str, bool, str]] = []
for path_str in paths:
original = Path(path_str)
absolute = original if original.is_absolute() else (Path.cwd() / original)
exists_on_disk = absolute.exists() or absolute.is_symlink()
if not exists_on_disk:
results.append((path_str, False, f"Path does not exist: {absolute}"))
logger.info("rm: nonexistent path %s", absolute)
continue
resolved = absolute.resolve()
allowed, reason = check_containment(resolved, roots)
if not allowed:
results.append((path_str, False, reason))
logger.warning(
"rm: containment refused %s (resolved %s): %s",
path_str,
resolved,
reason,
)
continue
blocked, carveout_reason = check_carveouts(resolved, project_root)
if blocked:
results.append((path_str, False, carveout_reason))
logger.warning(
"rm: carveout refused %s (resolved %s): %s",
path_str,
resolved,
carveout_reason,
)
continue
try:
if absolute.is_symlink():
absolute.unlink()
elif absolute.is_dir():
shutil.rmtree(absolute)
else:
absolute.unlink()
results.append((path_str, True, f"Deleted: {resolved}"))
logger.info("rm: deleted %s (resolved %s)", path_str, resolved)
except Exception as exc:
results.append((path_str, False, f"Delete failed: {exc}"))
logger.error("rm: delete failed for %s: %s", path_str, exc)
return results
+101
View File
@@ -0,0 +1,101 @@
# =================== AIPass ====================
# Name: rm.py
# Description: Module orchestrator for contained safe-delete
# Version: 1.0.0
# Created: 2026-06-02
# Modified: 2026-06-02
# =============================================
"""Module orchestrator for contained safe-delete.
Thin orchestrator that delegates to rm_handler for path containment
checks and deletion. Provider-agnostic alternative to shell ``rm``.
"""
from __future__ import annotations
from aipass.prax import logger
from aipass.cli.apps.modules import console
from aipass.drone.apps.handlers.json import json_handler
from aipass.drone.apps.handlers.rm_handler import (
safe_delete as _safe_delete,
)
DRONE_MODULE = {
"name": "rm",
"version": "1.0.0",
"description": "Contained safe-delete (project + tmp)",
}
def safe_delete(paths: list[str]) -> list[tuple[str, bool, str]]:
"""Delete paths with containment checks.
Returns list of ``(original_path, success, message)`` tuples.
"""
logger.info("rm: requested deletion of %d path(s)", len(paths))
return _safe_delete(paths)
def handle_command(command: str | None = None, args: list[str] | None = None) -> bool:
"""Entry point for ``drone rm`` module routing."""
if not args:
if command is None:
print_introspection()
return True
args = []
if command in ("--help", "-h") or (args and args[0] in ("--help", "-h")):
print_help()
return True
json_handler.log_operation("rm_command", {"command": command, "args": args})
paths: list[str] = []
if command is not None:
paths.append(command)
if args:
paths.extend(args)
if not paths:
print_help()
return True
results = _safe_delete(paths)
ok = True
for _path_str, success, message in results:
if success:
console.print(f"[green]✓[/green] {message}")
else:
console.print(f"[red]✗[/red] {message}")
ok = False
return ok
def print_introspection() -> None:
"""Display module overview (no args)."""
console.print()
console.print("[bold cyan]rm — Contained Safe-Delete[/bold cyan]")
console.print()
console.print("[dim]Deletes files and directories constrained to project root and system tmp.[/dim]")
console.print()
console.print("Run [green]'drone rm --help'[/green] for usage information")
console.print()
def print_help() -> None:
"""Display help (--help flag)."""
console.print("Usage: drone rm <path> [<path>...]")
console.print()
console.print("Contained safe-delete. Removes files and directories using pure Python")
console.print("(shutil.rmtree), constrained to the project root and system temp directory.")
console.print()
console.print("[bold]Rules:[/bold]")
console.print(" • Path must resolve under the project root or system temp dir")
console.print(" • Cannot delete the project root or temp root itself")
console.print(" • Symlinks are resolved; refuses if target escapes allowed roots")
console.print(" • Nonexistent paths produce a clean error")
console.print()
console.print("[bold]Examples:[/bold]")
console.print(" [green]drone rm /tmp/scratch_dir[/green]")
console.print(" [green]drone rm build/ dist/[/green]")
console.print(" [green]drone rm /tmp/aipass_test_abc123[/green]")
+571
View File
@@ -0,0 +1,571 @@
"""Tests for drone rm — contained safe-delete.
Red-team containment tests verify that paths outside allowed roots
are refused, including symlink escapes and traversal attempts.
Carve-out tests verify .git, .trinity, .aipass, .codex, .agents,
and sibling branches are protected even inside allowed roots.
"""
import os
import shutil
import tempfile
from pathlib import Path
from unittest.mock import patch
import pytest
from aipass.drone.apps.handlers.rm_handler import (
check_carveouts,
check_containment,
get_allowed_roots,
safe_delete,
)
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture()
def project_dir(tmp_path):
"""Fake project root with a registry file and src/aipass layout."""
(tmp_path / "AIPASS_REGISTRY.json").write_text("{}")
return tmp_path
@pytest.fixture()
def project_with_branches(project_dir):
"""Project root with src/aipass/<branch> layout for sibling tests."""
for branch in ("drone", "api", "flow"):
d = project_dir / "src" / "aipass" / branch
d.mkdir(parents=True)
(d / "README.md").write_text(f"# {branch}")
return project_dir
@pytest.fixture()
def _patch_roots(project_dir):
"""Patch get_allowed_roots to use deterministic test roots."""
tmpdir = Path(tempfile.gettempdir()).resolve()
slash_tmp = Path("/tmp").resolve()
roots = [project_dir.resolve()]
seen = set(roots)
for r in (slash_tmp, tmpdir):
if r not in seen:
seen.add(r)
roots.append(r)
with patch(
"aipass.drone.apps.handlers.rm_handler.get_allowed_roots",
return_value=roots,
):
yield
# ---------------------------------------------------------------------------
# get_allowed_roots
# ---------------------------------------------------------------------------
class TestGetAllowedRoots:
def test_includes_temp_dir(self):
roots = get_allowed_roots()
tmpdir = Path(tempfile.gettempdir()).resolve()
assert tmpdir in roots
def test_includes_slash_tmp(self):
roots = get_allowed_roots()
assert Path("/tmp").resolve() in roots
def test_includes_project_root_when_in_project(self, project_dir, monkeypatch):
monkeypatch.chdir(project_dir)
roots = get_allowed_roots()
assert project_dir.resolve() in roots
def test_temp_dir_always_present_even_without_project(self, tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
roots = get_allowed_roots()
tmpdir = Path(tempfile.gettempdir()).resolve()
assert tmpdir in roots
def test_tmpdir_and_slash_tmp_both_present_when_different(self, monkeypatch):
"""When $TMPDIR != /tmp, both must appear in roots."""
fake_tmpdir = "/tmp/claude-9999"
os.makedirs(fake_tmpdir, exist_ok=True)
try:
monkeypatch.setenv("TMPDIR", fake_tmpdir)
tempfile.tempdir = None
roots = get_allowed_roots()
resolved_roots = {r for r in roots}
assert Path("/tmp").resolve() in resolved_roots
assert Path(fake_tmpdir).resolve() in resolved_roots
finally:
tempfile.tempdir = None
def test_roots_are_deduplicated(self):
roots = get_allowed_roots()
assert len(roots) == len(set(roots))
# ---------------------------------------------------------------------------
# check_containment
# ---------------------------------------------------------------------------
class TestCheckContainment:
def test_allows_child_of_root(self, tmp_path):
root = tmp_path.resolve()
child = (tmp_path / "sub" / "file.txt").resolve()
allowed, reason = check_containment(child, [root])
assert allowed is True
assert reason == ""
def test_refuses_root_itself(self, tmp_path):
root = tmp_path.resolve()
allowed, reason = check_containment(root, [root])
assert allowed is False
assert "root directory itself" in reason
def test_refuses_outside_path(self, tmp_path):
root = tmp_path.resolve()
outside = Path("/etc/passwd").resolve()
allowed, reason = check_containment(outside, [root])
assert allowed is False
assert "outside allowed roots" in reason
def test_allows_second_root(self, tmp_path):
root1 = (tmp_path / "a").resolve()
root2 = (tmp_path / "b").resolve()
child = (tmp_path / "b" / "file.txt").resolve()
allowed, _ = check_containment(child, [root1, root2])
assert allowed is True
def test_refuses_empty_roots(self, tmp_path):
child = (tmp_path / "file.txt").resolve()
allowed, _reason = check_containment(child, [])
assert allowed is False
# ---------------------------------------------------------------------------
# ALLOW: valid deletions
# ---------------------------------------------------------------------------
class TestAllowDeletion:
@pytest.mark.usefixtures("_patch_roots")
def test_delete_dir_in_tmp(self):
target = Path(tempfile.mkdtemp())
try:
(target / "file.txt").write_text("data")
results = safe_delete([str(target)])
assert results[0][1] is True
assert not target.exists()
finally:
if target.exists():
shutil.rmtree(target)
@pytest.mark.usefixtures("_patch_roots")
def test_delete_nested_tmp_dir(self):
"""e.g. /tmp/claude-1000/<x>."""
parent = Path(tempfile.mkdtemp())
target = parent / "nested"
target.mkdir()
(target / "data.txt").write_text("hello")
try:
results = safe_delete([str(target)])
assert results[0][1] is True
assert not target.exists()
finally:
if parent.exists():
shutil.rmtree(parent)
@pytest.mark.usefixtures("_patch_roots")
def test_delete_file_in_project(self, project_dir):
target = project_dir / "build" / "output.o"
target.parent.mkdir(parents=True)
target.write_text("binary")
results = safe_delete([str(target)])
assert results[0][1] is True
assert not target.exists()
@pytest.mark.usefixtures("_patch_roots")
def test_delete_subdir_in_project(self, project_dir):
target = project_dir / "sub" / "scratch"
target.mkdir(parents=True)
(target / "temp.txt").write_text("scratch")
results = safe_delete([str(target)])
assert results[0][1] is True
assert not target.exists()
@pytest.mark.usefixtures("_patch_roots")
def test_delete_multiple_paths(self):
t1 = Path(tempfile.mkdtemp())
t2 = Path(tempfile.mkdtemp())
try:
results = safe_delete([str(t1), str(t2)])
assert all(r[1] for r in results)
assert not t1.exists()
assert not t2.exists()
finally:
for t in [t1, t2]:
if t.exists():
shutil.rmtree(t)
@pytest.mark.usefixtures("_patch_roots")
def test_pure_python_no_subprocess(self):
"""Verify shutil.rmtree is used, not subprocess rm."""
target = Path(tempfile.mkdtemp())
(target / "f.txt").write_text("x")
with patch("subprocess.run") as mock_run, patch("subprocess.Popen") as mock_popen:
results = safe_delete([str(target)])
assert results[0][1] is True
mock_run.assert_not_called()
mock_popen.assert_not_called()
@pytest.mark.usefixtures("_patch_roots")
def test_project_build_dir_allowed(self, project_dir):
"""Regression guard: ordinary project dirs are still deletable."""
target = project_dir / "build"
target.mkdir()
(target / "out.js").write_text("x")
results = safe_delete([str(target)])
assert results[0][1] is True
@pytest.mark.usefixtures("_patch_roots")
def test_project_dist_dir_allowed(self, project_dir):
"""Regression guard: dist/ is not a carve-out."""
target = project_dir / "dist"
target.mkdir()
(target / "bundle.js").write_text("x")
results = safe_delete([str(target)])
assert results[0][1] is True
@pytest.mark.usefixtures("_patch_roots")
def test_slash_tmp_literal_allowed(self):
"""/tmp/<x> must succeed even if $TMPDIR differs."""
target = Path("/tmp") / f"rm_test_{os.getpid()}"
target.mkdir(exist_ok=True)
try:
results = safe_delete([str(target)])
assert results[0][1] is True
assert not target.exists()
finally:
if target.exists():
shutil.rmtree(target)
@pytest.mark.usefixtures("_patch_roots")
def test_tmpdir_env_allowed(self):
"""$TMPDIR/<x> must succeed."""
target = Path(tempfile.mkdtemp())
try:
results = safe_delete([str(target)])
assert results[0][1] is True
assert not target.exists()
finally:
if target.exists():
shutil.rmtree(target)
# ---------------------------------------------------------------------------
# REFUSE: red-team containment
# ---------------------------------------------------------------------------
class TestRefuseDeletion:
@pytest.mark.usefixtures("_patch_roots")
def test_refuse_home_dir(self):
results = safe_delete([str(Path.home())])
assert results[0][1] is False
@pytest.mark.usefixtures("_patch_roots")
def test_refuse_etc(self):
results = safe_delete(["/etc"])
assert results[0][1] is False
@pytest.mark.usefixtures("_patch_roots")
def test_refuse_root_filesystem(self):
results = safe_delete(["/"])
assert results[0][1] is False
@pytest.mark.usefixtures("_patch_roots")
def test_refuse_project_root_itself(self, project_dir):
results = safe_delete([str(project_dir)])
assert results[0][1] is False
assert "root directory itself" in results[0][2]
@pytest.mark.usefixtures("_patch_roots")
def test_refuse_tmp_root_itself(self):
tmpdir = tempfile.gettempdir()
results = safe_delete([tmpdir])
assert results[0][1] is False
assert "root directory itself" in results[0][2]
@pytest.mark.usefixtures("_patch_roots")
def test_refuse_traversal_escape(self, project_dir, monkeypatch):
"""../../etc from inside project should resolve outside and be refused."""
subdir = project_dir / "deep" / "nested"
subdir.mkdir(parents=True)
monkeypatch.chdir(subdir)
results = safe_delete(["../../../../../../etc"])
assert results[0][1] is False
@pytest.mark.usefixtures("_patch_roots")
def test_refuse_absolute_outside_roots(self):
results = safe_delete(["/usr/local/bin"])
assert results[0][1] is False
@pytest.mark.usefixtures("_patch_roots")
def test_refuse_symlink_escape_from_tmp(self):
"""Symlink under /tmp pointing to /home/user should be refused."""
target_outside = Path.home()
link_dir = Path(tempfile.mkdtemp())
link = link_dir / "escape_link"
try:
link.symlink_to(target_outside)
results = safe_delete([str(link)])
assert results[0][1] is False
finally:
if link.exists() or link.is_symlink():
link.unlink()
if link_dir.exists():
shutil.rmtree(link_dir)
@pytest.mark.usefixtures("_patch_roots")
def test_nonexistent_path_clean_error(self):
results = safe_delete(["/tmp/this_path_does_not_exist_abc123xyz"])
assert results[0][1] is False
assert "does not exist" in results[0][2]
@pytest.mark.usefixtures("_patch_roots")
def test_mixed_valid_and_invalid(self, project_dir):
"""Valid paths succeed; invalid paths fail independently."""
valid = project_dir / "ok_to_delete"
valid.mkdir()
results = safe_delete([str(valid), "/etc/shadow"])
assert results[0][1] is True
assert results[1][1] is False
@pytest.mark.usefixtures("_patch_roots")
def test_refuse_home_patrick(self):
results = safe_delete(["/home/patrick"])
assert results[0][1] is False
@pytest.mark.usefixtures("_patch_roots")
def test_refuse_var_tmp(self):
"""/var/tmp is NOT in the default allowed set (Codex excludes it)."""
results = safe_delete(["/var/tmp"])
assert results[0][1] is False
# ---------------------------------------------------------------------------
# Carve-outs: .git, .trinity, .aipass, .codex, .agents, siblings
# ---------------------------------------------------------------------------
class TestCarveouts:
def test_refuse_dot_git_dir(self, project_dir):
"""<repo>/.git directory must be refused."""
git_dir = project_dir / ".git"
git_dir.mkdir()
resolved = git_dir.resolve()
blocked, reason = check_carveouts(resolved, project_dir.resolve())
assert blocked is True
assert ".git" in reason
def test_refuse_inside_dot_git(self, project_dir):
"""Files inside .git/ must be refused."""
git_dir = project_dir / ".git" / "objects"
git_dir.mkdir(parents=True)
resolved = git_dir.resolve()
blocked, reason = check_carveouts(resolved, project_dir.resolve())
assert blocked is True
assert ".git" in reason
def test_refuse_dot_trinity(self, project_dir):
trinity = project_dir / ".trinity"
trinity.mkdir()
resolved = trinity.resolve()
blocked, reason = check_carveouts(resolved, project_dir.resolve())
assert blocked is True
assert ".trinity" in reason
def test_refuse_inside_dot_trinity(self, project_dir):
passport = project_dir / ".trinity" / "passport.json"
passport.parent.mkdir(parents=True)
passport.write_text("{}")
resolved = passport.resolve()
blocked, reason = check_carveouts(resolved, project_dir.resolve())
assert blocked is True
assert ".trinity" in reason
def test_refuse_dot_aipass(self, project_dir):
aipass_dir = project_dir / ".aipass"
aipass_dir.mkdir()
resolved = aipass_dir.resolve()
blocked, reason = check_carveouts(resolved, project_dir.resolve())
assert blocked is True
assert ".aipass" in reason
def test_refuse_dot_codex(self, project_dir):
codex = project_dir / ".codex"
codex.mkdir()
resolved = codex.resolve()
blocked, reason = check_carveouts(resolved, project_dir.resolve())
assert blocked is True
assert ".codex" in reason
def test_refuse_dot_agents(self, project_dir):
agents = project_dir / ".agents"
agents.mkdir()
resolved = agents.resolve()
blocked, reason = check_carveouts(resolved, project_dir.resolve())
assert blocked is True
assert ".agents" in reason
def test_allow_normal_project_dir(self, project_dir):
"""build/ is not a carve-out."""
build = project_dir / "build"
build.mkdir()
resolved = build.resolve()
blocked, _reason = check_carveouts(resolved, project_dir.resolve())
assert blocked is False
def test_refuse_sibling_branch(self, project_with_branches, monkeypatch):
"""From drone CWD, deleting src/aipass/api must be refused."""
drone_dir = project_with_branches / "src" / "aipass" / "drone"
monkeypatch.chdir(drone_dir)
target = project_with_branches / "src" / "aipass" / "api"
resolved = target.resolve()
blocked, reason = check_carveouts(resolved, project_with_branches.resolve())
assert blocked is True
assert "sibling branch" in reason
assert "api" in reason
def test_allow_own_branch(self, project_with_branches, monkeypatch):
"""Deleting a subdir inside own branch must be allowed."""
drone_dir = project_with_branches / "src" / "aipass" / "drone"
monkeypatch.chdir(drone_dir)
target = drone_dir / "build"
target.mkdir()
resolved = target.resolve()
blocked, _reason = check_carveouts(resolved, project_with_branches.resolve())
assert blocked is False
def test_refuse_all_branches_when_outside(self, project_with_branches, monkeypatch):
"""When CWD isn't inside any branch, ALL src/aipass/<branch> are refused."""
monkeypatch.chdir(project_with_branches)
for branch in ("drone", "api", "flow"):
target = project_with_branches / "src" / "aipass" / branch
resolved = target.resolve()
blocked, _reason = check_carveouts(resolved, project_with_branches.resolve())
assert blocked is True, f"Expected {branch} to be blocked"
def test_git_file_worktree_pointer(self, project_dir):
"""A .git FILE (worktree pointer) should protect the resolved gitdir."""
real_git = project_dir / "real_git_dir"
real_git.mkdir()
git_file = project_dir / ".git"
git_file.write_text(f"gitdir: {real_git}")
resolved = git_file.resolve()
blocked, reason = check_carveouts(resolved, project_dir.resolve())
assert blocked is True
assert ".git" in reason
# ---------------------------------------------------------------------------
# Carve-outs via safe_delete (integration)
# ---------------------------------------------------------------------------
class TestCarveoutIntegration:
@pytest.mark.usefixtures("_patch_roots")
def test_safe_delete_refuses_dot_git(self, project_dir):
git_dir = project_dir / ".git"
git_dir.mkdir()
results = safe_delete([str(git_dir)])
assert results[0][1] is False
assert ".git" in results[0][2]
assert git_dir.exists()
@pytest.mark.usefixtures("_patch_roots")
def test_safe_delete_refuses_trinity(self, project_dir):
trinity = project_dir / ".trinity"
trinity.mkdir()
results = safe_delete([str(trinity)])
assert results[0][1] is False
assert trinity.exists()
@pytest.mark.usefixtures("_patch_roots")
def test_safe_delete_refuses_dot_aipass(self, project_dir):
aipass_dir = project_dir / ".aipass"
aipass_dir.mkdir()
results = safe_delete([str(aipass_dir)])
assert results[0][1] is False
assert aipass_dir.exists()
@pytest.mark.usefixtures("_patch_roots")
def test_safe_delete_allows_build_dir(self, project_dir):
"""Regression guard: build/ and dist/ still allowed."""
build = project_dir / "build"
build.mkdir()
results = safe_delete([str(build)])
assert results[0][1] is True
assert not build.exists()
# ---------------------------------------------------------------------------
# Edge cases
# ---------------------------------------------------------------------------
class TestEdgeCases:
@pytest.mark.usefixtures("_patch_roots")
def test_relative_path_resolved_from_cwd(self, project_dir, monkeypatch):
monkeypatch.chdir(project_dir)
target = project_dir / "relative_target"
target.mkdir()
results = safe_delete(["relative_target"])
assert results[0][1] is True
assert not target.exists()
@pytest.mark.usefixtures("_patch_roots")
def test_single_file_deletion(self):
fd, path = tempfile.mkstemp()
os.close(fd)
results = safe_delete([path])
assert results[0][1] is True
assert not Path(path).exists()
@pytest.mark.usefixtures("_patch_roots")
def test_empty_paths_list(self):
results = safe_delete([])
assert results == []
# ---------------------------------------------------------------------------
# Module orchestrator (rm.py)
# ---------------------------------------------------------------------------
class TestRmModule:
def test_handle_command_help(self):
from aipass.drone.apps.modules.rm import handle_command
result = handle_command("--help")
assert result is True
def test_handle_command_introspection(self):
from aipass.drone.apps.modules.rm import handle_command
result = handle_command(None, None)
assert result is True
def test_print_introspection(self):
from aipass.drone.apps.modules.rm import print_introspection
print_introspection()
def test_print_help(self):
from aipass.drone.apps.modules.rm import print_help
print_help()
+9
View File
@@ -35,6 +35,10 @@
{"file": "apps/handlers/security/git_gate.py", "standard": "unused_function", "reason": "handle() called dynamically by engine._run_handler via importlib.import_module + getattr from hooks.json. Verified fires in engine.jsonl (PreToolUse.git_gate)."},
{"file": "apps/handlers/security/git_gate.py", "standard": "json_structure", "reason": "Security gate uses stdlib json.dumps for hook protocol block responses — no JSON file ops needing json_handler."},
{"file": "apps/handlers/security/rm_gate.py", "standard": "dead_code", "reason": "Invoked dynamically by engine via importlib from hooks.json handler path 'aipass.hooks.apps.handlers.security.rm_gate.handle' — not statically imported by design. Wired in PreToolUse.rm_gate."},
{"file": "apps/handlers/security/rm_gate.py", "standard": "unused_function", "reason": "handle() called dynamically by engine._run_handler via importlib.import_module + getattr from hooks.json. Wired in PreToolUse.rm_gate."},
{"file": "apps/handlers/security/rm_gate.py", "standard": "json_structure", "reason": "Security gate uses stdlib json.dumps for hook protocol block responses — no JSON file ops needing json_handler."},
{"file": "apps/handlers/security/subagent_gate.py", "standard": "dead_code", "reason": "Invoked dynamically by engine via importlib from hooks.json handler path 'aipass.hooks.apps.handlers.security.subagent_gate.handle' — not statically imported by design. Verified wired in SubagentStop.subagent_stop_gate + fires in engine.jsonl."},
{"file": "apps/handlers/security/subagent_gate.py", "standard": "unused_function", "reason": "handle() called dynamically by engine._run_handler via importlib.import_module + getattr from hooks.json. Verified fires in engine.jsonl (SubagentStop.subagent_stop_gate)."},
{"file": "apps/handlers/security/subagent_gate.py", "standard": "json_structure", "reason": "Security gate uses stdlib json.dumps for hook protocol block responses — no JSON file ops needing json_handler."},
@@ -182,6 +186,11 @@
{"file": "tests/test_git_gate.py", "standard": "encapsulation", "reason": "Tests import handlers directly to test implementation details."},
{"file": "tests/test_git_gate.py", "standard": "meta", "reason": "Test files do not need Version/Modified metadata headers."},
{"file": "tests/test_rm_gate.py", "standard": "architecture", "reason": "Test files live in tests/, not in the 3-layer apps structure."},
{"file": "tests/test_rm_gate.py", "standard": "documentation", "reason": "Test methods use descriptive names as documentation per pytest convention."},
{"file": "tests/test_rm_gate.py", "standard": "encapsulation", "reason": "Tests import handlers directly to test implementation details."},
{"file": "tests/test_rm_gate.py", "standard": "meta", "reason": "Test files do not need Version/Modified metadata headers."},
{"file": "tests/test_sound.py", "standard": "architecture", "reason": "Test files live in tests/, not in the 3-layer apps structure."},
{"file": "tests/test_sound.py", "standard": "documentation", "reason": "Test methods use descriptive names as documentation per pytest convention."},
{"file": "tests/test_sound.py", "standard": "encapsulation", "reason": "Tests import sound module directly to test implementation details."},
+4 -3
View File
@@ -61,6 +61,7 @@ src/aipass/hooks/
│ │ ├── security/ # Enforcement hooks
│ │ │ ├── edit_gate.py # Blocks unsafe edits (cross-branch, inbox, diagnostics)
│ │ │ ├── git_gate.py # Enforces git access tiers
│ │ │ ├── rm_gate.py # Blocks raw recursive rm, teaches drone rm
│ │ │ └── subagent_gate.py # Blocks sub-agent stop until clean
│ │ ├── lifecycle/ # Session management hooks
│ │ │ ├── auto_fix.py # Post-edit diagnostics (ruff, pyright, py_compile)
@@ -77,7 +78,7 @@ src/aipass/hooks/
│ └── diagnostics.py # JSONL logging for hook execution
├── logs/
│ └── engine.jsonl # JSONL diagnostics (every hook execution)
├── tests/ # 253 tests across 19 test files
├── tests/ # 314 tests across 20 test files
└── STATUS.local.md
```
@@ -100,7 +101,7 @@ Handlers are called **dynamically at runtime** — the engine uses `importlib.im
| Event | Hooks | Description |
|---|---|---|
| UserPromptSubmit | identity, email, branch_loader, global_loader | Prompt injection + inbox check |
| PreToolUse | tool_sound, edit_gate, git_gate | Security gates + sound |
| PreToolUse | tool_sound, edit_gate, git_gate, rm_gate | Security gates + sound |
| PostToolUse | auto_fix, auto_watchdog | Diagnostics + watchdog |
| SubagentStop | subagent_gate | Seedgo validation |
| Stop | stop_sound | Achievement bell |
@@ -119,7 +120,7 @@ Handlers are called **dynamically at runtime** — the engine uses `importlib.im
All branches via hook dispatch. Every Claude Code session routes through the engine.
*Last Updated: 2026-05-28*
*Last Updated: 2026-06-02*
---
@@ -0,0 +1,108 @@
# =================== AIPass ====================
# Name: rm_gate.py
# Version: 1.0.0
# Description: Blocks raw recursive rm commands (PreToolUse)
# Branch: hooks
# Layer: apps/handlers/security
# Created: 2026-06-02
# Modified: 2026-06-02
# =============================================
"""Blocks raw recursive rm and teaches drone rm."""
import json
import re
from aipass.hooks.apps.sound import speak
from aipass.prax.apps.modules.logger import system_logger as logger
RM_REDIRECT = (
"Raw recursive rm is blocked. Use the safe contained delete instead:\n"
" drone rm <path> # safe delete (allows project + /tmp, refuses outside)\n"
"\n"
"This applies to all recursive rm variants (rm -rf, rm -r, rm -R, rm --recursive)."
)
_BLOCK_ALLOW = {"stdout": "", "exit_code": 0}
def _block(reason: str) -> dict:
return {"stdout": json.dumps({"decision": "block", "reason": reason}), "exit_code": 2}
def _strip_quotes(cmd: str) -> str:
"""Remove quoted strings so their contents aren't scanned."""
cmd = re.sub(r'"(?:[^"\\]|\\.)*"', '""', cmd)
cmd = re.sub(r"'(?:[^'\\]|\\.)*'", "''", cmd)
return cmd
def _split_clauses(cmd: str) -> list[str]:
"""Split on compound operators and subshell boundaries."""
parts = re.split(r"&&|\|\||[;|]", cmd)
clauses: list[str] = []
for part in parts:
clauses.extend(re.split(r"[$()`]", part))
return clauses
def _has_recursive_flag(tokens: list[str]) -> bool:
"""Return True if any token before '--' contains a recursive flag."""
for token in tokens:
if token == "--":
break
if token.startswith("-") and not token.startswith("--"):
if "r" in token[1:] or "R" in token[1:]:
return True
elif token == "--recursive":
return True
return False
def _clause_has_raw_recursive_rm(clause: str) -> bool:
"""Return True if a single clause contains a raw recursive rm."""
tokens = clause.split()
if not tokens:
return False
for i, tok in enumerate(tokens):
if tok != "rm" and not tok.endswith("/rm"):
continue
if i > 0 and tokens[i - 1] == "drone":
continue
if _has_recursive_flag(tokens[i + 1 :]):
return True
return False
def handle(hook_data: dict) -> dict:
"""Block raw recursive rm commands and teach drone rm.
Args:
hook_data: Parsed hook event dict from engine.
Returns:
Result dict with stdout (block JSON or empty) and exit_code.
"""
speak("rm gate")
try:
tool_name = hook_data.get("tool_name", "")
if tool_name != "Bash":
return _BLOCK_ALLOW
tool_input = hook_data.get("tool_input", {})
cmd = tool_input.get("command", "")
if not cmd:
return _BLOCK_ALLOW
scan = _strip_quotes(cmd)
for clause in _split_clauses(scan):
if _clause_has_raw_recursive_rm(clause):
return _block(RM_REDIRECT)
return _BLOCK_ALLOW
except Exception as exc:
logger.info("[HOOKS] rm_gate: unexpected error (allowing): %s", exc)
return _BLOCK_ALLOW
+228
View File
@@ -0,0 +1,228 @@
# =================== AIPass ====================
# Name: test_rm_gate.py
# Version: 1.0.0
# Description: Tests for rm_gate security handler
# Branch: hooks
# Created: 2026-06-02
# Modified: 2026-06-02
# =============================================
"""Tests for handlers/security/rm_gate.py."""
import json
from unittest.mock import patch
from aipass.hooks.apps.handlers.security.rm_gate import (
_clause_has_raw_recursive_rm,
_has_recursive_flag,
_split_clauses,
_strip_quotes,
handle,
)
class TestStripQuotes:
def test_double_quotes(self):
assert _strip_quotes('rm -rf "/tmp/foo bar"') == 'rm -rf ""'
def test_single_quotes(self):
assert _strip_quotes("rm -rf '/tmp/foo bar'") == "rm -rf ''"
def test_escaped_quote_in_double(self):
assert _strip_quotes(r'echo "he said \"hi\""') == 'echo ""'
def test_no_quotes(self):
assert _strip_quotes("rm -rf /tmp/foo") == "rm -rf /tmp/foo"
def test_mixed_quotes(self):
result = _strip_quotes("""echo "hello" && rm -rf '/tmp/x'""")
assert "rm" in result
assert "/tmp/x" not in result
class TestSplitClauses:
def test_and_operator(self):
clauses = _split_clauses("cd /tmp && rm -rf foo")
assert any("rm" in c for c in clauses)
def test_semicolon(self):
clauses = _split_clauses("echo hi; rm -rf /tmp/x")
assert any("rm" in c for c in clauses)
def test_pipe(self):
clauses = _split_clauses("ls | rm -rf /tmp/x")
assert any("rm" in c for c in clauses)
def test_or_operator(self):
clauses = _split_clauses("true || rm -rf /tmp/x")
assert any("rm" in c for c in clauses)
def test_subshell(self):
clauses = _split_clauses("echo $(rm -rf /tmp/x)")
assert any("rm" in c for c in clauses)
def test_backtick_subshell(self):
clauses = _split_clauses("echo `rm -rf /tmp/x`")
assert any("rm" in c for c in clauses)
class TestHasRecursiveFlag:
def test_rf(self):
assert _has_recursive_flag(["-rf", "/tmp/x"]) is True
def test_fr(self):
assert _has_recursive_flag(["-fr", "/tmp/x"]) is True
def test_r_alone(self):
assert _has_recursive_flag(["-r", "/tmp/x"]) is True
def test_uppercase_r(self):
assert _has_recursive_flag(["-R", "/tmp/x"]) is True
def test_rfv(self):
assert _has_recursive_flag(["-rfv", "/tmp/x"]) is True
def test_recursive_long(self):
assert _has_recursive_flag(["--recursive", "/tmp/x"]) is True
def test_no_recursive(self):
assert _has_recursive_flag(["-f", "/tmp/x"]) is False
def test_after_double_dash(self):
assert _has_recursive_flag(["--", "-rf", "/tmp/x"]) is False
def test_empty(self):
assert _has_recursive_flag([]) is False
class TestClauseHasRawRecursiveRm:
def test_basic_rm_rf(self):
assert _clause_has_raw_recursive_rm("rm -rf /tmp/x") is True
def test_rm_fr(self):
assert _clause_has_raw_recursive_rm("rm -fr /tmp/x") is True
def test_rm_rfv(self):
assert _clause_has_raw_recursive_rm("rm -rfv /tmp/x") is True
def test_rm_recursive_long(self):
assert _clause_has_raw_recursive_rm("rm --recursive /tmp/x") is True
def test_rm_uppercase_r(self):
assert _clause_has_raw_recursive_rm("rm -R /tmp/x") is True
def test_drone_rm_not_blocked(self):
assert _clause_has_raw_recursive_rm("drone rm /tmp/x") is False
def test_non_recursive_rm(self):
assert _clause_has_raw_recursive_rm("rm file.txt") is False
def test_rm_force_only(self):
assert _clause_has_raw_recursive_rm("rm -f file.txt") is False
def test_sudo_rm_rf(self):
assert _clause_has_raw_recursive_rm("sudo rm -rf /tmp/x") is True
def test_env_prefix_rm_rf(self):
assert _clause_has_raw_recursive_rm("env VAR=val rm -rf /tmp/x") is True
def test_absolute_path_rm(self):
assert _clause_has_raw_recursive_rm("/usr/bin/rm -rf /tmp/x") is True
def test_empty_clause(self):
assert _clause_has_raw_recursive_rm("") is False
def test_whitespace_clause(self):
assert _clause_has_raw_recursive_rm(" ") is False
class TestHandle:
CWD = "/home/patrick/Projects/AIPass/src/aipass/hooks"
def _bash(self, command: str) -> dict:
return handle({"tool_name": "Bash", "tool_input": {"command": command}, "cwd": self.CWD})
def _assert_blocked(self, result: dict):
assert result["exit_code"] == 2
parsed = json.loads(result["stdout"])
assert parsed["decision"] == "block"
assert "drone rm" in parsed["reason"]
def _assert_allowed(self, result: dict):
assert result["exit_code"] == 0
assert result["stdout"] == ""
def test_block_rm_rf(self):
self._assert_blocked(self._bash("rm -rf /tmp/x"))
def test_block_rm_fr(self):
self._assert_blocked(self._bash("rm -fr /tmp/x"))
def test_block_rm_rfv(self):
self._assert_blocked(self._bash("rm -rfv /tmp/x"))
def test_block_rm_recursive_long(self):
self._assert_blocked(self._bash("rm --recursive /tmp/x"))
def test_block_rm_uppercase_r(self):
self._assert_blocked(self._bash("rm -R /tmp/x"))
def test_block_rm_r(self):
self._assert_blocked(self._bash("rm -r /tmp/x"))
def test_allow_drone_rm(self):
self._assert_allowed(self._bash("drone rm /tmp/x"))
def test_allow_non_recursive_rm(self):
self._assert_allowed(self._bash("rm file.txt"))
def test_allow_rm_force_only(self):
self._assert_allowed(self._bash("rm -f file.txt"))
def test_block_compound_cd_and_rm(self):
self._assert_blocked(self._bash("cd /etc && rm -rf ."))
def test_block_compound_semicolon(self):
self._assert_blocked(self._bash("echo hi; rm -rf /tmp/x"))
def test_block_subshell_rm(self):
self._assert_blocked(self._bash("echo $(rm -rf /tmp/x)"))
def test_block_sudo_rm_rf(self):
self._assert_blocked(self._bash("sudo rm -rf /tmp/x"))
def test_block_absolute_path_rm(self):
self._assert_blocked(self._bash("/usr/bin/rm -rf /tmp/x"))
def test_rm_in_quoted_string_allowed(self):
self._assert_allowed(self._bash('echo "rm -rf /tmp/x"'))
def test_rm_in_single_quoted_string_allowed(self):
self._assert_allowed(self._bash("echo 'rm -rf /tmp/x'"))
def test_non_bash_tool_allowed(self):
result = handle({"tool_name": "Edit", "tool_input": {"file_path": "/tmp/x"}, "cwd": self.CWD})
self._assert_allowed(result)
def test_empty_command_allowed(self):
self._assert_allowed(self._bash(""))
def test_empty_hook_data(self):
result = handle({})
assert result["exit_code"] == 0
def test_no_tool_input(self):
result = handle({"tool_name": "Bash"})
assert result["exit_code"] == 0
@patch("aipass.hooks.apps.handlers.security.rm_gate.logger")
def test_exception_allows(self, mock_logger):
result = handle({"tool_name": "Bash", "tool_input": None, "cwd": self.CWD})
assert result["exit_code"] == 0
mock_logger.info.assert_called()
def test_block_variable_target(self):
self._assert_blocked(self._bash("rm -rf $DIR"))
def test_block_multiple_targets(self):
self._assert_blocked(self._bash("rm -rf /tmp/a /tmp/b"))