agentbbs/.github/threatcrush-to-sarif.py

#!/usr/bin/env python3
"""Convert ThreatCrush terminal output to SARIF 2.1.0.

Compatibility shim for CLI versions older than native ``--format sarif``.
When the CLI can emit SARIF itself the workflow uses that and never runs this
file; parsing a human-readable stream is strictly worse and exists only so a
repository is not left unscanned while waiting for a release.

It **fails closed**. If it cannot recognise the output it exits non-zero and
dumps what it saw. Emitting empty SARIF instead would report "0 findings",
which is indistinguishable from a clean scan and is the single most expensive
thing a security tool can get wrong.

Three details of the format, each of which is load-bearing:

* Severity is bare for ``CRITICAL`` and bracketed for ``[HIGH]``/``[MEDIUM]``/
  ``[LOW]``. One regex shape misses half the findings.
* ``File:`` paths are relative to the scan root, not the repository root. Left
  unprefixed, every finding resolves to nothing in the consumer's view of the
  repo. Hence ``--path-prefix``.
* Whole-file findings report line ``:0``. SARIF requires ``startLine >= 1``.

``Code:`` lines are redacted excerpts of the match. They are skipped rather
than parsed, both because matching them would double-count every finding and
because a redacted excerpt tells a reader nothing the ``Info:`` line does not.
"""

from __future__ import annotations

import argparse
import json
import re
import sys

ANSI = re.compile(r"\x1b\[[0-9;]*[A-Za-z]")

# `   CRITICAL  AWS Access Key`  /  `  [HIGH] Sensitive File`
SEVERITY_LINE = re.compile(r"^\s*(?:\[(CRITICAL|HIGH|MEDIUM|LOW|INFO)\]|(CRITICAL))\s+(.+?)\s*$")
FILE_LINE = re.compile(r"^\s*File:\s*(.+?):(\d+)\s*$")
INFO_LINE = re.compile(r"^\s*Info:\s*(.+?)\s*$")

# Proof that a scan ran to completion. Without one of these we are looking at a
# crash, a help screen, or an unrecognised release — never at a clean result.
FOOTER = re.compile(r"^\s*(?:\d+\s+issue\(s\)\s+found|.*No security issues found)")

LEVELS = {"CRITICAL": "error", "HIGH": "error", "MEDIUM": "warning", "LOW": "note", "INFO": "none"}
SECURITY_SEVERITY = {"CRITICAL": "9.0", "HIGH": "7.0", "MEDIUM": "5.0", "LOW": "3.0", "INFO": "1.0"}
RANK = {"info": 0, "low": 1, "medium": 2, "high": 3, "critical": 4}


class Unrecognised(Exception):
    """The output did not look like a completed ThreatCrush scan."""


def rule_id(title: str) -> str:
    """Derive a stable rule id from a finding title.

    Old CLIs print `AWS Access Key`, not `secret-aws-access-key`. Slugifying
    keeps SARIF results groupable and keeps fingerprints stable across runs,
    which is what stops the Security tab treating every run as brand-new alerts.
    """
    slug = re.sub(r"[^a-z0-9]+", "-", title.lower()).strip("-")
    return f"threatcrush-{slug}" if slug else "threatcrush-finding"


def parse(text: str) -> list[dict]:
    lines = ANSI.sub("", text).splitlines()
    if not any(FOOTER.match(line) for line in lines):
        raise Unrecognised("no scan-completion footer found")

    findings: list[dict] = []
    pending: dict | None = None

    for line in lines:
        severity_match = SEVERITY_LINE.match(line)
        if severity_match:
            severity = severity_match.group(1) or severity_match.group(2)
            pending = {"severity": severity.upper(), "title": severity_match.group(3).strip()}
            continue

        if pending is None:
            continue

        file_match = FILE_LINE.match(line)
        if file_match:
            pending["file"] = file_match.group(1).strip()
            pending["line"] = int(file_match.group(2))
            continue

        info_match = INFO_LINE.match(line)
        if info_match and "file" in pending:
            pending["message"] = info_match.group(1).strip()
            findings.append(pending)
            pending = None

    return findings


