Add epss to severity scoring

Add published_at date to the Vulnerability score model.
Add EPSS importer

Signed-off-by: ziadhany <[email protected]>

Author: ziadhany

vulnerabilities/api.py

@@ -24,7 +24,6 @@
 from rest_framework.throttling import AnonRateThrottle
 from rest_framework.throttling import UserRateThrottle
 
-from vulnerabilities.models import EPSS
 from vulnerabilities.models import Alias
 from vulnerabilities.models import Package
 from vulnerabilities.models import Vulnerability
@@ -38,7 +37,14 @@
 class VulnerabilitySeveritySerializer(serializers.ModelSerializer):
     class Meta:
         model = VulnerabilitySeverity
-        fields = ["value", "scoring_system", "scoring_elements"]
+        fields = ["value", "scoring_system", "scoring_elements", "published_at"]
+
+    def to_representation(self, instance):
+        data = super().to_representation(instance)
+        published_at = data.get("published_at", None)
+        if not published_at:
+            data.pop("published_at")
+        return data
 
 
 class VulnerabilityReferenceSerializer(serializers.ModelSerializer):
@@ -168,12 +174,6 @@ def to_representation(self, instance):
         return representation
 
 
-class EPSSSerializer(serializers.ModelSerializer):
-    class Meta:
-        model = EPSS
-        fields = ["score", "percentile", "created_at", "updated_at"]
-
-
 class VulnerabilitySerializer(BaseResourceSerializer):
     fixed_packages = MinimalPackageSerializer(
         many=True, source="filtered_fixed_packages", read_only=True
@@ -182,7 +182,6 @@ class VulnerabilitySerializer(BaseResourceSerializer):
 
     references = VulnerabilityReferenceSerializer(many=True, source="vulnerabilityreference_set")
     aliases = AliasSerializer(many=True, source="alias")
-    epss = EPSSSerializer(read_only=True)
     weaknesses = WeaknessSerializer(many=True)
 
     def to_representation(self, instance):
@@ -191,10 +190,6 @@ def to_representation(self, instance):
         weaknesses = data.get("weaknesses", [])
         data["weaknesses"] = [weakness for weakness in weaknesses if weakness is not None]
 
-        epss = data.get("epss", None)
-        if not epss:
-            data.pop("epss")
-
         return data
 
     class Meta:
@@ -208,7 +203,6 @@ class Meta:
             "affected_packages",
             "references",
             "weaknesses",
-            "epss",
         ]
 
 

vulnerabilities/import_runner.py

@@ -189,6 +189,7 @@ def process_inferences(inferences: List[Inference], advisory: Advisory, improver
                     defaults={
                         "value": str(severity.value),
                         "scoring_elements": str(severity.scoring_elements),
+                        "published_at": str(severity.published_at),
                     },
                 )
                 if updated:

vulnerabilities/importer.py

@@ -52,12 +52,17 @@ class VulnerabilitySeverity:
     system: ScoringSystem
     value: str
     scoring_elements: str = ""
+    published_at: Optional[datetime.datetime] = None
 
     def to_dict(self):
+        published_at_dict = (
+            {"published_at": self.published_at.isoformat()} if self.published_at else {}
+        )
         return {
             "system": self.system.identifier,
             "value": self.value,
             "scoring_elements": self.scoring_elements,
+            **published_at_dict,
         }
 
     @classmethod
@@ -70,6 +75,7 @@ def from_dict(cls, severity: dict):
             system=SCORING_SYSTEMS[severity["system"]],
             value=severity["value"],
             scoring_elements=severity.get("scoring_elements", ""),
+            published_at=severity.get("published_at"),
         )
 
 

vulnerabilities/importers/__init__.py

@@ -15,6 +15,7 @@
 from vulnerabilities.importers import debian
 from vulnerabilities.importers import debian_oval
 from vulnerabilities.importers import elixir_security
+from vulnerabilities.importers import epss
 from vulnerabilities.importers import fireeye
 from vulnerabilities.importers import gentoo
 from vulnerabilities.importers import github
@@ -71,6 +72,7 @@
     oss_fuzz.OSSFuzzImporter,
     ruby.RubyImporter,
     github_osv.GithubOSVImporter,
+    epss.EPSSImporter,
 ]
 
 IMPORTERS_REGISTRY = {x.qualified_name: x for x in IMPORTERS_REGISTRY}

