security: snapshot trusted verification inputs before agent runs

aicp/execution.py

@@ -6,7 +6,25 @@
 import time
 
 from .task_lifecycle import show_task
-from .verification import verify
+from .verification import run_checks, validate_checks
+
+
+def _load_trusted_inputs(project, name, checks_path):
+    """Snapshot the task and verification configuration before agent execution."""
+    project_root = project.resolve()
+    task_path = project / "tasks" / (name + ".json")
+    if checks_path.is_symlink() or not checks_path.is_file():
+        raise ValueError("Checks configuration must be a regular, non-symlink file")
+    if not checks_path.resolve().is_relative_to(project_root):
+        raise ValueError("Checks configuration must be inside project")
+    task = show_task(project, name)
+    task_bytes = task_path.read_bytes()
+    checks_bytes = checks_path.read_bytes()
+    config = json.loads(checks_bytes)
+    if not isinstance(config, dict) or set(config) != {"schema_version", "checks"} or config["schema_version"] != 1:
+        raise ValueError("Invalid checks configuration")
+    checks = validate_checks(config["checks"])
+    return task, task_path, task_bytes, checks_bytes, checks
 
 
 def run_task(project, name, tool, argv, checks_path, output, timeout=300):
@@ -18,7 +36,7 @@ def run_task(project, name, tool, argv, checks_path, output, timeout=300):
         raise ValueError("Agent timeout must be between 1 and 1800 seconds")
     if output.exists() or output.is_symlink():
         raise FileExistsError("Refusing to overwrite execution report")
-    task = show_task(project, name)
+    task, task_path, task_bytes, checks_bytes, trusted_checks = _load_trusted_inputs(project, name, checks_path)
     started = time.monotonic()
     try:
         proc = subprocess.run(argv, cwd=project, capture_output=True, timeout=timeout, check=False)
@@ -38,15 +56,28 @@ def run_task(project, name, tool, argv, checks_path, output, timeout=300):
         "stderr_bytes": len(stderr),
         "passed": status == "completed" and exit_code == 0,
     }
-    # Never execute verification if agent execution itself failed.
-    checks = verify(project, checks_path) if agent["passed"] else None
+    # Detect changes to the contract/configuration, and never execute a
+    # replacement check list supplied by the agent.
+    inputs_unchanged = (
+        task_path.is_file() and not task_path.is_symlink()
+        and checks_path.is_file() and not checks_path.is_symlink()
+        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
     report = {
         "schema_version": 1,
         "task": name,
         "objective": task["objective"],
         "agent": agent,
         "verification": checks,
-        "passed": bool(agent["passed"] and checks and checks["passed"]),
+        "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"]),
         "security_boundary": "trusted-host-execution-no-sandbox",
         "scope_enforced": False,
     }

docs/CLI.md

@@ -53,4 +53,4 @@ aicp run --path /path/to/project --name fix-retry \
   -- codex exec "Read tasks/fix-retry.json and implement the approved task"
 ```
 
-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 and can change the checks file before verification; independently review the resulting code and checks 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.
+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.

tests/test_execution.py

@@ -38,6 +38,37 @@ def test_agent_failure_skips_checks(self):
         self.assertFalse(result["passed"])
         self.assertIsNone(result["verification"])
 
+    def test_agent_modified_checks_are_rejected_without_running_them(self):
+        def tamper(*args, **kwargs):
+            self.checks.write_text('{"schema_version":1,"checks":[["python3","-c","print(123)"]]}', encoding="utf-8")
+            return type("Completed", (), {"returncode": 0, "stdout": b"", "stderr": b""})()
+
+        with patch("aicp.execution.subprocess.run", side_effect=tamper):
+            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.assertFalse(result["trusted_inputs"]["unchanged_after_agent"])
+        self.assertIsNone(result["verification"])
+
+    def test_agent_modified_task_is_rejected(self):
+        def tamper(*args, **kwargs):
+            task = self.project / "tasks" / "fix-retry.json"
+            task.write_text(task.read_text(encoding="utf-8") + " ", encoding="utf-8")
+            return type("Completed", (), {"returncode": 0, "stdout": b"", "stderr": b""})()
+
+        with patch("aicp.execution.subprocess.run", side_effect=tamper):
+            result = run_task(self.project, "fix-retry", "codex", ["codex"], self.checks, self.output)
+        self.assertFalse(result["passed"])
+        self.assertFalse(result["trusted_inputs"]["unchanged_after_agent"])
+
+    def test_invalid_checks_fail_before_agent_execution(self):
+        self.checks.write_text('{"schema_version":1,"checks":[]}', encoding="utf-8")
+        with patch("aicp.execution.subprocess.run") as agent:
+            with self.assertRaises(ValueError):
+                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)