def to_sarif(findings: list[dict], prefix: str, version: str) -> dict:
    rules: dict[str, dict] = {}
    results = []

    for finding in findings:
        rid = rule_id(finding["title"])
        rules.setdefault(
            rid,
            {
                "id": rid,
                "name": rid,
                "shortDescription": {"text": finding["title"]},
                "fullDescription": {"text": finding["title"]},
                "defaultConfiguration": {"level": LEVELS[finding["severity"]]},
                "properties": {
                    "tags": ["security", "threatcrush"],
                    "security-severity": SECURITY_SEVERITY[finding["severity"]],
                },
            },
        )

        uri = finding["file"].lstrip("./")
        if prefix:
            uri = f"{prefix.strip('/')}/{uri}"

        results.append(
            {
                "ruleId": rid,
                "level": LEVELS[finding["severity"]],
                "message": {"text": finding.get("message", finding["title"])},
                "locations": [
                    {
                        "physicalLocation": {
                            "artifactLocation": {"uri": uri, "uriBaseId": "%SRCROOT%"},
                            # Clamped: SARIF rejects 0, and a whole-file finding
                            # has no line to report.
                            "region": {"startLine": max(1, finding["line"])},
                        }
                    }
                ],
                "partialFingerprints": {
                    "primaryLocationLineHash": f"{rid}:{uri}:{max(1, finding['line'])}"
                },
                "properties": {"severity": finding["severity"].lower()},
            }
        )

    return {
        "$schema": "https://raw.githubusercontent.com/oasis-tcs/sarif-spec/master/Schemata/sarif-schema-2.1.0.json",
        "version": "2.1.0",
        "runs": [
            {
                "tool": {
                    "driver": {
                        "name": "ThreatCrush",
                        "version": version,
                        "informationUri": "https://threatcrush.com",
                        "rules": list(rules.values()),
                    }
                },
                "results": results,
                "columnKind": "utf16CodeUnits",
            }
        ],
    }


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--input", required=True, help="captured `threatcrush scan` output")
    parser.add_argument("--output", required=True, help="SARIF file to write")
    parser.add_argument("--path-prefix", default="", help="prepended to every file URI")
    parser.add_argument("--tool-version", default="unknown")
    parser.add_argument("--fail-on", default="", help="comma-separated severities that exit 1")
    args = parser.parse_args()

    with open(args.input, encoding="utf-8", errors="replace") as handle:
        text = handle.read()

    try:
        findings = parse(text)
    except Unrecognised as err:
        print(f"error: unrecognised ThreatCrush output ({err})", file=sys.stderr)
        print("--- first 40 lines ---", file=sys.stderr)
        for line in ANSI.sub("", text).splitlines()[:40]:
            print(line, file=sys.stderr)
        return 2

    with open(args.output, "w", encoding="utf-8") as handle:
        json.dump(to_sarif(findings, args.path_prefix, args.tool_version), handle, indent=2)
        handle.write("\n")

    print(f"converted {len(findings)} finding(s) to {args.output}")

    thresholds = [s.strip().lower() for s in args.fail_on.split(",") if s.strip()]
    if thresholds:
        unknown = [s for s in thresholds if s not in RANK]
        if unknown:
            # Silently ignoring a typo produces a gate that never fires, which
            # looks exactly like a passing build.
            print(f"error: unknown severity in --fail-on: {', '.join(unknown)}", file=sys.stderr)
            return 2
        floor = min(RANK[s] for s in thresholds)
        if any(RANK[f["severity"].lower()] >= floor for f in findings):
            print(f"::error::findings at or above {args.fail_on}")
            return 1

    return 0


if __name__ == "__main__":
    sys.exit(main())