Skip to content

Commit 14ffcb2

Browse files
ziadhanyziadhany
committed
Add Support for the EPSS History Pipeline Importer
Signed-off-by: ziad hany <ziadhany2016@gmail.com>
1 parent adc81cb commit 14ffcb2

7 files changed

Lines changed: 626 additions & 33 deletions

File tree

‎vulnerabilities/importers/__init__.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
from vulnerabilities.pipelines.v2_importers import (
2121
elixir_security_importer as elixir_security_importer_v2,
2222
)
23+
from vulnerabilities.pipelines.v2_importers import epss_history_importer_v2
2324
from vulnerabilities.pipelines.v2_importers import epss_importer_v2
2425
from vulnerabilities.pipelines.v2_importers import fireeye_importer_v2
2526
from vulnerabilities.pipelines.v2_importers import gentoo_importer as gentoo_importer_v2
@@ -78,6 +79,7 @@
7879
project_kb_msr2019_importer_v2.ProjectKBMSR2019Pipeline,
7980
ruby_importer_v2.RubyImporterPipeline,
8081
epss_importer_v2.EPSSImporterPipeline,
82+
epss_history_importer_v2.EPSSImporterHistoryPipeline,
8183
gentoo_importer_v2.GentooImporterPipeline,
8284
nginx_importer_v2.NginxImporterPipeline,
8385
debian_importer_v2.DebianImporterPipeline,

‎vulnerabilities/pipelines/v2_importers/epss_history_importer_v2.py‎

Lines changed: 92 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,92 @@
1+
# Copyright (c) nexB Inc. and others. All rights reserved.
2+
# VulnerableCode is a trademark of nexB Inc.
3+
# SPDX-License-Identifier: Apache-2.0
4+
# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
5+
# See https://github.com/aboutcode-org/vulnerablecode for support or download.
6+
# See https://aboutcode.org for more information about nexB OSS projects.
7+
#
8+
9+
import gzip
10+
from pathlib import Path
11+
from typing import Iterable
12+
13+
from aboutcode.pipeline import LoopProgress
14+
from fetchcode.vcs import fetch_via_vcs
15+
16+
from vulnerabilities.importer import AdvisoryDataV2
17+
from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2
18+
from vulnerabilities.pipelines.v2_importers.epss_importer_v2 import parse_epss_advisories
19+
20+
21+
class EPSSImporterHistoryPipeline(VulnerableCodeBaseImporterPipelineV2):
22+
"""Exploit Prediction Scoring System (EPSS) History Importer"""
23+
24+
pipeline_id = "epss_importer_v2"
25+
spdx_license_expression = "unknown"
26+
importer_name = "EPSS History Importer"
27+
datasource_id = "epss"
28+
repo_url = "https://github.com/empiricalsec/epss_scores"
29+
30+
exclude_from_package_todo = True
31+
run_once = True
32+
precedence = 200
33+
34+
@classmethod
35+
def steps(cls):
36+
return (
37+
cls.clone,
38+
cls.collect_and_store_advisories,
39+
cls.clean_up,
40+
)
41+
42+
def clone(self):
43+
self.log(f"Cloning `{self.repo_url}`")
44+
self.vcs_response = fetch_via_vcs(f"git+{self.repo_url}")
45+
46+
def advisories_count(self) -> int:
47+
advisory_dir = Path(self.vcs_response.dest_dir)
48+
return sum(1 for f in advisory_dir.rglob("*.csv.gz") if "beta_scores" not in f.parts)
49+
50+
def collect_advisories(self) -> Iterable[AdvisoryDataV2]:
51+
advisory_dir = Path(self.vcs_response.dest_dir)
52+
self.log(f"Scanning for EPSS CSV.gz files in: {advisory_dir}")
53+
54+
epss_files = sorted(
55+
(f for f in advisory_dir.rglob("*.csv.gz") if "beta_scores" not in f.parts),
56+
key=lambda f: f.name,
57+
)
58+
self.log(f"Found {len(epss_files)} EPSS files to process.")
59+
60+
if epss_files:
61+
self.log(f"Processing EPSS data from {epss_files[0].name} " f"to {epss_files[-1].name}")
62+
63+
progress = LoopProgress(
64+
total_iterations=len(epss_files),
65+
logger=self.log,
66+
)
67+
68+
for file_path in progress.iter(epss_files):
69+
relative_path = file_path.relative_to(advisory_dir)
70+
advisory_url = f"{self.repo_url}/blob/main/{relative_path.as_posix()}"
71+
72+
try:
73+
with gzip.open(file_path, mode="rt", encoding="utf-8") as f:
74+
lines = f.readlines()
75+
except (OSError, gzip.BadGzipFile) as e:
76+
self.log(f"Failed to read {file_path}: {e}. Skipping this file.")
77+
continue
78+
79+
yield from parse_epss_advisories(
80+
lines=lines, advisory_url=advisory_url, logger=self.log
81+
)
82+
83+
self.log(f"Finished processing all {len(epss_files)} EPSS files.")
84+
85+
def clean_up(self):
86+
if getattr(self, "vcs_response", None):
87+
self.vcs_response.delete()
88+
self.log("Successfully removed cloned EPSS repository")
89+
90+
def on_failure(self):
91+
self.log("EPSS importer pipeline failed, running cleanup")
92+
self.clean_up()

