# 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 Counter
import dataclasses
import json
import os
import sqlite3
import aiohttp
from bs4 import BeautifulSoup
import grpc.aio
import tqdm.asyncio
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"""
SWH_ORIGIN_URLS_BY_ECOSYSTEM = {
"CRAN": {"https://cran.r-project.org/package={package_name}"},
"GitHub Actions": {"https://github.com/{package_name}"},
"Go": {"https://pkg.go.dev/{package_name}"},
"npm": {"https://www.npmjs.com/package/{package_name}"},
"Pub": {"https://pub.dev/packages/{package_name}"},
"PyPI": {"https://pypi.org/project/{package_name}/"},
"Ubuntu": {
"deb://Ubuntu/packages/{package_name}",
"deb://Ubuntu-Legacy/packages/{package_name}",
},
}
QUERY_MAVEN = False
"""Sends requests to ``https://repo1.maven.org/maven2`` in order to check
whether origins have no sources, are missing from SWH, or are not on Maven;
and print which it is."""
DEFAULT_MAVEN_REPOSITORY = "https://repo1.maven.org/maven2"
"""
Which Maven repository to use when it is not specified in the vulnerability report
(which is true of all reports I have seen so far).
The OSV spec says ``https://repo.maven.apache.org/maven2`` but SWH archives
``https://repo1.maven.org/maven2`` instead, so let's use that instead,
because they are the same repository under the hood.
"""
[docs]
@dataclasses.dataclass
class DocProcessor:
stub: swhgraph_grpc.TraversalServiceStub
session: aiohttp.ClientSession
origin_denormalizer: OriginDenormalizer
found: set[tuple[str, str]] = dataclasses.field(default_factory=set)
unmapped: set[tuple[str, str]] = dataclasses.field(default_factory=set)
not_found: set[tuple[str, str]] = dataclasses.field(default_factory=set)
incorrect_ecosystem: set[tuple[str, str]] = dataclasses.field(default_factory=set)
confabulated: set[tuple[str, str]] = dataclasses.field(default_factory=set)
todo_mappings: set[tuple[str, str]] = dataclasses.field(default_factory=set)
total_found = 0
total_unmapped = 0
total_not_found = 0
total_incorrect_ecosystem = 0
total_confabulated = 0
total_todo = 0
[docs]
async def process_docs(self, arg):
filename, doc = arg
doc = json.loads(doc)
for affected in doc.get("affected", []):
if "package" in affected:
await self.process_package(affected["package"], doc)
[docs]
async def process_package(self, package: dict, doc: dict) -> None:
package_name = package["name"]
match package["ecosystem"].split(":"):
case ["Debian", _]:
"""
if tuple(map(int, version.split("."))) < (7, 0):
# SWH did not archive versions older than Wheezy...
if package["name"] in (
"phpwiki",
"kernel-latest-2.4-i386",
"gtkhtml",
):
continue # could not be loaded from snapshot.debian.org
if (
package["name"]
== "kernel-patch-2.4.17-mipsel"
):
continue # this package does not seem to exist
"""
urls = {
f"deb://Debian/packages/{package_name}",
f"deb://Debian-Security/packages/{package_name}",
f"http://snapshot.debian.org/package/{package_name}/",
}
if any(
await asyncio.gather(
*[
self.origin_denormalizer.is_origin_url_in_graph(url)
for url in urls
]
)
):
self.found.add(("Debian", package_name))
self.total_found += 1
else:
self.not_found.add(("Debian", package_name))
self.total_not_found += 1
"""
assert any(
await asyncio.gather(
*[
self.origin_denormalizer.is_origin_url_in_graph(url)
for url in urls
]
)
), (
f"Could not map {affected['package']} to any origin "
f"in SWH. Tried these origin URLs: {urls}"
)"""
case ["Ubuntu", "Pro", *_] | [
"Red Hat",
"enterprise_linux" | "rhel_aus" | "rhel_eus",
*_,
]:
# Not free software, can't be archived
self.unmapped.add(("Ubuntu:Pro", package_name))
self.total_unmapped += 1
case (
["Ubuntu" as ecosystem, _]
| ["Ubuntu" as ecosystem, _, "LTS"]
| [
(
"CRAN" | "GitHub Actions" | "Go" | "npm" | "Pub" | "PyPI"
) as ecosystem
]
):
urls = SWH_ORIGIN_URLS_BY_ECOSYSTEM[ecosystem]
if any(
await asyncio.gather(
*[
self.origin_denormalizer.denormalize_origin_url(
url.format(package_name=package_name)
)
for url in urls
]
)
):
self.found.add((ecosystem, package_name))
self.total_found += 1
elif ecosystem == "npm" and doc["id"].startswith("MAL-"):
self.confabulated.add((ecosystem, package_name))
self.total_confabulated += 1
else:
self.not_found.add((ecosystem, package_name))
self.total_not_found += 1
case ["Ubuntu" as ecosystem, _, "LTS", "for", "NVIDIA", "BlueField"]:
# variant of the Ubuntu ecosystem not allowed by the spec
pass
case ["Maven" as ecosystem, *repo]:
repository = ":".join(repo) or DEFAULT_MAVEN_REPOSITORY
group_id, artefact_id = package_name.split(":")
expected_url = (
f"{repository}/{group_id.replace('.', '/')}/{artefact_id}"
)
if await self.origin_denormalizer.denormalize_origin_url(expected_url):
self.found.add((ecosystem, package_name))
self.total_found += 1
elif group_id.startswith("org.xwiki."):
# ecosystem should be "Maven:https://maven.xwiki.org/releases",
# not "Maven"
self.incorrect_ecosystem.add((ecosystem, package_name))
self.total_incorrect_ecosystem += 1
elif group_id.startswith(
("org.jenkins-ci.", "org.jenkinsci.", "io.jenkins.")
) or group_id in ("org.jenkins-ci", "org.jenkinsci"):
# ecosystem should be "Maven:https://repo.jenkins-ci.org/releases",
# not "Maven"
self.incorrect_ecosystem.add((ecosystem, package_name))
self.total_incorrect_ecosystem += 1
elif group_id == "org.gradle":
# ecosystem should be
# "Maven:https://repo.gradle.org/ui/native/libs-releases",
# not "Maven"
self.incorrect_ecosystem.add((ecosystem, package_name))
self.total_incorrect_ecosystem += 1
else:
# query Maven to see if there is any source available for that package
if QUERY_MAVEN and not group_id.startswith("org.apache."):
# Skipping Apache because they have lots of packages with
# no sources
async with self.origin_denormalizer._session.get(
expected_url
) as r:
if 200 <= r.status < 300:
soup = BeautifulSoup(await r.read(), "html.parser")
for link in soup.find_all("a"):
href = link.get("href")
if (
not href
or href == "../"
or href.count("/") != 1
):
continue
# subdir
assert ".." not in href
version_url = f"{expected_url}/{href}"
async with self.origin_denormalizer._session.get(
version_url
) as r:
assert (
200 <= r.status < 300
), f"{version_url} returned {r.status}"
if b"sources.jar" in await r.read():
print(
f"Sources found in {version_url} but "
f"{expected_url} not in SWH"
)
break
else:
print(f"{expected_url} returned {r.status}")
self.not_found.add((ecosystem, package_name))
self.total_not_found += 1
case ["Packagist" as ecosystem]:
self.todo_mappings.add((ecosystem, package_name))
self.total_todo += 1
case ["Linux" as ecosystem]:
self.found.add((ecosystem, package_name))
self.total_found += 1
case ["SwiftURL" as ecosystem]:
# "The name is a Git URL to the source of the package"
if await self.origin_denormalizer.denormalize_origin_url(package_name):
self.found.add((ecosystem, package_name))
self.total_found += 1
else:
self.not_found.add((ecosystem, package_name))
self.total_not_found += 1
case [
(
"Alpine"
| "AlmaLinux"
| "Mageia"
| "Rocky Linux"
| "RubyGems"
| "openSUSE"
| "SUSE"
) as ecosystem,
_,
]:
# SWH does not archive these distributions yet
self.unmapped.add((ecosystem, package_name))
self.total_unmapped += 1
case ["Red Hat" as ecosystem, *_]:
# lots of variants that seem annoying to handle. SWH doesn't archive
# Red Hat yet anyway
self.unmapped.add((ecosystem, package_name))
self.total_unmapped += 1
case ["SUSE", *_]:
# variant of the SUSE ecosystem not allowed by the spec
pass
case [
(
"Android"
| "Bioconductor"
| "Bitnami"
| "Chainguard"
| "ConanCenter"
| "crates.io"
| "GHC"
| "Hackage"
| "Hex"
| "Kubernetes"
| "NuGet"
| "OSS-Fuzz"
| "Photon OS"
| "RubyGems"
| "Wolfi"
) as ecosystem
]:
# SWH does not archive these distributions yet
self.unmapped.add((ecosystem, package_name))
self.total_unmapped += 1
case ["GSD"]:
# invalid as an ecosystem (it's a database), but four vulnerabilities cite it
# as a package ecosystem anyway
pass
case ["UVI"]:
# not allowed by the spec, but there is one vulnerability citing it as
# a package ecosystem
pass
case other:
assert False, f"unknown ecosystem: {other}"
[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_packages.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
found_counts = Counter(ecosystem for (ecosystem, name) in proc.found)
todo_counts = Counter(ecosystem for (ecosystem, name) in proc.todo_mappings)
not_found_counts = Counter(ecosystem for (ecosystem, name) in proc.not_found)
incorrect_ecosystem = Counter(
ecosystem for (ecosystem, name) in proc.incorrect_ecosystem
)
confabulated_counts = Counter(ecosystem for (ecosystem, name) in proc.confabulated)
unmapped_counts = Counter(ecosystem for (ecosystem, name) in proc.unmapped)
print(
"| Packages... | Count | Count (unique) | Broken down by ecosystem |"
)
print(
"| ---------------------------------------------- | ----- | -------------- | ------------------------ |"
)
print(
f"| ... found | {proc.total_found} | {len(proc.found)} | {dict(found_counts)} | "
)
print(
f"| ... in the wrong ecosystem (*) | {proc.total_incorrect_ecosystem} | {len(proc.incorrect_ecosystem)} | {dict(incorrect_ecosystem)} |"
)
print(
f"| ... hallucinated (*) | {proc.total_confabulated} | {len(proc.confabulated)} | {dict(confabulated_counts)} |"
)
print(
f"| ... in ecosystems not supported by this script | {proc.total_todo} | {len(proc.todo_mappings)} | {dict(todo_counts)} |"
)
print(
f"| ... in ecosystems not supported by SWH | {proc.total_unmapped} |{len(proc.unmapped)} | {dict(unmapped_counts)} |"
)
print(
f"| ... not found in SWH for other reasons | {proc.total_not_found} | {len(proc.not_found)} | {dict(not_found_counts)} |"
)
if __name__ == "__main__":
with sqlite3.connect("all.sqlite") as conn:
asyncio.run(main(conn))
