# 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
import json
from pathlib import Path
import sqlite3
import tarfile
import zipfile
import pyzstd
import tqdm
[docs]
def init_db(conn: sqlite3.Connection) -> None:
cur = conn.cursor()
for statement in [
"""
CREATE TABLE IF NOT EXISTS vulnerabilities (
id INTEGER PRIMARY KEY AUTOINCREMENT,
filename TEXT,
doc TEXT
)
""",
"""
CREATE UNIQUE INDEX IF NOT EXISTS vulnerabilities_filename_idx
ON vulnerabilities (filename)
""",
"""
CREATE TABLE IF NOT EXISTS package_vulnerabilities (
package_ecosystem TEXT,
package_name TEXT,
vulnerability_filename TEXT
)
""",
"""
CREATE UNIQUE INDEX IF NOT EXISTS package_vulnerabilities_ecosystem_name
ON package_vulnerabilities (package_ecosystem, package_name, vulnerability_filename)
""",
"""
CREATE TABLE IF NOT EXISTS vulnerability_ids (
id TEXT, -- includes both id and aliases
vulnerability_filename TEXT
)
""",
"""
CREATE UNIQUE INDEX IF NOT EXISTS vulnerability_ids_idx
ON vulnerability_ids (id, vulnerability_filename)
""",
]:
cur.execute(statement)
[docs]
def create_indices(conn: sqlite3.Connection) -> None:
statements = []
for column in ["id", "published", "modified"]:
statements.append(f"""
CREATE INDEX IF NOT EXISTS vulnerabilities_{column}
ON vulnerabilities (doc->>'{column}')
""")
cur = conn.cursor()
for statement in tqdm.tqdm(statements, desc="Creating indices"):
cur.execute(statement)
[docs]
def main(conn: sqlite3.Connection) -> None:
init_db(conn)
import_osv_tarballs(conn, Path("./nvd_cve/"))
import_osv_zip(conn, Path("./all.zip"))
[docs]
def import_osv_tarballs(conn: sqlite3.Connection, path: Path) -> None:
"""Import a local download of tarballs of
https://storage.googleapis.com/cve-osv-conversion/osv-output/ 's objects
to the sqlite database"""
cur = conn.cursor()
nvd_shards = list(path.glob("*.tar.zst"))
num_nvd_vulnerabilities = 0
for shard in tqdm.tqdm(nvd_shards, desc="Counting OSV vulnerabilities"):
with pyzstd.open(shard) as fd:
with tarfile.open(fileobj=fd, mode="r") as tf:
num_nvd_vulnerabilities += sum(1 for _ in tf)
with tqdm.tqdm(
total=num_nvd_vulnerabilities, desc="Importing NVD vulnerabilities"
) as pb:
for shard in nvd_shards:
with pyzstd.open(shard) as fd:
with tarfile.open(fileobj=fd, mode="r") as tf:
for member in tf:
pb.update(1)
data = tf.extractfile(member)
assert data is not None, member.name
_insert_vulnerability(cur, member.name, data.read().decode())
[docs]
def import_osv_zip(conn: sqlite3.Connection, path: Path) -> None:
"""Import a local download of
https://osv-vulnerabilities.storage.googleapis.com/all.zip to the sqlite database"""
cur = conn.cursor()
with zipfile.ZipFile(path) as zf:
for filename in tqdm.tqdm(zf.namelist(), desc="Importing OSV vulnerabilities"):
_insert_vulnerability(cur, filename, zf.read(filename).decode())
create_indices(conn)
def _insert_vulnerability(cur: sqlite3.Cursor, filename: str, data: str) -> None:
doc = json.loads(data)
cur.execute(
"""
INSERT INTO vulnerabilities (filename, doc)
VALUES (?, ?)
ON CONFLICT(filename)
DO UPDATE SET doc=EXCLUDED.doc
WHERE doc != EXCLUDED.doc -- saves 35% runtime
""",
(filename, json.dumps(doc, indent=2, sort_keys=True)),
)
for affected in doc.get("affected", []):
if "package" in affected:
# eg. cURL reports don't have a package
cur.execute(
"""
INSERT INTO package_vulnerabilities (
package_ecosystem, package_name, vulnerability_filename
)
VALUES (?, ?, ?)
ON CONFLICT(package_ecosystem, package_name, vulnerability_filename)
DO NOTHING
""",
(
affected["package"]["ecosystem"],
affected["package"]["name"],
filename,
),
)
for id_ in [doc["id"], *doc.get("aliases", [])]:
cur.execute(
"""
INSERT INTO vulnerability_ids (
id, vulnerability_filename
)
VALUES (?, ?)
ON CONFLICT(id, vulnerability_filename)
DO NOTHING
""",
(id_, filename),
)
if __name__ == "__main__":
with sqlite3.connect("all.sqlite") as conn:
main(conn)
