Add support for storing exploitability and weighted severity

vulnerabilities/api.py

@@ -286,6 +286,9 @@ class Meta:
             "weaknesses",
             "exploits",
             "severity_range_score",
+            "exploitability",
+            "weighted_severity",
+            "risk_score",
         ]
 
 

vulnerabilities/api_v2.py

@@ -67,6 +67,9 @@ class VulnerabilityV2Serializer(serializers.ModelSerializer):
     weaknesses = WeaknessV2Serializer(many=True)
     references = VulnerabilityReferenceV2Serializer(many=True, source="vulnerabilityreference_set")
     severities = VulnerabilitySeverityV2Serializer(many=True)
+    exploitability = serializers.FloatField(read_only=True)
+    weighted_severity = serializers.FloatField(read_only=True)
+    risk_score = serializers.FloatField(read_only=True)
 
     class Meta:
         model = Vulnerability
@@ -77,6 +80,9 @@ class Meta:
             "severities",
             "weaknesses",
             "references",
+            "exploitability",
+            "weighted_severity",
+            "risk_score",
         ]
 
     def get_aliases(self, obj):

vulnerabilities/migrations/0082_vulnerability_exploitability_and_more.py

@@ -0,0 +1,43 @@
+# Generated by Django 4.2.16 on 2024-11-17 13:52
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("vulnerabilities", "0081_alter_packagechangelog_software_version_and_more"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="vulnerability",
+            name="exploitability",
+            field=models.DecimalField(
+                decimal_places=1,
+                help_text="Exploitability indicates the likelihood that a vulnerability in a software package could be used by malicious actors to compromise systems, applications, or networks. This metric is determined automatically based on the discovery of known exploits.",
+                max_digits=2,
+                null=True,
+            ),
+        ),
+        migrations.AddField(
+            model_name="vulnerability",
+            name="weighted_severity",
+            field=models.DecimalField(
+                decimal_places=1,
+                help_text="Weighted severity is the highest value calculated by multiplying each severity by its corresponding weight, divided by 10.",
+                max_digits=3,
+                null=True,
+            ),
+        ),
+        migrations.AlterField(
+            model_name="package",
+            name="risk_score",
+            field=models.DecimalField(
+                decimal_places=1,
+                help_text="Risk score between 0.00 and 10.00, where higher values indicate greater vulnerability risk for the package.",
+                max_digits=3,
+                null=True,
+            ),
+        ),
+    ]

vulnerabilities/models.py

@@ -243,6 +243,33 @@ class Vulnerability(models.Model):
         related_name="vulnerabilities",
     )
 
+    exploitability = models.DecimalField(
+        null=True,
+        max_digits=2,
+        decimal_places=1,
+        help_text="Exploitability indicates the likelihood that a vulnerability in a software package could be used by malicious actors to compromise systems, "
+        "applications, or networks. This metric is determined automatically based on the discovery of known exploits.",
+    )
+
+    weighted_severity = models.DecimalField(
+        null=True,
+        max_digits=3,
+        decimal_places=1,
+        help_text="Weighted severity is the highest value calculated by multiplying each severity by its corresponding weight, divided by 10.",
+    )
+
+    @property
+    def risk_score(self):
+        """
+        Risk expressed as a number ranging from 0 to 10.
+        Risk is calculated from weighted severity and exploitability values.
+        It is the maximum value of (the weighted severity multiplied by its exploitability) or 10
+        Risk = min(weighted severity * exploitability, 10)
+        """
+        if self.exploitability and self.weighted_severity:
+            risk_score = min(float(self.exploitability * self.weighted_severity), 10.0)
+            return round(risk_score, 1)
+
     objects = VulnerabilityQuerySet.as_manager()
 
     class Meta:
@@ -672,8 +699,8 @@ class Package(PackageURLMixin):
 
     risk_score = models.DecimalField(
         null=True,
-        max_digits=4,
-        decimal_places=2,
+        max_digits=3,
+        decimal_places=1,
         help_text="Risk score between 0.00 and 10.00, where higher values "
         "indicate greater vulnerability risk for the package.",
     )

vulnerabilities/pipelines/compute_package_risk.py

@@ -6,12 +6,14 @@
 # See https://github.com/aboutcode-org/vulnerablecode for support or download.
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
-
 from aboutcode.pipeline import LoopProgress
+from django.db.models import Prefetch
 
 from vulnerabilities.models import Package
+from vulnerabilities.models import Vulnerability
 from vulnerabilities.pipelines import VulnerableCodePipeline
 from vulnerabilities.risk import compute_package_risk
