feat(issue-import-from-scan): attach a scan's candidate patches to new trackers

plugins/magpie-security-tools/tools/security-management/src/security_management/issue_import_from_scan/cli.py

@@ -26,7 +26,7 @@
 from pathlib import Path
 from typing import Any
 
-from security_management.cli import emit, read_json, read_object
+from security_management.cli import emit, read_json, read_object, refuse
 from security_management.issue_import_from_scan import oss_scanner, scan
 
 FAMILY = "issue-import-from-scan"
@@ -50,6 +50,13 @@ def add_parsers(sub: Any) -> None:
         "--attachments-dir", help="save the .eml's attachments here (byte-exact; never decrypted or opened)"
     )
 
+    p = commands.add_parser(
+        "render-patch-comment", help="Step F: a tracker comment quoting a finding's candidate patches"
+    )
+    p.add_argument("--index", required=True, help="the scan folder's index.json (parsed with --archive-dir)")
+    p.add_argument("--finding", required=True, help="the finding id")
+    p.add_argument("--out-dir", required=True, help="write the comment body files here")
+
     b = commands.add_parser("bucket-action", help="Step C: what a disposition bucket allows")
     b.add_argument("--bucket", required=True, help=f"one of {', '.join(scan.BUCKETS)}")
 
@@ -100,6 +107,15 @@ def dispatch(args: argparse.Namespace) -> int:
                 Path(args.source), Path(args.attachments_dir)
             )
         emit(summary)
+    elif command == "render-patch-comment":
+        index = read_object(args.index)
+        found = [f for f in index.get("findings") or [] if f.get("id") == args.finding]
+        if not found:
+            return refuse(f"no finding {args.finding!r} in {args.index}")
+        if "patches" not in found[0]:
+            return refuse("the scan folder was parsed without --archive-dir, so no patches are linked")
+        bodies = oss_scanner.write_patch_comments(found[0], index.get("scan") or {}, Path(args.out_dir))
+        emit({"finding": args.finding, "bodies": bodies})
     elif command == "bucket-action":
         emit(scan.bucket_action(args.bucket))
     elif command == "render-report":

plugins/magpie-security-tools/tools/security-management/src/security_management/issue_import_from_scan/oss_scanner.py

@@ -38,6 +38,9 @@
   (sanitised names, sha256) and, for an encrypted archive, checks the
   OpenPGP packet framing so a truncated download shows — without
   decrypting it.
