diff --git a/plugins/security-guidance/hooks/llm.py b/plugins/security-guidance/hooks/llm.py index eff56de741..5b0f82c6f3 100644 --- a/plugins/security-guidance/hooks/llm.py +++ b/plugins/security-guidance/hooks/llm.py @@ -1009,6 +1009,14 @@ def _agentic_commit_review_enabled() -> bool: _SURVIVED_SCHEMA = review_api.SURVIVED_SCHEMA +def _findings_list(result: Optional[Dict[str, Any]]) -> List[Dict[str, Any]]: + """Return findings as a list, tolerating the common singleton shape.""" + findings = (result or {}).get("findings") or [] + if isinstance(findings, dict): + return [findings] + return findings if isinstance(findings, list) else [] + + def _agentic_spawn_env() -> Dict[str, str]: """opts.env for the SDK-spawned inner `claude` CLI. @@ -1136,18 +1144,15 @@ def agentic_review( "disk than in the diff, trust the diff.\n" ) + capped_diff_files = _cap_files_for_prompt(diff_files) diff_text = "\n\n".join( - f"=== DIFF: {fp} ===\n{content}" for fp, content in _cap_files_for_prompt(diff_files) + f"=== DIFF: {fp} ===\n{content}" for fp, content in capped_diff_files ) - user_prompt = ( - "Review this change for security vulnerabilities.\n\n" - f"Changed files (you may Read these and any other file in the repo):\n" - + "\n".join(f" - {p}" for p in touched_paths[:50]) - + context_note - + "\n\nUnified diff (only + lines are new):\n\n" - + diff_text - + "\n\nInvestigate per the method in your instructions, then return " - "the findings list." + user_prompt = review_api.build_investigate_prompt( + touched_paths, + capped_diff_files, + repo_root=context_dir, + context_note=context_note, ) # Always prefer the user's installed `claude` over the SDK's bundled CLI. @@ -1321,7 +1326,7 @@ def _run(system: str, prompt: str, *, schema: Dict[str, Any] # before refute drops most real findings; the eval-validated config keeps # mediums through to the final output. candidates = [ - f for f in (inv.get("findings") or []) + f for f in _findings_list(inv) if isinstance(f, dict) and f.get("severity") in ("critical", "high", "medium") ] metrics["pass1_candidates"] = len(candidates) @@ -1365,7 +1370,7 @@ def _scrub(s: object) -> str: ) if inv2: seen = {(c.get("filePath"), c.get("category")) for c in candidates} - for f in (inv2.get("findings") or []): + for f in _findings_list(inv2): if not isinstance(f, dict): continue if f.get("severity") not in ("critical", "high", "medium"): @@ -1694,4 +1699,3 @@ def analyze_security_concerns(files: List[Tuple[str, str]], is_diff: bool = Fals lines.append("") return "\n".join(lines) - diff --git a/plugins/security-guidance/hooks/review_api.py b/plugins/security-guidance/hooks/review_api.py index 499336b96f..d6e3494185 100644 --- a/plugins/security-guidance/hooks/review_api.py +++ b/plugins/security-guidance/hooks/review_api.py @@ -116,6 +116,11 @@ def cap_diff_for_prompt( Read every changed file in full and traced every new sink to a trusted source. +OUTPUT FORMAT: Return exactly one JSON object with a `findings` key. +findings MUST always be a JSON array. For one finding, return a one-item +array, never a single object. For no findings, return {"findings": []}. +Do not nest `findings` under another wrapper object. + BUDGET: you have at most ~15 tool calls. Spend them reading the changed files first, then 3-5 targeted Greps for callers/sinks. Do NOT exhaustively explore the repo — once you can name source→sink for each candidate (or rule it out), STOP. Partial findings are better than none.""" @@ -124,6 +129,10 @@ def cap_diff_for_prompt( "properties": { "findings": { "type": "array", + "description": ( + "Always a JSON array: [] for none, [object] for one, and " + "[object, ...] for multiple findings. Never return a single object." + ), "items": { "type": "object", "properties": { @@ -153,20 +162,107 @@ def cap_diff_for_prompt( } +def normalize_review_paths( + repo_root: str, touched_paths: list[str] +) -> tuple[list[str], list[str]]: + """Return (readable absolute paths, paths absent from this checkout). + + Git normally supplies repo-relative paths. Some imported diffs instead + carry root-anchored repo paths or absolute paths from another checkout. + Resolve those by the longest suffix that exists under ``repo_root`` and + never put an unresolved or out-of-worktree path in the reviewer's Read + allowlist. + """ + root = os.path.realpath(os.path.abspath(repo_root)) + readable: list[str] = [] + missing: list[str] = [] + seen_readable: set[str] = set() + seen_missing: set[str] = set() + + def _readable_in_repo(candidate: str) -> str | None: + resolved = os.path.realpath(candidate) + try: + in_repo = os.path.commonpath((root, resolved)) == root + except ValueError: + in_repo = False + if in_repo and os.path.isfile(resolved): + return resolved + return None + + for raw_path in touched_paths: + if not isinstance(raw_path, str) or not raw_path: + continue + + resolved = None + if not os.path.isabs(raw_path): + resolved = _readable_in_repo(os.path.join(root, raw_path)) + else: + resolved = _readable_in_repo(raw_path) + # A leading slash may mean "from the repo root" rather than the + # host filesystem root. Foreign checkout paths can also be + # recovered when a trailing repo-relative suffix exists here. + parts = [p for p in raw_path.replace("\\", "/").split("/") if p] + if resolved is None and ".." not in parts: + # Never recover a foreign path from its basename alone: an + # unrelated root-level file with the same name is not evidence + # that it is the file named by the diff. Require a suffix with + # at least two path components. + for index in range(max(0, len(parts) - 1)): + resolved = _readable_in_repo(os.path.join(root, *parts[index:])) + if resolved is not None: + break + + if resolved is not None: + if resolved not in seen_readable: + seen_readable.add(resolved) + readable.append(resolved) + elif raw_path not in seen_missing: + seen_missing.add(raw_path) + missing.append(raw_path) + + return readable, missing + + def build_investigate_prompt( touched_paths: list[str], diff_files: list[tuple[str, str]], *, + repo_root: str | None = None, context_note: str = "", ) -> str: + root = os.path.realpath(os.path.abspath(repo_root or os.getcwd())) + readable_paths, missing_paths = normalize_review_paths(root, touched_paths[:50]) capped, _ = cap_diff_for_prompt(diff_files) diff_text = "\n\n".join( f"=== DIFF: {fp} ===\n{content}" for fp, content in capped ) + readable_text = "\n".join(f" - {path}" for path in readable_paths) or " (none)" + path_instruction = ( + "All readable changed-file paths below are absolute and inside this " + "checkout. Use them directly; do not guess a different checkout root." + if readable_paths + else "No changed-file path below could be resolved inside this checkout. " + "Do not guess a different checkout root." + ) + missing_text = "" + if missing_paths: + missing_text = ( + "\n\nChanged paths not present in this checkout " + "(do not Read these paths):\n" + + "\n".join( + f" - {path} (not present in this checkout)" + for path in missing_paths + ) + ) return ( "Review this change for security vulnerabilities.\n\n" - "Changed files (you may Read these and any other file in the repo):\n" - + "\n".join(f" - {p}" for p in touched_paths[:50]) + f"Repository root: {root}\n" + + path_instruction + + "\n\n" + "Changed files present in this checkout " + "(you may Read these and any other file in the repo):\n" + + readable_text + + missing_text + context_note + "\n\nUnified diff (only + lines are new):\n\n" + diff_text diff --git a/plugins/security-guidance/hooks/test_review_paths.py b/plugins/security-guidance/hooks/test_review_paths.py new file mode 100644 index 0000000000..4b41a70561 --- /dev/null +++ b/plugins/security-guidance/hooks/test_review_paths.py @@ -0,0 +1,145 @@ +"""Regression tests for agentic review prompt path handling.""" +import os +import sys +import tempfile +import unittest + + +HOOKS_DIR = os.path.dirname(__file__) +if HOOKS_DIR not in sys.path: + sys.path.insert(0, HOOKS_DIR) + +import review_api +import llm + + +class ReviewPathNormalizationTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.repo_root = os.path.realpath(self.tmp.name) + os.makedirs(os.path.join(self.repo_root, "TaxEngine")) + with open( + os.path.join(self.repo_root, "TaxEngine", "Package.swift"), + "w", + encoding="utf-8", + ) as fixture: + fixture.write("// fixture\n") + + def tearDown(self): + self.tmp.cleanup() + + def test_relative_path_becomes_absolute(self): + readable, missing = review_api.normalize_review_paths( + self.repo_root, ["TaxEngine/Package.swift"] + ) + + self.assertEqual( + readable, + [os.path.join(self.repo_root, "TaxEngine", "Package.swift")], + ) + self.assertEqual(missing, []) + + def test_foreign_absolute_path_is_excluded_and_flagged(self): + readable, missing = review_api.normalize_review_paths( + self.repo_root, ["/home/user/App/missing.py"] + ) + + self.assertEqual(readable, []) + self.assertEqual(missing, ["/home/user/App/missing.py"]) + + def test_root_anchored_repo_path_is_normalized_when_present(self): + readable, missing = review_api.normalize_review_paths( + self.repo_root, ["/TaxEngine/Package.swift"] + ) + + self.assertEqual( + readable, + [os.path.join(self.repo_root, "TaxEngine", "Package.swift")], + ) + self.assertEqual(missing, []) + + def test_foreign_absolute_path_maps_by_existing_repo_suffix(self): + readable, missing = review_api.normalize_review_paths( + self.repo_root, + ["/Users/other-dev/App/TaxEngine/Package.swift"], + ) + + self.assertEqual( + readable, + [os.path.join(self.repo_root, "TaxEngine", "Package.swift")], + ) + self.assertEqual(missing, []) + + def test_foreign_path_does_not_map_by_bare_basename(self): + with open( + os.path.join(self.repo_root, "Config.swift"), + "w", + encoding="utf-8", + ) as fixture: + fixture.write("// unrelated fixture\n") + + foreign_path = "/other/vendor/Config.swift" + readable, missing = review_api.normalize_review_paths( + self.repo_root, [foreign_path] + ) + prompt = review_api.build_investigate_prompt( + [foreign_path], + [(foreign_path, "@@ -0,0 +1 @@\n+// foreign fixture")], + repo_root=self.repo_root, + ) + + self.assertEqual(readable, []) + self.assertEqual(missing, [foreign_path]) + self.assertIn( + f" - {foreign_path} (not present in this checkout)", prompt + ) + self.assertNotIn( + "All readable changed-file paths below are absolute", prompt + ) + self.assertIn( + "No changed-file path below could be resolved inside this checkout", + prompt, + ) + + def test_prompt_contains_repo_root_and_guarded_paths(self): + prompt = review_api.build_investigate_prompt( + ["/TaxEngine/Package.swift", "/home/user/App/missing.py"], + [("TaxEngine/Package.swift", "@@ -0,0 +1 @@\n+// fixture")], + repo_root=self.repo_root, + ) + + expected = os.path.join(self.repo_root, "TaxEngine", "Package.swift") + self.assertIn(f"Repository root: {self.repo_root}", prompt) + self.assertIn(f" - {expected}", prompt) + self.assertIn( + " - /home/user/App/missing.py (not present in this checkout)", + prompt, + ) + readable_section = prompt.split( + "Changed paths not present in this checkout", 1 + )[0] + self.assertNotIn(" - /home/user/App/missing.py", readable_section) + + +class FindingsContractTest(unittest.TestCase): + def test_prompt_requires_findings_array_for_zero_one_or_many_results(self): + self.assertIn( + "findings MUST always be a JSON array", + review_api.AGENTIC_INVESTIGATE_SYSTEM, + ) + findings_schema = review_api.FINDINGS_SCHEMA["properties"]["findings"] + self.assertEqual(findings_schema["type"], "array") + self.assertIn("Always a JSON array", findings_schema["description"]) + + def test_single_finding_object_is_normalized_to_list(self): + finding = { + "filePath": "src/auth.py", + "category": "Authorization", + "severity": "high", + } + + self.assertEqual(llm._findings_list({"findings": finding}), [finding]) + + +if __name__ == "__main__": + unittest.main()