# 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 asyncio
import dataclasses
import itertools
import json
import os
from pathlib import Path
import sqlite3
from typing import Iterable
import aiohttp
from google.protobuf.field_mask_pb2 import FieldMask
import grpc.aio
import grpc.experimental
import tqdm.asyncio
import swh.graph.grpc.swhgraph_pb2 as swhgraph
import swh.graph.grpc.swhgraph_pb2_grpc as swhgraph_grpc
from .map_origins import OriginDenormalizer
from .utils import Cache, concurrent_async_map
GRAPH_GRPC_URL = "localhost:50091"
# GRAPH_GRPC_URL = "localhost:12345"
[docs]
def init_db(conn: sqlite3.Connection) -> None:
cur = conn.cursor()
for statement in [
"""
CREATE TABLE IF NOT EXISTS unknown_swhids (
swhid TEXT PRIMARY KEY
)
""",
# """
# CREATE TABLE IF NOT EXISTS swhids (
# id INT PRIMARY KEY,
# swhid TEXT
# )
# """,
# """
# CREATE UNIQUE INDEX IF NOT EXISTS swhids_swhid
# ON swhids (swhid)
# """,
# """
# CREATE TABLE IF NOT EXISTS vulnerability_swhids (
# vulnerability_filename TEXT,
# swhid_id INT
# )
# """,
# """
# CREATE INDEX IF NOT EXISTS vulnerability_swhids_vulnerability_filename
# ON vulnerability_swhids (vulnerability_filename)
# """,
# """
# CREATE INDEX IF NOT EXISTS vulnerability_swhids_swhid_id
# ON vulnerability_swhids (swhid_id)
# """,
]:
cur.execute(statement)
[docs]
@dataclasses.dataclass
class DocProcessor:
stub: swhgraph_grpc.TraversalServiceStub
session: aiohttp.ClientSession
conn: sqlite3.Connection
origin_denormalizer: OriginDenormalizer
root_revisions_cache: Cache[dict[tuple[str, ...], list[str]]]
total_invalid_commit_ids = 0
total_unknown_commits = 0
total_skipped_due_to_invalid_commit = 0
total_skipped_due_to_unknown_commit = 0
total_skipped_due_to_unimplemented_event_types = 0
total_vulnerable = 0
total_successes = 0
[docs]
def add_unknown_swhid(self, swhid) -> None:
assert swhid.startswith("swh:1:rev:"), swhid
return
self.conn.execute(
"""
INSERT INTO unknown_swhids(swhid)
VALUES(?)
ON CONFLICT (swhid) DO NOTHING
""",
(swhid,),
)
[docs]
def get_unknown_swhids(self, swhids: Iterable[str]) -> set[str]:
"""Given an iterable of SWHIDs, returns which ones are known to be unknown"""
swhids = tuple(swhids)
cur = self.conn.cursor()
cur.execute(
f"""
SELECT swhid
FROM unknown_swhids
WHERE swhid IN ({','.join('?' for _ in swhids)})
""",
swhids,
)
return {swhid for (swhid,) in cur}
[docs]
async def process_docs(self, arg):
filename, doc = arg
doc = json.loads(doc)
for affected in doc.get("affected", []):
for range_ in affected.get("ranges", []):
if range_["type"] == "GIT":
commits_by_event_type = await self.expand_git_range(range_)
if commits_by_event_type is not None:
await self.process_git_range(commits_by_event_type)
[docs]
async def expand_git_range(self, range_: dict) -> dict[str, set[str]] | None:
root_revisions = None
commits_by_event_type: dict[str, set[str]] = {}
for event in range_["events"]:
for event_type, commit_id in event.items():
if commit_id == "0":
if event_type != "introduced":
raise ValueError(
"Pseudo-version '0' is only allowed in 'introduced' events"
)
commits_by_event_type.setdefault(event_type, set()).add(commit_id)
# forbidden by the spec
if commits_by_event_type.keys() >= {"last_affected", "fixed"}:
raise ValueError(
"'events' array contains both 'last_affected' and 'fixed' events"
)
commit_ids = set(itertools.chain.from_iterable(commits_by_event_type.values()))
if "0" in commit_ids:
commit_ids.remove("0")
# "introduced allows a version of the value "0" to represent a version
# that sorts before any other version."
root_revisions = set()
for commit_id in commit_ids:
commit_swhid = "swh:1:rev:" + commit_id
if commit_id not in self.root_revisions_cache:
try:
revs = self.stub.Traverse(
swhgraph.TraversalRequest(
src=[commit_swhid],
edges="rev:rev",
direction=swhgraph.GraphDirection.FORWARD,
mask=FieldMask(paths=["swhid"]),
return_nodes=swhgraph.NodeFilter(
max_traversal_successors=0, # return only leaves
),
)
)
commit_root_revisions = [rev.swhid async for rev in revs]
except grpc.aio.AioRpcError as e:
if e.code() == grpc.StatusCode.NOT_FOUND:
self.add_unknown_swhid(commit_swhid)
self.total_unknown_commits += 1
continue
elif e.code() == grpc.StatusCode.INVALID_ARGUMENT:
self.total_invalid_commit_ids += 1
self.total_skipped_due_to_invalid_commit += 1
return None
else:
raise
else:
self.root_revisions_cache[commit_id] = commit_root_revisions
await self.root_revisions_cache.save()
else:
commit_root_revisions = self.root_revisions_cache[commit_id]
root_revisions.update(commit_root_revisions)
# replace "0" with the set of root commits
commit_swhids_by_event_type = {}
for event_type, commit_set in commits_by_event_type.items():
if "0" in commit_set:
commit_set.clear()
commit_swhids_by_event_type[event_type] = {
f"swh:1:rev:{commit_id}"
for commit_id in commit_set
if commit_id != "0"
} | root_revisions
else:
commit_swhids_by_event_type[event_type] = {
f"swh:1:rev:{commit_id}" for commit_id in commit_set
}
return commit_swhids_by_event_type
[docs]
async def process_git_range(
self, commits_by_event_type: dict[str, set[str]]
) -> None:
unknown_swhids = self.get_unknown_swhids(
set(itertools.chain.from_iterable(commits_by_event_type.values()))
)
self.total_unknown_commits += len(
commits_by_event_type["introduced"] & unknown_swhids
)
introduction_commits = commits_by_event_type["introduced"] - unknown_swhids
if not introduction_commits:
# not worth running this
self.total_skipped_due_to_unknown_commit += 1
return
assert "swh:1:rev:0" not in introduction_commits
while introduction_commits - unknown_swhids:
# keep trying in case some of the commits are missing from the graph
try:
reachable_by_introduction = {
rev.swhid.removeprefix("swh:1:rev:")
async for rev in self.stub.Traverse(
swhgraph.TraversalRequest(
src=introduction_commits,
edges="rev:rev",
direction=swhgraph.GraphDirection.BACKWARD,
mask=FieldMask(paths=["swhid"]),
ignore_node=(
commits_by_event_type.get("fixed", set())
| commits_by_event_type.get("last_affected", set())
| commits_by_event_type.get("limit", set())
)
- unknown_swhids,
)
)
}
except grpc.aio.AioRpcError as e:
if e.code() == grpc.StatusCode.NOT_FOUND:
# one of the commits is missing from the graph. remove it and try again.
# this is correct because if an introduction/fix commit is missing then
# so are its children, so we don't mislabel them.
details = e.details()
assert details is not None
unknown_swhid = details.removeprefix("Unknown SWHID: ")
unknown_swhids.add(unknown_swhid)
self.add_unknown_swhid(unknown_swhid)
else:
raise
else:
break
else:
self.total_skipped_due_to_unknown_commit += 1
return
event_types = set(commits_by_event_type)
if event_types == {"introduced", "fixed"} or event_types == {"introduced"}:
if "fixed" in commits_by_event_type:
self.total_unknown_commits += len(
commits_by_event_type["fixed"] & unknown_swhids
)
fix_commits = commits_by_event_type["fixed"] - unknown_swhids
else:
fix_commits = set()
assert "swh:1:rev:0" not in fix_commits
if fix_commits:
reachable_by_fix = {
rev.swhid.removeprefix("swh:1:rev:")
async for rev in self.stub.Traverse(
swhgraph.TraversalRequest(
src=fix_commits,
edges="rev:rev",
direction=swhgraph.GraphDirection.BACKWARD,
mask=FieldMask(paths=["swhid"]),
ignore_node=introduction_commits,
)
)
}
else:
reachable_by_fix = set()
vulnerable = reachable_by_introduction - reachable_by_fix
self.total_successes += 1
self.total_vulnerable += len(vulnerable)
# self.vulnerable_commits.update(vulnerable)
elif event_types == {"introduced", "last_affected"}:
self.total_unknown_commits += len(
commits_by_event_type["last_affected"] & unknown_swhids
)
last_affected_commits = (
commits_by_event_type["last_affected"] - unknown_swhids
)
assert "swh:1:rev:0" not in last_affected_commits
if last_affected_commits:
reachable_by_last_affected = {
rev.swhid.removeprefix("swh:1:rev:")
async for rev in self.stub.Traverse(
swhgraph.TraversalRequest(
src=last_affected_commits,
edges="rev:rev",
direction=swhgraph.GraphDirection.BACKWARD,
mask=FieldMask(paths=["swhid"]),
ignore_node=introduction_commits,
)
)
}
else:
reachable_by_last_affected = set()
vulnerable = (
reachable_by_introduction - reachable_by_last_affected
) | last_affected_commits
self.total_successes += 1
self.total_vulnerable += len(vulnerable)
# self.vulnerable_commits.update(vulnerable)
elif event_types == {"introduced", "limit"}:
self.total_unknown_commits += len(
commits_by_event_type["limit"] & unknown_swhids
)
limit_commits = commits_by_event_type["limit"] - unknown_swhids
assert "swh:1:rev:0" not in limit_commits
if limit_commits:
coreachable_by_limit = {
rev.swhid.removeprefix("swh:1:rev:")
async for rev in self.stub.Traverse(
swhgraph.TraversalRequest(
src=limit_commits,
edges="rev:rev",
direction=swhgraph.GraphDirection.FORWARD,
mask=FieldMask(paths=["swhid"]),
ignore_node=introduction_commits,
)
)
}
else:
reachable_by_last_affected = set()
vulnerable = reachable_by_introduction & coreachable_by_limit
self.total_successes += 1
self.total_vulnerable += len(vulnerable)
# self.vulnerable_commits.update(vulnerable)
else:
self.total_skipped_due_to_unimplemented_event_types += 1
raise NotImplementedError(f"Set of event types: {event_types}")
[docs]
async def main(conn: sqlite3.Connection) -> None:
cur = conn.cursor()
cur.execute("SELECT COUNT(*) FROM vulnerabilities")
((row_count,),) = cur
headers = {
"User-Agent": "osv/map_git.py +https://gitlab.softwareheritage.org/vlorentz/osv"
}
if "SWH_BEARER_TOKEN" in os.environ:
headers["Authorization"] = f"Bearer {os.environ['SWH_BEARER_TOKEN']}"
init_db(conn)
options = [
# makes the overall script over 5 times faster
(grpc.experimental.ChannelOptions.SingleThreadedUnaryStream, 1),
]
async with (
grpc.aio.insecure_channel(GRAPH_GRPC_URL, options=options) as channel,
aiohttp.ClientSession(headers=headers) as session,
):
stub = swhgraph_grpc.TraversalServiceStub(channel)
origin_denormalizer = OriginDenormalizer(stub=stub, session=session)
await origin_denormalizer.load()
root_revisions_cache = Cache(
Path("./root_revisions.json"),
dict[tuple[str, ...], list[str]](),
json.dumps,
json.loads,
)
await root_revisions_cache.load()
proc = DocProcessor(
origin_denormalizer=origin_denormalizer,
stub=stub,
conn=conn,
session=session,
root_revisions_cache=root_revisions_cache,
)
cur.execute(
"SELECT filename, doc FROM vulnerabilities WHERE random() % 100 == 0"
)
async for x in tqdm.asyncio.tqdm(
concurrent_async_map(
proc.process_docs, cur, max_concurrency=500, buffer=10000
),
total=row_count,
desc="Reading",
):
pass
cur.execute("SELECT COUNT(*) FROM unknown_swhids")
((unknown_swhids_count,),) = cur
print(f"Total unknown commits: {proc.total_unknown_commits}")
print(f"Unique unknown commits: {unknown_swhids_count}")
print(f"Total vulnerable commits×vulnerabilities: {proc.total_vulnerable}")
# print(f"Total vulnerable commits: {len(proc.vulnerable_commits)}")
print(f"Total successes: {proc.total_successes}")
print(
f"Total ranges skipped due to unimplemented event types: {proc.total_skipped_due_to_unimplemented_event_types}"
)
print(
f"Total ranges skipped due to unknown commits: {proc.total_skipped_due_to_unknown_commit}"
)
print(
f"Total ranges skipped due to invalid commits: {proc.total_skipped_due_to_invalid_commit}"
)
if __name__ == "__main__":
with sqlite3.connect("all.sqlite") as conn:
asyncio.run(main(conn))
