Resolve migration conflict & add weighted_severity, exploitability to api_v2

Signed-off-by: ziad hany <[email protected]>

Author: ziadhany

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/0078_vulnerability_exploitability_and_more.py

@@ -0,0 +1,43 @@
+# Generated by Django 4.2.16 on 2024-11-16 20:41
+
+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 \n        by malicious actors to compromise systems, applications, or networks. \n        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/migrations/0082_vulnerability_exploitability_and_more.py

@@ -243,6 +243,34 @@ 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 +700,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/models.py

@@ -34,10 +34,11 @@ def steps(cls):
         )
 
     def compute_and_store_vulnerability_risk_score(self):
-        affected_vulnerabilities = (
-            Vulnerability.objects.filter(affectedbypackagerelatedvulnerability__isnull=False)
-            .prefetch_related("references")
-            .only("references", "exploits")
+        affected_vulnerabilities = Vulnerability.objects.filter(
+            affectedbypackagerelatedvulnerability__isnull=False
+        ).prefetch_related(
+            "references",
+            "exploits",
         )
 
         self.log(
@@ -51,8 +52,8 @@ def compute_and_store_vulnerability_risk_score(self):
         batch_size = 5000
 
         for vulnerability in progress.iter(affected_vulnerabilities.paginated()):
-            references = vulnerability.references
-            severities = vulnerability.severities.select_related("reference")
+            severities = vulnerability.severities.all()
+            references = vulnerability.references.all()
 
             (
                 vulnerability.weighted_severity,
@@ -76,17 +77,8 @@ def compute_and_store_vulnerability_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",
-            )
-        ).distinct()
-
-        affected_packages = Package.objects.filter(
-            affected_by_vulnerabilities__isnull=False
-        ).distinct()
+            Package.objects.filter(affected_by_vulnerabilities__isnull=False).only("id").distinct()
+        )
 
         self.log(f"Calculating risk for {affected_packages.count():,d} affected package records")
 

vulnerabilities/pipelines/compute_package_risk.py

@@ -6,24 +6,22 @@
 # See https://github.com/aboutcode-org/vulnerablecode for support or download.
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
-
-
+from typing import List
 from urllib.parse import urlparse
 
 from django.db.models import Prefetch
 
 from vulnerabilities.models import AffectedByPackageRelatedVulnerability
-from vulnerabilities.models import Exploit
-from vulnerabilities.models import Package
 from vulnerabilities.models import Vulnerability
 from vulnerabilities.models import VulnerabilityReference
+from vulnerabilities.models import VulnerabilitySeverity
 from vulnerabilities.severity_systems import EPSS
 from vulnerabilities.weight_config import WEIGHT_CONFIG
 
 DEFAULT_WEIGHT = 5
 
 
-def get_weighted_severity(severities):
+def get_weighted_severity(severities: List[VulnerabilitySeverity]):
     """
     Weighted Severity is the maximum value obtained when each Severity is multiplied
     by its associated Weight/10.
@@ -57,7 +55,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):
@@ -99,12 +99,8 @@ def compute_vulnerability_risk_factors(references, severities, exploits):
 
     Risk = min(weighted severity * exploitability, 10)
     """
-    severities = severities.all()
-    exploits = exploits.all()
-    reference = references.all()
-
     weighted_severity = get_weighted_severity(severities)
-    exploitability = get_exploitability_level(exploits, reference, severities)
+    exploitability = get_exploitability_level(exploits, references, severities)
     return weighted_severity, exploitability
 
 
@@ -130,4 +126,4 @@ def compute_package_risk(package):
     if not result:
         return
 
-    return f"{max(result):.2f}"
+    return round(max(result), 1)

vulnerabilities/risk.py

@@ -131,7 +131,7 @@ def test_exploitability_level(
 @pytest.mark.django_db
 def test_get_weighted_severity(vulnerability):
     severities = vulnerability.severities.all()
-    assert get_weighted_severity(severities) == 6.210000000000001
+    assert get_weighted_severity(severities) == 6.2
 
     severity2 = VulnerabilitySeverity.objects.create(
         url="https://security-tracker.debian.org/tracker/CVE-2019-13057",
@@ -146,21 +146,17 @@ def test_get_weighted_severity(vulnerability):
 
 @pytest.mark.django_db
 def test_compute_vulnerability_risk_factors(vulnerability):
-    assert compute_vulnerability_risk_factors(
-        vulnerability.references, vulnerability.severities, vulnerability.exploits
-    ) == (6.210000000000001, 2)
-    assert compute_vulnerability_risk_factors(
-        vulnerability.references, vulnerability.severities, None
-    ) == (
-        6.210000000000001,
-        0.5,
-    )
-    assert compute_vulnerability_risk_factors(
-        vulnerability.references, None, vulnerability.exploits
-    ) == (
-        0,
+    severities = vulnerability.severities.all()
+    references = vulnerability.references.all()
+
+    assert compute_vulnerability_risk_factors(references, severities, vulnerability.exploits) == (
+        6.2,
         2,
     )
+
+    assert compute_vulnerability_risk_factors(references, severities, None) == (6.2, 0.5)
+    assert compute_vulnerability_risk_factors(references, None, vulnerability.exploits) == (0, 2)
+
     assert compute_vulnerability_risk_factors(None, None, None) == (0, 0.5)
 
 
@@ -169,15 +165,15 @@ def test_get_vulnerability_risk_score(vulnerability):
     vulnerability.weighted_severity = 6.0
     vulnerability.exploitability = 2
 
-    assert vulnerability.risk_score == "10"  # max risk_score can be reached
+    assert vulnerability.risk_score == 10.0  # max risk_score can be reached
 
     vulnerability.weighted_severity = 6
     vulnerability.exploitability = 0.5
-    assert vulnerability.risk_score == "3"
+    assert vulnerability.risk_score == 3.0
 
     vulnerability.weighted_severity = 5.6
     vulnerability.exploitability = 0.5
-    assert vulnerability.risk_score == "2.8"
+    assert vulnerability.risk_score == 2.8
 
     vulnerability.weighted_severity = None
     vulnerability.exploitability = 0.5