aboutcode-org · TG1999 · Apr 1, 2025 · Feb 11, 2025 · Feb 11, 2025 · Feb 12, 2025
diff --git a/vulnerabilities/improvers/__init__.py b/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 populate_vulnerability_summary_pipeline
 from vulnerabilities.pipelines import remove_duplicate_advisories
 
 IMPROVERS_REGISTRY = [
@@ -47,6 +48,7 @@
     collect_commits.CollectFixCommitsPipeline,
     add_cvss31_to_CVEs.CVEAdvisoryMappingPipeline,
     remove_duplicate_advisories.RemoveDuplicateAdvisoriesPipeline,
+    populate_vulnerability_summary_pipeline.PopulateVulnerabilitySummariesPipeline,
 ]
 
 IMPROVERS_REGISTRY = {

diff --git a/vulnerabilities/pipelines/populate_vulnerability_summary_pipeline.py b/vulnerabilities/pipelines/populate_vulnerability_summary_pipeline.py
@@ -0,0 +1,71 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/aboutcode-org/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+import logging
+
+from aboutcode.pipeline import LoopProgress
+from django.db.models import Q
+
+from vulnerabilities.models import Advisory
+from vulnerabilities.models import Vulnerability
+from vulnerabilities.pipelines import VulnerableCodePipeline
+
+
+class PopulateVulnerabilitySummariesPipeline(VulnerableCodePipeline):
+    """Pipeline to populate missing vulnerability summaries from advisories."""

10000
+
+    pipeline_id = "populate_vulnerability_summaries"
+
+    @classmethod
+    def steps(cls):
+        return (cls.populate_missing_summaries,)
+
+    def populate_missing_summaries(self):
+        """Find vulnerabilities with missing summaries and populate them using advisories with the same aliases."""
+        vulnerabilities_qs = Vulnerability.objects.filter(summary="")
+        self.log(
+            f"Processing {vulnerabilities_qs.count()} vulnerabilities without summaries",
+            level=logging.INFO,
+        )
+
+        progress = LoopProgress(total_iterations=vulnerabilities_qs.count(), logger=self.log)
+
+        vulnerabilities_to_be_updated = []
+
+        for vulnerability in progress.iter(vulnerabilities_qs.iterator()):
+            cve_alias = vulnerability.aliases.filter(alias__startswith="CVE-").first()
+
+            if not cve_alias:
+                self.log(
+                    f"Vulnerability {vulnerability.vulnerability_id} has no CVE alias",
+                    level=logging.DEBUG,
+                )
+                continue
+
+            matching_advisories = Advisory.objects.filter(
+                aliases=cve_alias, created_by="nvd_importer"
+            ).exclude(summary="")
+
+            if matching_advisories.exists():
+                best_advisory = matching_advisories.order_by("-date_collected").first()
+                # Note: we filtered above to only get non-empty summaries
+                vulnerability.summary = best_advisory.summary
+                vulnerabilities_to_be_updated.append(vulnerability)
+                self.log(
+                    f"Updated summary for vulnerability {vulnerability.vulnerability_id}",
+                    level=logging.INFO,
+                )
+            else:
+                self.log(f"No advisory found for alias {cve_alias}", level=logging.DEBUG)
+        Vulnerability.objects.bulk_update(vulnerabilities_to_be_updated, ["summary"])
+        self.log(
+            f"Successfully populated {len(vulnerabilities_to_be_updated)} vulnerabilities with summary",
+            level=logging.INFO,
+        )
+        self.log("Pipeline completed", level=logging.INFO)
diff --git a/vulnerabilities/tests/pipelines/test_populate_vulnerability_summary_pipeline.py b/vulnerabilities/tests/pipelines/test_populate_vulnerability_summary_pipeline.py
@@ -0,0 +1,160 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/aboutcode-org/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+import datetime
+from pathlib import Path
+
+import pytz
+from django.test import TestCase
+
+from vulnerabilities.models import Advisory
+from vulnerabilities.models import Alias
+from vulnerabilities.models import Vulnerability
+from vulnerabilities.pipelines.populate_vulnerability_summary_pipeline import (
+    PopulateVulnerabilitySummariesPipeline,
+)
+
+
+class PopulateVulnerabilitySummariesPipelineTest(TestCase):
+    def setUp(self):
+        self.data = Path(__file__).parent.parent / "test_data"
+
+    def test_populate_missing_summaries_from_nvd(self):
+        """
+        Test that vulnerabilities without summaries get them from NVD advisories.
+        """
+
+        # Create a vulnerability without a summary
+        vulnerability = Vulnerability.objects.create(
+            vulnerability_id="VCID-1234",
+            summary="",
+        )
+        alias = Alias.objects.create(alias="CVE-2024-1234", vulnerability=vulnerability)
+
+        # Create an NVD advisory with a summary
+        adv = Advisory.objects.create(
+            summary="Test vulnerability summary",
+            created_by="nvd_importer",
+            date_collected=datetime.datetime(2024, 1, 1, tzinfo=pytz.UTC),
+            unique_content_id="Test",
+        )
+      
10000
  adv.aliases.add(alias)
+
+        # Run the pipeline
+        pipeline = PopulateVulnerabilitySummariesPipeline()
+        pipeline.populate_missing_summaries()
+
+        # Check that the vulnerability now has a summary
+        vulnerability.refresh_from_db()
+        self.assertEqual(vulnerability.summary, "Test vulnerability summary")
+
+    def test_no_matching_advisory(self):
+        """
+        Test handling of vulnerabilities that have no matching NVD advisory.
+        """
+        # Create a vulnerability without a summary
+        vulnerability = Vulnerability.objects.create(
+            vulnerability_id="VCID-1234",
+            summary="",
+        )
+        Alias.objects.create(alias="CVE-2024-1234", vulnerability=vulnerability)
+
+        # Run the pipeline
+        pipeline = PopulateVulnerabilitySummariesPipeline()
+        pipeline.populate_missing_summaries()
+
+        # Check that the vulnerability still has no summary
+        vulnerability.refresh_from_db()
+        self.assertEqual(vulnerability.summary, "")
+
+    def test_vulnerability_without_alias(self):
+        """
+        Test handling of vulnerabilities that have no aliases.
+        """
+
+        # Create a vulnerability without a summary or alias
+        vulnerability = Vulnerability.objects.create(
+            vulnerability_id="VCID-1234",
+            summary="",
+        )
+
+        # Run the pipeline
+        pipeline = PopulateVulnerabilitySummariesPipeline()
+        pipeline.populate_missing_summaries()
+
+        # Check that the vulnerability still has no summary
+        vulnerability.refresh_from_db()
+        self.assertEqual(vulnerability.summary, "")
+
+    def test_non_nvd_advisory_ignored(self):
+        """
+        Test that advisories from sources other than NVD are ignored.
+        """
+
+        # Create a vulnerability without a summary
+        vulnerability = Vulnerability.objects.create(
+            vulnerability_id="VCID-1234",
+            summary="",
+        )
+        alias = Alias.objects.create(alias="CVE-2024-1234", vulnerability=vulnerability)
+
+        # Create a non-NVD advisory with a summary
+        adv = Advisory.objects.create(
+            summary="Test vulnerability summary",
+            created_by="other_importer",
+            date_collected=datetime.datetime(2024, 1, 1, tzinfo=pytz.UTC),
+            unique_content_id="Test",
+        )
+
+        adv.aliases.add(alias)
+
+        # Run the pipeline
+        pipeline = PopulateVulnerabilitySummariesPipeline()
+        pipeline.populate_missing_summaries()
+
+        # Check that the vulnerability still has no summary
+        vulnerability.refresh_from_db()
+        self.assertEqual(vulnerability.summary, "")
+
+    def test_multiple_matching_advisories(self):
+        """
+        Test that the most recent matching advisory is used when there are multiple.
+        """
+        vulnerability = Vulnerability.objects.create(
+            vulnerability_id="VCID-1234",
+            summary="",
+        )
+        alias = Alias.objects.create(alias="CVE-2024-1234", vulnerability=vulnerability)
+
+        # Create two NVD advisories with the same alias
+        adv1 = Advisory.objects.create(
+            summary="First matching advisory",
+            created_by="nvd_importer",
+            date_collected=datetime.datetime(2024, 1, 1, tzinfo=pytz.UTC),
+            unique_content_id="Test",
+        )
+
+        adv1.aliases.add(alias)
+
+        adv2 = Advisory.objects.create(
+            summary="Second matching advisory",
+            created_by="nvd_importer",
+            date_collected=datetime.datetime(2024, 1, 2, tzinfo=pytz.UTC),
+            unique_content_id="Test-1",
+        )
+
+        adv2.aliases.add(alias)
+
+        # Run the pipeline
+        pipeline = PopulateVulnerabilitySummariesPipeline()
+        pipeline.populate_missing_summaries()
+
+        # Check that the vulnerability now has the most recent summary
+        vulnerability.refresh_from_db()
+        self.assertEqual(vulnerability.summary, "Second matching advisory")