|
| 1 | +#!/usr/bin/env python3 |
| 2 | +"""Enrich the Python audit.json with severity data from the OSV API. |
| 3 | +
|
| 4 | +pip-audit does not include severity in its output. This script queries |
| 5 | +https://api.osv.dev/v1/vulns/{id} for each vulnerability (preferring GHSA |
| 6 | +aliases which have richer metadata) and writes the severity string back into |
| 7 | +public/data/ldpov/python/audit.json. |
| 8 | +
|
| 9 | +Run after run_ldpov_pypi_scan.sh: |
| 10 | + python3 enrich_python_severity.py |
| 11 | +""" |
| 12 | + |
| 13 | +import json |
| 14 | +import os |
| 15 | +import time |
| 16 | +import urllib.request |
| 17 | + |
| 18 | +SCRIPT_DIR = os.path.dirname(__file__) |
| 19 | +AUDIT_PATH = os.path.join(SCRIPT_DIR, "..", "public", "data", "ldpov", "python", "audit.json") |
| 20 | + |
| 21 | + |
| 22 | +def fetch_severity(osv_id: str) -> str | None: |
| 23 | + url = f"https://api.osv.dev/v1/vulns/{osv_id}" |
| 24 | + try: |
| 25 | + with urllib.request.urlopen(url, timeout=10) as r: |
| 26 | + rec = json.loads(r.read()) |
| 27 | + return (rec.get("database_specific") or {}).get("severity") or None |
| 28 | + except Exception: |
| 29 | + return None |
| 30 | + |
| 31 | + |
| 32 | +def main() -> None: |
| 33 | + with open(AUDIT_PATH) as f: |
| 34 | + data = json.load(f) |
| 35 | + |
| 36 | + # Collect unique vulns that need enrichment |
| 37 | + vuln_lookup: dict[str, str] = {} # audit_id -> best OSV id to query |
| 38 | + for dep in data.get("dependencies", []): |
| 39 | + for v in dep.get("vulns", []): |
| 40 | + if v.get("severity"): |
| 41 | + continue # already enriched |
| 42 | + vid = v["id"] |
| 43 | + aliases = v.get("aliases", []) |
| 44 | + ghsa = next((a for a in aliases if a.startswith("GHSA-")), None) |
| 45 | + vuln_lookup[vid] = ghsa or vid |
| 46 | + |
| 47 | + if not vuln_lookup: |
| 48 | + print("All vulns already have severity. Nothing to do.") |
| 49 | + return |
| 50 | + |
| 51 | + print(f"Querying OSV API for {len(vuln_lookup)} vulns...", flush=True) |
| 52 | + |
| 53 | + severity_map: dict[str, str | None] = {} |
| 54 | + errors = 0 |
| 55 | + for i, (audit_id, osv_id) in enumerate(vuln_lookup.items(), 1): |
| 56 | + sev = fetch_severity(osv_id) |
| 57 | + severity_map[audit_id] = sev |
| 58 | + if i % 20 == 0 or i == len(vuln_lookup): |
| 59 | + print(f" {i}/{len(vuln_lookup)}", flush=True) |
| 60 | + time.sleep(0.05) |
| 61 | + |
| 62 | + # Patch in place |
| 63 | + for dep in data.get("dependencies", []): |
| 64 | + for v in dep.get("vulns", []): |
| 65 | + vid = v["id"] |
| 66 | + if vid in severity_map: |
| 67 | + v["severity"] = severity_map[vid] |
| 68 | + |
| 69 | + with open(AUDIT_PATH, "w") as f: |
| 70 | + json.dump(data, f, separators=(",", ":")) |
| 71 | + |
| 72 | + dist: dict[str, int] = {} |
| 73 | + for s in severity_map.values(): |
| 74 | + dist[s or "unknown"] = dist.get(s or "unknown", 0) + 1 |
| 75 | + print(f"Done. Severity distribution: {dist}") |
| 76 | + if errors: |
| 77 | + print(f" {errors} OSV lookups failed (severity left as null)") |
| 78 | + |
| 79 | + |
| 80 | +if __name__ == "__main__": |
| 81 | + main() |
0 commit comments