swh.osv.luigi.vuln_reports module#

Luigi tasks for downloading and processing vulnerability reports#

class swh.osv.luigi.vuln_reports.DownloadNvdTarballs(*args, **kwargs)[source]#

Bases: Task

Downloads NVD tarballs from Google Cloud Storage.

vulnerabilities_dir#

Parameter whose value is a path.

In the task definition, use

class MyTask(luigi.Task):
    existing_file_path = luigi.PathParameter(exists=True)
    new_file_path = luigi.PathParameter()

    def run(self):
        # Get data from existing file
        with self.existing_file_path.open("r", encoding="utf-8") as f:
            data = f.read()

        # Output message in new file
        self.new_file_path.parent.mkdir(parents=True, exist_ok=True)
        with self.new_file_path.open("w", encoding="utf-8") as f:
            f.write("hello from a PathParameter => ")
            f.write(data)

At the command line, use

$ luigi --module my_tasks MyTask --existing-file-path <path> --new-file-path <path>

output() → LocalTarget[source]#: nvd_cve/.stamp file indicating NVD tarballs have been downloaded.

run() → None[source]#: Downloads NVD tarballs from GCS and creates a stamp file.

class swh.osv.luigi.vuln_reports.DownloadOsvZip(*args, **kwargs)[source]#

Bases: Task

Downloads the all.zip file from the OSV vulnerability feed.

vulnerabilities_dir#

Parameter whose value is a path.

In the task definition, use

class MyTask(luigi.Task):
    existing_file_path = luigi.PathParameter(exists=True)
    new_file_path = luigi.PathParameter()

    def run(self):
        # Get data from existing file
        with self.existing_file_path.open("r", encoding="utf-8") as f:
            data = f.read()

        # Output message in new file
        self.new_file_path.parent.mkdir(parents=True, exist_ok=True)
        with self.new_file_path.open("w", encoding="utf-8") as f:
            f.write("hello from a PathParameter => ")
            f.write(data)

At the command line, use

$ luigi --module my_tasks MyTask --existing-file-path <path> --new-file-path <path>

output() → LocalTarget[source]#: all.zip file containing all OSV vulnerabilities.

run() → None[source]#: Downloads all.zip from the OSV vulnerability feed.

class swh.osv.luigi.vuln_reports.DownloadVulnerabilities(*args, **kwargs)[source]#

Bases: Task

Downloads vulnerability reports and imports them into a SQLite database.

local_graph_path#

Parameter whose value is a path.

In the task definition, use

class MyTask(luigi.Task):
    existing_file_path = luigi.PathParameter(exists=True)
    new_file_path = luigi.PathParameter()

    def run(self):
        # Get data from existing file
        with self.existing_file_path.open("r", encoding="utf-8") as f:
            data = f.read()

        # Output message in new file
        self.new_file_path.parent.mkdir(parents=True, exist_ok=True)
        with self.new_file_path.open("w", encoding="utf-8") as f:
            f.write("hello from a PathParameter => ")
            f.write(data)

At the command line, use

$ luigi --module my_tasks MyTask --existing-file-path <path> --new-file-path <path>

graph_name#: Parameter whose value is a str.

vulnerabilities_dir#

Parameter whose value is a path.

In the task definition, use

class MyTask(luigi.Task):
    existing_file_path = luigi.PathParameter(exists=True)
    new_file_path = luigi.PathParameter()

    def run(self):
        # Get data from existing file
        with self.existing_file_path.open("r", encoding="utf-8") as f:
            data = f.read()

        # Output message in new file
        self.new_file_path.parent.mkdir(parents=True, exist_ok=True)
        with self.new_file_path.open("w", encoding="utf-8") as f:
            f.write("hello from a PathParameter => ")
            f.write(data)

At the command line, use

$ luigi --module my_tasks MyTask --existing-file-path <path> --new-file-path <path>

requires() → Dict[str, Task][source]#: Returns instances of swh.graph.luigi.compressed_graph.LocalGraph, DownloadNvdTarballs and DownloadOsvZip.

output() → LocalTarget[source]#: all.sqlite file containing all vulnerability data.

run() → None[source]#: Downloads vulnerability reports and imports them into a SQLite database.

class swh.osv.luigi.vuln_reports.UploadVulnerabilitiesSqliteToS3(*args, **kwargs)[source]#

Bases: _BaseUploadToS3

Uploads the vulnerabilities SQLite database to S3.

requires() → Dict[str, Task][source]#: Returns an instance of DownloadVulnerabilities.

output() → Target[source]#: S3 target for the uploaded all.sqlite file.