Add Support for the EPSS History Pipeline Importer

vulnerabilities/importers/__init__.py

@@ -20,6 +20,7 @@
 from vulnerabilities.pipelines.v2_importers import (
     elixir_security_importer as elixir_security_importer_v2,
 )
+from vulnerabilities.pipelines.v2_importers import epss_history_importer_v2
 from vulnerabilities.pipelines.v2_importers import epss_importer_v2
 from vulnerabilities.pipelines.v2_importers import fireeye_importer_v2
 from vulnerabilities.pipelines.v2_importers import gentoo_importer as gentoo_importer_v2
@@ -78,6 +79,7 @@
         project_kb_msr2019_importer_v2.ProjectKBMSR2019Pipeline,
         ruby_importer_v2.RubyImporterPipeline,
         epss_importer_v2.EPSSImporterPipeline,
+        epss_history_importer_v2.EPSSImporterHistoryPipeline,
         gentoo_importer_v2.GentooImporterPipeline,
         nginx_importer_v2.NginxImporterPipeline,
         debian_importer_v2.DebianImporterPipeline,

vulnerabilities/pipelines/v2_importers/epss_history_importer_v2.py

@@ -0,0 +1,92 @@
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/aboutcode-org/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+import gzip
+from pathlib import Path
+from typing import Iterable
+
+from aboutcode.pipeline import LoopProgress
+from fetchcode.vcs import fetch_via_vcs
+
+from vulnerabilities.importer import AdvisoryDataV2
+from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2
+from vulnerabilities.pipelines.v2_importers.epss_importer_v2 import parse_epss_advisories
+
+
+class EPSSImporterHistoryPipeline(VulnerableCodeBaseImporterPipelineV2):
+    """Exploit Prediction Scoring System (EPSS) History Importer"""
+
+    pipeline_id = "epss_importer_v2"
+    spdx_license_expression = "unknown"
+    importer_name = "EPSS History Importer"
+    datasource_id = "epss"
+    repo_url = "https://github.com/empiricalsec/epss_scores"
+
+    exclude_from_package_todo = True
+    run_once = True
+    precedence = 200
+
+    @classmethod
+    def steps(cls):
+        return (
+            cls.clone,
+            cls.collect_and_store_advisories,
+            cls.clean_up,
+        )
+
+    def clone(self):
+        self.log(f"Cloning `{self.repo_url}`")
+        self.vcs_response = fetch_via_vcs(f"git+{self.repo_url}")
+
+    def advisories_count(self) -> int:
+        advisory_dir = Path(self.vcs_response.dest_dir)
+        return sum(1 for f in advisory_dir.rglob("*.csv.gz") if "beta_scores" not in f.parts)
+
+    def collect_advisories(self) -> Iterable[AdvisoryDataV2]:
+        advisory_dir = Path(self.vcs_response.dest_dir)
+        self.log(f"Scanning for EPSS CSV.gz files in: {advisory_dir}")
+
+        epss_files = sorted(
+            (f for f in advisory_dir.rglob("*.csv.gz") if "beta_scores" not in f.parts),
+            key=lambda f: f.name,
+        )
+        self.log(f"Found {len(epss_files)} EPSS files to process.")
+
+        if epss_files:
+            self.log(f"Processing EPSS data from {epss_files[0].name} " f"to {epss_files[-1].name}")
+
+        progress = LoopProgress(
+            total_iterations=len(epss_files),
+            logger=self.log,
+        )
+
+        for file_path in progress.iter(epss_files):
+            relative_path = file_path.relative_to(advisory_dir)
+            advisory_url = f"{self.repo_url}/blob/main/{relative_path.as_posix()}"
+
+            try:
+                with gzip.open(file_path, mode="rt", encoding="utf-8") as f:
+                    lines = f.readlines()
+            except (OSError, gzip.BadGzipFile) as e:
+                self.log(f"Failed to read {file_path}: {e}. Skipping this file.")
+                continue
+
+            yield from parse_epss_advisories(
+                lines=lines, advisory_url=advisory_url, logger=self.log
+            )
+
+        self.log(f"Finished processing all {len(epss_files)} EPSS files.")
+
+    def clean_up(self):
+        if getattr(self, "vcs_response", None):
+            self.vcs_response.delete()
+            self.log("Successfully removed cloned EPSS repository")
+
+    def on_failure(self):
+        self.log("EPSS importer pipeline failed, running cleanup")
+        self.clean_up()

