#!/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/<year>/` 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
# Credit trailers we attribute separately from authorship. Reported-by is the
# real "found the bug" credit; Suggested-by / Co-developed-by / Tested-by are
# weaker (proposed an approach / partial authorship / tested a patch). We tally
# each per author but NEVER rank by them -- the board ranks fix authorship.
CREDIT_TRAILERS = ("reported-by:", "suggested-by:", "co-developed-by:", "tested-by:")
_KIND_FIELD = {"reported-by": "reported_by", "suggested-by": "suggested_by",
               "co-developed-by": "co_developed_by", "tested-by": "tested_by"}


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 _split_ident(raw: str) -> tuple[str, str]:
    """'Name <email>' -> (name, lowercased-email)."""
    name, email = raw.strip(), ""
    if "<" in raw and ">" in raw:
        name = raw[:raw.index("<")].strip()
        email = raw[raw.index("<") + 1:raw.index(">")].strip().lower()
    return name, email


def resolve_authors(shas: list[str], tree: Path) -> dict[str, dict]:
    """{sha: {name, email, credits}} for shas present as commit objects in `tree`.

    `credits` is a list of (kind, name, email) from the commit's credit trailers
    (Reported-by / Suggested-by / Co-developed-by / Tested-by), so the caller can
    attribute finding-credit separately from fix authorship.
    """
    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 = _split_ident(aline)
        credits = []
        for line in body.splitlines():
            low = line.strip().lower()
            for kind in CREDIT_TRAILERS:
                if low.startswith(kind):
                    cn, ce = _split_ident(line.strip()[len(kind):])
                    credits.append((kind[:-1], cn, ce))  # drop trailing ':'
                    break
        res[sha] = {"name": name, "email": email, "credits": credits}
    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'<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 {W} {H}" '
             f'width="{W}" height="{H}" font-family="ui-sans-serif,-apple-system,'
             f'\'Segoe UI\',Roboto,Helvetica,Arial,sans-serif" '
             f'role="img" aria-label="Top {len(top)} Linux kernel CVE fix authors '
             f'for {esc(year)}">')
    p.append('<defs>'
             '<linearGradient id="bar" x1="0" y1="0" x2="1" y2="0">'
             '<stop offset="0" stop-color="#22d3ee"/>'
             '<stop offset="1" stop-color="#818cf8"/></linearGradient>'
             '<linearGradient id="lead" x1="0" y1="0" x2="1" y2="0">'
             '<stop offset="0" stop-color="#fbbf24"/>'
             '<stop offset="1" stop-color="#fb923c"/></linearGradient></defs>')
    # card
    p.append(f'<rect x="0.5" y="0.5" width="{W - 1}" height="{H - 1}" rx="16" '
             f'fill="#0f172a" stroke="#1e293b"/>')
    # title + subtitle + as-of
    p.append(f'<text x="{label_x}" y="38" fill="#f1f5f9" font-size="21" '
             f'font-weight="600">linux kernel CVEs by fix author</text>')
    p.append(f'<text x="{label_x}" y="60" fill="#94a3b8" font-size="12.5">'
             f'top {len(top)} of {total_authors:,} authors &#183; {esc(year)} '
             f'&#183; {total_cves:,} CVEs</text>')
    p.append(f'<text x="{W - 30}" y="38" fill="#cbd5e1" font-size="12.5" '
             f'text-anchor="end" font-weight="600">as of {esc(as_of)}</text>')
    p.append(f'<text x="{W - 30}" y="60" fill="#64748b" font-size="11" '
             f'text-anchor="end">CVE count per fixing-commit author</text>')

    for i, r in enumerate(top):
        y = top_pad + i * rh
        cy = y + rh / 2
        w = max(3, round(r["cve_count"] / peak * bar_w_max))
        lead = i == 0
        fill = "url(#lead)" if lead else "url(#bar)"
        name_ink = "#fde68a" if lead else "#e2e8f0"
        # rank
        p.append(f'<text x="{label_x + 12}" y="{cy + 4:.0f}" fill="#64748b" '
                 f'font-size="12" text-anchor="end">{r["rank"]}</text>')
        # name
        p.append(f'<text x="{label_x + 22}" y="{cy + 4:.0f}" fill="{name_ink}" '
                 f'font-size="13"{" font-weight=\"600\"" if lead else ""}>'
                 f'{esc(truncate(r["author"]))}</text>')
        # bar
        p.append(f'<rect x="{bar_x}" y="{cy - bar_h / 2:.1f}" width="{w}" '
                 f'height="{bar_h}" rx="4" fill="{fill}"/>')
        # value
        p.append(f'<text x="{bar_x + w + val_pad}" y="{cy + 4:.0f}" '
                 f'fill="{name_ink}" font-size="12.5" '
                 f'font-weight="{"600" if lead else "500"}">{r["cve_count"]}</text>')
    p.append('</svg>')
    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 <clone>/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
        info = sha_author[sha]
        name, email = info["name"], info["email"]
        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"],
            "credits": [{"kind": k, "name": n, "email": e}
                        for k, n, e in info["credits"]],
            "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,
                 "reported_by": 0, "suggested_by": 0, "co_developed_by": 0,
                 "tested_by": 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"]
    # Credit pass: attribute finding-credit (from trailers) to authors, on OTHER
    # people's fixes. Only annotate people already on the board (authored >= 1);
    # ranking is unaffected -- these are context columns, not a ranking metric.
    for row in cve_rows:
        author_key = row["author_email"] or row["author_name"].lower()
        for c in row["credits"]:
            ck = c["email"] or c["name"].lower()
            if not ck or ck == author_key or ck not in agg:
                continue
            agg[ck][_KIND_FIELD[c["kind"]]] += 1
    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),
          "reported_by": e["reported_by"], "suggested_by": e["suggested_by"],
          "co_developed_by": e["co_developed_by"], "tested_by": e["tested_by"],
          "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.",
            "reported_by/suggested_by/co_developed_by/tested_by": "finding-credit "
                "from commit trailers, on OTHER authors' fixes (not counted in "
                "cve_count, never ranked). Reported-by is the real 'found the bug' "
                "credit; the rest are weaker (suggested an approach / partial "
                "authorship / tested a patch).",
        },
        "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", "reported_by", "suggested_by",
                    "co_developed_by", "tested_by"])
        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"], r["reported_by"],
                        r["suggested_by"], r["co_developed_by"], r["tested_by"]])

    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())