Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 17 additions & 13 deletions plugins/security-guidance/hooks/llm.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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"):
Expand Down Expand Up @@ -1694,4 +1699,3 @@ def analyze_security_concerns(files: List[Tuple[str, str]], is_diff: bool = Fals
lines.append("")

return "\n".join(lines)

100 changes: 98 additions & 2 deletions plugins/security-guidance/hooks/review_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""


Expand All @@ -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": {
Expand Down Expand Up @@ -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
Expand Down
145 changes: 145 additions & 0 deletions plugins/security-guidance/hooks/test_review_paths.py
Original file line number Diff line number Diff line change
@@ -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()