Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
210 changes: 210 additions & 0 deletions .github/threatcrush-to-sarif.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,210 @@
#!/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())
Loading
Loading