# Copyright (C) 2025-2026 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
from collections import defaultdict
import datetime
import json
import logging
import math
import os
from pathlib import Path
import sqlite3
import textwrap
from dateutil import parser as _dateutil_parser
import matplotlib.pyplot as plt
import tqdm
from .utils import format_number
CURRENT_YEAR = datetime.datetime.now().year
def _safe_show(fig: plt.Figure) -> None:
"""Show figure only if ``kitty`` is available, otherwise skip.
The ``matplotlib-backend-kitty`` backend invokes the ``kitty`` terminal to
display images; in environments without ``kitty`` this leads to
FileNotFoundError or PIL errors when the backend attempts to render
to a buffer. Only call ``fig.show()`` when ``kitty`` is present.
"""
if "kitty" in os.environ.get("TERM", "").split("-"):
# nothing to do in headless or non-kitty environments
return
try:
fig.show()
except Exception: # defensive: don't let display errors stop processing
logging.exception("interactive show() failed:")
[docs]
def main(conn: sqlite3.Connection) -> None:
num_affected = []
published_year = []
modified_year = []
published_to_modified_days = []
num_by_database: defaultdict[str, int] = defaultdict(int)
# {database: {range_type: count}}
num_range_types_by_database: defaultdict[str, defaultdict[str, int]] = defaultdict(
lambda: defaultdict(int)
)
# {database: {event_type: count}}
num_event_types_by_database: defaultdict[str, defaultdict[str, int]] = defaultdict(
lambda: defaultdict(int)
)
# {range_type: {event_type: count}}
num_event_types_by_range_type: defaultdict[str, defaultdict[str, int]] = (
defaultdict(lambda: defaultdict(int))
)
cur = conn.cursor()
cur.execute("SELECT COUNT(*) FROM vulnerabilities")
((row_count,),) = cur
cur.execute("SELECT filename, doc FROM vulnerabilities")
for filename, doc in tqdm.tqdm(cur, total=row_count, desc="Reading"):
doc = json.loads(doc)
num_affected.append(len(doc.get("affected", [])))
# Parse published/modified timestamps robustly with dateutil.
# Some records contain more than 6 fractional digits (nanoseconds),
# and `datetime.fromisoformat` rejects those; dateutil handles them.
published = (
_dateutil_parser.parse(doc["published"]) if doc.get("published") else None
)
modified = (
_dateutil_parser.parse(doc["modified"]) if doc.get("modified") else None
)
if published is not None:
published_year.append(published.year)
if modified is not None and modified.year > 1950:
modified_year.append(modified.year)
if published is not None and modified is not None:
published_to_modified_days.append((modified - published).days)
if filename.startswith("osv-output/"):
database = "(NVD CVE)"
else:
assert "/" not in filename, filename
database, _ = filename.split("-", 1)
for affected in doc.get("affected", []):
for range_ in affected.get("ranges", []):
range_type = (
(
"ECOSYSTEM (with versions)"
if affected.get("versions", [])
else "ECOSYSTEM (without versions)"
)
if range_["type"] == "ECOSYSTEM"
else range_["type"]
)
if range_["type"] == "ECOSYSTEM":
if affected.get("versions", []):
if database == "UBUNTU":
range_type = "ECOSYSTEM (with versions + UBUNTU database)"
elif database == "GO":
range_type = "ECOSYSTEM (with versions + GO database)"
else:
range_type = (
"ECOSYSTEM "
"(with versions + except GO and UBUNTU databases)"
)
else:
if database == "UBUNTU":
range_type = (
"ECOSYSTEM (without versions + UBUNTU database)"
)
elif database == "GO":
range_type = "ECOSYSTEM (without versions + GO database)"
else:
range_type = (
"ECOSYSTEM "
"(without versions + except GO and UBUNTU databases)"
)
else:
if affected.get("versions", []):
if database == "GO":
range_type = (
range_["type"] + " (with versions + GO database)"
)
elif database == "GHSA":
range_type = (
range_["type"] + " (with versions + GHSA database)"
)
elif database == "MAL":
range_type = (
range_["type"] + " (with versions + MAL database)"
)
else:
range_type = (
range_["type"]
+ " (with versions + except GHSA/GO/MAL databases)"
)
else:
if database == "GO":
range_type = (
range_["type"] + " (without versions + GO database)"
)
elif database == "GHSA":
range_type = (
range_["type"] + " (without versions + GHSA database)"
)
elif database == "MAL":
range_type = (
range_["type"] + " (without versions + MAL database)"
)
else:
range_type = (
range_["type"]
+ " (without versions + except GHSA/GO/MAL databases)"
)
if (
affected.get("package", {})
.get("ecosystem", "")
.startswith("Debian:")
):
range_type += " (Debian packages)"
else:
range_type += " (except Debian packages)"
event_types = {list(event)[0] for event in range_["events"]}
assert event_types
for event_type in event_types:
num_event_types_by_range_type[range_type][event_type] += 1
num_event_types_by_database[database][event_type] += 1
"""
if (
"introduced" in event_types
and "fixed" not in event_types
and "limit" not in event_types
and "last_affected" not in event_types
and "(without versions" in range_type
and "except GHSA/GO/MAL databases" in range_type
and "except Debian packages" in range_type
):
print(filename)
print(affected)
print(doc)
"""
range_types = {
(
(
"ECOSYSTEM (with versions)"
if affected.get("versions", [])
else "ECOSYSTEM (without versions)"
)
if range_["type"] == "ECOSYSTEM"
else range_["type"]
)
for affected in doc.get("affected", [])
for range_ in affected.get("ranges", [])
}
num_by_database[database] += 1
if not range_types:
range_types = {"none"}
for range_type in range_types:
num_range_types_by_database[database][range_type] += 1
graphs_dir = Path("output/stats")
graphs_dir.mkdir(exist_ok=True, parents=True)
print("=" * 100)
print("Publication and modification time:")
fig, axs = plt.subplot_mosaic(
[["pub", "mod"], ["dif", "dif"]],
)
axs["pub"].hist(
published_year, bins=max(published_year) - min(published_year) + 1, log=True
)
axs["pub"].set_ylabel("Count (log scale)")
axs["pub"].set_xlabel("Year")
axs["pub"].set_title("Publication years")
axs["mod"].hist(
modified_year, bins=max(modified_year) - min(modified_year) + 1, log=True
)
axs["mod"].set_ylabel("Count (log scale)")
axs["mod"].set_xlabel("Year")
axs["mod"].set_title("Modification years (excluding pre-1950)")
axs["dif"].hist(published_to_modified_days, bins=100, log=True)
axs["dif"].set_ylabel("Count (log scale)")
axs["dif"].set_xlabel("Days")
axs["dif"].set_title("Days between publication and modification")
_safe_show(fig)
fig.savefig(graphs_dir / "publication_and_modification_times.svg")
print("=" * 100)
print("Number of 'affected' items per file (log-lin):")
fig, ax = plt.subplots()
ax.hist(
num_affected,
bins=list(
sorted(
{
int(1.1**i)
for i in range(math.ceil(math.log(max(num_affected), 1.1)) + 2)
}
)
),
log=True,
)
ax.set_xscale("log")
ax.set_ylabel("Count (log scale)")
ax.set_xlabel("Affected items per file (log scale)")
ax.set_title("Number of 'affected' items per file (log-log scale)")
_safe_show(fig)
fig.savefig(graphs_dir / "affected_items_per_file_log_scale.svg")
print("Range types by database")
databases = sorted(num_by_database)
sorted_range_types = sorted(
{
range_type
for range_types in num_range_types_by_database.values()
for range_type in range_types
}
)
bar_group_width = 1.0
bar_group_locations = [bar_group_width * i for (i, _) in enumerate(databases)]
bar_width = bar_group_width / (1 + len(sorted_range_types))
fig, ax = plt.subplots()
for i, range_type in enumerate(sorted_range_types):
offset = bar_width * i
rects = ax.bar(
[x + offset for x in bar_group_locations],
[
num_range_types_by_database[database][range_type]
for database in databases
],
bar_width,
label=range_type,
)
ax.bar_label(rects, padding=3, fmt=lambda n: "" if n == 0 else format_number(n))
ax.set_ylabel("Number of ranges")
ax.set_xlabel("Database")
ax.set_title("Event types per database (lin scale)")
ax.set_xticks(
[
x + bar_width * (len(sorted_range_types) - 1) / 2
for x in bar_group_locations
],
databases,
)
ax.legend(loc="upper center")
fig.draw_without_rendering()
_safe_show(fig)
fig.savefig(graphs_dir / "event_types_per_database_lin_scale.svg")
print()
print()
print("Event types for each 'database'; at most one type per range is counted")
databases = sorted(num_event_types_by_database)
sorted_event_types = sorted(
{
event_type
for event_types in num_event_types_by_database.values()
for event_type in event_types
}
)
bar_group_width = 1.0
bar_group_locations = [bar_group_width * i for (i, _) in enumerate(databases)]
bar_width = bar_group_width / (1 + len(sorted_event_types))
fig, ax = plt.subplots()
for i, event_type in enumerate(sorted_event_types):
offset = bar_width * i
rects = ax.bar(
[x + offset for x in bar_group_locations],
[
num_event_types_by_database[database][event_type]
for database in databases
],
bar_width,
label=event_type,
log=True,
)
ax.bar_label(rects, padding=3, fmt=format_number)
ax.set_ylabel("Number of events")
ax.set_xlabel("Database")
ax.set_title("Event types per 'affected' type (log scale)")
ax.set_xticks(
[
x + bar_width * (len(sorted_event_types) - 1) / 2
for x in bar_group_locations
],
databases,
)
ax.legend(loc="upper left")
fig.draw_without_rendering()
_safe_show(fig)
fig.savefig(graphs_dir / "event_types_per_affected_type_log_scale.svg")
print()
print()
print("Event types for each 'affected' type; at most one type per range is counted")
sorted_range_types = sorted(num_event_types_by_range_type)
sorted_event_types = sorted(
{
event_type
for event_types in num_event_types_by_range_type.values()
for event_type in event_types
}
)
bar_group_width = 1.0
bar_group_locations = [
bar_group_width * i for (i, _) in enumerate(sorted_range_types)
]
bar_width = bar_group_width / (1 + len(sorted_event_types))
fig, ax = plt.subplots()
for i, event_type in enumerate(sorted_event_types):
offset = bar_width * i
rects = ax.bar(
[x + offset for x in bar_group_locations],
[
num_event_types_by_range_type[range_type][event_type]
for range_type in sorted_range_types
],
bar_width,
label=event_type,
log=True,
)
ax.bar_label(rects, padding=3, fmt=format_number)
ax.set_ylabel("Number of events")
ax.set_xlabel("Range type")
ax.set_title("Event types per range type (log scale)")
ax.set_xticks(
[
x + bar_width * (len(sorted_event_types) - 1) / 2
for x in bar_group_locations
],
[
"\n".join(textwrap.wrap(range_type, width=16))
for range_type in sorted_range_types
],
fontsize="x-small",
)
ax.legend(loc="upper right")
fig.draw_without_rendering()
_safe_show(fig)
fig.savefig(graphs_dir / "event_types_per_range_type_log_scale.svg")
if __name__ == "__main__":
import matplotlib
matplotlib.use("module://matplotlib-backend-kitty")
matplotlib.rcParams["svg.hashsalt"] = "fixed-salt" # make SVG output deterministic
with sqlite3.connect("all.sqlite") as conn:
main(conn)
