diff --git a/aicp/execution.py b/aicp/execution.py index 750fde1..2ee3daf 100644 --- a/aicp/execution.py +++ b/aicp/execution.py @@ -6,6 +6,7 @@ import time from .task_lifecycle import show_task +from .scope_audit import git_snapshot, audit_scope from .verification import run_checks, validate_checks @@ -37,6 +38,9 @@ def run_task(project, name, tool, argv, checks_path, output, timeout=300): if output.exists() or output.is_symlink(): raise FileExistsError("Refusing to overwrite execution report") task, task_path, task_bytes, checks_bytes, trusted_checks = _load_trusted_inputs(project, name, checks_path) + baseline = git_snapshot(project) + if baseline: + raise ValueError('Scope audit requires a clean Git worktree before execution') started = time.monotonic() try: proc = subprocess.run(argv, cwd=project, capture_output=True, timeout=timeout, check=False) @@ -64,20 +68,23 @@ def run_task(project, name, tool, argv, checks_path, output, timeout=300): and task_path.read_bytes() == task_bytes and checks_path.read_bytes() == checks_bytes ) - checks = run_checks(project, trusted_checks) if agent["passed"] and inputs_unchanged else None + after = git_snapshot(project) + scope = audit_scope(baseline, after, task["scope"]["allowed_paths"]) + checks = run_checks(project, trusted_checks) if agent["passed"] and inputs_unchanged and scope["passed"] else None report = { "schema_version": 1, "task": name, "objective": task["objective"], "agent": agent, "verification": checks, + "scope_audit": scope, "trusted_inputs": { "task_sha256": hashlib.sha256(task_bytes).hexdigest(), "checks_sha256": hashlib.sha256(checks_bytes).hexdigest(), "unchanged_after_agent": inputs_unchanged, "verification_source": "pre-agent-snapshot", }, - "passed": bool(agent["passed"] and inputs_unchanged and checks and checks["passed"]), + "passed": bool(agent["passed"] and inputs_unchanged and scope["passed"] and checks and checks["passed"]), "security_boundary": "trusted-host-execution-no-sandbox", "scope_enforced": False, } diff --git a/aicp/scope_audit.py b/aicp/scope_audit.py new file mode 100644 index 0000000..d4a2cfc --- /dev/null +++ b/aicp/scope_audit.py @@ -0,0 +1,51 @@ +"""Best-effort Git working-tree change auditing (not filesystem enforcement).""" +import subprocess +from pathlib import PurePosixPath + + +def git_snapshot(project): + proc = subprocess.run( + ["git", "status", "--porcelain=v1", "-z", "--untracked-files=all"], + cwd=project, capture_output=True, check=False, timeout=30, + ) + if proc.returncode: + raise ValueError("Git status failed; scope audit requires a Git worktree") + entries = proc.stdout.split(b"\0") + paths = set() + i = 0 + while i < len(entries): + entry = entries[i] + i += 1 + if not entry: + continue + if len(entry) < 4: + raise ValueError("Unexpected Git status output") + path = entry[3:].decode("utf-8", errors="surrogateescape") + paths.add(path) + if entry[:1] in (b"R", b"C") or entry[1:2] in (b"R", b"C"): + if i >= len(entries) or not entries[i]: + raise ValueError("Incomplete Git rename entry") + paths.add(entries[i].decode("utf-8", errors="surrogateescape")) + i += 1 + return paths + + +def audit_scope(before, after, allowed_paths): + added = sorted(after - before) + allowed = [] + outside = [] + for path in added: + parts = PurePosixPath(path).parts + match = any( + path == prefix.rstrip("/") or path.startswith(prefix.rstrip("/") + "/") + for prefix in allowed_paths + ) + (allowed if match else outside).append(path) + return { + "newly_reported_paths": added, + "within_allowed_paths": allowed, + "outside_allowed_paths": outside, + "preexisting_changed_paths": sorted(before), + "passed": not outside, + "limitations": "git-status-delta-only; does not detect modifications to already dirty files, committed changes, ignored files, or reverted writes", + } diff --git a/docs/CLI.md b/docs/CLI.md index 9769895..b9b682e 100644 --- a/docs/CLI.md +++ b/docs/CLI.md @@ -54,3 +54,7 @@ aicp run --path /path/to/project --name fix-retry \ ``` The report records agent exit status, output hashes, verification results, and whether both stages passed. This is **not a sandbox**, and task allowed paths are **not enforced**. The agent executes on your host with your permissions. Task and check definitions are now validated and snapshotted **before** agent execution; changes to either input after the agent runs cause the run to fail without executing verification. Checks run from the pre-agent command snapshot. This does **not** make the underlying test code, dependencies, environment, or workspace immutable, and does not protect against concurrent filesystem races. Independently review the resulting code and tests before trusting the outcome. Do not pass secrets in command arguments, run on sensitive workspaces, or treat a passing report as permission to merge. CI uses mocked agent execution only. The command does not create branches, PRs, or merges. + +### Git workspace scope audit + +Supervised `aicp run` now requires a **clean Git worktree** at startup and records newly changed tracked/untracked paths after the agent exits. If any reported path is outside the task's `allowed_paths`, verification is skipped and the run fails. The JSON report includes `scope_audit` with paths and status. This is a **best-effort audit, not permission enforcement**: Git status cannot detect ignored files, committed changes, writes that are reverted, or writes outside the repository. The agent retains full host permissions. Use a disposable isolated environment, and independently review diffs and tests. diff --git a/tests/test_execution.py b/tests/test_execution.py index b78cbbe..55b0c69 100644 --- a/tests/test_execution.py +++ b/tests/test_execution.py @@ -18,6 +18,9 @@ def setUp(self): self.checks = self.project / "checks.json" self.checks.write_text('{"schema_version":1,"checks":[["python3","--version"]]}', encoding="utf-8") self.output = self.project / "report.json" + git_patch = patch("aicp.execution.git_snapshot", return_value=set()) + git_patch.start() + self.addCleanup(git_patch.stop) def test_successful_agent_and_checks(self): with patch("aicp.execution.subprocess.run") as run: @@ -69,6 +72,25 @@ def test_invalid_checks_fail_before_agent_execution(self): run_task(self.project, "fix-retry", "codex", ["codex"], self.checks, self.output) agent.assert_not_called() + def test_out_of_scope_changes_block_verification(self): + with patch("aicp.execution.git_snapshot", side_effect=[set(), {"src/fix.py", "secrets.txt"}]): + with patch("aicp.execution.subprocess.run") as agent: + agent.return_value.returncode = 0 + agent.return_value.stdout = b"" + agent.return_value.stderr = b"" + with patch("aicp.execution.run_checks") as checks: + result = run_task(self.project, "fix-retry", "codex", ["codex"], self.checks, self.output) + checks.assert_not_called() + self.assertFalse(result["passed"]) + self.assertEqual(result["scope_audit"]["outside_allowed_paths"], ["secrets.txt"]) + + def test_dirty_worktree_rejected_before_agent(self): + with patch("aicp.execution.git_snapshot", return_value={"existing.txt"}): + with patch("aicp.execution.subprocess.run") as agent: + with self.assertRaisesRegex(ValueError, "clean Git worktree"): + run_task(self.project, "fix-retry", "codex", ["codex"], self.checks, self.output) + agent.assert_not_called() + def test_invalid_agent_command_rejected(self): with self.assertRaises(ValueError): run_task(self.project, "fix-retry", "codex", ["sh", "-c", "true"], self.checks, self.output) diff --git a/tests/test_scope_audit.py b/tests/test_scope_audit.py new file mode 100644 index 0000000..928010c --- /dev/null +++ b/tests/test_scope_audit.py @@ -0,0 +1,22 @@ +"""Scope auditing unit tests.""" +import unittest +from aicp.scope_audit import audit_scope + + +class ScopeAuditTests(unittest.TestCase): + def test_prefixes_are_directory_boundaries(self): + result = audit_scope(set(), {"src/a.py", "src-extra/b.py", "tests/test_a.py"}, ["src/", "tests/"]) + self.assertEqual(result["outside_allowed_paths"], ["src-extra/b.py"]) + self.assertFalse(result["passed"]) + + def test_exact_file_allowed(self): + result = audit_scope(set(), {"README.md"}, ["README.md"]) + self.assertTrue(result["passed"]) + + def test_clean_delta(self): + result = audit_scope(set(), set(), ["src/"]) + self.assertTrue(result["passed"]) + + +if __name__ == "__main__": + unittest.main()