mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 03:49:26 +00:00
Marius CiepluchandClaude Opus 5
eab89bf203
Task 3.2. Each obligation finding prints its operation, the missing discharger with the provenance that failed, the evidence behind the label, the known fix learned from the corpus and the hunter's intent verdict; the Obligations section lists the sibling sets evaluated and the declared policy version. SARIF carries the same as properties. Obligation findings rank by resource sensitivity times label weight, declared above consistency above function-local, and stay strictly below every proven finding. Proof rungs live on verdicts rather than on static findings, so the cap takes the proven ids from the run's triggerable verdicts. Co-Authored-By: Claude Opus 5 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG
479 lines
20 KiB
Python
479 lines
20 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from collections.abc import Mapping, Sequence
|
|
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,
|
|
verifications: list[VerificationResult] | None = None,
|
|
*,
|
|
redact: bool = True,
|
|
complexity_map: object | None = None,
|
|
verdicts: Sequence[object] | None = None,
|
|
overlay: Sequence[object] | None = None,
|
|
mechanisms: Mapping[str, Mapping[str, object]] | None = None,
|
|
obligations: Mapping[str, Mapping[str, object]] | None = None,
|
|
obligations_summary: 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]] = {}
|
|
cited_obligations: dict[str, Mapping[str, object]] = {}
|
|
lines = ["# OpenUltraSAST Report", "", f"Findings: {len(findings)}", "", "## Inventory", ""]
|
|
if not findings:
|
|
lines.append("No quick-mode findings were emitted.")
|
|
lines.append("")
|
|
for finding in findings:
|
|
verification = verification_by_id.get(finding.finding_id)
|
|
disposition = overlay_by_id.get(finding.finding_id)
|
|
lines.extend(
|
|
[
|
|
f"## {finding.title}",
|
|
"",
|
|
f"- ID: `{finding.finding_id}`",
|
|
f"- Path: `{finding.path}`",
|
|
f"- Line: `{finding.line}`",
|
|
f"- Function: `{finding.function_name or 'unknown'}`",
|
|
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}`",
|
|
f"- Tags: `{', '.join(finding.tags)}`",
|
|
]
|
|
)
|
|
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))}`")
|
|
info_o = (obligations or {}).get(finding.finding_id)
|
|
if info_o is not None:
|
|
cited_obligations[finding.finding_id] = info_o
|
|
_append_obligation_lines(lines, info_o, (mechanisms or {}).get(str(info_o.get("known_fix") or "")))
|
|
lines.extend(["", finding.rationale, ""])
|
|
if cited:
|
|
_append_mechanisms(lines, cited)
|
|
if cited_obligations:
|
|
_append_obligations(lines, cited_obligations, obligations_summary)
|
|
if overlay:
|
|
_append_overlay(lines, overlay)
|
|
if complexity_map is not None:
|
|
_append_complexity_map(lines, complexity_map)
|
|
if verdicts is not None:
|
|
_append_worth_fixing(lines, verdicts)
|
|
report = "\n".join(lines).rstrip() + "\n"
|
|
if redact:
|
|
from .redaction import redact_secrets
|
|
|
|
report = redact_secrets(report)
|
|
path.write_text(report)
|
|
|
|
|
|
def _append_complexity_map(lines: list[str], complexity_map: object) -> None:
|
|
hotspots = tuple(getattr(complexity_map, "hotspots", ()) or ())
|
|
heuristic_only = getattr(complexity_map, "heuristic_only", None)
|
|
summary = f"Hotspots: {len(hotspots)}"
|
|
if heuristic_only is not None:
|
|
summary += f" (heuristic_only={str(bool(heuristic_only)).lower()})"
|
|
lines.extend(["## Complexity map", "", summary, ""])
|
|
if not hotspots:
|
|
lines.extend(["No complexity hotspots were emitted.", ""])
|
|
return
|
|
for hotspot in hotspots:
|
|
identity = _report_identity(getattr(hotspot, "path", ""), getattr(hotspot, "function_name", None))
|
|
lines.extend(
|
|
[
|
|
f"### `{identity}`",
|
|
"",
|
|
f"- Score: `{getattr(hotspot, 'score', '')}`",
|
|
f"- Band: `{getattr(hotspot, 'band', '')}`",
|
|
]
|
|
)
|
|
hint = getattr(hotspot, "test_hint", None)
|
|
if hint is not None:
|
|
lines.append(f"- Gap: `{getattr(hint, 'gap', '')}`")
|
|
kind = getattr(hint, "test_kind", None)
|
|
if kind:
|
|
lines.append(f"- Recommended test: `{kind}`")
|
|
reason = getattr(hint, "reason", "")
|
|
if reason:
|
|
lines.append(f"- Hint: {reason}")
|
|
inventory_ids = tuple(getattr(hotspot, "inventory_finding_ids", ()) or ())
|
|
if inventory_ids:
|
|
lines.append(f"- Inventory: `{', '.join(str(item) for item in inventory_ids)}`")
|
|
rationale = str(getattr(hotspot, "rationale", "") or "")
|
|
lines.extend(["", rationale, ""] if rationale else [""])
|
|
|
|
|
|
def _append_worth_fixing(lines: list[str], verdicts: Sequence[object]) -> None:
|
|
worth = [item for item in verdicts if getattr(item, "worth_fixing", False)]
|
|
lines.extend(["## Worth fixing", "", f"Count: {len(worth)}", ""])
|
|
if not worth:
|
|
lines.extend(["No worth-fixing verdicts were produced.", ""])
|
|
return
|
|
for item in worth:
|
|
identity = _report_identity(getattr(item, "path", ""), getattr(item, "function_name", None))
|
|
lines.extend(
|
|
[
|
|
f"### `{identity}`",
|
|
"",
|
|
f"- Verdict: `{getattr(item, 'verdict', '')}`",
|
|
f"- Reason: `{getattr(item, 'reason', '')}`",
|
|
"",
|
|
]
|
|
)
|
|
|
|
|
|
def _report_identity(path: object, function_name: object) -> str:
|
|
path_text = str(path or "")
|
|
if function_name:
|
|
return f"{path_text}::{function_name}"
|
|
return path_text
|
|
|
|
|
|
def write_sarif_report(
|
|
findings: list[StaticFinding],
|
|
verifications: list[VerificationResult],
|
|
path: Path,
|
|
overlay: Sequence[object] | None = None,
|
|
mechanisms: Mapping[str, Mapping[str, object]] | None = None,
|
|
obligations: Mapping[str, Mapping[str, object]] | None = None,
|
|
) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
verification_by_id = _verification_by_id(verifications)
|
|
overlay_by_id = _overlay_by_id(overlay or [])
|
|
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),
|
|
overlay_by_id.get(finding.finding_id),
|
|
mechanisms=mechanisms,
|
|
obligation=(obligations or {}).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,
|
|
score: dict[str, object] | None = None,
|
|
degradations: list[dict[str, object]] | None = None,
|
|
fusion: list[dict[str, object]] | None = None,
|
|
stages: dict[str, object] | None = None,
|
|
complexity: dict[str, object] | None = None,
|
|
worth_fixing: dict[str, object] | None = None,
|
|
provenance: dict[str, object] | None = None,
|
|
variants: dict[str, object] | None = None,
|
|
obligations: dict[str, object] | None = None,
|
|
) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
verification_by_id = _verification_by_id(verifications)
|
|
payload: dict[str, object] = {
|
|
"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": _artifact_refs(run.root, artifact_paths),
|
|
}
|
|
for finding in findings
|
|
],
|
|
}
|
|
if score is not None:
|
|
payload["score"] = score
|
|
if degradations is not None:
|
|
payload["degradations"] = degradations
|
|
if fusion is not None:
|
|
payload["fusion"] = fusion
|
|
if stages is not None:
|
|
payload["stages"] = stages
|
|
if complexity is not None:
|
|
payload["complexity"] = complexity
|
|
if worth_fixing is not None:
|
|
payload["worth_fixing"] = worth_fixing
|
|
if provenance is not None:
|
|
payload["provenance"] = provenance
|
|
if variants is not None:
|
|
payload["variants"] = variants
|
|
if obligations is not None:
|
|
payload["obligations"] = obligations
|
|
path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
|
|
|
|
|
|
def scan_exit_code(
|
|
findings: list[StaticFinding],
|
|
verifications: list[VerificationResult],
|
|
policy: str,
|
|
worth_fixing_verdicts: Sequence[object] | None = None,
|
|
) -> 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
|
|
if policy == "worth-fixing":
|
|
return 1 if any(getattr(verdict, "worth_fixing", False) for verdict in worth_fixing_verdicts or ()) 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,
|
|
overlay: object | None = None,
|
|
*,
|
|
mechanisms: Mapping[str, Mapping[str, object]] | None = None,
|
|
obligation: Mapping[str, object] | None = None,
|
|
) -> dict[str, object]:
|
|
properties: dict[str, object] = {
|
|
"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,
|
|
}
|
|
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)
|
|
if obligation is not None:
|
|
properties["obligation_kind"] = str(obligation.get("obligation", ""))
|
|
properties["obligation_resource"] = obligation.get("resource")
|
|
properties["obligation_missing"] = str(obligation.get("missing", ""))
|
|
properties["obligation_provenance"] = obligation.get("provenance")
|
|
properties["obligation_label"] = str(obligation.get("label", ""))
|
|
properties["obligation_evidence"] = _strings_of(obligation, "evidence")
|
|
properties["obligation_known_fix"] = obligation.get("known_fix")
|
|
properties["obligation_intent"] = obligation.get("intent")
|
|
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": properties,
|
|
}
|
|
|
|
|
|
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 _overlay_by_id(records: Sequence[object]) -> dict[str, object]:
|
|
mapped: dict[str, object] = {}
|
|
for record in records:
|
|
proposal_id = getattr(record, "proposal_id", None)
|
|
if isinstance(proposal_id, str) and proposal_id:
|
|
mapped[proposal_id] = record
|
|
return mapped
|
|
|
|
|
|
def _append_overlay(lines: list[str], records: Sequence[object]) -> None:
|
|
lines.extend(["## Overlay", "", f"Records: {len(records)}", ""])
|
|
labels = ("promoted", "demoted", "unadjudicated", "coverage")
|
|
counts = {label: 0 for label in labels}
|
|
for record in records:
|
|
disposition = str(getattr(record, "disposition", ""))
|
|
if disposition == "promote":
|
|
counts["promoted"] += 1
|
|
elif disposition == "demote":
|
|
counts["demoted"] += 1
|
|
elif disposition == "unadjudicated":
|
|
counts["unadjudicated"] += 1
|
|
elif disposition == "coverage":
|
|
counts["coverage"] += 1
|
|
lines.append("- Labels: " + ", ".join(f"{label}={counts[label]}" for label in labels))
|
|
lines.append("")
|
|
for record in records:
|
|
disposition = str(getattr(record, "disposition", ""))
|
|
proposal_id = str(getattr(record, "proposal_id", ""))
|
|
reason = str(getattr(record, "reason", ""))
|
|
lines.extend(
|
|
[
|
|
f"### `{disposition}` `{proposal_id}`",
|
|
"",
|
|
f"- Path: `{getattr(record, 'path', '')}`",
|
|
f"- Line: `{getattr(record, 'line', None)}`",
|
|
f"- Reason: `{reason}`",
|
|
"",
|
|
]
|
|
)
|
|
|
|
|
|
def _artifact_refs(root: Path, artifact_paths: dict[str, Path]) -> dict[str, str]:
|
|
refs = {
|
|
"findings_json": _relative_artifact(root, artifact_paths["findings"]),
|
|
"verification_json": _relative_artifact(root, artifact_paths["verification"]),
|
|
"markdown": _relative_artifact(root, artifact_paths["markdown"]),
|
|
"sarif": _relative_artifact(root, artifact_paths["sarif"]),
|
|
}
|
|
if "trajectories" in artifact_paths:
|
|
refs["trajectories_jsonl"] = _relative_artifact(root, artifact_paths["trajectories"])
|
|
return refs
|
|
|
|
|
|
def _relative_artifact(root: Path, path: Path) -> str:
|
|
try:
|
|
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:<id>` 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 []
|
|
|
|
|
|
def _append_obligation_lines(lines: list[str], info: Mapping[str, object], mechanism: Mapping[str, object] | None) -> None:
|
|
resource = info.get("resource") or "a resource"
|
|
provenance = f" (`{info.get('provenance')}`)" if info.get("provenance") else ""
|
|
evidence = ", ".join(_strings_of(info, "evidence")) or "the operation fact alone"
|
|
lines.append(f"- Obligation: `{info.get('obligation')}` on `{resource}`")
|
|
lines.append(f"- Missing discharger: `{info.get('missing')}`{provenance}")
|
|
lines.append(f"- Evidence: `{info.get('label')}` — {evidence}")
|
|
if info.get("known_fix"):
|
|
guard = mechanism.get("guard") if mechanism else info.get("missing")
|
|
pairs = ", ".join(_pairs_of(mechanism)) if mechanism else "the mechanism store"
|
|
lines.append(f"- Known fix: `{guard}` learned from `{pairs}` ({info.get('known_fix')})")
|
|
if info.get("intent"):
|
|
lines.append(f"- Intent: `{info.get('intent')}`")
|
|
|
|
|
|
def _append_obligations(lines: list[str], cited: Mapping[str, Mapping[str, object]], summary: Mapping[str, object] | None = None) -> None:
|
|
lines.extend(
|
|
["## Obligations", "", "Operations reached without a dominating discharger; suspicions labeled by why the obligation exists.", ""]
|
|
)
|
|
if summary is not None:
|
|
version = summary.get("policy_version")
|
|
lines.append(f"Sibling sets evaluated: {summary.get('sibling_sets', 0)} (under-populated: {summary.get('under_populated', 0)})")
|
|
lines.append(f"Policy version: {f'`{version}`' if version else 'none'}")
|
|
lines.append("")
|
|
sets = summary.get("sets")
|
|
if isinstance(sets, list | tuple):
|
|
for group in sets:
|
|
if isinstance(group, Mapping):
|
|
lines.append(f"- `{group.get('module')}` / `{group.get('resource')}`: {group.get('handlers')} handlers")
|
|
lines.append("")
|
|
for finding_id, info in sorted(cited.items()):
|
|
resource = info.get("resource") or "a resource"
|
|
lines.append(f"- `{finding_id}`: `{info.get('obligation')}` on `{resource}` without `{info.get('missing')}` ({info.get('label')})")
|
|
lines.append("")
|
|
|
|
|
|
def _strings_of(info: Mapping[str, object], key: str) -> list[str]:
|
|
value = info.get(key)
|
|
return [str(item) for item in value] if isinstance(value, list | tuple) else []
|