+from vulnerabilities.risk import compute_vulnerability_risk_factors
 
 
 class ComputePackageRiskPipeline(VulnerableCodePipeline):
@@ -26,15 +28,73 @@ class ComputePackageRiskPipeline(VulnerableCodePipeline):
 
     @classmethod
     def steps(cls):
-        return (cls.add_package_risk_score,)
+        return (
+            cls.compute_and_store_vulnerability_risk_score,
+            cls.compute_and_store_package_risk_score,
+        )
+
+    def compute_and_store_vulnerability_risk_score(self):
+        affected_vulnerabilities = (
+            Vulnerability.objects.filter(affecting_packages__isnull=False)
+            .prefetch_related(
+                "references",
+                "severities",
+                "exploits",
+            )
+            .distinct()
+        )
+
+        self.log(
+            f"Calculating risk for {affected_vulnerabilities.count():,d} vulnerability with a affected packages records"
+        )
+
+        progress = LoopProgress(total_iterations=affected_vulnerabilities.count(), logger=self.log)
+
+        updatables = []
+        updated_vulnerability_count = 0
+        batch_size = 5000
+
+        for vulnerability in progress.iter(affected_vulnerabilities.paginated(per_page=batch_size)):
+            severities = vulnerability.severities.all()
+            references = vulnerability.references.all()
+            exploits = vulnerability.exploits.all()
+
+            weighted_severity, exploitability = compute_vulnerability_risk_factors(
+                references=references,
+                severities=severities,
+                exploits=exploits,
+            )
+            vulnerability.weighted_severity = weighted_severity
+            vulnerability.exploitability = exploitability
+
+            updatables.append(vulnerability)
+
+            if len(updatables) >= batch_size:
+                updated_vulnerability_count += bulk_update(
+                    model=Vulnerability,
+                    items=updatables,
+                    fields=["weighted_severity", "exploitability"],
+                    logger=self.log,
+                )
+
+        updated_vulnerability_count += bulk_update(
+            model=Vulnerability,
+            items=updatables,
+            fields=["weighted_severity", "exploitability"],
+            logger=self.log,
+        )
+
+        self.log(
+            f"Successfully added risk score for {updated_vulnerability_count:,d} vulnerability"
+        )
 
-    def add_package_risk_score(self):
+    def compute_and_store_package_risk_score(self):
         affected_packages = (
             Package.objects.filter(affected_by_vulnerabilities__isnull=False).prefetch_related(
-                "affectedbypackagerelatedvulnerability_set__vulnerability",
-                "affectedbypackagerelatedvulnerability_set__vulnerability__references",
-                "affectedbypackagerelatedvulnerability_set__vulnerability__severities",
-                "affectedbypackagerelatedvulnerability_set__vulnerability__exploits",
+                Prefetch(
+                    "affectedbypackagerelatedvulnerability_set__vulnerability",
+                    queryset=Vulnerability.objects.only("weighted_severity", "exploitability"),
+                ),
             )
         ).distinct()
 
@@ -60,24 +120,28 @@ def add_package_risk_score(self):
             updatables.append(package)
 
             if len(updatables) >= batch_size:
