chore: remove enriched_cvss fields

Author: StefanFl

backend/application/core/api/serializers_observation.py

@@ -87,6 +87,7 @@ class ObservationSerializer(ModelSerializer):
     issue_tracker_issue_url = SerializerMethodField()
     assessment_needs_approval = SerializerMethodField()
     vulnerability_id_aliases = SerializerMethodField()
+    cve_found_in = SerializerMethodField()
 
     class Meta:
         model = Observation
@@ -127,6 +128,9 @@ def get_assessment_needs_approval(self, observation: Observation) -> Optional[in
     def get_vulnerability_id_aliases(self, observation: Observation) -> list[dict[str, str]]:
         return _get_vulnerability_id_aliases(observation)
 
+    def get_cve_found_in(self, observation: Observation) -> list[dict[str, str]]:
+        return _get_cve_found_in_sources(observation)
+
     def validate_product(self, product: Product) -> Product:
         if product and product.is_product_group:
             raise ValidationError("Product must not be a product group")
@@ -149,6 +153,7 @@ class ObservationListSerializer(ModelSerializer):
     origin_source_file_url = SerializerMethodField()
     origin_component_purl_namespace = SerializerMethodField()
     vulnerability_id_aliases = SerializerMethodField()
+    cve_found_in = SerializerMethodField()
 
     class Meta:
         model = Observation
@@ -176,6 +181,9 @@ def get_origin_component_purl_namespace(self, observation: Observation) -> Optio
     def get_vulnerability_id_aliases(self, observation: Observation) -> list[dict[str, str]]:
         return _get_vulnerability_id_aliases(observation)
 
+    def get_cve_found_in(self, observation: Observation) -> list[dict[str, str]]:
+        return _get_cve_found_in_sources(observation)
+
 
 def _get_origin_source_file_url(observation: Observation) -> Optional[str]:
     origin_source_file_url = None
@@ -244,6 +252,14 @@ def _get_vulnerability_id_aliases(observation: Observation) -> list[dict[str, st
     return return_list
 
 
+def _get_cve_found_in_sources(observation: Observation) -> list[dict[str, str]]:
+    sources_list = get_comma_separated_as_list(observation.cve_found_in)
+    return_list = []
+    for source in sources_list:
+        return_list.append({"source": source})
+    return return_list
+
+
 class ObservationUpdateSerializer(ModelSerializer):
     def validate(self, attrs: dict) -> dict:
         self.instance: Observation
@@ -509,6 +525,7 @@ class ObservationBulkMarkDuplicatesSerializer(Serializer):
 class NestedObservationSerializer(ModelSerializer):
     scanner_name = SerializerMethodField()
     origin_component_name_version = SerializerMethodField()
+    cve_found_in = SerializerMethodField()
 
     class Meta:
         model = Observation
@@ -520,6 +537,9 @@ def get_scanner_name(self, observation: Observation) -> str:
     def get_origin_component_name_version(self, observation: Observation) -> str:
         return get_origin_component_name_version(observation)
 
+    def get_cve_found_in(self, observation: Observation) -> list[dict[str, str]]:
+        return _get_cve_found_in_sources(observation)
+
 
 class ObservationLogSerializer(ModelSerializer):
     observation_data = ObservationSerializer(source="observation")

backend/application/core/migrations/0061_observation_cve_found_in_and_more.py

@@ -15,14 +15,4 @@ class Migration(migrations.Migration):
             name="cve_found_in",
             field=models.CharField(blank=True, max_length=255),
         ),
-        migrations.AddField(
-            model_name="observation",
-            name="enriched_cvss_score",
-            field=models.DecimalField(decimal_places=1, max_digits=3, null=True),
-        ),
-        migrations.AddField(
-            model_name="observation",
-            name="enriched_cvss_vector",
-            field=models.CharField(blank=True, max_length=255),
-        ),
     ]

backend/application/core/models.py

@@ -430,8 +430,6 @@ class Observation(Model):
     cvss3_vector = CharField(max_length=255, blank=True)
     cvss4_score = DecimalField(max_digits=3, decimal_places=1, null=True)
     cvss4_vector = CharField(max_length=255, blank=True)
-    enriched_cvss_score = DecimalField(max_digits=3, decimal_places=1, null=True)
-    enriched_cvss_vector = CharField(max_length=255, blank=True)
     cve_found_in = CharField(max_length=255, blank=True)
 
     cwe = IntegerField(null=True, validators=[MinValueValidator(1), MaxValueValidator(999999)])

backend/application/core/services/observation.py

@@ -78,9 +78,6 @@ def get_current_severity(observation: Observation) -> str:
     if observation.parser_severity:
         return observation.parser_severity
 
-    if observation.enriched_cvss_score is not None:
-        return get_cvss_severity(observation.enriched_cvss_score)
-
     if observation.cvss4_score is not None:
         return get_cvss_severity(observation.cvss4_score)
 
@@ -191,8 +188,6 @@ def normalize_cvss_vectors(observation: Observation) -> None:
         observation.cvss3_vector = ""
     if observation.cvss4_vector is None:
         observation.cvss4_vector = ""
-    if observation.enriched_cvss_vector is None:
-        observation.enriched_cvss_vector = ""
     if observation.cve_found_in is None:
         observation.cve_found_in = ""
 

backend/application/epss/migrations/0003_enriched_cvss_alter_epss_score_cve.py

@@ -13,9 +13,16 @@ class Migration(migrations.Migration):
         migrations.CreateModel(
             name="Enriched_CVSS",
             fields=[
-                ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
                 ("cve", models.CharField(max_length=255, unique=True)),
-                ("enriched_cvss_vector", models.CharField(blank=True, max_length=255)),
                 ("base_cvss_vector", models.CharField(blank=True, max_length=255)),
                 ("cisa_kev", models.BooleanField(default=False)),
                 ("vulncheck_kev", models.BooleanField(default=False)),

backend/application/epss/models.py

@@ -40,7 +40,6 @@ def load(cls) -> "EPSS_Status":
 
 class Enriched_CVSS(Model):
     cve = CharField(max_length=255, unique=True)
-    enriched_cvss_vector = CharField(max_length=255, blank=True)
     base_cvss_vector = CharField(max_length=255, blank=True)
     cisa_kev = BooleanField(default=False)
     vulncheck_kev = BooleanField(default=False)

backend/application/epss/services/cvss_bt.py

@@ -46,13 +46,9 @@ def import_cvss_bt() -> None:
         settings = Settings.load()
         if int(cve_year) <= current_year - settings.cvss_enrichment_max_age_years:
             continue
-        enriched_cvss_vector = row.get("cvss-bt_vector", "")
-        if not enriched_cvss_vector.startswith("CVSS:3") and not enriched_cvss_vector.startswith("CVSS:4"):
-            continue
 
         enriched_cvss = Enriched_CVSS(
             cve=cve,
-            enriched_cvss_vector=enriched_cvss_vector,
             base_cvss_vector=row.get("base_vector", ""),
             cisa_kev=row.get("cisa_kev", "").lower() == "true",
             vulncheck_kev=row.get("vulncheck_kev", "").lower() == "true",
@@ -97,8 +93,6 @@ def enriched_cvss_apply_observations(settings: Settings) -> None:
                 "cvss3_vector",
                 "cvss4_score",
                 "cvss4_vector",
-                "enriched_cvss_score",
-                "enriched_cvss_vector",
                 "cve_found_in",
                 "current_severity",
             ],
@@ -115,37 +109,32 @@ def apply_enriched_cvss(observation: Observation, settings: Settings) -> bool:
         if not enriched_cvss:
             return False
 
-        if observation.cvss3_vector and not enriched_cvss.enriched_cvss_vector.startswith(observation.cvss3_vector):
-            return False
-        if observation.cvss4_vector and not enriched_cvss.enriched_cvss_vector.startswith(observation.cvss4_vector):
-            return False
-
-        observation.enriched_cvss_vector = enriched_cvss.enriched_cvss_vector
-        if enriched_cvss.enriched_cvss_vector.startswith("CVSS:3"):
-            cvss = CVSS3(observation.enriched_cvss_vector)
-            observation.enriched_cvss_score = cvss.temporal_score
-        else:
-            cvss = CVSS4(observation.enriched_cvss_vector)
-            observation.enriched_cvss_score = cvss.base_score
+        cvss3_vector_before = observation.cvss3_vector
+        cvss4_vector_before = observation.cvss4_vector
+        cve_found_in_before = observation.cve_found_in
 
         if not observation.cvss3_vector and enriched_cvss.base_cvss_vector.startswith("CVSS:3"):
             observation.cvss3_vector = enriched_cvss.base_cvss_vector
             cvss = CVSS3(observation.cvss3_vector)
-            observation.cvss3_score = cvss.temporal_score
+            observation.cvss3_score = cvss.base_score
         if not observation.cvss4_vector and enriched_cvss.base_cvss_vector.startswith("CVSS:4"):
             observation.cvss4_vector = enriched_cvss.base_cvss_vector
             cvss = CVSS4(observation.cvss4_vector)
             observation.cvss4_score = cvss.base_score
 
         _add_cve_found_in(observation, enriched_cvss)
 
-        observation.current_severity = get_current_severity(observation)
+        if (
+            observation.cvss3_vector != cvss3_vector_before
+            or observation.cvss4_vector != cvss4_vector_before
+            or observation.cve_found_in != cve_found_in_before
+        ):
+            observation.current_severity = get_current_severity(observation)
+            return True
 
-        return True
+        return False
     else:
-        if observation.enriched_cvss_score or observation.enriched_cvss_vector or observation.cve_found_in:
-            observation.enriched_cvss_score = None
-            observation.enriched_cvss_vector = ""
+        if observation.cve_found_in:
             observation.cve_found_in = ""
             observation.current_severity = get_current_severity(observation)
             return True
@@ -157,8 +146,6 @@ def _add_cve_found_in(observation: Observation, enriched_cvss: Enriched_CVSS) ->
     cve_found_in = []
     if enriched_cvss.cisa_kev:
         cve_found_in.append("CISA KEV")
-    if enriched_cvss.vulncheck_kev:
-        cve_found_in.append("VulnCheck KEV")
     if enriched_cvss.exploitdb:
         cve_found_in.append("ExploitDB")
     if enriched_cvss.metasploit:
@@ -167,4 +154,6 @@ def _add_cve_found_in(observation: Observation, enriched_cvss: Enriched_CVSS) ->
         cve_found_in.append("Nuclei")
     if enriched_cvss.poc_github:
         cve_found_in.append("PoC GitHub")
+    if enriched_cvss.vulncheck_kev:
+        cve_found_in.append("VulnCheck KEV")
     observation.cve_found_in = ", ".join(cve_found_in)