Coverage for src/ai_jury/patches.py: 97%
177 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-30 06:29 +0000
1"""Suggested-patch output for verified findings (issue #10).
3The jury identifies issues; this renders a *separate*, opt-in "suggested
4patches" section that turns verified findings into concrete, inspectable fix
5suggestions. It is deliberately conservative:
7- only VERIFIED findings (a consensus group the verifier confirmed) produce a
8 suggestion — unverified or rejected findings never do;
9- suggestions are rendered as clearly-labelled blocks tied to one finding;
10- nothing is ever applied automatically (read-only by design); the output is for
11 a human to inspect, copy, or adapt.
13Pure and deterministic: given the same groups it renders the same markdown.
14"""
16from __future__ import annotations
18from dataclasses import dataclass
19from pathlib import Path
21from .consensus import BUCKET_REJECTED, FindingGroup
22from .findings import fence_safe, flatten_inline
23from .redaction import redact
26@dataclass
27class PatchSuggestion:
28 file: str
29 line: int | None
30 severity: str
31 claim: str
32 suggested_fix: str
34 def location(self) -> str:
35 loc = self.file or "?"
36 if self.line is not None:
37 loc = f"{loc}:{self.line}"
38 return loc
41def patch_suggestions(groups: list[FindingGroup]) -> list[PatchSuggestion]:
42 """Return one suggestion per VERIFIED group that carries a suggested fix.
44 A group qualifies only when the verifier marked it ``verified`` (not
45 unsupported/disputed and not merely unverified) AND its representative
46 finding has a non-empty ``suggested_fix``. Order follows the input group
47 order (already severity-sorted by the consensus pass).
48 """
49 out: list[PatchSuggestion] = []
50 for g in groups:
51 if getattr(g, "status", "") != "verified" or g.bucket == BUCKET_REJECTED:
52 continue
53 rep = g.representative
54 fix = (getattr(rep, "suggested_fix", "") or "").strip()
55 if not rep or not fix:
56 continue
57 out.append(
58 PatchSuggestion(
59 file=rep.file or "",
60 line=rep.line,
61 severity=g.severity,
62 claim=(rep.claim or "").strip(),
63 suggested_fix=fix,
64 )
65 )
66 return out
69def render_patch_suggestions(groups: list[FindingGroup]) -> str:
70 """Render the "Suggested patches" markdown section, or "" when there are none.
72 Kept separate from the default report so the standard review flow stays
73 read-only; the CLI emits this only under ``--suggest-patches``.
74 """
75 suggestions = patch_suggestions(groups)
76 if not suggestions:
77 return ""
78 lines = [
79 "## Suggested patches",
80 "",
81 "_Opt-in, read-only suggestions for **verified** findings only. Inspect "
82 "before applying — nothing here is applied automatically._",
83 "",
84 ]
85 for s in suggestions:
86 # Flatten the heading text and break any fence-closer inside the
87 # suggestion body so attacker-influenced finding text can't inject a
88 # forged verdict/heading into the posted comment (audit 2026-06-13 r3).
89 lines.append(
90 f"### {flatten_inline(s.location())} — [{s.severity}] {flatten_inline(s.claim)}"
91 )
92 lines.append("")
93 lines.append("> Verified by the jury.")
94 lines.append("")
95 lines.append("```suggestion")
96 lines.append(fence_safe(s.suggested_fix))
97 lines.append("```")
98 lines.append("")
99 return "\n".join(lines).rstrip() + "\n"
102def parse_patch_suggestions(text: str) -> list[PatchSuggestion]:
103 """Parse PatchSuggestion objects from a markdown report or suggested-patches block."""
104 import re
106 out: list[PatchSuggestion] = []
107 # Pattern matches: ### file.py:123 — [severity] claim
108 heading_re = re.compile(
109 r"^###\s+([^—\n]+?)(?::(\d+))?\s+—\s+\[([^\]]+)\]\s+(.+)$", re.MULTILINE
110 )
111 suggestion_block_re = re.compile(r"```suggestion\n(.*?)\n```", re.DOTALL)
113 matches = list(heading_re.finditer(text))
114 for i, m in enumerate(matches):
115 file_path = m.group(1).strip()
116 line_num = int(m.group(2)) if m.group(2) else None
117 severity = m.group(3).strip()
118 claim = m.group(4).strip()
120 start = m.end()
121 end = matches[i + 1].start() if i + 1 < len(matches) else len(text)
122 sub_content = text[start:end]
124 fix_match = suggestion_block_re.search(sub_content)
125 if fix_match:
126 fix = fix_match.group(1).strip()
127 out.append(
128 PatchSuggestion(
129 file=file_path,
130 line=line_num,
131 severity=severity,
132 claim=claim,
133 suggested_fix=fix,
134 )
135 )
136 return out
139def _patch_body(fix: str) -> str:
140 """The patch text as a patch *file* would hold it — newline-terminated.
142 ``parse_patch_suggestions`` strips the fenced block, which takes the trailing
143 newline with it. Git then rejects bodies whose last line is a header rather
144 than content ("git diff header lacks filename information"), so a suggestion
145 carrying a rename or mode section could never be read at all — including by
146 the preview, which would report "nothing git could read" for a patch that is
147 merely missing its terminator. Content semantics are unaffected: a file whose
148 last line has no newline is expressed by git's own ``\\ No newline at end of
149 file`` marker, not by the patch text's terminator.
150 """
151 return fix if fix.endswith("\n") else fix + "\n"
154#: What a caller gets back when `git` could not be started at all. Unlike the `gh`
155#: wrappers in `github.py` there is no `shutil.which` guard here, so a machine without
156#: git on PATH raises `FileNotFoundError` from the spawn — and `jury apply` reached the
157#: user as a traceback rather than as the refusal both of these functions otherwise
158#: return. A spawn can also fail with git perfectly present, when the fork is refused
159#: (`ENOMEM`, `EAGAIN`); both are `OSError`.
160_GIT_SPAWN_FAILED = "Cannot run git: {detail}"
163def _git_apply(argv: list[str], fix: str, root: Path):
164 """Run ``git apply`` with ``fix`` on stdin, or None when git could not be started."""
165 import subprocess
167 try:
168 return subprocess.run(
169 argv,
170 input=_patch_body(fix),
171 text=True,
172 cwd=str(root),
173 capture_output=True,
174 )
175 except OSError:
176 return None
179def _probe_patch(fix: str, root: Path):
180 """Ask git what ``fix`` would do, writing nothing (``--check``).
182 Returns None when git could not be started; every caller treats that as "this
183 patch cannot be vouched for", which is the same answer a failed probe gives.
184 """
185 return _git_apply(["git", "apply", "--numstat", "-z", "--summary", "--check", "-"], fix, root)
188def preview_patch_suggestion(
189 suggestion: PatchSuggestion, root_dir: Path | None = None
190) -> tuple[list[str], str | None]:
191 """Paths ``suggestion`` would touch, and why it would be refused (if it would).
193 The operator-facing half of the containment check, sharing its probe so the
194 preview cannot disagree with what an apply would do (#605). Any hand-rolled
195 containment check is a bet that every way a patch can name a file was
196 enumerated; showing the operator git's own answer before writing is what makes
197 losing that bet survivable rather than silent.
199 Returns ``(paths, refusal)``. ``refusal`` is ``None`` when the suggestion would
200 be applied; ``paths`` is what git says it would touch, which for a refused
201 patch is exactly the evidence the operator needs to see.
202 """
203 root = (root_dir or Path.cwd()).resolve()
204 try:
205 target = (root / suggestion.file).resolve()
206 target.relative_to(root)
207 except (ValueError, RuntimeError):
208 return [], f"Path traversal rejected: {suggestion.file}"
209 sensitive = _sensitive_target(target, root)
210 if sensitive is not None:
211 # Preview and apply must agree (#605): the same refusal an apply would raise.
212 return [], sensitive
213 if not target.exists() or not target.is_file():
214 return [], f"File not found: {suggestion.file}"
216 fix = suggestion.suggested_fix
217 if not _looks_like_patch(fix):
218 # A literal line replacement touches exactly the file it names.
219 return [suggestion.file], None
221 probe = _probe_patch(fix, root)
222 if probe is None:
223 return [], _GIT_SPAWN_FAILED.format(detail="it could not be started")
224 if probe.returncode != 0: 224 ↛ 225line 224 didn't jump to line 225 because the condition on line 224 was never true
225 detail = redact(probe.stderr.strip())[0] or "patch does not apply cleanly"
226 return [], f"Git apply failed: {detail}"
227 paths = [
228 record.split("\t", 2)[2]
229 for record in probe.stdout.split("\0")[:-1]
230 if len(record.split("\t", 2)) == 3
231 ]
232 return paths, _containment_refusal(fix, root=root, target=target, file=suggestion.file)
235def _containment_refusal(fix: str, *, root: Path, target: Path, file: str) -> str | None:
236 """Why ``fix`` may not be applied, or ``None`` when it touches only ``target``.
238 Asks git, rather than reading the patch by hand. The previous check inspected
239 only ``---``/``+++`` header lines, but git carries filenames in several other
240 constructs and honours all of them: ``rename from``/``rename to``,
241 ``copy from``/``copy to``, ``old mode``/``new mode``, and a ``GIT binary
242 patch`` section which has no ``---``/``+++`` lines at all. A patch whose
243 headers named the suggested file could rename an unrelated path and still be
244 reported as "Applied git patch to <file>" (#603, reproduced).
246 That check was a **blocklist** — enumerate the dangerous header forms — and it
247 missed because git has more of them than the enumeration covered. Adding
248 ``rename from`` to the same loop repeats the design and misses the next one.
249 So the question goes to the parser that will actually apply the patch:
250 ``--check`` writes nothing, ``--numstat -z`` lists every path the patch would
251 touch, and ``--summary`` names the operations. Validation and application now
252 share one parser, which is what closes the gap rather than narrowing it.
253 """
254 probe = _probe_patch(fix, root)
255 if probe is None:
256 # Containment is decided by this probe, so "git would not start" is a refusal,
257 # never a pass: an unvalidated patch must not reach `git apply` below.
258 return _GIT_SPAWN_FAILED.format(detail="it could not be started")
259 if probe.returncode != 0:
260 detail = redact(probe.stderr.strip())[0] or "patch does not apply cleanly"
261 return f"Git apply failed: {detail}"
263 # NUL-terminated numstat records, then the summary block as trailing text.
264 chunks = probe.stdout.split("\0")
265 summary = chunks.pop() if chunks else ""
266 if not chunks:
267 # An allowlist answers "which paths does this touch?" — and "none that I
268 # could see" is not the same answer as "only the target". A patch git reads
269 # as touching nothing cannot be the fix this suggestion claims to be.
270 return f"Patch touches no files; nothing to apply to {file}"
271 for record in chunks:
272 fields = record.split("\t", 2)
273 if len(fields) != 3:
274 # Not a shape this parser understands. Refusing is the only safe
275 # reading: an unparsed record is a path that went unchecked.
276 return f"Unrecognized patch summary from git, refusing to apply to {file}"
277 try:
278 touched = (root / fields[2]).resolve()
279 touched.relative_to(root)
280 except (ValueError, RuntimeError):
281 return f"Path traversal rejected in patch: {fields[2]}"
282 if touched != target:
283 return f"Patch touches {fields[2]}, not {file}"
285 # `--numstat` reports a rename's *destination* but never its source, so a patch
286 # can delete a path that no numstat record mentions. A single-file suggestion
287 # has no business renaming or copying anything, so the operation itself is
288 # refused rather than its paths re-derived from prose.
289 for line in summary.splitlines():
290 operation = line.strip().split(" ", 1)[0]
291 if operation in ("rename", "copy"):
292 return f"Patch {operation}s a file; a suggestion for {file} may only edit it"
293 return None
296#: Line prefixes that mean "git will read this body as a patch".
297#:
298#: The old test — ``startswith("---") or "@@" in fix`` — recognised only a plain
299#: unified diff. A rename-only or binary-only body has neither, so it missed the
300#: git branch entirely and fell through to the line-replacement path, which wrote
301#: the diff *text* into the file and reported success (found while fixing #603).
302#: That is the same blocklist mistake as the containment check, one step earlier:
303#: a form this list does not name is not merely unvalidated, it is written
304#: literally. Everything git can read as a patch must reach the git branch, where
305#: :func:`_containment_refusal` decides whether it may be applied.
306_PATCH_MARKERS = ("diff --git ", "--- ", "+++ ", "@@", "GIT binary patch")
309def _looks_like_patch(fix: str) -> bool:
310 """Whether ``fix`` should be handled as a git patch rather than as literal text."""
311 if fix.startswith("---") or "@@" in fix:
312 return True
314 # bolt: avoid splitlines() memory allocation and any() generator on large patches via C-optimized search
315 if fix.startswith(_PATCH_MARKERS):
316 return True
317 for marker in _PATCH_MARKERS: # noqa: SIM110 - bolt: avoiding any() generator overhead is intentional
318 if f"\n{marker}" in fix:
319 return True
320 return False
323#: Directories a suggested patch may never write into, even though they sit inside the
324#: working tree. ``.git`` holds config and hooks that run a command on the next git
325#: operation; ``.github`` holds the workflows that run in CI. ``target`` is already
326#: ``resolve()``-d, so a symlink that redirects here is caught by the same check (#831).
327_SENSITIVE_DIRS = frozenset({".git", ".github"})
330def _sensitive_target(target: Path, root: Path) -> str | None:
331 # Precondition: ``target`` is already under ``root`` (the caller's traversal check
332 # returned otherwise), so ``relative_to`` cannot raise here. Case-fold the comparison:
333 # on a case-insensitive filesystem (macOS APFS, Windows NTFS) ``.Git/config`` names the
334 # real ``.git/config`` while ``resolve()`` keeps the casing it was given.
335 parts = target.relative_to(root).parts
336 for part in parts:
337 if part.lower() in _SENSITIVE_DIRS:
338 return (
339 f"refusing to write inside {part}/ ({'/'.join(parts)}): a suggested patch may "
340 "not touch the repository's git internals or CI configuration"
341 )
342 return None
345def apply_patch_suggestion(
346 suggestion: PatchSuggestion, root_dir: Path | None = None
347) -> tuple[bool, str]:
348 """Safely apply a patch suggestion to the targeted file."""
349 root = (root_dir or Path.cwd()).resolve()
350 try:
351 target = (root / suggestion.file).resolve()
352 target.relative_to(root)
353 except (ValueError, RuntimeError):
354 return False, f"Path traversal rejected: {suggestion.file}"
356 sensitive = _sensitive_target(target, root)
357 if sensitive is not None:
358 return False, sensitive
360 if not target.exists() or not target.is_file():
361 return False, f"File not found: {suggestion.file}"
363 fix = suggestion.suggested_fix
364 if _looks_like_patch(fix):
365 refusal = _containment_refusal(fix, root=root, target=target, file=suggestion.file)
366 if refusal is not None:
367 return False, refusal
369 # Same body the probe validated — a different one here would mean the
370 # containment check answered a question about a patch that is not applied.
371 proc = _git_apply(["git", "apply", "-"], fix, root)
372 if proc is None:
373 return False, _GIT_SPAWN_FAILED.format(detail="it could not be started")
374 if proc.returncode == 0: 374 ↛ 376line 374 didn't jump to line 376 because the condition on line 374 was always true
375 return True, f"Applied git patch to {suggestion.file}"
376 return (
377 False,
378 f"Git apply failed: {redact(proc.stderr.strip())[0] or 'patch does not apply cleanly'}",
379 )
381 try:
382 lines = target.read_text(encoding="utf-8").splitlines(keepends=True)
383 except (OSError, UnicodeDecodeError) as exc:
384 return False, f"Cannot read {suggestion.file}: {redact(str(exc))[0]}"
386 if suggestion.line is not None and 1 <= suggestion.line <= len(lines):
387 idx = suggestion.line - 1
388 lines[idx] = fix + ("\n" if not fix.endswith("\n") else "")
389 try:
390 target.write_text("".join(lines), encoding="utf-8")
391 except OSError as exc:
392 return False, f"Cannot write {suggestion.file}: {redact(str(exc))[0]}"
393 return True, f"Applied line replacement at {suggestion.file}:{suggestion.line}"
395 return False, f"Cannot apply non-diff suggestion without line match in {suggestion.file}"