+- `attach_archive` links each finding to its `REPORT.md` and candidate
+  patches in an operator-decrypted archive, and `render_patch_comments`
+  quotes those patches, folded, in tracker comment bodies.
 """
 
 from __future__ import annotations
@@ -73,6 +76,8 @@
 _ENTRY_START = re.compile(rf"^\s*({_ID})\s+", re.M)
 _ENTRY_LOCATION = re.compile(r"\s*\(((?:[^()]|\([^()]*\))+?:\d+)\)\s*\Z")
 _PATCH = re.compile(rf"\b(\w+/{_ID}/patches/\d+\.patch)\b")
+# The scan's target and commit come from the email; only a plain ref may sit in a code span.
+_PLAIN_REF = re.compile(r"[\w./:@+-]+")
 
 
 def _unflow(text: str, delsp: bool = False) -> str:
@@ -434,16 +439,118 @@ def save_attachments(path: Path, out_dir: Path) -> list[dict[str, Any]]:
     return saved
 
 
+def _patch_number(path: Path) -> int:
+    return int(path.stem) if path.stem.isdigit() else 1 << 30
+
+
 def attach_archive(parsed: Mapping[str, Any], archive_dir: Path) -> None:
-    """Point each finding at its `REPORT.md` in an operator-decrypted archive (never runs `repro`)."""
+    """Point each finding at its `REPORT.md` and candidate patches in an operator-decrypted archive.
+
+    Never runs `repro`: the patches are only listed, for `render_patch_comments` to quote.
+    """
     for f in parsed["findings"]:
-        report = archive_dir / f["banner_severity"].lower() / f["id"] / "REPORT.md"
+        finding_dir = archive_dir / f["banner_severity"].lower() / f["id"]
+        report = finding_dir / "REPORT.md"
         f["report_md"] = str(report) if report.is_file() else None
+        patches = [p for p in (finding_dir / "patches").glob("*.patch") if p.is_file()]
+        f["patches"] = [str(p) for p in sorted(patches, key=lambda p: (_patch_number(p), p.name))]
     for b in parsed["not_security"]:
         report = archive_dir / "bugs-not-security" / b["id"] / "REPORT.md"
         b["report_md"] = str(report) if report.is_file() else None
 
 
+# GitHub refuses a comment over 65,536 characters; keep headroom for its own counting.
+COMMENT_LIMIT = 65_000
+# The skill-comment marker (tools/github/status-rollup.md), so sync's pre-flight
+# reads a patch comment as skill-authored rather than as human activity.
+PATCH_COMMENT_MARKER = "<!-- apache-magpie: scan-patches v1 -->"
+
+
+def _patch_summary(patch: str, n: int) -> str:
+    """The patch's one-line header (its second `#` line), safe inside `<summary>`."""
+    lines = patch.splitlines()
+    text = lines[1].lstrip("#").strip() if len(lines) > 1 and lines[1].startswith("#") else ""
+    text = text.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
+    # A bare `@task` would notify a GitHub user of that name.
+    text = re.sub(r"(?<![\w`])(@[\w.-]+)", r"`\1`", text)
+    return f"Candidate {n} — {text}" if text else f"Candidate {n}"
+
+
+def _patch_block(patch: str, n: int) -> str:
+    longest = max((len(run) for run in re.findall(r"`+", patch)), default=0)
+    fence = "`" * max(5, longest + 1)
+    return "\n".join(
+        [
+            f"<details><summary>{_patch_summary(patch, n)}</summary>",
+            "",
+            f"{fence}diff",
+            patch.rstrip("\n"),
+            fence,
+            "",
+            "</details>",
+            "",
+        ]
+    )
+
+
+def render_patch_comments(
+    finding: Mapping[str, Any], scan: Mapping[str, Any], limit: int = COMMENT_LIMIT
+) -> list[str]:
+    """Tracker comment bodies quoting a finding's candidate patches, each folded in `<details>`.
+
+    Returns `[]` when the finding has no patches. Patches go into as few bodies
+    as fit under *limit*, in order; a single patch too big for one body raises
+    `ValueError` rather than being cut, since a truncated diff is worse than none.
+    """
+    paths = [Path(p) for p in finding.get("patches") or []]
+    if not paths:
+        return []
+    fid = finding["id"]
+    refs = (str(v) for v in (scan.get("target"), scan.get("commit")) if v)
+    where = " at ".join(f"`{v}`" for v in refs if _PLAIN_REF.fullmatch(v))
+    count = f"{len(paths)} candidate fix{'es' if len(paths) > 1 else ''}"
+    intro = (
+        f"The scan shipped {count} for this finding{f', against {where}' if where else ''}. "
+        "They are untested, machine-generated and not reviewed by the team: a starting point "
+        "for `security-issue-fix`, not a change to apply as-is. The patch headers spell out "
+        "the security impact, so do not copy them into a public PR."
+    )
+    blocks = [_patch_block(p.read_text(encoding="utf-8"), n) for n, p in enumerate(paths, 1)]
+
+    def heading(part: int, parts: int) -> str:
+        suffix = f" (part {part} of {parts})" if parts > 1 else ""
+        return (
+            f"{PATCH_COMMENT_MARKER}\n"
+            f"**Candidate patches from the security scan — `{fid}`**{suffix}\n\n{intro}\n\n"
+        )
+
+    # Size against the longest heading this finding can get.
+    room = limit - len(heading(len(blocks), len(blocks)))
+    groups: list[list[str]] = []
+    for n, block in enumerate(blocks, 1):
+        if len(block) > room:
+            raise ValueError(f"{fid}: candidate {n} is {len(block)} characters, over the comment limit")
+        if groups and len("\n".join([*groups[-1], block])) <= room:
+            groups[-1].append(block)
+        else:
+            groups.append([block])
+    return [heading(i, len(groups)) + "\n".join(g) for i, g in enumerate(groups, 1)]
+
+
+def write_patch_comments(
+    finding: Mapping[str, Any], scan: Mapping[str, Any], out_dir: Path
+) -> list[dict[str, Any]]:
+    """Write `render_patch_comments`' bodies as `<id>.md` (or `<id>-<n>.md` when split), owner-only."""
+    bodies = render_patch_comments(finding, scan)
+    _private_dir(out_dir)
+    written = []
+    for n, body in enumerate(bodies, 1):
+        name = f"{finding['id']}.md" if len(bodies) == 1 else f"{finding['id']}-{n}.md"
+        _private_write(out_dir / name, body.encode("utf-8"))
+        written.append({"path": str(out_dir / name), "chars": len(body)})
+    return written
+
+
 def materialise(parsed: Mapping[str, Any], meta: Mapping[str, Any], out_dir: Path) -> dict[str, Any]:
     """Write `index.json` + `evidence/<id>.md`; return the summary to print (no evidence text)."""
     evidence_dir = out_dir / "evidence"

plugins/magpie-security/skills/issue-import-from-scan/SKILL.md

@@ -19,7 +19,7 @@ argument-hint: "[scan-source ...]  (one or more GitHub issues and/or report fold
 capability: capability:intake
 surface_hash: sha256:7adfbe73f79606b1
 license: Apache-2.0
