From 624024e72a2b3fac4a3041b41a201843067920b5 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sat, 5 Sep 2026 15:35:08 +0200 Subject: [PATCH] feat(corpus-seeded-mechanisms): reports name the mechanism, its known fix guard and pair provenance Task 5.1 (Req 6.1, 6.3). - reports.py: the markdown and SARIF writers take the mechanisms a run cites; a finding cites one through its mechanism: tag (variant) or its merged overlay record's mechanism_id; markdown shows Mechanism, Known fix guard and Learned from pairs per finding plus a Mechanisms section, SARIF properties carry mechanism_id/summary/guard/pairs; variant findings keep suspicion, merged inventory findings keep their own evidence level; reports without mechanisms are unchanged. - cli: _search_variants returns the cited mechanisms from the store and the scan passes them to both writers. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG --- src/openultrasast/cli.py | 27 ++++-- src/openultrasast/reports.py | 60 +++++++++++++- tests/test_mechanism_reports.py | 141 ++++++++++++++++++++++++++++++++ tests/test_variant_scan.py | 6 ++ 4 files changed, 225 insertions(+), 9 deletions(-) create mode 100644 tests/test_mechanism_reports.py diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py index 5d3e46f..fdcbb7f 100644 --- a/src/openultrasast/cli.py +++ b/src/openultrasast/cli.py @@ -405,6 +405,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut wrote_overlay = False variants_payload: dict[str, object] | None = None variant_findings: list[StaticFinding] = [] + mechanisms_cited: dict[str, dict[str, object]] = {} if Stage.MAP in plan.requested: built_map = runtime.run_stage( "map", @@ -438,7 +439,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut findings = findings + extra write_findings(findings, findings_path) if config.variants.enabled: - variant_findings, overlay_records, variants_payload = _search_variants( + variant_findings, overlay_records, variants_payload, mechanisms_cited = _search_variants( root=run.target, targets=targets, overlay_records=overlay_records, @@ -545,6 +546,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut complexity_map=built_map, verdicts=verdict_records if wrote_verdicts else None, overlay=overlay_records if wrote_overlay else None, + mechanisms=mechanisms_cited or None, ), ) runtime.run_stage( @@ -554,6 +556,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut verifications, sarif_path, overlay=overlay_records if wrote_overlay else None, + mechanisms=mechanisms_cited or None, ), ) runtime.run_stage( @@ -865,19 +868,23 @@ def _search_variants( facts: SemanticFacts | FactLoadError, max_mechanisms: int, runtime: HarnessRuntime, -) -> tuple[list[StaticFinding], list[OverlayRecord], dict[str, object]]: - """Third MAP proposer (corpus-seeded-mechanisms Req 3): shapes from the mechanism store, merged with overlay flows.""" +) -> tuple[list[StaticFinding], list[OverlayRecord], dict[str, object], dict[str, dict[str, object]]]: + """Third MAP proposer (corpus-seeded-mechanisms Req 3): shapes from the mechanism store, merged with overlay flows. + + Also returns the mechanisms the findings cite (summary, guard, pairs, cwe) for the reports (Req 6.1). + """ from .semantic.mechanisms import corpus_mechanisms from .semantic.variant_search import hits_to_findings, search_tree store = MechanismStore(root / CALIBRATION_DIR / "mechanisms.jsonl") empty: dict[str, object] = {"mechanisms_searched": 0, "files_searched": 0, "findings": 0, "merged_into_overlay": 0} if isinstance(facts, FactLoadError): - return [], overlay_records, empty + return [], overlay_records, empty, {} result = runtime.run_stage("variants", lambda: search_tree(root, targets, store, facts, max_mechanisms=max_mechanisms)) for degradation in result.degradations: runtime.state["degradations"].append(degradation) - summaries = {record.id: (record.summary, record.pairs, record.cwe) for record in corpus_mechanisms(store.load())} + records_by_id = {record.id: record for record in corpus_mechanisms(store.load())} + summaries = {record.id: (record.summary, record.pairs, record.cwe) for record in records_by_id.values()} findings, records = hits_to_findings(result.hits, overlay_records, summaries) merged = sum(1 for before, after in zip(overlay_records, records, strict=True) if before.mechanism_id is None and after.mechanism_id) payload: dict[str, object] = { @@ -886,7 +893,15 @@ def _search_variants( "findings": len(findings), "merged_into_overlay": merged, } - return findings, records, payload + cited_ids = {tag.split(":", 1)[1] for finding in findings for tag in finding.tags if tag.startswith("mechanism:")} | { + str(record.mechanism_id) for record in records if record.mechanism_id + } + cited: dict[str, dict[str, object]] = { + mechanism_id: {"summary": record.summary, "guard": record.guard, "pairs": list(record.pairs), "cwe": record.cwe} + for mechanism_id, record in records_by_id.items() + if mechanism_id in cited_ids + } + return findings, records, payload, cited def _hotspot_from_variant(finding: StaticFinding) -> Hotspot: diff --git a/src/openultrasast/reports.py b/src/openultrasast/reports.py index 184414f..7af33d8 100644 --- a/src/openultrasast/reports.py +++ b/src/openultrasast/reports.py @@ -1,7 +1,7 @@ from __future__ import annotations import json -from collections.abc import Sequence +from collections.abc import Mapping, Sequence from dataclasses import asdict from pathlib import Path @@ -19,10 +19,12 @@ def write_markdown_report( complexity_map: object | None = None, verdicts: Sequence[object] | None = None, overlay: Sequence[object] | None = None, + mechanisms: Mapping[str, Mapping[str, object]] | None = None, ) -> None: path.parent.mkdir(parents=True, exist_ok=True) verification_by_id = _verification_by_id(verifications or []) overlay_by_id = _overlay_by_id(overlay or []) + cited: dict[str, Mapping[str, object]] = {} lines = ["# OpenUltraSAST Report", "", f"Findings: {len(findings)}", "", "## Inventory", ""] if not findings: lines.append("No quick-mode findings were emitted.") @@ -51,7 +53,16 @@ def write_markdown_report( if disposition is not None: lines.append(f"- Disposition: `{getattr(disposition, 'disposition', '')}`") lines.append(f"- Overlay reason: `{getattr(disposition, 'reason', '')}`") + mechanism_id = _mechanism_id_for(finding, disposition) + if mechanism_id and mechanisms is not None and mechanism_id in mechanisms: + info = mechanisms[mechanism_id] + cited[mechanism_id] = info + lines.append(f"- Mechanism: `{mechanism_id}` — {info.get('summary', '')}") + lines.append(f"- Known fix guard: `{info.get('guard', 'none')}`") + lines.append(f"- Learned from pairs: `{', '.join(_pairs_of(info))}`") lines.extend(["", finding.rationale, ""]) + if cited: + _append_mechanisms(lines, cited) if overlay: _append_overlay(lines, overlay) if complexity_map is not None: @@ -133,6 +144,7 @@ def write_sarif_report( verifications: list[VerificationResult], path: Path, overlay: Sequence[object] | None = None, + mechanisms: Mapping[str, Mapping[str, object]] | None = None, ) -> None: path.parent.mkdir(parents=True, exist_ok=True) verification_by_id = _verification_by_id(verifications) @@ -151,7 +163,9 @@ def write_sarif_report( } }, "results": [ - _sarif_result(finding, verification_by_id.get(finding.finding_id), overlay_by_id.get(finding.finding_id)) + _sarif_result( + finding, verification_by_id.get(finding.finding_id), overlay_by_id.get(finding.finding_id), mechanisms=mechanisms + ) for finding in findings ], } @@ -249,7 +263,13 @@ def _sarif_rules(findings: list[StaticFinding]) -> dict[str, dict[str, object]]: return rules -def _sarif_result(finding: StaticFinding, verification: VerificationResult | None, overlay: object | None = None) -> dict[str, object]: +def _sarif_result( + finding: StaticFinding, + verification: VerificationResult | None, + overlay: object | None = None, + *, + mechanisms: Mapping[str, Mapping[str, object]] | None = None, +) -> dict[str, object]: properties: dict[str, object] = { "finding_id": finding.finding_id, "severity": finding.severity, @@ -263,6 +283,14 @@ def _sarif_result(finding: StaticFinding, verification: VerificationResult | Non if overlay is not None: properties["disposition"] = getattr(overlay, "disposition", "") properties["overlay_reason"] = getattr(overlay, "reason", "") + mechanism_id = _mechanism_id_for(finding, overlay) + if mechanism_id: + properties["mechanism_id"] = mechanism_id + info = (mechanisms or {}).get(mechanism_id) + if info is not None: + properties["mechanism_summary"] = str(info.get("summary", "")) + properties["mechanism_guard"] = str(info.get("guard", "none")) + properties["mechanism_pairs"] = _pairs_of(info) return { "ruleId": _rule_id(finding), "level": _sarif_level(finding.severity), @@ -354,3 +382,29 @@ def _relative_artifact(root: Path, path: Path) -> str: return path.relative_to(root).as_posix() except ValueError: return path.as_posix() + + +def _mechanism_id_for(finding: StaticFinding, overlay: object | None) -> str | None: + """Mechanism cited by a finding: a `mechanism:` tag on a variant finding, or the merged overlay record's id.""" + for tag in finding.tags: + if tag.startswith("mechanism:"): + return tag.split(":", 1)[1] + merged = getattr(overlay, "mechanism_id", None) + return str(merged) if merged else None + + +def _append_mechanisms(lines: list[str], cited: Mapping[str, Mapping[str, object]]) -> None: + lines.extend( + ["## Mechanisms", "", "Known mechanisms this report cites (corpus-seeded; variant findings stay suspicions until verified).", ""] + ) + for mechanism_id, info in sorted(cited.items()): + pairs = ", ".join(_pairs_of(info)) or "unknown" + lines.append( + f"- `{mechanism_id}`: {info.get('summary', '')}; known fix guard `{info.get('guard', 'none')}`; learned from `{pairs}`" + ) + lines.append("") + + +def _pairs_of(info: Mapping[str, object]) -> list[str]: + pairs = info.get("pairs") + return [str(p) for p in pairs] if isinstance(pairs, list | tuple) else [] diff --git a/tests/test_mechanism_reports.py b/tests/test_mechanism_reports.py new file mode 100644 index 0000000..c19b702 --- /dev/null +++ b/tests/test_mechanism_reports.py @@ -0,0 +1,141 @@ +"""corpus-seeded-mechanisms task 5.1: reports name the mechanism, the known fix guard, and the pair provenance.""" + +from __future__ import annotations + +import json +from pathlib import Path + +from openultrasast.findings import StaticFinding +from openultrasast.reports import write_markdown_report, write_sarif_report +from openultrasast.semantic.overlay import OverlayRecord + + +def _variant() -> StaticFinding: + return StaticFinding( + finding_id="variant:corpus:abc:app.py:5", + path="app.py", + title="Variant of known mechanism: run_shell", + severity="medium", + confidence="low", + evidence_level="suspicion", + rationale="parameter into run_shell; fix adds allowlist_test. Structural variant of a mechanism learned from p1, p2.", + line=5, + function_name="handler", + reachability_status="unknown", + reachability_evidence=[], + reachability_conditions=[], + tags=["variant", "mechanism:corpus:abc"], + ranking_priority=0.0, + ) + + +MECHANISMS = { + "corpus:abc": { + "summary": "source_reaches_sink: parameter into run_shell (CWE-78); fix adds allowlist_test", + "guard": "allowlist_test", + "pairs": ["p1", "p2"], + "cwe": "CWE-78", + } +} + + +def test_markdown_names_mechanism_guard_and_provenance_and_keeps_the_suspicion_label(tmp_path: Path) -> None: + path = tmp_path / "report.md" + write_markdown_report([_variant()], path, mechanisms=MECHANISMS) + text = path.read_text() + assert "- Mechanism: `corpus:abc`" in text + assert "source_reaches_sink: parameter into run_shell" in text + assert "- Known fix guard: `allowlist_test`" in text + assert "- Learned from pairs: `p1, p2`" in text + assert "- Evidence level: `suspicion`" in text + assert "## Mechanisms" in text # a section summarising every mechanism the report cites + + +def test_sarif_carries_mechanism_properties_for_variant_and_merged_overlay_findings(tmp_path: Path) -> None: + merged = OverlayRecord( + proposal_id="python-os-command:app.py:9", + path="app.py", + line=9, + disposition="promote", + reason="request reaches os.system", + cwe="CWE-78", + sources=("request",), + sinks=("os.system",), + sanitizers=(), + evidence_level="static_corroboration", + origin="inventory", + engine="python-ast", + language="python", + mechanism_id="corpus:abc", + ) + inventory = StaticFinding( + **{ + **_variant().__dict__, + "finding_id": "python-os-command:app.py:9", + "line": 9, + "tags": ["injection"], + "evidence_level": "static_corroboration", + } + ) + path = tmp_path / "report.sarif" + write_sarif_report([_variant(), inventory], [], path, overlay=[merged], mechanisms=MECHANISMS) + results = json.loads(path.read_text())["runs"][0]["results"] + by_id = {item["properties"]["finding_id"]: item["properties"] for item in results} + variant = by_id["variant:corpus:abc:app.py:5"] + assert ( + variant["mechanism_id"] == "corpus:abc" + and variant["mechanism_guard"] == "allowlist_test" + and variant["mechanism_pairs"] == ["p1", "p2"] + ) + assert variant["evidence_level"] == "suspicion" + merged_props = by_id["python-os-command:app.py:9"] + assert merged_props["mechanism_id"] == "corpus:abc" and merged_props["mechanism_guard"] == "allowlist_test" # from the overlay record + + +def test_reports_without_mechanisms_are_unchanged(tmp_path: Path) -> None: + plain = StaticFinding(**{**_variant().__dict__, "finding_id": "python-os-command:app.py:9", "tags": ["injection"]}) + path = tmp_path / "report.md" + write_markdown_report([plain], path) + text = path.read_text() + assert "Mechanism:" not in text and "## Mechanisms" not in text + + +def test_standard_scan_report_names_the_mechanism_behind_a_variant_finding(tmp_path: Path, monkeypatch) -> None: # type: ignore[no-untyped-def] + from openultrasast.cli import main + from openultrasast.semantic.mechanisms import MechanismStore, append_from_pair + from openultrasast.semantic.variants import Shape + + repo = tmp_path / "repo" + repo.mkdir() + (repo / "app.py").write_text("from shelltools import run_shell\n\n\ndef handler(q):\n run_shell(q)\n") + shape = Shape( + language="python", + sink_name="run_shell", + arity=1, + source_positions=(0,), + source_kinds=("parameter",), + guard="allowlist_test", + mechanism="source_reaches_sink", + ) + record = append_from_pair( + MechanismStore(repo / ".openultrasast" / "calibration" / "mechanisms.jsonl"), + shape, + summary="parameter into run_shell", + cwe="CWE-78", + pair="p1", + provenance="human", + tier="seeded", + ) + monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs") + monkeypatch.setenv("OPENULTRASAST_SANDBOX_PROBE", "0") + assert main(["scan", str(repo), "--mode", "standard"]) == 0 + run_dir = sorted((repo / ".runs").iterdir())[-1] + report = (run_dir / "report.md").read_text() + assert ( + f"- Mechanism: `{record.id}`" in report + and "- Known fix guard: `allowlist_test`" in report + and "- Learned from pairs: `p1`" in report + ) + sarif = json.loads((run_dir / "report.sarif").read_text()) + props = [item["properties"] for item in sarif["runs"][0]["results"] if item["properties"]["finding_id"].startswith("variant:")] + assert props and props[0]["mechanism_guard"] == "allowlist_test" and props[0]["evidence_level"] == "suspicion" diff --git a/tests/test_variant_scan.py b/tests/test_variant_scan.py index c6b3f3d..76f7963 100644 --- a/tests/test_variant_scan.py +++ b/tests/test_variant_scan.py @@ -90,6 +90,12 @@ def test_variant_hit_merges_into_an_overlay_flow_at_the_same_call_site(tmp_path: assert not any(f["finding_id"].startswith("variant:") for f in findings) # merged, not duplicated manifest = json.loads((run_dir / "manifest.json").read_text()) assert manifest["variants"]["merged_into_overlay"] == 1 and manifest["variants"]["findings"] == 0 + # Req 6.1 end to end: the merged inventory finding names the mechanism in both reports and keeps its own evidence level + report = (run_dir / "report.md").read_text() + assert f"- Mechanism: `{record.id}`" in report and "- Learned from pairs: `p`" in report + sarif = json.loads((run_dir / "report.sarif").read_text()) + props = [item["properties"] for item in sarif["runs"][0]["results"] if item["properties"].get("mechanism_id") == record.id] + assert props and props[0]["mechanism_guard"] == "none" and props[0]["evidence_level"] != "suspicion" def test_quick_scan_never_searches_variants(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: