feat: enrich observations with exploit information from cvss-bt (#2672)

* feat: enrich cvss with cvss-bt

* chore: remove enriched_cvss fields

* chore: fix import linter

* chore: documentation (start)

* chore: renaming

* chore: documentation (finish)

* chore: unittest for exploit enrichment

* chore: more work on unittests

* chore: code quality

* chore: unittests shall not run settings signals task

backend/application/commons/api/views.py

@@ -67,7 +67,6 @@ def get(self, request: Request) -> Response:
         features = []
 
         settings = Settings.load()
-
         if settings.feature_disable_user_login:
             features.append("feature_disable_user_login")
 
@@ -82,6 +81,8 @@ def get(self, request: Request) -> Response:
                 features.append("feature_automatic_api_import")
             if settings.feature_automatic_osv_scanning:
                 features.append("feature_automatic_osv_scanning")
+            if settings.feature_exploit_information:
+                features.append("feature_exploit_information")
 
         content: dict[str, Union[int, list[str]]] = {
             "features": features,
@@ -111,8 +112,8 @@ def patch(self, request: Request, pk: int = None) -> Response:  # pylint: disabl
         if not request_serializer.is_valid():
             raise ValidationError(request_serializer.errors)
 
-        settings = request_serializer.create(request_serializer.validated_data)
-        settings.save()
+        settings = Settings.load()
+        request_serializer.update(settings, request_serializer.validated_data)
 
         response_serializer = SettingsSerializer(settings)
         return Response(response_serializer.data)

backend/application/commons/migrations/0015_settings_exploit_information_max_age_years_and_more.py

@@ -0,0 +1,31 @@
+# Generated by Django 5.1.6 on 2025-03-06 04:40
+
+import django.core.validators
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("commons", "0014_settings_feature_automatic_osv_scanning"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="settings",
+            name="exploit_information_max_age_years",
+            field=models.IntegerField(
+                default=10,
+                help_text="Maximum age of CVEs for enrichment in years",
+                validators=[
+                    django.core.validators.MinValueValidator(0),
+                    django.core.validators.MaxValueValidator(999999),
+                ],
+            ),
+        ),
+        migrations.AddField(
+            model_name="settings",
+            name="feature_exploit_information",
+            field=models.BooleanField(default=True, help_text="Enable CVSS enrichment"),
+        ),
+    ]

backend/application/commons/models.py

@@ -222,6 +222,12 @@ class Settings(Model):
         help_text="Hour crontab expression for importing licenses (UTC)",
     )
     feature_automatic_osv_scanning = BooleanField(default=True, help_text="Enable automatic OSV scanning")
