Open_Ultra_SAST/src/openultrasast/reports.py

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"- Rung: `{finding.rung}`",
                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,
                "rung": finding.rung,
                "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,
        "rung": finding.rung,
        "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 []