titan-iac/ci/scripts/semgrep_report.py

#!/usr/bin/env python3
"""Build a compact Semgrep gate report from Semgrep JSON output."""

from __future__ import annotations

import argparse
import json
from pathlib import Path
from typing import Any

BLOCKING_SEVERITIES = {"ERROR"}
SEVERITIES = ("ERROR", "WARNING", "INFO", "UNKNOWN")
SONAR_SEVERITY_BY_SEMGREP = {
    "ERROR": "CRITICAL",
    "WARNING": "MAJOR",
    "INFO": "MINOR",
    "UNKNOWN": "INFO",
}


def _read_json(path: Path) -> dict[str, Any]:
    """Read a JSON object from disk, returning an error-shaped payload on failure."""
    if not path.exists():
        return {"errors": [{"message": f"report missing: {path}"}], "results": []}
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        return {"errors": [{"message": f"invalid JSON: {exc}"}], "results": []}
    if not isinstance(payload, dict):
        return {"errors": [{"message": "report payload is not an object"}], "results": []}
    return payload


def _finding_severity(finding: dict[str, Any]) -> str:
    """Return a normalized Semgrep finding severity."""
    extra = finding.get("extra")
    severity = extra.get("severity") if isinstance(extra, dict) else None
    normalized = str(severity or "UNKNOWN").strip().upper()
    return normalized if normalized in SEVERITIES else "UNKNOWN"


def build_report(
    semgrep_payload: dict[str, Any],
    *,
    semgrep_exit_code: int,
    blocking_severities: set[str] | None = None,
) -> dict[str, Any]:
    """Summarize Semgrep evidence into the quality-gate report contract."""
    blocking = blocking_severities or BLOCKING_SEVERITIES
    raw_results = semgrep_payload.get("results", [])
    raw_errors = semgrep_payload.get("errors", [])
    results = [item for item in raw_results if isinstance(item, dict)] if isinstance(raw_results, list) else []
    errors = [item for item in raw_errors if isinstance(item, dict)] if isinstance(raw_errors, list) else []

    severity_counts = {severity: 0 for severity in SEVERITIES}
    blocking_findings = 0
    for finding in results:
        severity = _finding_severity(finding)
        severity_counts[severity] += 1
        if severity in blocking:
            blocking_findings += 1

    engine_error = semgrep_exit_code not in {0, 1} or bool(errors)
    status = "failed" if engine_error or blocking_findings else "ok"
    return {
        "status": status,
        "scanner": "semgrep",
        "semgrep_rc": semgrep_exit_code,
        "findings_total": len(results),
        "blocking_findings": blocking_findings,
        "errors_total": len(errors),
        "severity_counts": severity_counts,
        "blocking_severities": sorted(blocking),
    }


def _line_number(value: Any, default: int = 1) -> int:
    """Return a Sonar-compatible one-indexed line number."""
    try:
        line = int(value)
    except (TypeError, ValueError):
        return default
    return max(line, default)


def _sonar_issue_from_finding(finding: dict[str, Any]) -> dict[str, Any] | None:
    """Convert one Semgrep finding into SonarQube generic issue format."""
    path = str(finding.get("path") or "").strip()
    if not path:
        return None
    extra = finding.get("extra") if isinstance(finding.get("extra"), dict) else {}
    severity = _finding_severity(finding)
    start = finding.get("start") if isinstance(finding.get("start"), dict) else {}
    end = finding.get("end") if isinstance(finding.get("end"), dict) else {}
    start_line = _line_number(start.get("line") if isinstance(start, dict) else None)
    end_line = _line_number(end.get("line") if isinstance(end, dict) else None, start_line)
    if end_line < start_line:
        end_line = start_line
    return {
        "engineId": "semgrep",
        "ruleId": str(finding.get("check_id") or "semgrep.unknown"),
        "type": "VULNERABILITY" if severity == "ERROR" else "CODE_SMELL",
        "severity": SONAR_SEVERITY_BY_SEMGREP[severity],
        "primaryLocation": {
            "message": str(extra.get("message") or finding.get("check_id") or "Semgrep finding"),
            "filePath": path,
            "textRange": {
                "startLine": start_line,
                "endLine": end_line,
            },
        },
    }


def build_sonar_issues(semgrep_payload: dict[str, Any]) -> dict[str, list[dict[str, Any]]]:
    """Build SonarQube generic external issues from Semgrep JSON output."""
    raw_results = semgrep_payload.get("results", [])
    results = [item for item in raw_results if isinstance(item, dict)] if isinstance(raw_results, list) else []
    issues = [issue for finding in results if (issue := _sonar_issue_from_finding(finding))]
    return {"issues": issues}


def main(argv: list[str] | None = None) -> int:
    """CLI entrypoint used by Jenkins after Semgrep finishes."""
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--semgrep-json", required=True)
    parser.add_argument("--exit-code", type=int, default=0)
    parser.add_argument("--output", required=True)
    parser.add_argument("--sonar-issues-output")
    parser.add_argument(
        "--blocking-severity",
        action="append",
        default=[],
        help="Severity that should mark the report failed. Defaults to ERROR.",
    )
    args = parser.parse_args(argv)

    blocking = {item.strip().upper() for item in args.blocking_severity if item.strip()}
    payload = _read_json(Path(args.semgrep_json))
    report = build_report(
        payload,
        semgrep_exit_code=args.exit_code,
        blocking_severities=blocking or BLOCKING_SEVERITIES,
    )
    output = Path(args.output)
    output.parent.mkdir(parents=True, exist_ok=True)
    output.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    if args.sonar_issues_output:
        sonar_output = Path(args.sonar_issues_output)
        sonar_output.parent.mkdir(parents=True, exist_ok=True)
        sonar_output.write_text(
            json.dumps(build_sonar_issues(payload), indent=2, sort_keys=True) + "\n",
            encoding="utf-8",
        )
    return 0


if __name__ == "__main__":  # pragma: no cover
    raise SystemExit(main())