+    feature_exploit_information = BooleanField(default=True, help_text="Enable CVSS enrichment")
+    exploit_information_max_age_years = IntegerField(
+        default=10,
+        validators=[MinValueValidator(0), MaxValueValidator(999999)],
+        help_text="Maximum age of CVEs for enrichment in years",
+    )
 
     def save(self, *args: Any, **kwargs: Any) -> None:
         """

backend/application/commons/signals.py

@@ -1,17 +1,39 @@
 from typing import Any
 
+import environ
 from django.db.models.signals import post_save
 from django.dispatch import receiver
+from huey.contrib.djhuey import db_task, lock_task
 
 from application.commons.models import Settings
 from application.core.models import Product
 from application.core.services.security_gate import check_security_gate
+from application.epss.models import Exploit_Information
+from application.epss.services.cvss_bt import (
+    apply_exploit_information_observations,
+    import_cvss_bt,
+)
 
 
 @receiver(post_save, sender=Settings)
 def settings_post_save(  # pylint: disable=unused-argument
     sender: Any, instance: Settings, created: bool, **kwargs: Any
 ) -> None:
     # parameters are needed according to Django documentation
+    env = environ.Env()
+    if not env.bool("SO_UNITTESTS", False):
+        settings_post_save_task(instance, created)
+
+
+@db_task()
+@lock_task("settings_post_save_task_lock")
+def settings_post_save_task(settings: Settings, created: bool) -> None:
     for product in Product.objects.filter(is_product_group=False):
         check_security_gate(product)
+
+    if not created:
+        if settings.feature_exploit_information and not Exploit_Information.objects.exists():
+            import_cvss_bt()
+        if not settings.feature_exploit_information and Exploit_Information.objects.exists():
+            Exploit_Information.objects.all().delete()
+            apply_exploit_information_observations(settings)

backend/application/core/api/filters.py

@@ -1,5 +1,5 @@
 from datetime import timedelta
-from typing import Any
+from typing import Any, Optional
 
 from django.db.models import Q, QuerySet
 from django.utils import timezone
@@ -216,6 +216,7 @@ class ObservationFilter(FilterSet):
         field_name="product__product_group",
         queryset=Product.objects.filter(is_product_group=True),
     )
+    cve_known_exploited = BooleanFilter(field_name="cve_known_exploited", method="get_cve_known_exploited")
 
     ordering = OrderingFilter(
         # tuple-mapping retains order
@@ -284,6 +285,19 @@ def get_age(
         time_threshold = today - timedelta(days=int(days))
         return queryset.filter(last_observation_log__gte=time_threshold)
 
+    def get_cve_known_exploited(
+        self,
+        queryset: QuerySet,
+        name: Any,  # pylint: disable=unused-argument
+        value: Optional[bool],
+    ) -> QuerySet:
+        # name is used as a positional argument
+        if value is True:
+            return queryset.exclude(cve_found_in="")
+        if value is False:
+            return queryset.filter(cve_found_in="")
+        return queryset
+
 
 class ObservationLogFilter(FilterSet):
     age = ChoiceFilter(field_name="age", method="get_age", choices=Age_Choices.AGE_CHOICES)

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

@@ -0,0 +1,18 @@
+# Generated by Django 5.1.6 on 2025-03-04 05:07
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("core", "0060_product_automatic_osv_scanning_enabled"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="observation",
+            name="cve_found_in",
+            field=models.CharField(blank=True, max_length=255),
+        ),
+    ]

backend/application/core/models.py

@@ -22,11 +22,6 @@
 from django.utils import timezone
 
 from application.access_control.models import Authorization_Group, User
-from application.core.services.observation import (
-    get_identity_hash,
-    normalize_observation_fields,
-    set_product_flags,
-)
 from application.core.types import (
     Assessment_Status,
     OSVLinuxDistribution,
@@ -372,31 +367,37 @@ class Observation(Model):
     title = CharField(max_length=255)
     description = TextField(max_length=2048, blank=True)
     recommendation = TextField(max_length=2048, blank=True)
+
     current_severity = CharField(max_length=12, choices=Severity.SEVERITY_CHOICES)
     numerical_severity = IntegerField(validators=[MinValueValidator(1), MaxValueValidator(6)])
     parser_severity = CharField(max_length=12, choices=Severity.SEVERITY_CHOICES, blank=True)
     rule_severity = CharField(max_length=12, choices=Severity.SEVERITY_CHOICES, blank=True)
     assessment_severity = CharField(max_length=12, choices=Severity.SEVERITY_CHOICES, blank=True)
+
     current_status = CharField(max_length=16, choices=Status.STATUS_CHOICES)
     parser_status = CharField(max_length=16, choices=Status.STATUS_CHOICES, blank=True)
     vex_status = CharField(max_length=16, choices=Status.STATUS_CHOICES, blank=True)
     rule_status = CharField(max_length=16, choices=Status.STATUS_CHOICES, blank=True)
     assessment_status = CharField(max_length=16, choices=Status.STATUS_CHOICES, blank=True)
+
     scanner_observation_id = CharField(max_length=255, blank=True)
     vulnerability_id = CharField(max_length=255, blank=True)
     vulnerability_id_aliases = CharField(max_length=512, blank=True)
+
     origin_component_name = CharField(max_length=255, blank=True)
     origin_component_version = CharField(max_length=255, blank=True)
     origin_component_name_version = CharField(max_length=513, blank=True)
     origin_component_purl = CharField(max_length=255, blank=True)
     origin_component_purl_type = CharField(max_length=16, blank=True)
     origin_component_cpe = CharField(max_length=255, blank=True)
     origin_component_dependencies = TextField(max_length=32768, blank=True)
+
     origin_docker_image_name = CharField(max_length=255, blank=True)
     origin_docker_image_tag = CharField(max_length=255, blank=True)
     origin_docker_image_name_tag = CharField(max_length=513, blank=True)
     origin_docker_image_name_tag_short = CharField(max_length=513, blank=True)
     origin_docker_image_digest = CharField(max_length=255, blank=True)
+
     origin_endpoint_url = TextField(max_length=2048, blank=True)
     origin_endpoint_scheme = CharField(max_length=255, blank=True)
     origin_endpoint_hostname = CharField(max_length=255, blank=True)
@@ -405,26 +406,34 @@ class Observation(Model):
     origin_endpoint_params = TextField(max_length=2048, blank=True)
     origin_endpoint_query = TextField(max_length=2048, blank=True)
     origin_endpoint_fragment = TextField(max_length=2048, blank=True)
+
     origin_service_name = CharField(max_length=255, blank=True)
     origin_service = ForeignKey(Service, on_delete=PROTECT, null=True)
+
     origin_source_file = CharField(max_length=255, blank=True)
     origin_source_line_start = IntegerField(null=True, validators=[MinValueValidator(0), MaxValueValidator(999999)])
     origin_source_line_end = IntegerField(null=True, validators=[MinValueValidator(0), MaxValueValidator(999999)])
+
     origin_cloud_provider = CharField(max_length=255, blank=True)
     origin_cloud_account_subscription_project = CharField(max_length=255, blank=True)
     origin_cloud_resource = CharField(max_length=255, blank=True)
     origin_cloud_resource_type = CharField(max_length=255, blank=True)
     origin_cloud_qualified_resource = CharField(max_length=255, blank=True)
+
     origin_kubernetes_cluster = CharField(max_length=255, blank=True)
     origin_kubernetes_namespace = CharField(max_length=255, blank=True)
     origin_kubernetes_resource_type = CharField(max_length=255, blank=True)
     origin_kubernetes_resource_name = CharField(max_length=255, blank=True)
     origin_kubernetes_qualified_resource = CharField(max_length=255, blank=True)
+
     cvss3_score = DecimalField(max_digits=3, decimal_places=1, null=True)
     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)
+    cve_found_in = CharField(max_length=255, blank=True)
+
     cwe = IntegerField(null=True, validators=[MinValueValidator(1), MaxValueValidator(999999)])
+
     epss_score = DecimalField(
         max_digits=6,
         decimal_places=3,
@@ -437,6 +446,7 @@ class Observation(Model):
         null=True,
         validators=[MinValueValidator(Decimal(0)), MaxValueValidator(Decimal(100))],
     )
+
     found = DateField(null=True)
     scanner = CharField(max_length=255, blank=True)
     upload_filename = CharField(max_length=255, blank=True)
@@ -446,6 +456,7 @@ class Observation(Model):
     modified = DateTimeField(auto_now=True)
     last_observation_log = DateTimeField(default=timezone.now)
     identity_hash = CharField(max_length=64)
+
     general_rule = ForeignKey(
         "rules.Rule",
         related_name="general_rules",
@@ -460,10 +471,13 @@ class Observation(Model):
         null=True,
         on_delete=PROTECT,
     )
+
     issue_tracker_issue_id = CharField(max_length=255, blank=True)
     issue_tracker_issue_closed = BooleanField(default=False)
     issue_tracker_jira_initial_status = CharField(max_length=255, blank=True)
+
     has_potential_duplicates = BooleanField(default=False)
+
     current_vex_justification = CharField(max_length=64, choices=VexJustification.VEX_JUSTIFICATION_CHOICES, blank=True)
     parser_vex_justification = CharField(max_length=64, choices=VexJustification.VEX_JUSTIFICATION_CHOICES, blank=True)
     vex_vex_justification = CharField(max_length=64, choices=VexJustification.VEX_JUSTIFICATION_CHOICES, blank=True)
@@ -478,6 +492,7 @@ class Observation(Model):
         null=True,
         on_delete=SET_NULL,
     )
+
     risk_acceptance_expiry_date = DateField(null=True)
 
     class Meta:
@@ -509,13 +524,6 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
         self.unsaved_references: list[str] = []
         self.unsaved_evidences: list[list[str]] = []
 
-    def save(self, *args: Any, **kwargs: Any) -> None:
-        normalize_observation_fields(self)
-        self.identity_hash = get_identity_hash(self)
-        set_product_flags(self)
-
-        return super().save(*args, **kwargs)
-
 
 class Observation_Log(Model):
     observation = ForeignKey(Observation, related_name="observation_logs", on_delete=CASCADE)