# Copyright (C) 2026 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information
"""
Luigi tasks for building the vulnerability-to-commit mapping
============================================================
"""
from os import PathLike
from pathlib import Path
from typing import Dict, Iterator, Tuple
import luigi
from swh.graph.luigi.compressed_graph import LocalGraph
from swh.osv.luigi import DownloadVulnerabilities
from swh.osv.luigi.cherrypicks import ListCherrypicks
from swh.osv.luigi.reachability import ComputeConnectedComponents
from swh.osv.luigi.utils import _BaseUploadToS3
[docs]
class Vuln2CommitParquet(luigi.Task):
"""Creates a Parquet dataset that maps each vulnerability to the commits
that fix it."""
local_graph_path = luigi.PathParameter()
graph_name = luigi.StrParameter(default="graph")
vulnerabilities_dir = luigi.PathParameter()
include_vcs_matches = luigi.BoolParameter(default=True)
use_cherrypicks = luigi.BoolParameter(default=False)
@property
def resources(self):
"""Declares the task uses every CPU available"""
import socket
hostname = socket.getfqdn()
return {
f"{hostname}_max_cpu": 1,
}
[docs]
def requires(self) -> Dict[str, luigi.Task]:
"""Returns instances of :class:`swh.graph.luigi.compressed_graph.LocalGraph`,
:class:`swh.osv.luigi.DownloadVulnerabilities`,
:class:`swh.osv.luigi.reachability.ComputeConnectedComponents` and,
if :attr:`use_cherrypicks` is True, an instance of
:class:`swh.osv.luigi.cherrypicks.ListCherrypicks`."""
result: Dict[str, luigi.Task] = {
"graph": LocalGraph(local_graph_path=self.local_graph_path),
"sqlite": DownloadVulnerabilities(
local_graph_path=self.local_graph_path,
graph_name=self.graph_name,
vulnerabilities_dir=self.vulnerabilities_dir,
),
"subgraphwccs": ComputeConnectedComponents(
local_graph_path=self.local_graph_path,
graph_name=self.graph_name,
vulnerabilities_dir=self.vulnerabilities_dir,
include_vcs_matches=self.include_vcs_matches,
),
}
if self.use_cherrypicks:
result["cherrypicks"] = ListCherrypicks(
local_graph_path=self.local_graph_path,
graph_name=self.graph_name,
vulnerabilities_dir=self.vulnerabilities_dir,
include_vcs_matches=self.include_vcs_matches,
)
return result
[docs]
def output(self) -> dict[str, luigi.LocalTarget]:
"""Directory of Parquet files mapping vulnerability filenames to revision ids."""
dir_name = (
"vuln2commit_using_cherrypicks"
if self.use_cherrypicks
else "vuln2commit_without_cherrypicks"
)
path = self.vulnerabilities_dir / dir_name
return {
"dir": luigi.LocalTarget(path),
"stamp": luigi.LocalTarget(path / ".stamp"),
}
[docs]
def run(self) -> None:
"""Runs ``build-vuln2commit-parquet``."""
from .shell import Rust
args = [
"build-vuln2commit-parquet",
"--graph",
self.local_graph_path / self.graph_name,
"--db",
self.input()["sqlite"],
"--subgraphwccs",
self.input()["subgraphwccs"],
"--out-dir",
self.output()["dir"].path,
]
if self.use_cherrypicks:
cherrypicks_dir = Path(self.input()["cherrypicks"].path)
args += [
"--cherrypicks2originals",
cherrypicks_dir / "cherrypicks2originals",
"--originals2cherrypicks",
cherrypicks_dir / "originals2cherrypicks",
]
# fmt: off
Rust(*args).run()
# fmt: on
Path(self.output()["stamp"].path).touch()
[docs]
class UploadVuln2CommitParquetToS3(_BaseUploadToS3):
"""Uploads the vuln2commit Parquet dataset to S3."""
include_vcs_matches = luigi.BoolParameter(default=True)
use_cherrypicks = luigi.BoolParameter(default=False)
[docs]
def requires(self) -> Dict[str, luigi.Task]:
"""Returns an instance of :class:`Vuln2CommitParquet`."""
return {
"vuln2commit": Vuln2CommitParquet(
local_graph_path=self.local_graph_path,
graph_name=self.graph_name,
vulnerabilities_dir=self.vulnerabilities_dir,
include_vcs_matches=self.include_vcs_matches,
use_cherrypicks=self.use_cherrypicks,
),
}
[docs]
def output(self) -> luigi.Target:
"""S3 target for the vuln2commit Parquet directory stamp."""
import luigi.contrib.s3
dir_name = (
"vuln2commit_using_cherrypicks"
if self.use_cherrypicks
else "vuln2commit_without_cherrypicks"
)
return luigi.contrib.s3.S3Target(f"{self._s3_prefix()}/{dir_name}/.stamp")
def _files_to_upload(self) -> Iterator[Tuple[PathLike, str]]:
dir_name = (
"vuln2commit_using_cherrypicks"
if self.use_cherrypicks
else "vuln2commit_without_cherrypicks"
)
local_dir = Path(self.input()["vuln2commit"]["dir"].path)
for path in sorted(local_dir.rglob("*")):
if path.is_dir():
continue
yield (
path,
f"{dir_name}/{path.relative_to(local_dir)}",
)
