feat(validation): validate findings after a scan

plugins/codex-security/mcp-app/tests/test_artifact_scan_draft_recovery.ts

@@ -1065,6 +1065,13 @@ for (const layout of ["standard", "diff", "worker"] as const) {
           f.write(input),
         );
       await f.write(f.draft({ deferred: [pending] }));
+      const checkpoints = path.join(f.root, "checkpoints");
+      for (const name of await readdir(checkpoints))
+        await utimes(path.join(checkpoints, name), 1, 1);
+      for (const name of layout === "worker"
+        ? ["result.json", "checkpoint-head.json"]
+        : ["coverage.json", "scan-manifest.json", "findings.json"])
+        await utimes(path.join(f.root, name), 1, 1);
       await fail(closing);
       await f.write(f.draft({}, true));
       assert.deepEqual((await f.read()).deferred, []);

plugins/codex-security/mcp-app/tests/test_deep_scan_executor.ts

@@ -2365,6 +2365,16 @@ async function testWorkerCyberAccessSettings() {
         api_key_model_discovery: true,
       },
     },
+    {
+      name: "red",
+      configuration:
+        '[codex_security]\ncyber_access_program = "daybreak_red"\n[features]\napi_key_cyber_access_programs = true\napi_key_model_discovery = true\n',
+      program: "daybreak_red",
+      features: {
+        api_key_cyber_access_programs: true,
+        api_key_model_discovery: true,
+      },
+    },
     {
       name: "explicit-false",
       configuration:

plugins/codex-security/scripts/workbench_cli.py

@@ -101,6 +101,7 @@ def add_command(name: str, *required: str) -> argparse.ArgumentParser:
 
     get_scan = add_command("get-scan", "--scan-id")
     get_scan.add_argument("--occurrence-id")
+    get_scan.add_argument("--check-target", action="store_true")
 
     add_command("rename-scan", "--scan-id", "--name")
 

plugins/codex-security/scripts/workbench_db.py

@@ -114,6 +114,7 @@
     require_git_worktree_head,
     require_remediation_target,
     require_scan_target_identity,
+    require_scan_target_unchanged,
     restore_directory_junctions,
     scan_target_warning,
     worktree_content_digest,
@@ -3441,6 +3442,8 @@ def main(*, before_archive: Callable[[], None] | None = None) -> None:
         elif args.command == "start-headless-standard-scan":
             result = _start_prompt_driven_scan(connection, args, headless_standard=True)
         elif args.command == "get-scan":
+            if args.check_target:
+                require_scan_target_unchanged(require_scan(connection, args.scan_id))
             result = scan_context(connection, args.scan_id, args.occurrence_id)
         elif args.command == "rename-scan":
             result = scan_history.rename_scan(

plugins/codex-security/scripts/workbench_finding_workflows.py

@@ -6,12 +6,19 @@
 import hashlib
 import json
 import sqlite3
+import stat
 import sys
 from pathlib import Path
 from typing import Any
 
 sys.path.insert(0, str(Path(__file__).resolve().parent))
-from workbench_target import directory_content_digest, git_bytes, git_revision
+from workbench_target import (
+    UnreadableLocalFile,
+    UnsupportedLocalFileType,
+    directory_content_digest,
+    git_bytes,
+    git_revision,
+)
 
 WORKFLOW_BINDINGS = {
     "repositoryPath": "repository_path",
@@ -123,7 +130,24 @@ def finding_workflow(
     if payload["action"] == "get":
         return {"workflow": read_workflow(connection, workflow_id)}
     if payload["action"] == "source":
-        target = Path(payload["repository"]).resolve(strict=True)
+        evidence = payload.get("evidence") is True
+        if evidence:
+            try:
+                target = Path(payload["repository"])
+                metadata = target.lstat()
+                if (
+                    not target.is_absolute()
+                    or not stat.S_ISDIR(metadata.st_mode)
+                    or getattr(metadata, "st_reparse_tag", 0) & 0x20000000
+                    or target.resolve(strict=True) != target
+                ):
+                    raise UnreadableLocalFile("The validation evidence directory was replaced.")
+            except (OSError, RuntimeError, UnreadableLocalFile):
+                if payload.get("optional") is True:
+                    return {"source": None}
+                raise
+        else:
+            target = Path(payload["repository"]).resolve(strict=True)
         private_paths = payload.get("privateStatePaths", [])
         if not isinstance(private_paths, list) or any(
             not isinstance(path, str) or not Path(path).is_absolute() for path in private_paths
@@ -133,18 +157,31 @@ def finding_workflow(
         excluded = tuple(path for path in excluded if path.is_relative_to(target))
         if target in excluded:
             raise SystemExit("Private runtime storage must not exclude the repository root.")
+        try:
+            content = directory_content_digest(
+                target,
+                excluded=excluded,
+                include_ignored=True,
+                include_git_metadata=evidence,
+            )
+        except (UnsupportedLocalFileType, UnreadableLocalFile, OSError):
+            if payload.get("optional") is True:
+                return {"source": None}
+            raise
         return {
             "source": {
                 "repository": str(target),
-                "revision": "unversioned" if payload.get("gitDisabled") else git_revision(target),
+                "revision": (
+                    "unversioned"
+                    if evidence or payload.get("gitDisabled")
+                    else git_revision(target)
+                ),
                 "refsDigest": hashlib.sha256(
                     b""
-                    if payload.get("gitDisabled")
+                    if evidence or payload.get("gitDisabled")
                     else (git_bytes(target, "show-ref") or b"").strip()
                 ).hexdigest(),
-                "content": directory_content_digest(
-                    target, excluded=excluded, include_ignored=True
-                ),
+                "content": content,
                 **({"privateStatePaths": [str(path) for path in excluded]} if excluded else {}),
             }
         }
@@ -158,15 +195,20 @@ def finding_workflow(
         binding = payload["binding"]
         source = binding["source"]
         scope = binding["scope"]
+        on_conflict = (
+            "DO UPDATE SET result_json = excluded.result_json, created_at = excluded.created_at"
+            if payload.get("replace") is True and binding.get("stage") == "scan-validation"
+            else "DO NOTHING"
+        )
         with connection:
             connection.execute(
-                """INSERT INTO finding_workflow_reviews
+                f"""INSERT INTO finding_workflow_reviews
                 (workflow_id, review_key, review_contract_version, codex_version,
                  source_repository_path, source_revision, source_refs_digest, source_content_digest,
                  scope_repository_id, scope_all_repositories, model, effort, settings_digest,
                  prompt_digest, contract_digest, result_json, created_at)
                 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
-                ON CONFLICT(workflow_id, review_key) DO NOTHING""",
+                ON CONFLICT(workflow_id, review_key) {on_conflict}""",
                 (
                     workflow_id,
                     payload["key"],

plugins/codex-security/scripts/workbench_target.py

@@ -27,6 +27,14 @@
 _WINDOWS = os.name == "nt"
 
 
+class UnreadableLocalFile(SystemExit):
+    """Source contents are unavailable for an optional review cache."""
+
+
+class UnsupportedLocalFileType(SystemExit):
+    """An entry cannot be read as source content for a review cache."""
+
+
 def committed_diff_snapshot_digest(kind: str, base_revision: str, head_revision: str) -> str:
     digest = hashlib.sha256(
         f"codex-security-diff/v1\0{kind}\0{base_revision}\0{head_revision}".encode()
@@ -418,15 +426,18 @@ def worktree_content_digest(target: Path) -> str:
 
 
 def remediation_checkout_snapshot(
-    scan: sqlite3.Row, *, expected_revision: str | None = None
+    scan: sqlite3.Row,
+    *,
+    expected_revision: str | None = None,
+    head_changed_message: str = (
+        "Repository HEAD changed. Regenerate the remediation patch against the current checkout."
+    ),
 ) -> tuple[str, str | None]:
     target = require_scan_target_identity(scan)
     revision = git_revision(target)
     required_revision = expected_revision or scan["target_revision"]
     if revision != required_revision:
-        raise SystemExit(
-            "Repository HEAD changed. Regenerate the remediation patch against the current checkout."
-        )
+        raise SystemExit(head_changed_message)
     content_digest = (
         worktree_content_digest(target)
         if revision != "unversioned"
@@ -435,6 +446,36 @@ def remediation_checkout_snapshot(
     return revision, content_digest
 
 
+def require_scan_target_unchanged(scan: sqlite3.Row) -> None:
+    head_changed_message = "Repository HEAD changed. Run a new scan before validating its findings."
+    if (
+        scan["diff_target_kind"] not in {"working_tree", "commit", "range"}
+        and scan["target_snapshot_digest"] is None
+        and scan["mode"] != "diff"
+        and scan["target_revision"] != "unversioned"
+    ):
+        # Legacy Git records bind only the revision, not working-tree contents.
+        target = require_scan_target_identity(scan)
+        if git_revision(target) != scan["target_revision"]:
+            raise SystemExit(head_changed_message)
+        require_clean_submodule_worktrees(target)
+        return
+    _, current_digest = remediation_checkout_snapshot(
+        scan, head_changed_message=head_changed_message
+    )
+    if scan["diff_target_kind"] == "working_tree":
+        expected_digest = scan["diff_content_digest"]
+    elif scan["diff_target_kind"] in {"commit", "range"}:
+        expected_digest = clean_worktree_content_digest()
+    else:
+        expected_digest = scan["target_snapshot_digest"]
+    if current_digest != expected_digest:
+        raise SystemExit(
+            "Scan target contents changed since the scan. "
+            "Run a new scan before validating its findings."
+        )
+
+
 def worktree_content_digest_for_context(
     repository: Path,
     pathspec: str,
@@ -800,6 +841,7 @@ def source_directory_snapshot_paths(
     target: Path,
     excluded: tuple[Path, ...] = (),
     *,
+    include_git_metadata: bool = False,
     onerror: Callable[[OSError], None] | None = None,
 ) -> list[Path]:
     paths: list[Path] = []
@@ -809,15 +851,17 @@ def source_directory_snapshot_paths(
         parent = Path(directory)
         for name in directories[:]:
             path = parent / name
-            if name == ".git" or path in excluded:
+            if (name == ".git" and not include_git_metadata) or path in excluded:
                 directories.remove(name)
                 continue
             paths.append(path)
             # Name-surrogate reparse points include Windows directory junctions.
             if getattr(path.lstat(), "st_reparse_tag", 0) & 0x20000000:
                 directories.remove(name)
         paths.extend(
-            parent / name for name in files if name != ".git" and parent / name not in excluded
+            parent / name
+            for name in files
+            if (name != ".git" or include_git_metadata) and parent / name not in excluded
         )
     return sorted(paths)
 
@@ -827,6 +871,7 @@ def directory_content_digest(
     *,
     excluded: tuple[Path, ...] = (),
     include_ignored: bool = False,
+    include_git_metadata: bool = False,
 ) -> str:
     def raise_walk_error(error: OSError) -> None:
         raise error
@@ -835,7 +880,9 @@ def raise_walk_error(error: OSError) -> None:
         path.relative_to(target) for path in excluded if path.is_relative_to(target)
     ]
     paths = (
-        source_directory_snapshot_paths(target, excluded, onerror=raise_walk_error)
+        source_directory_snapshot_paths(
+            target, excluded, include_git_metadata=include_git_metadata, onerror=raise_walk_error
+        )
         if include_ignored
         else git_directory_snapshot_paths(target)
     )
@@ -850,7 +897,7 @@ def raise_walk_error(error: OSError) -> None:
         try:
             metadata = path.lstat()
         except OSError as exc:
-            raise SystemExit(f"Could not read local file: {relative_path}") from exc
+            raise UnreadableLocalFile(f"Could not read local file: {relative_path}") from exc
         mode = metadata.st_mode
         link_target = (
             os.readlink(path)
@@ -874,12 +921,12 @@ def raise_walk_error(error: OSError) -> None:
                         content_digest.update(chunk)
                         content_size += len(chunk)
             except OSError as exc:
-                raise SystemExit(f"Could not read local file: {relative_path}") from exc
+                raise UnreadableLocalFile(f"Could not read local file: {relative_path}") from exc
             update_digest_field(digest, b"kind", b"file")
             update_digest_field(digest, b"size", str(content_size).encode())
             update_digest_field(digest, b"content-sha256", content_digest.digest())
         else:
-            raise SystemExit(f"Unsupported local file type: {relative_path}")
+            raise UnsupportedLocalFileType(f"Unsupported local file type: {relative_path}")
     return f"codex-security-snapshot/v1:sha256:{digest.hexdigest()}"