diff --git a/.aipass/hooks.json b/.aipass/hooks.json index 71f9fa42..b7e2d45f 100644 --- a/.aipass/hooks.json +++ b/.aipass/hooks.json @@ -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", diff --git a/CHANGELOG.md b/CHANGELOG.md index 17dabed6..a9d9c011 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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` diff --git a/src/aipass/drone/.seedgo/bypass.json b/src/aipass/drone/.seedgo/bypass.json index 3a634e38..d3103bfa 100644 --- a/src/aipass/drone/.seedgo/bypass.json +++ b/src/aipass/drone/.seedgo/bypass.json @@ -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", diff --git a/src/aipass/drone/apps/drone.py b/src/aipass/drone/apps/drone.py index 0343a515..386e7261 100644 --- a/src/aipass/drone/apps/drone.py +++ b/src/aipass/drone/apps/drone.py @@ -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 ", "Remove a custom command") + table.add_row("rm [...]", "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 [...]`` — 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) diff --git a/src/aipass/drone/apps/handlers/rm_handler.py b/src/aipass/drone/apps/handlers/rm_handler.py new file mode 100644 index 00000000..0c7a2ab2 --- /dev/null +++ b/src/aipass/drone/apps/handlers/rm_handler.py @@ -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 diff --git a/src/aipass/drone/apps/modules/rm.py b/src/aipass/drone/apps/modules/rm.py new file mode 100644 index 00000000..572aff2f --- /dev/null +++ b/src/aipass/drone/apps/modules/rm.py @@ -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 [...]") + 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]") diff --git a/src/aipass/drone/tests/test_rm.py b/src/aipass/drone/tests/test_rm.py new file mode 100644 index 00000000..159e4cbc --- /dev/null +++ b/src/aipass/drone/tests/test_rm.py @@ -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/ 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/.""" + 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/ 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/ 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): + """/.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/ 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() diff --git a/src/aipass/hooks/.seedgo/bypass.json b/src/aipass/hooks/.seedgo/bypass.json index 0cc5435b..f08d720a 100644 --- a/src/aipass/hooks/.seedgo/bypass.json +++ b/src/aipass/hooks/.seedgo/bypass.json @@ -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."}, diff --git a/src/aipass/hooks/README.md b/src/aipass/hooks/README.md index aac8db71..b108928c 100644 --- a/src/aipass/hooks/README.md +++ b/src/aipass/hooks/README.md @@ -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* --- diff --git a/src/aipass/hooks/apps/handlers/security/rm_gate.py b/src/aipass/hooks/apps/handlers/security/rm_gate.py new file mode 100644 index 00000000..f51cc878 --- /dev/null +++ b/src/aipass/hooks/apps/handlers/security/rm_gate.py @@ -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 # 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 diff --git a/src/aipass/hooks/tests/test_rm_gate.py b/src/aipass/hooks/tests/test_rm_gate.py new file mode 100644 index 00000000..78633f33 --- /dev/null +++ b/src/aipass/hooks/tests/test_rm_gate.py @@ -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"))