#!/usr/bin/env python3 """Build a Linux kernel CVE-by-fix-author leaderboard for a year. For every published Linux kernel CNA CVE record in a `vulns.git` clone, resolve the author of the mainline fix commit and rank contributors by CVE count. Emits a frequency table (authors) and a per-CVE attribution table (with links) as CSV and JSON, a top-20 SVG bar chart, and a manifest carrying provenance and a UTC timestamp. Every run is reproducible from the same corpus + kernel objects. Author is read from the `author` header of each CVE's mainline fix commit via `git cat-file --batch`, never `git log --author`: `git log --author` walks one branch's ancestry from HEAD and drops fixes that landed through a maintainer tree not on the local tip. Stable backports preserve the original `author`, so the first (mainline) SHA in each CVE's `.sha1` sidecar yields the true author, and any tree that holds the objects works. One CVE counts once, to the author of its mainline fix. Inputs (both public): - a clone of https://git.kernel.org/pub/scm/linux/security/vulns.git (the `cve/published//` tree, with `.json` + `.sha1` per CVE) - a Linux kernel clone holding the fix objects (mainline is usually enough; add a stable clone to resolve the last few backport-only SHAs) Usage: build_leaderboard.py --year 2026 \ --corpus /path/to/vulns/cve/published \ --tree /path/to/linux [--tree /path/to/linux-stable] \ --out ./out Paths also read from $KERNEL_VULNS_DIR (corpus) and $LINUX_TREE (one tree). Standard library + `git` only. MIT-licensed; share freely. """ from __future__ import annotations import argparse import collections import csv import datetime as dt import hashlib import html import json import os import shutil import subprocess import sys from pathlib import Path SITE_ROOT = Path(__file__).resolve().parents[2] DEFAULT_CORPUS = Path(os.environ.get("KERNEL_VULNS_DIR", "vulns/cve/published")) DEFAULT_TREES = [Path(os.environ.get("LINUX_TREE", "linux"))] DEFAULT_OUT = SITE_ROOT / "public" / "wiki" / "linux-kernel-cve-leaderboard" TOP_N_CHART = 20 def load_records(year_dir: Path) -> dict[str, dict]: """{cve: {sha, subject, cvss_score, cvss_severity, cvss_vector}} for a year.""" out: dict[str, dict] = {} for sha_path in sorted(year_dir.glob("*.sha1")): cve = sha_path.name[:-5] lines = [l.strip() for l in sha_path.read_text().splitlines() if l.strip()] if not lines: continue rec = {"sha": lines[0], "subject": "", "cvss_score": "", "cvss_severity": "", "cvss_vector": ""} json_path = year_dir / f"{cve}.json" if json_path.exists(): try: cna = json.loads(json_path.read_text())["containers"]["cna"] except (OSError, KeyError, json.JSONDecodeError): cna = {} desc = "" for d in cna.get("descriptions", []): if d.get("lang", "en").startswith("en"): desc = d.get("value", "") break for line in desc.splitlines(): s = line.strip() if s and not s.lower().startswith("in the linux kernel"): rec["subject"] = s break for m in cna.get("metrics", []): c = m.get("cvssV3_1") or m.get("cvssV4_0") or m.get("cvssV3_0") if c: rec["cvss_score"] = c.get("baseScore", "") rec["cvss_severity"] = c.get("baseSeverity", "") rec["cvss_vector"] = c.get("vectorString", "") break out[cve] = rec return out def resolve_authors(shas: list[str], tree: Path) -> dict[str, tuple[str, str]]: """{sha: (name, email)} for shas present as commit objects in `tree`.""" payload = ("\n".join(shas) + "\n").encode() proc = subprocess.run(["git", "-C", str(tree), "cat-file", "--batch"], input=payload, capture_output=True, timeout=900) out, off, idx, res = proc.stdout, 0, 0, {} keys = list(shas) while off < len(out) and idx < len(keys): nl = out.find(b"\n", off) if nl < 0: break header = out[off:nl].split(b" ") off = nl + 1 sha = keys[idx] idx += 1 if not (len(header) == 3 and header[1] == b"commit" and header[2].isdigit()): continue # not here; retried against the next tree size = int(header[2]) body = out[off:off + size].decode("utf-8", "replace") off += size + 1 aline = next((l[len("author "):] for l in body.splitlines() if l.startswith("author ")), "") name, email = aline, "" if "<" in aline and ">" in aline: name = aline[:aline.index("<")].strip() email = aline[aline.index("<") + 1:aline.index(">")].strip().lower() res[sha] = (name, email) return res def git_head(tree: Path) -> str: try: return subprocess.run(["git", "-C", str(tree), "rev-parse", "HEAD"], capture_output=True, text=True, timeout=30).stdout.strip() except Exception: return "" def sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def render_svg(ranked: list[dict], year: str, as_of: str, total_cves: int, total_authors: int) -> str: """Self-contained dark-card horizontal bar chart of the top TOP_N_CHART.""" top = ranked[:TOP_N_CHART] W, rh, top_pad, bot_pad = 880, 30, 82, 26 H = top_pad + len(top) * rh + bot_pad label_x, bar_x, val_pad = 30, 232, 12 bar_w_max = W - bar_x - 58 bar_h = 17 peak = max(r["cve_count"] for r in top) or 1 def esc(s): return html.escape(str(s)) def truncate(name, n=24): return name if len(name) <= n else name[:n - 1] + "…" p = [] p.append(f'') return "\n".join(p) + "\n" def main() -> int: ap = argparse.ArgumentParser( description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--year", default="2026") ap.add_argument("--corpus", type=Path, default=DEFAULT_CORPUS, help="vulns.git cve/published root ($KERNEL_VULNS_DIR)") ap.add_argument("--tree", type=Path, action="append", help="Linux tree(s) holding the fix objects ($LINUX_TREE)") ap.add_argument("--out", type=Path, default=DEFAULT_OUT, help="artifact output directory") a = ap.parse_args() trees = a.tree or DEFAULT_TREES year = a.year year_dir = a.corpus / year if not year_dir.is_dir(): sys.exit(f"no such published-year dir: {year_dir}\n" f"clone vulns.git and pass --corpus /cve/published") generated = dt.datetime.now(dt.timezone.utc) as_of = generated.strftime("%Y-%m-%d") records = load_records(year_dir) if not records: sys.exit(f"no CVE records with a fix SHA under {year_dir}") todo = {r["sha"] for r in records.values()} sha_author: dict[str, tuple[str, str]] = {} for t in trees: if not (t / ".git").exists(): print(f"warn: not a git tree, skipping: {t}", file=sys.stderr) continue got = resolve_authors(list(todo), t) sha_author.update(got) todo -= set(got) if not todo: break cve_rows, unresolved = [], [] for cve, rec in sorted(records.items()): sha = rec["sha"] if sha not in sha_author: unresolved.append(cve) continue name, email = sha_author[sha] cve_rows.append({ "cve": cve, "author_name": name, "author_email": email, "fix_sha": sha, "subject": rec["subject"], "cvss_score": rec["cvss_score"], "cvss_severity": rec["cvss_severity"], "cvss_vector": rec["cvss_vector"], "nvd_url": f"https://nvd.nist.gov/vuln/detail/{cve}", "commit_url": f"https://git.kernel.org/stable/c/{sha}", }) agg = collections.defaultdict( lambda: {"count": 0, "names": collections.Counter(), "cves": [], "critical": 0, "high": 0, "scored": 0, "cvss_sum": 0.0}) for row in cve_rows: key = row["author_email"] or row["author_name"].lower() e = agg[key] e["count"] += 1 e["names"][row["author_name"]] += 1 e["cves"].append(row["cve"]) sev = (row["cvss_severity"] or "").upper() if sev == "CRITICAL": e["critical"] += 1 elif sev == "HIGH": e["high"] += 1 if isinstance(row["cvss_score"], (int, float)): e["scored"] += 1 e["cvss_sum"] += row["cvss_score"] ranked = sorted( ({"author": e["names"].most_common(1)[0][0], "email": key, "cve_count": e["count"], "critical": e["critical"], "high": e["high"], "scored": e["scored"], "cvss_sum_indicative": round(e["cvss_sum"], 1), "cves": sorted(e["cves"])} for key, e in agg.items()), key=lambda r: (-r["cve_count"], r["author"].lower())) for r in ranked: # standard competition ranking (1,2,2,4) r["rank"] = 1 + sum(1 for x in ranked if x["cve_count"] > r["cve_count"]) provenance = { "generated_utc": generated.isoformat(), "as_of": as_of, "year": year, "corpus": str(year_dir), "kernel_trees": [str(t) for t in trees], "kernel_tree_heads": {str(t): git_head(t) for t in trees if (t / ".git").exists()}, "method": ("fix-commit author, object-addressed via git cat-file --batch; " "first (mainline) SHA per CVE; backports preserve author"), "field_notes": { "author_name/author_email/fix_sha/subject": "ground truth from the " "commit object and the CNA record.", "cvss_*": "indicative only. The Linux kernel CNA does not assign CVSS; " "these come from the corpus record and may be research-augmented.", }, "total_cves": len(records), "resolved_cves": len(cve_rows), "unresolved_cves": unresolved, "distinct_authors": len(ranked), } out = a.out out.mkdir(parents=True, exist_ok=True) (out / "snapshots").mkdir(exist_ok=True) authors_json = out / f"authors-{year}.json" authors_json.write_text(json.dumps({**provenance, "authors": ranked}, indent=2) + "\n") authors_csv = out / f"authors-{year}.csv" with authors_csv.open("w", newline="") as f: w = csv.writer(f) w.writerow(["rank", "author", "email", "cve_count", "critical", "high", "scored", "cvss_sum_indicative"]) for r in ranked: w.writerow([r["rank"], r["author"], r["email"], r["cve_count"], r["critical"], r["high"], r["scored"], r["cvss_sum_indicative"]]) cves_csv = out / f"cves-{year}.csv" with cves_csv.open("w", newline="") as f: w = csv.writer(f) cols = ["cve", "author_name", "author_email", "fix_sha", "cvss_score", "cvss_severity", "cvss_vector", "subject", "nvd_url", "commit_url"] w.writerow(cols) for row in cve_rows: w.writerow([row[c] for c in cols]) combined = out / f"leaderboard-{year}.json" combined.write_text(json.dumps({**provenance, "authors": ranked, "cves": cve_rows}, indent=2) + "\n") svg = out / f"leaderboard-top{TOP_N_CHART}-{year}.svg" svg.write_text(render_svg(ranked, year, as_of, len(cve_rows), len(ranked))) # ship the generator itself as a downloadable, so the build stays # self-describing without depending on the (private) source repo. script_copy = out / "build_leaderboard.py" shutil.copyfile(Path(__file__), script_copy) artifacts = [authors_json, authors_csv, cves_csv, combined, svg, script_copy] manifest_body = { **provenance, "artifacts": [{"file": p.name, "bytes": p.stat().st_size, "sha256": sha256(p)} for p in artifacts], } (out / f"MANIFEST-{year}.json").write_text(json.dumps(manifest_body, indent=2) + "\n") stamp = generated.strftime("%Y%m%dT%H%M%SZ") (out / "snapshots" / f"run-{year}-{stamp}.json").write_text( json.dumps(manifest_body, indent=2) + "\n") print(f"[{year}] as of {as_of}: {len(cve_rows)}/{len(records)} CVEs resolved, " f"{len(ranked)} authors, {len(unresolved)} unresolved") print("top 3: " + "; ".join( f"{r['rank']}. {r['author']} ({r['cve_count']})" for r in ranked[:3])) print(f"artifacts -> {out}") return 0 if __name__ == "__main__": raise SystemExit(main())