-                updated_package_count += bulk_update_package_risk_score(
-                    packages=updatables,
+                updated_package_count += bulk_update(
+                    model=Package,
+                    items=updatables,
+                    fields=["risk_score"],
                     logger=self.log,
                 )
-        updated_package_count += bulk_update_package_risk_score(
-            packages=updatables,
+        updated_package_count += bulk_update(
+            model=Package,
+            items=updatables,
+            fields=["risk_score"],
             logger=self.log,
         )
         self.log(f"Successfully added risk score for {updated_package_count:,d} package")
 
 
-def bulk_update_package_risk_score(packages, logger):
-    package_count = 0
-    if packages:
+def bulk_update(model, items, fields, logger):
+    item_count = 0
+    if items:
         try:
-            Package.objects.bulk_update(objs=packages, fields=["risk_score"])
-            package_count += len(packages)
+            model.objects.bulk_update(objs=items, fields=fields)
+            item_count += len(items)
         except Exception as e:
-            logger(f"Error updating packages: {e}")
-        packages.clear()
-    return package_count
+            logger(f"Error updating {model.__name__}: {e}")
+        items.clear()
+    return item_count

vulnerabilities/risk.py

@@ -6,8 +6,6 @@
 # See https://github.com/aboutcode-org/vulnerablecode for support or download.
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
-
-
 from urllib.parse import urlparse
 
 from vulnerabilities.models import VulnerabilityReference
@@ -23,6 +21,8 @@ def get_weighted_severity(severities):
     by its associated Weight/10.
     Example of Weighted Severity: max(7*(10/10), 8*(3/10), 6*(8/10)) = 7
     """
+    if not severities:
+        return 0
 
     score_map = {
         "low": 3,
@@ -49,7 +49,9 @@ def get_weighted_severity(severities):
             vul_score_value = score_map.get(vul_score, 0) * max_weight
 
         score_list.append(vul_score_value)
-    return max(score_list) if score_list else 0
+
+    max_score = max(score_list) if score_list else 0
+    return round(max_score, 1)
 
 
 def get_exploitability_level(exploits, references, severities):
@@ -83,35 +85,30 @@ def get_exploitability_level(exploits, references, severities):
     return exploit_level
 
 
-def compute_vulnerability_risk(vulnerability):
+def compute_vulnerability_risk_factors(references, severities, exploits):
     """
     Risk may be expressed as a number ranging from 0 to 10.
     Risk is calculated from weighted severity and exploitability values.
     It is the maximum value of (the weighted severity multiplied by its exploitability) or 10
 
     Risk = min(weighted severity * exploitability, 10)
     """
-    severities = vulnerability.severities.all()
-    exploits = vulnerability.exploits.all()
-    reference = vulnerability.references.all()
-    if reference.exists() or severities.exists() or exploits.exists():
-        weighted_severity = get_weighted_severity(severities)
-        exploitability = get_exploitability_level(exploits, reference, severities)
-        return min(weighted_severity * exploitability, 10)
+    weighted_severity = get_weighted_severity(severities)
+    exploitability = get_exploitability_level(exploits, references, severities)
+    return weighted_severity, exploitability
 
 
 def compute_package_risk(package):
     """
     Calculate the risk for a package by iterating over all vulnerabilities that affects this package
     and determining the associated risk.
     """
-
     result = []
-    for package_vulnerability in package.affectedbypackagerelatedvulnerability_set.all():
-        if risk := compute_vulnerability_risk(package_vulnerability.vulnerability):
-            result.append(risk)
+    for relation in package.affectedbypackagerelatedvulnerability_set.all():
+        if risk := relation.vulnerability.risk_score:
+            result.append(float(risk))
 
     if not result:
         return
 
-    return f"{max(result):.2f}"
+    return round(max(result), 1)

vulnerabilities/templates/vulnerability_details.html

@@ -121,6 +121,38 @@
                                 <td class="two-col-left">Status</td>
                                 <td class="two-col-right">{{ status }}</td>
                             </tr>
+
+                            <tr>
+                                <td class="two-col-left"
+                                    data-tooltip="Exploitability indicates the likelihood that a vulnerability in a software package 
+                                    could be used by malicious actors to compromise systems,
+                                    applications, or networks. This metric is determined automatically based on the discovery of known exploits.">
+                                    Exploitability</td>
+                                <td class="two-col-right wrap-strings">
+                                    {{ vulnerability.exploitability }}
+                                </td>
+                            </tr>
+
+                            <tr>
+                                <td class="two-col-left"
+                                    data-tooltip="Weighted severity is the highest value calculated by multiplying each severity by its corresponding weight, divided by 10."
+                                >Weighted Severity</td>
+                                <td class="two-col-right wrap-strings">
+                                    {{ vulnerability.weighted_severity }}
+                                </td>
+                            </tr>
+
+                            <tr>
+                                <td class="two-col-left"
+                                    data-tooltip="Risk expressed as a number ranging from 0 to 10. It is calculated by multiplying
+                                    the weighted severity and exploitability values, capped at a maximum of 10.
+                                    "
+                                >Risk</td>
+                                <td class="two-col-right wrap-strings">
+                                    {{ vulnerability.risk_score }}
+                                </td>
+                            </tr>
+
                         </tbody>
                     </table>
                 </div>

vulnerabilities/tests/pipelines/test_compute_package_risk.py

@@ -6,6 +6,7 @@
 # See https://github.com/aboutcode-org/vulnerablecode for support or download.
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
+from decimal import Decimal
 
 import pytest
 
@@ -30,4 +31,4 @@ def test_simple_risk_pipeline(vulnerability):
     improver.execute()
 
     pkg = Package.objects.get(type="pypi", name="foo", version="2.3.0")
-    assert str(pkg.risk_score) == str(3.11)
+    assert pkg.risk_score == Decimal("3.1")  # max( 6.9 * 9/10 , 6.5 * 9/10 ) * .5 = 3.105