security: constrain agent command execution

backend/app/command_policy.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+import re
+import shlex
+
+SAFE_DIRECT_EXECUTABLES = {
+    "pytest",
+    "ruff",
+    "mypy",
+    "pyright",
+}
+SAFE_PACKAGE_SCRIPTS = {"test", "lint", "typecheck", "check"}
+SHELL_META = re.compile(r"[;&|`$<>\n\r]")
+
+
+def _validate_package_manager(argv: list[str]) -> None:
+    executable = argv[0]
+    if len(argv) < 2:
+        raise ValueError(f"{executable} requires an approved script command")
+
+    command = argv[1]
+    if command in SAFE_PACKAGE_SCRIPTS:
+        return
+    if command == "run" and len(argv) >= 3 and argv[2] in SAFE_PACKAGE_SCRIPTS:
+        return
+
+    raise ValueError(
+        f"{executable} commands are limited to test/lint/typecheck/check scripts"
+    )
+
+
+def parse_safe_command(command: str) -> list[str]:
+    """Parse a single test/tool command without invoking a shell.
+
+    This blocks shell composition and obvious command-runner escape hatches. It
+    is not a sandbox: approved test and lint tools can still execute repository
+    code or project-defined scripts.
+    """
+    value = command.strip()
+    if not value:
+        raise ValueError("command is empty")
+    if SHELL_META.search(value):
+        raise ValueError("shell operators and redirection are not allowed")
+
+    try:
+        argv = shlex.split(value, posix=True)
+    except ValueError as exc:
+        raise ValueError(f"invalid command syntax: {exc}") from exc
+
+    if not argv:
+        raise ValueError("command is empty")
+
+    executable = argv[0]
+    if "/" in executable or "\\" in executable:
+        raise ValueError("path-qualified executables are not allowed")
+
+    if executable in SAFE_DIRECT_EXECUTABLES:
+        return argv
+
+    if executable in {"python", "python3"}:
+        if len(argv) < 3 or argv[1] != "-m" or argv[2] not in {"pytest", "unittest"}:
+            raise ValueError("python commands are limited to -m pytest or -m unittest")
+        return argv
+
+    if executable in {"npm", "pnpm", "yarn"}:
+        _validate_package_manager(argv)
+        return argv
+
+    if executable == "go":
+        if len(argv) < 2 or argv[1] != "test":
+            raise ValueError("go commands are limited to go test")
+        return argv
+
+    if executable == "cargo":
+        if len(argv) < 2 or argv[1] not in {"test", "check", "clippy"}:
+            raise ValueError("cargo commands are limited to test/check/clippy")
+        return argv
+
+    raise ValueError(f"executable is not allowed: {executable}")

backend/app/main.py

@@ -14,6 +14,8 @@
 from fastapi.middleware.cors import CORSMiddleware
 from pydantic import BaseModel, Field
 
+from app.command_policy import parse_safe_command
+
 APP_NAME = "PatchPilot"
 ROOT = Path(__file__).resolve().parents[2]
 WORKSPACES = ROOT / ".workspaces"
@@ -331,6 +333,14 @@ async def run_agent(req: RunRequest) -> dict[str, Any]:
     plan_raw = await ask_model(req.provider, req.model, PLANNER, f"Repository: {req.repo_url}\nTask: {req.task}\n\nFiles:\n{chr(10).join(files)}\n\nContext:\n{context}")
     plan = parse_json_object(plan_raw)
     test_command = plan.get("test_command")
+    test_argv: list[str] | None = None
+    if test_command is not None:
+        if not isinstance(test_command, str):
+            raise HTTPException(502, "Planner returned a non-string test command")
+        try:
+            test_argv = parse_safe_command(test_command)
+        except ValueError as exc:
+            raise HTTPException(502, f"Planner returned an unsafe test command: {exc}") from exc
 
     coder_input = f"TASK:\n{req.task}\n\nIMPLEMENTATION PLAN:\n{json.dumps(plan, indent=2)}\n\nCURRENT REPOSITORY CONTEXT:\n{context}"
     raw_edits = await ask_model(req.provider, req.model, CODER, coder_input)
@@ -344,9 +354,9 @@ async def run_agent(req: RunRequest) -> dict[str, Any]:
     last_test_code = 0
 
     for iteration in range(1, req.max_iterations + 1):
-        if not test_command:
+        if not test_argv:
             break
-        test_code, test_output = run(["bash", "-lc", test_command], repo, 180)
+        test_code, test_output = run(test_argv, repo, 180)
         last_test_code, last_test_output = test_code, test_output
         history.append({"iteration": iteration, "action": "test", "returncode": test_code, "output": test_output[-5000:]})
         if test_code == 0 or iteration >= req.max_iterations:
@@ -373,7 +383,7 @@ async def run_agent(req: RunRequest) -> dict[str, Any]:
         "plan": plan.get("plan", []),
         "touched_files": plan.get("touched_files", []),
         "test_command": test_command,