‎vulnerabilities/pipelines/v2_importers/epss_importer_v2.py‎

Lines changed: 43 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -53,38 +53,48 @@ def fetch_db(self):
5353

5454
def collect_advisories(self) -> Iterable[AdvisoryDataV2]:
5555
if not self.lines:
56-
logger.error("No EPSS data loaded")
56+
self.log("No EPSS data loaded")
5757
raise ValueError("EPSS data is empty")
5858

59-
epss_reader = csv.reader(self.lines)
60-
model_version, score_date = next(
61-
epss_reader
62-
) # score_date='score_date:2024-05-19T00:00:00+0000'
63-
published_at = datetime.strptime(score_date[11::], "%Y-%m-%dT%H:%M:%S%z")
64-
65-
next(epss_reader) # skip the header row
66-
for epss_row in epss_reader:
67-
cve, score, percentile = epss_row
68-
69-
if not cve or not score or not percentile:
70-
logger.error(f"Invalid epss row: {epss_row}")
71-
continue
72-
73-
severity = VulnerabilitySeverity(
74-
system=severity_systems.EPSS,
75-
value=score,
76-
scoring_elements=percentile,
77-
published_at=published_at,
78-
)
79-
80-
references = ReferenceV2(
81-
url=f"https://api.first.org/data/v1/epss?cve={cve}",
82-
)
83-
84-
yield AdvisoryDataV2(
85-
advisory_id=cve,
86-
severities=[severity],
87-
references=[references],
88-
url=self.advisory_url,
89-
original_advisory_text=",".join(epss_row),
90-
)
59+
yield from parse_epss_advisories(
60+
lines=self.lines, advisory_url=self.advisory_url, logger=self.log
61+
)
62+
63+
64+
def parse_epss_advisories(
65+
lines,
66+
advisory_url,
67+
logger,
68+
) -> Iterable[AdvisoryDataV2]:
69+
epss_reader = csv.reader(lines)
70+
model_version, score_date = next(
71+
epss_reader
72+
) # score_date='score_date:2024-05-19T00:00:00+0000'
73+
published_at = datetime.strptime(score_date[11::], "%Y-%m-%dT%H:%M:%S%z")
74+
75+
next(epss_reader) # skip the header row
76+
for epss_row in epss_reader:
77+
cve, score, percentile = epss_row
78+
79+
if not cve or not score or not percentile:
80+
logger(f"Invalid epss row: {epss_row}")
81+
continue
82+
83+
severity = VulnerabilitySeverity(
84+
system=severity_systems.EPSS,
85+
value=score,
86+
scoring_elements=percentile,
87+
published_at=published_at,
88+
)
89+
90+
references = ReferenceV2(
91+
url=f"https://api.first.org/data/v1/epss?cve={cve}",
92+
)
93+
94+
yield AdvisoryDataV2(
95+
advisory_id=cve,
96+
severities=[severity],
97+
references=[references],
98+
url=advisory_url,
99+
original_advisory_text=",".join(epss_row),
100+
)

‎vulnerabilities/tests/pipelines/v2_importers/test_epss_history.py‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
#
2+
# Copyright (c) nexB Inc. and others. All rights reserved.
3+
# VulnerableCode is a trademark of nexB Inc.
4+
# SPDX-License-Identifier: Apache-2.0
5+
# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
6+
# See https://github.com/aboutcode-org/vulnerablecode for support or download.
7+
# See https://aboutcode.org for more information about nexB OSS projects.
8+
#
9+
10+
from pathlib import Path
11+
from unittest.mock import MagicMock
12+
13+
import pytest
14+
15+
from vulnerabilities.pipelines.v2_importers.epss_history_importer_v2 import (
16+
EPSSImporterHistoryPipeline,
17+
)
18+
from vulnerabilities.tests import util_tests
19+
20+
TEST_DATA = Path(__file__).parent.parent.parent / "test_data" / "epss_history"
21+
22+
TEST_CVE_FILES = [
23+
TEST_DATA / "2026/epss_scores-2026-01-01.csv.gz",
24+
TEST_DATA / "2025/epss_scores-2025-12-01.csv.gz",
25+
]
26+
27+
28+
@pytest.mark.django_db
29+
def test_epss_advisories_history_pipeline():
30+
pipeline = EPSSImporterHistoryPipeline()
31+
pipeline.vcs_response = MagicMock()
32+
pipeline.vcs_response.dest_dir = str(TEST_DATA)
33+
results = list(pipeline.collect_advisories())
34+
35+
result_dicts = [adv.to_dict() for adv in results]
36+
expected_file = Path(TEST_DATA / "epss-expected.json")
37+
util_tests.check_results_against_json(result_dicts, expected_file)

‎vulnerabilities/tests/test_data/epss_history/2025/epss_scores-2025-01-01.csv.gz‎

235 Bytes
Binary file not shown.

‎vulnerabilities/tests/test_data/epss_history/2026/epss_scores-2026-01-01.csv.gz‎

232 Bytes
Binary file not shown.

0 commit comments

Comments
 (0)