vulnerabilities/importers/epss.py

@@ -0,0 +1,67 @@
+#
+# 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/nexB/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+import csv
+import gzip
+import logging
+import urllib.request
+from datetime import datetime
+from typing import Iterable
+
+from vulnerabilities import severity_systems
+from vulnerabilities.importer import AdvisoryData
+from vulnerabilities.importer import Importer
+from vulnerabilities.importer import Reference
+from vulnerabilities.importer import VulnerabilitySeverity
+
+logger = logging.getLogger(__name__)
+
+
+class EPSSImporter(Importer):
+    """Exploit Prediction Scoring System (EPSS) Importer"""
+
+    advisory_url = "https://epss.cyentia.com/epss_scores-current.csv.gz"
+    spdx_license_expression = "unknown"
+    importer_name = "EPSS Importer"
+
+    def advisory_data(self) -> Iterable[AdvisoryData]:
+        response = urllib.request.urlopen(self.advisory_url)
+        with gzip.open(response, "rb") as f:
+            lines = [l.decode("utf-8") for l in f.readlines()]
+
+        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.error(f"Invalid epss row: {epss_row}")
+                continue
+
+            severity = VulnerabilitySeverity(
+                system=severity_systems.EPSS,
+                value=score,
+                scoring_elements=percentile,
+                published_at=published_at,
+            )
+
+            references = Reference(
+                url=f"https://api.first.org/data/v1/epss?cve={cve}",
+                severities=[severity],
+            )
+
+            yield AdvisoryData(
+                aliases=[cve],
+                references=[references],
+                url=self.advisory_url,
+            )

vulnerabilities/improvers/__init__.py

@@ -8,7 +8,6 @@
 #
 
 from vulnerabilities.improvers import valid_versions
-from vulnerabilities.improvers import vulnerability_epss
 from vulnerabilities.improvers import vulnerability_status
 
 IMPROVERS_REGISTRY = [
@@ -27,7 +26,6 @@
     valid_versions.OSSFuzzImprover,
     valid_versions.RubyImprover,
     valid_versions.GithubOSVImprover,
-    vulnerability_epss.EPSSImprover,
     vulnerability_status.VulnerabilityStatusImprover,
 ]
 

vulnerabilities/improvers/vulnerability_epss.py

@@ -0,0 +1,43 @@
+# Generated by Django 4.1.13 on 2024-05-22 21:51
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("vulnerabilities", "0056_alter_packagechangelog_software_version_and_more"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="vulnerabilityseverity",
+            name="published_at",
+            field=models.DateTimeField(
+                blank=True,
+                help_text="UTC Date of publication of the vulnerability severity",
+                null=True,
+            ),
+        ),
+        migrations.AlterField(
+            model_name="vulnerabilityseverity",
+            name="scoring_system",
+            field=models.CharField(
+                choices=[
+                    ("cvssv2", "CVSSv2 Base Score"),
+                    ("cvssv3", "CVSSv3 Base Score"),
+                    ("cvssv3.1", "CVSSv3.1 Base Score"),
+                    ("rhbs", "RedHat Bugzilla severity"),
+                    ("rhas", "RedHat Aggregate severity"),
+                    ("archlinux", "Archlinux Vulnerability Group Severity"),
+                    ("cvssv3.1_qr", "CVSSv3.1 Qualitative Severity Rating"),
+                    ("generic_textual", "Generic textual severity rating"),
+                    ("apache_httpd", "Apache Httpd Severity"),
+                    ("apache_tomcat", "Apache Tomcat Severity"),
+                    ("epss", "Exploit Prediction Scoring System"),
+                ],
+                help_text="Identifier for the scoring system used. Available choices are: cvssv2: CVSSv2 Base Score,\ncvssv3: CVSSv3 Base Score,\ncvssv3.1: CVSSv3.1 Base Score,\nrhbs: RedHat Bugzilla severity,\nrhas: RedHat Aggregate severity,\narchlinux: Archlinux Vulnerability Group Severity,\ncvssv3.1_qr: CVSSv3.1 Qualitative Severity Rating,\ngeneric_textual: Generic textual severity rating,\napache_httpd: Apache Httpd Severity,\napache_tomcat: Apache Tomcat Severity,\nepss: Exploit Prediction Scoring System ",
+                max_length=50,
+            ),
+        ),
+    ]