File size: 3,983 Bytes
7b92798 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 | """Parse SARIF, npm audit JSON, and other scanner formats into normalized alert text."""
from __future__ import annotations
import json
import re
from typing import Any
def detect_format(text: str, filename: str = "") -> str:
name = filename.lower()
stripped = text.strip()
if name.endswith(".sarif") or name.endswith(".json"):
if '"runs"' in stripped and '"results"' in stripped:
return "sarif"
if '"vulnerabilities"' in stripped or '"metadata"' in stripped and '"vulnerabilities"' in stripped:
return "npm_audit"
if stripped.startswith("{"):
try:
data = json.loads(stripped)
if isinstance(data, dict):
if "runs" in data:
return "sarif"
if "vulnerabilities" in data or data.get("auditReportVersion"):
return "npm_audit"
except json.JSONDecodeError:
pass
return "text"
def parse_upload(text: str, filename: str = "") -> dict[str, Any]:
fmt = detect_format(text, filename)
if fmt == "sarif":
return _parse_sarif(text, filename)
if fmt == "npm_audit":
return _parse_npm_audit(text, filename)
return {
"format": "text",
"filename": filename,
"finding_count": 1,
"alert_text": text.strip(),
"title": "Uploaded security report",
}
def _parse_sarif(raw: str, filename: str) -> dict[str, Any]:
data = json.loads(raw)
lines: list[str] = ["SARIF security report", "=" * 40, ""]
count = 0
for run in data.get("runs", []):
tool = run.get("tool", {}).get("driver", {})
tool_name = tool.get("name", "scanner")
lines.append(f"Tool: {tool_name}")
for res in run.get("results", [])[:40]:
count += 1
rule = res.get("ruleId", "unknown")
level = res.get("level", "warning")
msg = res.get("message", {})
text = msg.get("text") if isinstance(msg, dict) else str(msg)
locs = res.get("locations", [])
path = ""
if locs:
phys = locs[0].get("physicalLocation", {}).get("artifactLocation", {})
path = phys.get("uri", "")
lines.append(f"\n[{count}] {level.upper()} — {rule}")
if path:
lines.append(f" Location: {path}")
lines.append(f" {text}")
lines.append("")
alert = "\n".join(lines)
return {
"format": "sarif",
"filename": filename,
"finding_count": count,
"alert_text": alert,
"title": f"SARIF report ({count} finding(s))",
}
def _parse_npm_audit(raw: str, filename: str) -> dict[str, Any]:
data = json.loads(raw)
vulns = data.get("vulnerabilities") or {}
lines = ["npm audit report", "=" * 40, ""]
count = 0
for name, v in list(vulns.items())[:50]:
count += 1
sev = v.get("severity", "unknown")
via = v.get("via", [])
title = name
if via and isinstance(via[0], dict):
title = via[0].get("title", name)
fix = v.get("fixAvailable")
fix_str = "fix available" if fix else "no automatic fix"
lines.append(f"{name} — {sev} — {title} ({fix_str})")
if count == 0 and data.get("metadata", {}).get("vulnerabilities"):
meta = data["metadata"]["vulnerabilities"]
lines.append(
f"Summary: {meta.get('high', 0)} high, {meta.get('moderate', 0)} moderate, "
f"{meta.get('low', 0)} low"
)
count = sum(meta.values()) if isinstance(meta, dict) else 1
alert = "\n".join(lines)
return {
"format": "npm_audit",
"filename": filename,
"finding_count": max(count, 1),
"alert_text": alert,
"title": f"npm audit ({count} package(s))",
}
def extract_cve_ids(text: str) -> list[str]:
return list(dict.fromkeys(re.findall(r"CVE-\d{4}-\d+", text, re.I)))
|