"""the-breach-record-in-one-spreadsheet.py — summarise the public breach catalogue by year, data class, and size. This reads the Have I Been Pwned breach catalogue, which is *metadata about breaches*: one record per loaded breach with a date, a loaded-address count, and the list of data classes exposed. It contains no email addresses, no passwords, and no credentials of any kind, and this script never looks up a person, a domain, or an account. It counts records and does arithmetic. What it produces: 1. --out CSV one row per breach year: how many breaches, how many addresses, the median and 90th percentile size, the largest single breach, the verification and flag counts, how many carried each of the headline data classes, and the median days from the stated breach date to the date the record was loaded. 2. --classes the data-class league table: how often each class appears across the whole catalogue, and its prevalence before 2017 against 2017 and later. 3. --concentration the size-concentration table: what share of all loaded addresses sits in the largest 1, 10, 25 and 50 percent of breaches. 4. --chart-size a concentration curve (share of catalogue against share of addresses). 5. --chart-classes prevalence of the top data classes, earlier era against later era. Inputs: --catalogue FILE a JSON array saved from the documented public endpoint https://haveibeenpwned.com/api/v3/breaches (no authentication required for this endpoint; see https://haveibeenpwned.com/API/v3) --fetch retrieve that endpoint now instead of reading a file (the only network call this script can make, and it is off unless you ask for it) --era-split YEAR the year that divides "earlier" from "later" in the class table (default 2017) Run: python code/the-breach-record-in-one-spreadsheet.py --fetch --catalogue catalogue.json python code/the-breach-record-in-one-spreadsheet.py --catalogue catalogue.json \ --out datasets/the-breach-record-in-one-spreadsheet.csv --classes --concentration \ --chart-size content/images/the-breach-record-in-one-spreadsheet/size-concentration.png \ --chart-classes content/images/the-breach-record-in-one-spreadsheet/data-classes-by-era.png Counting notes, so the numbers are reproducible rather than merely quoted: * Breaches are grouped by BreachDate, which the API documentation calls a guide only, because breaches are often discovered long after the incident. Grouping by AddedDate gives a different and equally valid picture; --by added switches to it. * PwnCount is documented as "the total number of email addresses loaded into the system", which is usually lower than the figure reported in the press. Summing it across breaches double counts any address that appears in more than one breach, so the total is addresses-in-breaches, not people. * A breach is counted once per data class it lists, so class percentages are shares of breaches, not shares of addresses or of records. Standard library only (Python 3.9+); matplotlib is needed for the two chart options alone (tested with matplotlib 3.10.9 on Python 3.13.12). """ from __future__ import annotations import argparse import collections import csv import datetime import json import statistics import sys API_URL = "https://haveibeenpwned.com/api/v3/breaches" # The headline classes broken out as their own columns in the per-year CSV. Every other class is still # counted in the league table; these are the ones a defender triages differently. HEADLINE_CLASSES = [ ("with_passwords", "Passwords"), ("with_email", "Email addresses"), ("with_names", "Names"), ("with_phone", "Phone numbers"), ("with_dob", "Dates of birth"), ("with_physical", "Physical addresses"), ("with_govid", "Government issued IDs"), ("with_card", "Partial credit card data"), ] # "Identity data" composite: the classes that cannot be rotated after a breach the way a password can. IDENTITY_CLASSES = ("Phone numbers", "Dates of birth", "Physical addresses", "Government issued IDs") FIELDS = [ "year", "breaches", "addresses", "median_size", "p90_size", "largest_name", "largest_addresses", "verified", "unverified", "sensitive", "spam_list", "malware", "stealer_log", "with_passwords", "with_email", "with_names", "with_phone", "with_dob", "with_physical", "with_govid", "with_card", "identity_any", "median_days_to_load", "source", "retrieved_at", ] SOURCE_TAG = "haveibeenpwned.com/api/v3/breaches" def fetch_catalogue() -> list: """Retrieve the public breach catalogue. Metadata only; no authentication, no personal data.""" import urllib.request req = urllib.request.Request(API_URL, headers={"User-Agent": "breach-record-summary/1.0"}) with urllib.request.urlopen(req, timeout=60) as r: return json.loads(r.read().decode("utf-8")) def load(path: str) -> list: with open(path, encoding="utf-8") as fh: rows = json.load(fh) if not isinstance(rows, list) or not rows: raise SystemExit(f"{path}: expected a non-empty JSON array of breach records") return rows def year_of(rec: dict, basis: str) -> str: return (rec["BreachDate"] if basis == "breach" else rec["AddedDate"])[:4] def days_to_load(rec: dict): try: b = datetime.date.fromisoformat(rec["BreachDate"]) a = datetime.date.fromisoformat(rec["AddedDate"][:10]) except (ValueError, KeyError): return None return (a - b).days def percentile(values, q: float) -> int: """Nearest-rank percentile, so small groups still return a real observed value.""" s = sorted(values) if not s: return 0 k = max(1, min(len(s), int(round(q / 100 * len(s))))) return int(s[k - 1]) def per_year(rows: list, basis: str, retrieved: str) -> list: groups = collections.defaultdict(list) for r in rows: groups[year_of(r, basis)].append(r) out = [] for year in sorted(groups): g = groups[year] sizes = [r["PwnCount"] for r in g] largest = max(g, key=lambda r: r["PwnCount"]) lags = [d for d in (days_to_load(r) for r in g) if d is not None] row = { "year": year, "breaches": len(g), "addresses": sum(sizes), "median_size": int(statistics.median(sizes)), "p90_size": percentile(sizes, 90), "largest_name": largest["Name"], "largest_addresses": largest["PwnCount"], "verified": sum(1 for r in g if r.get("IsVerified")), "unverified": sum(1 for r in g if not r.get("IsVerified")), "sensitive": sum(1 for r in g if r.get("IsSensitive")), "spam_list": sum(1 for r in g if r.get("IsSpamList")), "malware": sum(1 for r in g if r.get("IsMalware")), "stealer_log": sum(1 for r in g if r.get("IsStealerLog")), "identity_any": sum(1 for r in g if any(c in r["DataClasses"] for c in IDENTITY_CLASSES)), "median_days_to_load": int(statistics.median(lags)) if lags else "", "source": SOURCE_TAG, "retrieved_at": retrieved, } for col, cls in HEADLINE_CLASSES: row[col] = sum(1 for r in g if cls in r["DataClasses"]) out.append(row) return out def class_table(rows: list, split: int) -> list: early = [r for r in rows if int(r["BreachDate"][:4]) < split] late = [r for r in rows if int(r["BreachDate"][:4]) >= split] counts, e_counts, l_counts = collections.Counter(), collections.Counter(), collections.Counter() for r in rows: for c in r["DataClasses"]: counts[c] += 1 for r in early: for c in r["DataClasses"]: e_counts[c] += 1 for r in late: for c in r["DataClasses"]: l_counts[c] += 1 table = [] for cls, n in counts.most_common(): table.append({ "data_class": cls, "breaches": n, "share_all": 100 * n / len(rows), "share_early": 100 * e_counts[cls] / len(early) if early else 0.0, "share_late": 100 * l_counts[cls] / len(late) if late else 0.0, }) return table, len(early), len(late) def concentration(rows: list) -> list: sizes = sorted((r["PwnCount"] for r in rows), reverse=True) total = sum(sizes) out, cum = [], 0 marks = {max(1, int(round(len(sizes) * p / 100))): p for p in (1, 5, 10, 25, 50, 100)} for i, x in enumerate(sizes, 1): cum += x if i in marks: out.append({"top_breaches": i, "share_of_catalogue": marks[i], "share_of_addresses": 100 * cum / total}) return out def curve_points(rows: list): sizes = sorted((r["PwnCount"] for r in rows), reverse=True) total = sum(sizes) xs, ys, cum = [0.0], [0.0], 0 for i, x in enumerate(sizes, 1): cum += x xs.append(100 * i / len(sizes)) ys.append(100 * cum / total) return xs, ys PALETTE = ("#fcfcfb", "#0b0b0b", "#52514e", "#898781", "#e1e0d9", "#c3c2b7", "#2a78d6", "#c8102e") def chart_size(rows: list, path: str) -> None: """Concentration curve: share of the catalogue against share of loaded addresses.""" import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt surface, ink, ink2, muted, grid, axis, bar, hot = PALETTE xs, ys = curve_points(rows) fig, ax = plt.subplots(figsize=(9.0, 5.4), dpi=110) fig.patch.set_facecolor(surface) ax.set_facecolor(surface) ax.plot([0, 100], [0, 100], color=axis, linewidth=1.0, linestyle=(0, (4, 3)), zorder=2) ax.plot(xs, ys, color=bar, linewidth=2.0, zorder=3) ax.fill_between(xs, ys, color=bar, alpha=0.09, zorder=1) for mark in (1, 10, 25): idx = max(1, int(round(len(rows) * mark / 100))) y = ys[idx] ax.plot([mark], [y], marker="o", markersize=5, color=hot, zorder=4) ax.annotate(f"largest {mark}% of breaches = {y:.0f}% of addresses", xy=(mark, y), xytext=(mark + 4, y - 7), fontsize=9, color=ink2, arrowprops={"arrowstyle": "-", "color": axis, "linewidth": 0.8}) ax.set_xlim(0, 100) ax.set_ylim(0, 101) ax.set_xlabel("share of the catalogue, largest breaches first", color=muted, fontsize=9) ax.set_ylabel("cumulative share of loaded addresses", color=muted, fontsize=9) ax.tick_params(colors=muted, length=0) ax.grid(color=grid, linewidth=0.8, zorder=0) for side in ("top", "right"): ax.spines[side].set_visible(False) for side in ("left", "bottom"): ax.spines[side].set_color(axis) fig.text(0.01, 0.985, "The breach record is a handful of enormous corpora and a long tail", ha="left", va="top", fontsize=12.5, color=ink) fig.text(0.01, 0.938, f"{len(rows):,} breaches in the public catalogue; the dashed line is what an " "even distribution would look like.", ha="left", va="top", fontsize=9, color=ink2) fig.tight_layout(rect=(0, 0, 1, 0.9)) fig.savefig(path, facecolor=surface) print(f"wrote chart to {path}", file=sys.stderr) def chart_classes(rows: list, path: str, split: int, top: int = 14) -> None: """Prevalence of the top data classes, earlier era against later era.""" import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt surface, ink, ink2, muted, grid, axis, bar, hot = PALETTE table, n_early, n_late = class_table(rows, split) table = table[:top][::-1] labels = [t["data_class"] for t in table] ypos = range(len(table)) fig, ax = plt.subplots(figsize=(9.4, 0.52 * len(table) + 2.0), dpi=110) fig.patch.set_facecolor(surface) ax.set_facecolor(surface) ax.barh([y + 0.19 for y in ypos], [t["share_early"] for t in table], height=0.36, color=axis, zorder=3, label=f"breaches dated before {split} (n={n_early})") ax.barh([y - 0.19 for y in ypos], [t["share_late"] for t in table], height=0.36, color=bar, zorder=3, label=f"breaches dated {split} or later (n={n_late})") ax.set_yticks(list(ypos)) ax.set_yticklabels(labels, color=ink, fontsize=9) ax.set_xlim(0, 105) ax.set_xticks([0, 25, 50, 75, 100]) ax.set_xticklabels(["0%", "25%", "50%", "75%", "100%"], color=muted, fontsize=9) ax.set_xlabel("share of breaches in the era that listed this data class", color=muted, fontsize=9) ax.tick_params(colors=muted, length=0) ax.grid(axis="x", color=grid, linewidth=0.8, zorder=0) for side in ("top", "right", "left"): ax.spines[side].set_visible(False) ax.spines["bottom"].set_color(axis) leg = ax.legend(loc="lower right", frameon=False, fontsize=9) for text in leg.get_texts(): text.set_color(ink2) fig.text(0.01, 0.985, "Passwords are a shrinking share of the record; identity data is a growing one", ha="left", va="top", fontsize=12.5, color=ink) fig.text(0.01, 0.945, "Share of breaches listing each data class, by the era the breach is dated to.", ha="left", va="top", fontsize=9, color=ink2) fig.tight_layout(rect=(0, 0, 1, 0.93)) fig.savefig(path, facecolor=surface) print(f"wrote chart to {path}", file=sys.stderr) def main() -> int: ap = argparse.ArgumentParser(description=__doc__.split("\n")[0]) ap.add_argument("--catalogue", help="JSON file saved from the public breaches endpoint") ap.add_argument("--fetch", action="store_true", help="retrieve the catalogue now (writes --catalogue if given)") ap.add_argument("--by", choices=("breach", "added"), default="breach", help="group by breach date or load date") ap.add_argument("--era-split", type=int, default=2017, help="year dividing earlier from later (default 2017)") ap.add_argument("--out", help="CSV path for the per-year summary") ap.add_argument("--classes", action="store_true", help="print the data-class league table") ap.add_argument("--concentration", action="store_true", help="print the size-concentration table") ap.add_argument("--chart-size", help="PNG path for the concentration curve; requires matplotlib") ap.add_argument("--chart-classes", help="PNG path for the data-class chart; requires matplotlib") args = ap.parse_args() sys.stdout.reconfigure(encoding="utf-8") if args.fetch: rows = fetch_catalogue() if args.catalogue: with open(args.catalogue, "w", encoding="utf-8") as fh: json.dump(rows, fh) print(f"saved {len(rows)} records to {args.catalogue}", file=sys.stderr) elif args.catalogue: rows = load(args.catalogue) else: ap.error("give --catalogue FILE, --fetch, or both") retrieved = datetime.datetime.now(datetime.timezone.utc).date().isoformat() total = sum(r["PwnCount"] for r in rows) print(f"{len(rows):,} breaches, {total:,} loaded addresses, grouped by {args.by} date " f"(catalogue read {retrieved})") summary = per_year(rows, args.by, retrieved) if args.out: with open(args.out, "w", newline="", encoding="utf-8") as fh: w = csv.DictWriter(fh, fieldnames=FIELDS) w.writeheader() w.writerows(summary) print(f"wrote {len(summary)} year rows to {args.out}", file=sys.stderr) print() print(f"{'year':<6}{'breaches':>9}{'addresses':>16}{'median':>12}{'passwords':>11}{'identity':>10}") for r in summary: print(f"{r['year']:<6}{r['breaches']:>9}{r['addresses']:>16,}{r['median_size']:>12,}" f"{100 * r['with_passwords'] / r['breaches']:>10.0f}%{100 * r['identity_any'] / r['breaches']:>9.0f}%") if args.classes: table, n_early, n_late = class_table(rows, args.era_split) print(f"\ndata classes: {len(table)} distinct; prevalence before {args.era_split} " f"(n={n_early}) against {args.era_split}+ (n={n_late})") print(f"{'class':<28}{'breaches':>9}{'all':>8}{'early':>8}{'later':>8}") for t in table[:20]: print(f"{t['data_class'][:27]:<28}{t['breaches']:>9}{t['share_all']:>7.1f}%" f"{t['share_early']:>7.1f}%{t['share_late']:>7.1f}%") if args.concentration: print("\nsize concentration") for c in concentration(rows): print(f" largest {c['share_of_catalogue']:>3}% of breaches " f"({c['top_breaches']:>5,} of {len(rows):,}) = {c['share_of_addresses']:>5.1f}% of addresses") if args.chart_size: chart_size(rows, args.chart_size) if args.chart_classes: chart_classes(rows, args.chart_classes, args.era_split) return 0 if __name__ == "__main__": raise SystemExit(main())