diff --git a/plugins/magpie-security-tools/tools/security-management/src/security_management/issue_import_from_scan/cli.py b/plugins/magpie-security-tools/tools/security-management/src/security_management/issue_import_from_scan/cli.py index 889a70e72..3f4f0d768 100644 --- a/plugins/magpie-security-tools/tools/security-management/src/security_management/issue_import_from_scan/cli.py +++ b/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": diff --git a/plugins/magpie-security-tools/tools/security-management/src/security_management/issue_import_from_scan/oss_scanner.py b/plugins/magpie-security-tools/tools/security-management/src/security_management/issue_import_from_scan/oss_scanner.py index e3767366c..de7d4601d 100644 --- a/plugins/magpie-security-tools/tools/security-management/src/security_management/issue_import_from_scan/oss_scanner.py +++ b/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 = "" + + +def _patch_summary(patch: str, n: int) -> str: + """The patch's one-line header (its second `#` line), safe inside ``.""" + lines = patch.splitlines() + text = lines[1].lstrip("#").strip() if len(lines) > 1 and lines[1].startswith("#") else "" + text = text.replace("&", "&").replace("<", "<").replace(">", ">") + # A bare `@task` would notify a GitHub user of that name. + text = re.sub(r"(? str: + longest = max((len(run) for run in re.findall(r"`+", patch)), default=0) + fence = "`" * max(5, longest + 1) + return "\n".join( + [ + f"
{_patch_summary(patch, n)}", + "", + f"{fence}diff", + patch.rstrip("\n"), + fence, + "", + "
", + "", + ] + ) + + +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 `
`. + + 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 `.md` (or `-.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/.md`; return the summary to print (no evidence text).""" evidence_dir = out_dir / "evidence" diff --git a/plugins/magpie-security/skills/issue-import-from-scan/SKILL.md b/plugins/magpie-security/skills/issue-import-from-scan/SKILL.md index 30251c82f..d6d751e17 100644 --- a/plugins/magpie-security/skills/issue-import-from-scan/SKILL.md +++ b/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 --- " + + +def _patch_summary(patch: str, n: int) -> str: + """The patch's one-line header (its second `#` line), safe inside ``.""" + lines = patch.splitlines() + text = lines[1].lstrip("#").strip() if len(lines) > 1 and lines[1].startswith("#") else "" + text = text.replace("&", "&").replace("<", "<").replace(">", ">") + # A bare `@task` would notify a GitHub user of that name. + text = re.sub(r"(? str: + longest = max((len(run) for run in re.findall(r"`+", patch)), default=0) + fence = "`" * max(5, longest + 1) + return "\n".join( + [ + f"
{_patch_summary(patch, n)}", + "", + f"{fence}diff", + patch.rstrip("\n"), + fence, + "", + "
", + "", + ] + ) + + +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 `
`. + + 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 `.md` (or `-.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/.md`; return the summary to print (no evidence text).""" evidence_dir = out_dir / "evidence" diff --git a/tools/security-management/tests/test_issue_import_from_scan_oss_scanner.py b/tools/security-management/tests/test_issue_import_from_scan_oss_scanner.py index 7cb0f361d..3a182880a 100644 --- a/tools/security-management/tests/test_issue_import_from_scan_oss_scanner.py +++ b/tools/security-management/tests/test_issue_import_from_scan_oss_scanner.py @@ -443,3 +443,116 @@ def test_flowed_quotes_join_only_at_the_same_depth() -> None: text = "> pass \n> word\n>> deeper \n> shallower" assert oss_scanner._unflow(text, delsp=True) == "> password\n>> deeper\n> shallower" assert oss_scanner._unflow("plain \nline") == "plain line" + + +def patch(summary: str, extra: str = "") -> str: + return f"# src/widget/auth.py\n# {summary}\n#\n# why it works\n{extra}diff --git a/x b/x\n+fixed\n" + + +def archive_with_patches(tmp_path: Path, patches: dict[str, str]) -> Path: + archive = tmp_path / "archive" + folder = archive / "high" / "ANT-2026-AAAAAAA1" / "patches" + folder.mkdir(parents=True) + for name, text in patches.items(): + (folder / name).write_text(text) + return archive + + +def parse_with_archive(capsys, tmp_path: Path, archive: Path) -> Path: + run( + capsys, + "parse-email", + "--source", + str(eml(tmp_path)), + "--out-dir", + str(tmp_path / "s"), + "--archive-dir", + str(archive), + ) + return tmp_path / "s" / "index.json" + + +def render(capsys, tmp_path: Path, index: Path) -> tuple[int, dict[str, Any]]: + return run( + capsys, + "render-patch-comment", + "--index", + str(index), + "--finding", + "ANT-2026-AAAAAAA1", + "--out-dir", + str(tmp_path / "c"), + ) + + +def test_archive_patches_are_linked_in_numeric_order(capsys, tmp_path) -> None: + archive = archive_with_patches(tmp_path, {f"{n}.patch": patch(f"fix {n}") for n in (1, 2, 10)}) + index = json.loads(parse_with_archive(capsys, tmp_path, archive).read_text()) + patches = {f["id"]: f["patches"] for f in index["findings"]} + assert [Path(p).name for p in patches["ANT-2026-AAAAAAA1"]] == ["1.patch", "2.patch", "10.patch"] + assert patches["ANT-2026-BBBBBBB2"] == [] + + +def test_patch_comment_folds_each_candidate(capsys, tmp_path) -> None: + archive = archive_with_patches( + tmp_path, + {"1.patch": patch("Stop rendering @task.bash "), "2.patch": patch("Smaller variant")}, + ) + code, out = render(capsys, tmp_path, parse_with_archive(capsys, tmp_path, archive)) + assert code == 0 + (entry,) = out["bodies"] + body = Path(entry["path"]).read_text() + assert body.startswith( + "\n" + "**Candidate patches from the security scan — `ANT-2026-AAAAAAA1`**\n" + ) + assert "against `gh:example/widget@main` at `0123456789ab`" in body + assert "untested, machine-generated" in body and "public PR" in body + assert "
Candidate 1 — Stop rendering `@task.bash` <args>" in body + assert "
Candidate 2 — Smaller variant" in body + assert body.count("`````diff\n") == 2 and body.count("
") == 2 + assert "
None: + path = tmp_path / "1.patch" + path.write_text(patch("Fix")) + finding = {"id": "ANT-2026-AAAAAAA1", "patches": [str(path)]} + scan = {"target": "x` @someone [link](https://example.com) `", "commit": "0123456789ab"} + (body,) = oss_scanner.render_patch_comments(finding, scan) + assert "against `0123456789ab`" in body + assert "@someone" not in body and "example.com" not in body + + +def test_patch_fence_outgrows_backticks_in_the_patch() -> None: + body = oss_scanner._patch_block(patch("Uses fences", "# ``````python\n"), 1) + assert "```````diff\n" in body + + +def test_no_patches_means_no_comment() -> None: + assert oss_scanner.render_patch_comments({"id": "ANT-2026-AAAAAAA1", "patches": []}, {}) == [] + + +def test_patch_comment_refuses_without_archive(capsys, tmp_path) -> None: + run(capsys, "parse-email", "--source", str(eml(tmp_path)), "--out-dir", str(tmp_path / "s")) + code, out = render(capsys, tmp_path, tmp_path / "s" / "index.json") + assert code == 2 and "--archive-dir" in out["refused"] + + +def test_large_patches_split_across_comments(tmp_path) -> None: + paths = [] + for n in (1, 2, 3): + path = tmp_path / f"{n}.patch" + path.write_text(patch(f"fix {n}", "+x\n" * 300)) + paths.append(str(path)) + finding = {"id": "ANT-2026-AAAAAAA1", "patches": paths} + bodies = oss_scanner.render_patch_comments(finding, {}, limit=2500) + assert len(bodies) > 1 + assert all(len(b) <= 2500 for b in bodies) + assert bodies[0].startswith( + "\n" + f"**Candidate patches from the security scan — `ANT-2026-AAAAAAA1`** (part 1 of {len(bodies)})" + ) + assert "".join(bodies).count("
") == 3 + with pytest.raises(ValueError, match="candidate 1"): + oss_scanner.render_patch_comments(finding, {}, limit=500)