# 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
from collections import defaultdict
import dataclasses
import json
import os
import sqlite3
import urllib.parse
import aiohttp
from google.protobuf.field_mask_pb2 import FieldMask
import grpc.aio
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 concurrent_async_map
GRAPH_GRPC_URL = "localhost:50091"
SWH_API_SEMAPHORE = asyncio.Semaphore(10)
"""Prevents more than 10 concurrent requests to the SWH API"""
[docs]
class InvalidCommitId(ValueError):
pass
[docs]
async def is_commit_in_graph(
stub: swhgraph_grpc.TraversalServiceStub, commit_id: str
) -> bool:
try:
await stub.GetNode(
swhgraph.GetNodeRequest(
swhid=f"swh:1:rev:{commit_id}", mask=FieldMask(paths=["swhid"])
)
)
except grpc.aio.AioRpcError as e:
if e.code() == grpc.StatusCode.NOT_FOUND:
return False
elif e.code() == grpc.StatusCode.INVALID_ARGUMENT:
raise InvalidCommitId(commit_id)
else:
raise
else:
return True
[docs]
@dataclasses.dataclass
class DocProcessor:
stub: swhgraph_grpc.TraversalServiceStub
session: aiohttp.ClientSession
origin_denormalizer: OriginDenormalizer
found: set[str] = dataclasses.field(default_factory=set)
unmapped: set[str] = dataclasses.field(default_factory=set)
not_found: set[str] = dataclasses.field(default_factory=set)
total_found = 0
total_unmapped = 0
total_not_found = 0
found_git_repos: set[str] = dataclasses.field(default_factory=set)
not_found_git_repos: set[str] = dataclasses.field(default_factory=set)
not_found_git_repos_by_domain: defaultdict[str, int] = dataclasses.field(
default_factory=lambda: defaultdict(int)
)
total_found_git_repos = 0
total_not_found_git_repos = 0
found_git_commits: set[str] = dataclasses.field(default_factory=set)
found_git_commits_not_in_graph: set[str] = dataclasses.field(default_factory=set)
not_found_git_commits: set[str] = dataclasses.field(default_factory=set)
not_found_git_commits_by_domain: defaultdict[str, int] = dataclasses.field(
default_factory=lambda: defaultdict(int)
)
not_found_git_commits_by_origin: defaultdict[str, int] = dataclasses.field(
default_factory=lambda: defaultdict(int)
)
total_found_git_commits = 0
total_found_git_commits_not_in_graph = 0
total_not_found_git_commits = 0
invalid_git_commit_ids: set[str] = dataclasses.field(default_factory=set)
total_invalid_git_commit_ids = 0
[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":
await self.process_git_range(range_)
[docs]
async def process_git_range(self, range_: dict) -> None:
url = range_["repo"]
url = await self.origin_denormalizer.denormalize_origin_url(url)
if url:
self.found_git_repos.add(url)
self.total_found_git_repos += 1
else:
url = range_["repo"]
if url not in self.not_found_git_repos:
netloc = urllib.parse.urlparse(url).netloc
self.not_found_git_repos_by_domain[netloc] += 1
# print(f'"{url}","git"')
self.not_found_git_repos.add(url)
self.total_not_found_git_repos += 1
for event in range_["events"]:
commit_ids = set(event.values())
for commit_id in commit_ids:
if commit_id == "0":
# used as a placeholder when the commit that introduced a vuln is not known,
# and should be interpreted as meaning every commit since the beginning is
# affected
continue
await self._process_commit(url, commit_id)
async def _process_commit(self, url: str, commit_id: str) -> None:
netloc = urllib.parse.urlparse(url).netloc
try:
commit_id_in_graph = await is_commit_in_graph(self.stub, commit_id)
except InvalidCommitId:
self.invalid_git_commit_ids.add(commit_id)
self.total_invalid_git_commit_ids += 1
return
if commit_id_in_graph:
self.found_git_commits.add(commit_id)
self.total_found_git_commits += 1
return
if commit_id in self.found_git_commits_not_in_graph:
# save an API request, we already sent the same request before
found_in_archive = True
elif commit_id in self.not_found_git_commits:
# ditto
found_in_archive = False
else:
for _ in range(10):
async with SWH_API_SEMAPHORE:
async with self.session.head(
f"https://archive.softwareheritage.org/api/1/revision/{commit_id}/"
) as resp:
if resp.status == 200:
found_in_archive = True
break
elif resp.status == 404:
found_in_archive = False
break
elif 500 <= resp.status < 600:
print(f"Server error {resp.status}: " + await resp.text())
print("Sleeping 10s")
else:
raise Exception(
f"Unexpected status code: {resp.status} "
+ await resp.text()
)
await asyncio.sleep(10)
else:
raise Exception(
f"https://archive.softwareheritage.org/api/1/revision/{commit_id}/ "
f"keeps returning 5xx errors"
)
if found_in_archive:
self.found_git_commits_not_in_graph.add(commit_id)
self.total_found_git_commits_not_in_graph += 1
else:
if commit_id not in self.not_found_git_commits:
self.not_found_git_commits_by_domain[netloc] += 1
self.not_found_git_commits_by_origin[url] += 1
# print(f'"{url}","git"')
self.not_found_git_commits.add(url)
self.total_not_found_git_commits += 1
[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']}"
async with (
grpc.aio.insecure_channel(GRAPH_GRPC_URL) as channel,
aiohttp.ClientSession(headers=headers) as session,
):
stub = swhgraph_grpc.TraversalServiceStub(channel)
origin_denormalizer = OriginDenormalizer(stub=stub, session=session)
await origin_denormalizer.load()
proc = DocProcessor(
origin_denormalizer=origin_denormalizer,
stub=stub,
session=session,
)
cur.execute("SELECT filename, doc FROM vulnerabilities")
async for x in tqdm.asyncio.tqdm(
concurrent_async_map(proc.process_docs, cur),
total=row_count,
desc="Reading",
):
pass
print()
print("| GIT repositories... | Count (absolute) | Count (unique) |")
print("| ------------------------- | ---------------- | -------------- |")
print(
f"| ... in OSV and in SWH | {proc.total_found_git_repos} | "
f"{len(proc.found_git_repos)} |"
)
print(
f"| ... in OSV and not in SWH | {proc.total_not_found_git_repos} | "
f"{len(proc.not_found_git_repos)} |"
)
print()
print(
"| Domains with GIT repositories missing from SWH | Number of missing repositories |"
)
print(
"| ---------------------------------------------- | ------------------------------ |"
)
for count, domain in sorted(
(
(count, domain)
for (domain, count) in proc.not_found_git_repos_by_domain.items()
),
reverse=True,
):
print(f"| {domain} | {count} |")
print()
print("| GIT commits... | Count (absolute) | Count (unique) |")
print("| ----------------------------- | ---------------- | -------------- |")
print(
f"| invalid (not 40 hex chars) | {proc.total_invalid_git_commit_ids} |"
f"{len(proc.invalid_git_commit_ids)} |"
)
print(
f"| ...in swh-graph | {proc.total_found_git_commits} |"
f"{len(proc.found_git_commits)} |"
)
print(
f"| ... archive but not swh-graph | "
f"{proc.total_found_git_commits_not_in_graph} |"
f"{len(proc.found_git_commits_not_in_graph)} |"
)
print(
f"| ... not in SWH | {proc.total_not_found_git_commits} |"
f"{len(proc.not_found_git_commits)} |"
)
print()
print("| Domains with GIT commits missing from SWH | Number of missing commits |")
print("| ----------------------------------------- | ------------------------- |")
for count, domain in sorted(
(
(count, domain)
for (domain, count) in proc.not_found_git_commits_by_domain.items()
),
reverse=True,
):
print(f"| {domain} | {count} |")
print()
print("| Origins with GIT commits missing from SWH | Number of missing commits |")
print("| ----------------------------------------- | ------------------------- |")
for count, domain in sorted(
(
(count, domain)
for (domain, count) in proc.not_found_git_commits_by_origin.items()
),
reverse=True,
):
print(f"| {domain} | {count} |")
if __name__ == "__main__":
with sqlite3.connect("all.sqlite") as conn:
asyncio.run(main(conn))