vulnerabilities/pipelines/v2_importers/epss_importer_v2.py

@@ -53,38 +53,48 @@ def fetch_db(self):
 
     def collect_advisories(self) -> Iterable[AdvisoryDataV2]:
         if not self.lines:
-            logger.error("No EPSS data loaded")
+            self.log("No EPSS data loaded")
             raise ValueError("EPSS data is empty")
 
-        epss_reader = csv.reader(self.lines)
-        model_version, score_date = next(
-            epss_reader
-        )  # score_date='score_date:2024-05-19T00:00:00+0000'
-        published_at = datetime.strptime(score_date[11::], "%Y-%m-%dT%H:%M:%S%z")
-
-        next(epss_reader)  # skip the header row
-        for epss_row in epss_reader:
-            cve, score, percentile = epss_row
-
-            if not cve or not score or not percentile:
-                logger.error(f"Invalid epss row: {epss_row}")
-                continue
-
-            severity = VulnerabilitySeverity(
-                system=severity_systems.EPSS,
-                value=score,
-                scoring_elements=percentile,
-                published_at=published_at,
-            )
-
-            references = ReferenceV2(
-                url=f"https://api.first.org/data/v1/epss?cve={cve}",
-            )
-
-            yield AdvisoryDataV2(
-                advisory_id=cve,
-                severities=[severity],
-                references=[references],
-                url=self.advisory_url,
-                original_advisory_text=",".join(epss_row),
-            )
+        yield from parse_epss_advisories(
+            lines=self.lines, advisory_url=self.advisory_url, logger=self.log
+        )
+
+
+def parse_epss_advisories(
+    lines,
+    advisory_url,
+    logger,
+) -> Iterable[AdvisoryDataV2]:
+    epss_reader = csv.reader(lines)
+    model_version, score_date = next(
+        epss_reader
+    )  # score_date='score_date:2024-05-19T00:00:00+0000'
+    published_at = datetime.strptime(score_date[11::], "%Y-%m-%dT%H:%M:%S%z")
+
+    next(epss_reader)  # skip the header row
+    for epss_row in epss_reader:
+        cve, score, percentile = epss_row
+
+        if not cve or not score or not percentile:
+            logger(f"Invalid epss row: {epss_row}")
+            continue
+
+        severity = VulnerabilitySeverity(
+            system=severity_systems.EPSS,
+            value=score,
+            scoring_elements=percentile,
+            published_at=published_at,
+        )
+
+        references = ReferenceV2(
+            url=f"https://api.first.org/data/v1/epss?cve={cve}",
+        )
+
+        yield AdvisoryDataV2(
+            advisory_id=cve,
+            severities=[severity],
+            references=[references],
+            url=advisory_url,
+            original_advisory_text=",".join(epss_row),
+        )

vulnerabilities/tests/pipelines/v2_importers/test_epss_history.py

@@ -0,0 +1,37 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/aboutcode-org/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+from vulnerabilities.pipelines.v2_importers.epss_history_importer_v2 import (
+    EPSSImporterHistoryPipeline,
+)
+from vulnerabilities.tests import util_tests
+
+TEST_DATA = Path(__file__).parent.parent.parent / "test_data" / "epss_history"
+
+TEST_CVE_FILES = [
+    TEST_DATA / "2026/epss_scores-2026-01-01.csv.gz",
+    TEST_DATA / "2025/epss_scores-2025-12-01.csv.gz",
+]
+
+
+@pytest.mark.django_db
+def test_epss_advisories_history_pipeline():
+    pipeline = EPSSImporterHistoryPipeline()
+    pipeline.vcs_response = MagicMock()
+    pipeline.vcs_response.dest_dir = str(TEST_DATA)
+    results = list(pipeline.collect_advisories())
+
+    result_dicts = [adv.to_dict() for adv in results]
+    expected_file = Path(TEST_DATA / "epss-expected.json")
+    util_tests.check_results_against_json(result_dicts, expected_file)