feat(reports): add SARIF manifest artifacts

This commit is contained in:
Marius Ciepluch
2026-06-18 10:16:25 +02:00
parent 1e6c1f4436
commit cad70721cb
4 changed files with 296 additions and 16 deletions
+29 -7
View File
@@ -11,7 +11,7 @@ from .index import build_code_chunks
from .mapping import analyze_entry_points, attach_reachability_hints, ingest_sarif, write_entry_points, write_static_hints
from .preprocess import preprocess_repository, write_preprocess_artifact
from .rank import rank_targets, write_rankings
from .reports import write_markdown_report
from .reports import scan_exit_code, write_manifest, write_markdown_report, write_sarif_report
from .run import create_scan_run
from .verification import verify_findings, write_verification_results
@@ -24,6 +24,7 @@ def main(argv: list[str] | None = None) -> int:
scan.add_argument("path", type=Path)
scan.add_argument("--mode", choices=("quick", "standard", "deep"), default="quick")
scan.add_argument("--config", type=Path, default=Path("openultrasast.toml"))
scan.add_argument("--fail-on", choices=("never", "findings", "verified"), default="never")
index = subparsers.add_parser("index", help="chunk a local repository for embedding index construction")
index.add_argument("path", type=Path)
@@ -32,13 +33,13 @@ def main(argv: list[str] | None = None) -> int:
args = parser.parse_args(argv)
if args.command == "scan":
return _scan(args.path, args.config, args.mode)
return _scan(args.path, args.config, args.mode, args.fail_on)
if args.command == "index":
return _index(args.path, args.config, args.chunk_lines)
return 2
def _scan(path: Path, config_path: Path, mode: str) -> int:
def _scan(path: Path, config_path: Path, mode: str, fail_on: str) -> int:
if mode != "quick":
raise SystemExit(f"mode {mode!r} is specified but only quick preprocess artifacts are implemented")
if not path.exists() or not path.is_dir():
@@ -66,17 +67,38 @@ def _scan(path: Path, config_path: Path, mode: str) -> int:
rankings = runtime.run_stage("rank", lambda: rank_targets(targets))
write_rankings(rankings, run.root / "rank" / "ranking.json")
findings = runtime.run_stage("quick_findings", lambda: quick_scan_findings(run.target, targets, rankings))
write_findings(findings, run.root / "findings.json")
findings_path = run.root / "findings.json"
verification_path = run.root / "verification.json"
markdown_path = run.root / "report.md"
sarif_path = run.root / "report.sarif"
manifest_path = run.root / "manifest.json"
write_findings(findings, findings_path)
verifications = runtime.run_stage("verify", lambda: verify_findings(findings))
write_verification_results(verifications, run.root / "verification.json")
runtime.run_stage("report", lambda: write_markdown_report(findings, run.root / "report.md"))
write_verification_results(verifications, verification_path)
runtime.run_stage("report", lambda: write_markdown_report(findings, markdown_path, verifications))
runtime.run_stage("sarif", lambda: write_sarif_report(findings, verifications, sarif_path))
runtime.run_stage(
"manifest",
lambda: write_manifest(
run=run,
findings=findings,
verifications=verifications,
artifact_paths={
"findings": findings_path,
"verification": verification_path,
"markdown": markdown_path,
"sarif": sarif_path,
},
path=manifest_path,
),
)
runtime.finish(status="succeeded")
print(f"scan_id={run.scan_id}")
print(f"run_dir={run.root}")
print(f"file_targets={len(targets)}")
print(f"ranked_targets={len(rankings)}")
print(f"findings={len(findings)}")
return 0
return scan_exit_code(findings, verifications, fail_on)
def _load_static_hints(sarif_paths: tuple[str, ...]) -> list[object]:
+144 -1
View File
@@ -1,16 +1,22 @@
from __future__ import annotations
import json
from dataclasses import asdict
from pathlib import Path
from .findings import StaticFinding
from .run import ScanRun
from .verification import VerificationResult
def write_markdown_report(findings: list[StaticFinding], path: Path) -> None:
def write_markdown_report(findings: list[StaticFinding], path: Path, verifications: list[VerificationResult] | None = None) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
verification_by_id = _verification_by_id(verifications or [])
lines = ["# OpenUltraSAST Report", "", f"Findings: {len(findings)}", ""]
if not findings:
lines.append("No quick-mode findings were emitted.")
for finding in findings:
verification = verification_by_id.get(finding.finding_id)
lines.extend(
[
f"## {finding.title}",
@@ -22,6 +28,7 @@ def write_markdown_report(findings: list[StaticFinding], path: Path) -> None:
f"- Severity: `{finding.severity}`",
f"- Confidence: `{finding.confidence}`",
f"- Evidence level: `{finding.evidence_level}`",
f"- Verification: `{verification.status if verification else 'not_run'}`",
f"- Reachability: `{finding.reachability_status}`",
f"- Conditions: `{', '.join(finding.reachability_conditions) or 'none'}`",
f"- Ranking priority: `{finding.ranking_priority}`",
@@ -32,3 +39,139 @@ def write_markdown_report(findings: list[StaticFinding], path: Path) -> None:
]
)
path.write_text("\n".join(lines).rstrip() + "\n")
def write_sarif_report(findings: list[StaticFinding], verifications: list[VerificationResult], path: Path) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
verification_by_id = _verification_by_id(verifications)
rules = _sarif_rules(findings)
payload = {
"version": "2.1.0",
"$schema": "https://json.schemastore.org/sarif-2.1.0.json",
"runs": [
{
"tool": {
"driver": {
"name": "OpenUltraSAST",
"informationUri": "https://github.com/norandom/OpenUltraSAST",
"rules": list(rules.values()),
}
},
"results": [_sarif_result(finding, verification_by_id.get(finding.finding_id)) for finding in findings],
}
],
}
path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
def write_manifest(
*,
run: ScanRun,
findings: list[StaticFinding],
verifications: list[VerificationResult],
artifact_paths: dict[str, Path],
path: Path,
) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
verification_by_id = _verification_by_id(verifications)
payload = {
"scan_id": run.scan_id,
"target": str(run.target),
"artifacts": {name: _relative_artifact(run.root, artifact) for name, artifact in sorted(artifact_paths.items())},
"findings": [
{
"finding_id": finding.finding_id,
"path": finding.path,
"line": finding.line,
"severity": finding.severity,
"evidence_level": finding.evidence_level,
"verification_status": verification_by_id[finding.finding_id].status
if finding.finding_id in verification_by_id
else "not_run",
"artifact_refs": {
"findings_json": _relative_artifact(run.root, artifact_paths["findings"]),
"verification_json": _relative_artifact(run.root, artifact_paths["verification"]),
"markdown": _relative_artifact(run.root, artifact_paths["markdown"]),
"sarif": _relative_artifact(run.root, artifact_paths["sarif"]),
},
}
for finding in findings
],
}
path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
def scan_exit_code(findings: list[StaticFinding], verifications: list[VerificationResult], policy: str) -> int:
if policy == "never":
return 0
if policy == "findings":
return 1 if findings else 0
if policy == "verified":
return 1 if any(result.verified for result in verifications) else 0
raise ValueError(f"unknown fail policy: {policy}")
def _sarif_rules(findings: list[StaticFinding]) -> dict[str, dict[str, object]]:
rules: dict[str, dict[str, object]] = {}
for finding in findings:
rule_id = _rule_id(finding)
rules.setdefault(
rule_id,
{
"id": rule_id,
"name": finding.title,
"shortDescription": {"text": finding.title},
"properties": {"tags": finding.tags},
},
)
return rules
def _sarif_result(finding: StaticFinding, verification: VerificationResult | None) -> dict[str, object]:
return {
"ruleId": _rule_id(finding),
"level": _sarif_level(finding.severity),
"message": {"text": finding.rationale},
"locations": [
{
"physicalLocation": {
"artifactLocation": {"uri": finding.path},
"region": {"startLine": finding.line or 1},
}
}
],
"partialFingerprints": {"openultrasastFindingId": finding.finding_id},
"properties": {
"finding_id": finding.finding_id,
"severity": finding.severity,
"confidence": finding.confidence,
"evidence_level": finding.evidence_level,
"verification": asdict(verification) if verification else None,
"reachability_status": finding.reachability_status,
"reachability_conditions": finding.reachability_conditions,
"ranking_priority": finding.ranking_priority,
},
}
def _rule_id(finding: StaticFinding) -> str:
return finding.finding_id.split(":", 1)[0]
def _sarif_level(severity: str) -> str:
if severity in {"critical", "high"}:
return "error"
if severity == "medium":
return "warning"
return "note"
def _verification_by_id(verifications: list[VerificationResult]) -> dict[str, VerificationResult]:
return {result.finding_id: result for result in verifications}
def _relative_artifact(root: Path, path: Path) -> str:
try:
return path.relative_to(root).as_posix()
except ValueError:
return path.as_posix()