Fillup missing vulnerabilities summary

Signed-off-by: Tushar Goel <[email protected]>

Author: TG1999

vulnerabilities/improvers/__init__.py

@@ -18,6 +18,7 @@
 from vulnerabilities.pipelines import enhance_with_kev
 from vulnerabilities.pipelines import enhance_with_metasploit
 from vulnerabilities.pipelines import flag_ghost_packages
+from vulnerabilities.pipelines import fill_vulnerability_summary_pipeline
 
 IMPROVERS_REGISTRY = [
     valid_versions.GitHubBasicImprover,
@@ -45,6 +46,7 @@
     compute_package_version_rank.ComputeVersionRankPipeline,
     collect_commits.CollectFixCommitsPipeline,
     add_cvss31_to_CVEs.CVEAdvisoryMappingPipeline,
+    fill_vulnerability_summary_pipeline.FillVulnerabilitySummariesPipeline,
 ]
 
 IMPROVERS_REGISTRY = {

vulnerabilities/pipelines/fill_vulnerability_summary_pipeline.py

@@ -0,0 +1,42 @@
+import logging
+from vulnerabilities.models import Vulnerability, Advisory
+from vulnerabilities.pipelines import VulnerableCodePipeline
+from django.db.models import Q
+
+
+class FillVulnerabilitySummariesPipeline(VulnerableCodePipeline):
+    """Pipeline to fill missing vulnerability summaries from advisories."""
+
+    pipeline_id = "fill_vulnerability_summaries"
+
+    @classmethod
+    def steps(cls):
+        return (cls.fill_missing_summaries,)
+
+    def fill_missing_summaries(self):
+        """Find vulnerabilities without summaries and fill them using advisories with the same aliases."""
+        vulnerabilities_qs = Vulnerability.objects.filter(summary="").prefetch_related("aliases")
+        self.log(f"Processing {vulnerabilities_qs.count()} vulnerabilities without summaries", level=logging.INFO)
+        nvd_importer_advisories = Advisory.objects.filter(created_by="nvd_importer", summary__isnull=False).exclude(summary="")
+        self.log(f"Found {nvd_importer_advisories.count()} advisories from NVD importer", level=logging.INFO)
+        
+        for vulnerability in vulnerabilities_qs:
+            aliases = vulnerability.aliases.values_list("alias", flat=True)
+            alias = aliases.first()
+
+            # check if the vulnerability has an alias
+            if not alias:
+                self.log(f"Vulnerability {vulnerability.vulnerability_id} has no alias", level=logging.INFO)
+                continue
+
+            # check if the vulnerability has an alias that matches an advisory
+            matching_advisories = nvd_importer_advisories.filter(Q(aliases__contains=alias))
+            
+            if matching_advisories.exists():
+                best_advisory = matching_advisories.first()  # Take the first matching advisory with a summary
+                vulnerability.summary = best_advisory.summary
+                vulnerability.save()
+                if self.log:
+                    self.log(f"Updated summary for vulnerability {vulnerability.vulnerability_id}", level=logging.INFO)
+            else:
+                self.log(f"No advisory found for alias {alias}", level=logging.INFO)