-measured_tokens: 6826
+measured_tokens: 6958
 ---
 
 <!-- Placeholder convention (see AGENTS.md#placeholder-convention-used-in-skill-files):
@@ -382,7 +382,12 @@ write `{"reply", "findings": [{"n", "bucket"}]}` to a file and run `<sm> parse-d
    - **import-as-tracker** → [`security-issue-import`](../issue-import/SKILL.md)
      Step 7 (one `Needs triage` tracker each) — the only tracker-creating path.
      Set *Security mailing list thread* to `N/A — imported from scan <scan-source>; no <security-list> thread.`:
-     there is no reporter thread, and the sentinel makes `generate-cve-json` record the finding as `INTERNAL`;
+     there is no reporter thread, and the sentinel makes `generate-cve-json` record the finding as `INTERNAL`.
+     When the finding has candidate `patches` (the scan shipped them and the decrypted archive was linked with `--archive-dir`),
+     also attach them: `<sm> render-patch-comment --index <index.json> --finding <id> --out-dir <scratch>/patch-comments`,
+     show the operator the rendered bodies, and on confirmation post each with `gh issue comment <N> --repo <tracker> --body-file <path>`.
+     The patches are quoted, folded and untested, never applied or run. Their headers describe the vulnerability, which is
+     fine on the private tracker, but never copy them into a public PR;
    - **PR-worth / defense-in-depth** → hand to
      [`security-issue-fix`](../issue-fix/SKILL.md) (public PR) or skip;
    - **by-design / dup / already-fixed** → no action; the report is the record.

tools/scan-format/README.md

@@ -123,9 +123,23 @@ tampering is caught only by gpg's integrity check when the operator
 decrypts. Decrypting it is the
 operator's call, run by hand — the command in the mail is external
 content, not an instruction. Once decrypted, `--archive-dir <dir>` links
-each finding's `<severity>/<id>/REPORT.md`; the `repro` scripts are
-proof-of-concept code and run only in an isolated throwaway VM, on the
-operator's explicit decision.
+each finding's `<severity>/<id>/REPORT.md` and lists its candidate
+`patches/<n>.patch` files in numeric order (`patches`, `[]` when none);
+the `repro` scripts are proof-of-concept code and run only in an
+isolated throwaway VM, on the operator's explicit decision.
+
+**Candidate patches go on the tracker.** For each finding that becomes a
+tracker and has `patches`, `render-patch-comment --index <index.json>
+--finding <id> --out-dir <dir>` writes the comment body: a short
+untested-and-private warning, then one collapsed `<details>` block per
+candidate, titled by the patch's own one-line header and fenced as
+`diff` with a fence longer than any backtick run inside it. It only
+reads the patch files, never applies or runs them. Bodies stay under
+GitHub's 65,536-character comment limit: the patches split across
+`<id>-1.md`, `<id>-2.md`, … when they do not fit in one, and a single
+patch too large for a comment is refused rather than cut. The patch headers
+explain the vulnerability, so the comment belongs on the private tracker
+only, never in a public PR.
 
 **Commit-message credit.** The mail asks fixers to cite the finding id
 (*"Discovered by Anthropic's OSS Scanner, as vulnerability ANT-…"*) in
@@ -173,6 +187,7 @@ Every adapter normalises a finding to:
 | `attacker_capability` | what the attacker must already have — the load-bearing input for the trust-boundary mapping |
 | `impact` | claimed impact |
 | `remediation` | the scanner's proposed fix |
+| `patches` | local paths of candidate patch files, when the scan ships them (optional; `[]` when none) |
 
 The skill treats `severity`/`level` as starting hypotheses and re-derives
 the real disposition from the project's Security Model + precedents.

tools/security-management/src/security_management/issue_import_from_scan/cli.py

@@ -26,7 +26,7 @@
 from pathlib import Path
 from typing import Any
 
-from security_management.cli import emit, read_json, read_object
+from security_management.cli import emit, read_json, read_object, refuse
 from security_management.issue_import_from_scan import oss_scanner, scan
 
 FAMILY = "issue-import-from-scan"
@@ -50,6 +50,13 @@ def add_parsers(sub: Any) -> None:
         "--attachments-dir", help="save the .eml's attachments here (byte-exact; never decrypted or opened)"
     )
 
+    p = commands.add_parser(
+        "render-patch-comment", help="Step F: a tracker comment quoting a finding's candidate patches"
+    )
+    p.add_argument("--index", required=True, help="the scan folder's index.json (parsed with --archive-dir)")
+    p.add_argument("--finding", required=True, help="the finding id")
+    p.add_argument("--out-dir", required=True, help="write the comment body files here")
+
     b = commands.add_parser("bucket-action", help="Step C: what a disposition bucket allows")
     b.add_argument("--bucket", required=True, help=f"one of {', '.join(scan.BUCKETS)}")
 
@@ -100,6 +107,15 @@ def dispatch(args: argparse.Namespace) -> int:
                 Path(args.source), Path(args.attachments_dir)
             )
         emit(summary)
+    elif command == "render-patch-comment":
+        index = read_object(args.index)
+        found = [f for f in index.get("findings") or [] if f.get("id") == args.finding]
+        if not found:
+            return refuse(f"no finding {args.finding!r} in {args.index}")
+        if "patches" not in found[0]:
+            return refuse("the scan folder was parsed without --archive-dir, so no patches are linked")
+        bodies = oss_scanner.write_patch_comments(found[0], index.get("scan") or {}, Path(args.out_dir))
+        emit({"finding": args.finding, "bodies": bodies})
     elif command == "bucket-action":
         emit(scan.bucket_action(args.bucket))
     elif command == "render-report":