Spaces:
Sleeping
Sleeping
| """ | |
| Normalizes output from *any* scanner into a single Finding shape, so the triage | |
| swarm is decoupled from your 29 tools. Adapters included: | |
| - SARIF 2.1.0 (Semgrep, CodeQL, Trivy, Bandit, ...) | |
| - Generic JSON list (best-effort field mapping — covers most custom scanners) | |
| - Raw text (one finding per non-empty line) | |
| Point any of your scanners at this and the rest just works. | |
| """ | |
| from __future__ import annotations | |
| import hashlib | |
| import json | |
| from typing import Any | |
| def _fid(*parts: str) -> str: | |
| return hashlib.sha1("|".join(parts).encode()).hexdigest()[:10] | |
| def finding(rule: str, message: str, *, path: str = "", scanner: str = "", | |
| raw_severity: str = "", snippet: str = "") -> dict: | |
| return { | |
| "id": _fid(scanner, rule, path, message[:80]), | |
| "rule": rule or "finding", | |
| "message": message.strip(), | |
| "path": path, | |
| "scanner": scanner, | |
| "raw_severity": raw_severity, | |
| "snippet": snippet, | |
| } | |
| def from_sarif(doc: dict) -> list[dict]: | |
| out: list[dict] = [] | |
| for run in doc.get("runs", []): | |
| tool = (run.get("tool", {}).get("driver", {}) or {}).get("name", "sarif") | |
| rules = {r.get("id"): r for r in | |
| (run.get("tool", {}).get("driver", {}).get("rules", []) or [])} | |
| for res in run.get("results", []): | |
| rule_id = res.get("ruleId") or res.get("rule", {}).get("id") or "rule" | |
| msg = (res.get("message", {}) or {}).get("text", "") or rule_id | |
| level = res.get("level") or rules.get(rule_id, {}).get("defaultConfiguration", {}).get("level", "") | |
| path, snippet = "", "" | |
| locs = res.get("locations") or [] | |
| if locs: | |
| phys = locs[0].get("physicalLocation", {}) | |
| path = (phys.get("artifactLocation", {}) or {}).get("uri", "") | |
| snippet = (phys.get("region", {}) or {}).get("snippet", {}).get("text", "") | |
| out.append(finding(rule_id, msg, path=path, scanner=tool, | |
| raw_severity=level, snippet=snippet)) | |
| return out | |
| def from_generic(items: list[dict]) -> list[dict]: | |
| """Best-effort mapping for custom scanner JSON (handles common key names).""" | |
| def pick(d: dict, *keys: str, default: str = "") -> str: | |
| for k in keys: | |
| if k in d and d[k] not in (None, ""): | |
| return str(d[k]) | |
| return default | |
| out = [] | |
| for it in items: | |
| if not isinstance(it, dict): | |
| out.append(finding("finding", str(it))) | |
| continue | |
| out.append(finding( | |
| pick(it, "rule", "rule_id", "check", "id", "title", "name", default="finding"), | |
| pick(it, "message", "description", "desc", "detail", "title", default=json.dumps(it)[:200]), | |
| path=pick(it, "path", "file", "location", "uri", "target"), | |
| scanner=pick(it, "scanner", "tool", "source", "engine"), | |
| raw_severity=pick(it, "severity", "level", "risk", "confidence"), | |
| snippet=pick(it, "snippet", "code", "evidence", "match"), | |
| )) | |
| return out | |
| def from_text(text: str, scanner: str = "text") -> list[dict]: | |
| return [finding("line", ln.strip(), scanner=scanner) | |
| for ln in text.splitlines() if ln.strip()] | |
| def normalize(payload: Any) -> list[dict]: | |
| """Auto-detect format and return a list of normalized findings.""" | |
| if isinstance(payload, str): | |
| s = payload.strip() | |
| try: | |
| payload = json.loads(s) | |
| except json.JSONDecodeError: | |
| return from_text(s) | |
| if isinstance(payload, dict): | |
| if "runs" in payload and "version" in payload: # SARIF | |
| return from_sarif(payload) | |
| for key in ("findings", "results", "issues", "vulnerabilities", "alerts"): | |
| if isinstance(payload.get(key), list): | |
| return from_generic(payload[key]) | |
| return from_generic([payload]) | |
| if isinstance(payload, list): | |
| return from_generic(payload) | |
| return [] | |