-        "tests_passed": bool(test_command) and last_test_code == 0,
+        "tests_passed": bool(test_argv) and last_test_code == 0,
         "test_output": last_test_output[-12000:],
         "diff": diff_text,
         "history": history,
@@ -384,11 +394,11 @@ async def run_agent(req: RunRequest) -> dict[str, Any]:
 @app.post("/api/execute")
 async def execute(req: CommandRequest) -> dict[str, Any]:
     repo = workspace_repo(req.workspace_id)
-    command = req.command.strip()
-    blocked = ["rm -rf /", "mkfs", ":(){ :|:& };:", "shutdown", "reboot"]
-    if any(token in command for token in blocked):
-        raise HTTPException(400, "Command blocked by PatchPilot safety guard")
-    code, output = run(["bash", "-lc", command], repo, 180)
+    try:
+        argv = parse_safe_command(req.command)
+    except ValueError as exc:
+        raise HTTPException(400, f"Command blocked by PatchPilot safety policy: {exc}") from exc
+    code, output = run(argv, repo, 180)
     return {"returncode": code, "output": output[-12000:]}
 
 

backend/tests/test_command_policy.py

@@ -0,0 +1,83 @@
+import unittest
+
+from app.command_policy import parse_safe_command
+
+
+class CommandPolicyTests(unittest.TestCase):
+    def test_allows_plain_pytest_command(self) -> None:
+        self.assertEqual(
+            parse_safe_command("pytest -q tests/test_api.py"),
+            ["pytest", "-q", "tests/test_api.py"],
+        )
+
+    def test_allows_python_module_test_runner(self) -> None:
+        self.assertEqual(
+            parse_safe_command("python -m unittest discover -s tests"),
+            ["python", "-m", "unittest", "discover", "-s", "tests"],
+        )
+
+    def test_allows_approved_package_script(self) -> None:
+        self.assertEqual(
+            parse_safe_command("npm run lint"),
+            ["npm", "run", "lint"],
+        )
+        self.assertEqual(
+            parse_safe_command("pnpm test -- --runInBand"),
+            ["pnpm", "test", "--", "--runInBand"],
+        )
+
+    def test_allows_go_and_cargo_validation_commands(self) -> None:
+        self.assertEqual(parse_safe_command("go test ./..."), ["go", "test", "./..."])
+        self.assertEqual(parse_safe_command("cargo clippy --all-targets"), ["cargo", "clippy", "--all-targets"])
+
+    def test_rejects_shell_chaining(self) -> None:
+        with self.assertRaisesRegex(ValueError, "shell operators"):
+            parse_safe_command("pytest -q; rm -rf .git")
+
+    def test_rejects_command_substitution(self) -> None:
+        with self.assertRaisesRegex(ValueError, "shell operators"):
+            parse_safe_command("pytest $(cat /etc/passwd)")
+
+    def test_rejects_unlisted_executable(self) -> None:
+        with self.assertRaisesRegex(ValueError, "not allowed"):
+            parse_safe_command("git status")
+
+    def test_rejects_absolute_path_to_allowlisted_executable(self) -> None:
+        with self.assertRaisesRegex(ValueError, "path-qualified"):
+            parse_safe_command("/tmp/pytest -q")
+
+    def test_rejects_relative_path_to_allowlisted_executable(self) -> None:
+        with self.assertRaisesRegex(ValueError, "path-qualified"):
+            parse_safe_command("./pytest -q")
+
+    def test_rejects_windows_path_to_allowlisted_executable(self) -> None:
+        with self.assertRaisesRegex(ValueError, "path-qualified"):
+            parse_safe_command(r"tools\\pytest -q")
+
+    def test_restricts_direct_python_scripts(self) -> None:
+        with self.assertRaisesRegex(ValueError, "limited"):
+            parse_safe_command("python dangerous.py")
+
+    def test_rejects_direct_node_execution(self) -> None:
+        with self.assertRaisesRegex(ValueError, "not allowed"):
+            parse_safe_command("node -e 'console.log(1)'")
+
+    def test_rejects_package_manager_exec_escape_hatches(self) -> None:
+        for command in (
+            "npm exec node -- -e console.log(1)",
+            "pnpm exec node -e console.log(1)",
+            "yarn exec node -e console.log(1)",
+        ):
+            with self.subTest(command=command):
+                with self.assertRaisesRegex(ValueError, "limited"):
+                    parse_safe_command(command)
+
+    def test_rejects_go_run_and_cargo_run(self) -> None:
+        with self.assertRaisesRegex(ValueError, "go test"):
+            parse_safe_command("go run ./cmd/tool")
+        with self.assertRaisesRegex(ValueError, "test/check/clippy"):
+            parse_safe_command("cargo run --bin tool")
+
+
+if __name__ == "__main__":
+    unittest.main()