Add pipeline for adding advisory ID and tests

Signed-off-by: Tushar Goel <tushar.goel.dav@gmail.com>

Author: TG1999

vulnerabilities/importer.py

@@ -382,7 +382,7 @@ def get_advisory_id(self, aliases: list[str]) -> str:
         Return the Advisory ID for the given aliases.
         """
         raise NotImplementedError
-    
+
     def get_cve_id(self, aliases: list[str]) -> str:
         """
         Return the CVE ID for the given aliases.

vulnerabilities/importers/apache_kafka.py

@@ -101,7 +101,7 @@ class ApacheKafkaImporter(Importer):
     def fetch_advisory_page(self):
         page = requests.get(self.GH_PAGE_URL)
         return page.content
-    
+
     def get_advisory_id(self, aliases: list[str]) -> str:
         """
         Return the Advisory ID for the given aliases.

vulnerabilities/importers/elixir_security.py

@@ -40,7 +40,7 @@ def advisory_data(self) -> Set[AdvisoryData]:
         finally:
             if self.vcs_response:
                 self.vcs_response.delete()
-    
+
     def get_advisory_id(self, aliases: list[str]) -> str:
         """
         Return the Advisory ID for the given aliases.

vulnerabilities/importers/postgresql.py

@@ -34,7 +34,7 @@ def get_advisory_id(self, aliases: list[str]) -> str:
         """
         Return the Advisory ID for the given aliases.
         """
-        return self.get_cve_id(aliases)    
+        return self.get_cve_id(aliases)
 
     def advisory_data(self):
         known_urls = {self.root_url}

vulnerabilities/importers/retiredotnet.py

@@ -43,7 +43,7 @@ def advisory_data(self) -> Iterable[AdvisoryData]:
         finally:
             if self.vcs_response:
                 self.vcs_response.delete()
-    
+
     def get_advisory_id(self, aliases: list[str]) -> str:
         """
         Return the Advisory ID for the given aliases.

vulnerabilities/importers/ruby.py

@@ -71,7 +71,7 @@ def advisory_data(self) -> Iterable[AdvisoryData]:
         finally:
             if self.vcs_response:
                 self.vcs_response.delete()
-    
+
     def get_advisory_id(self, aliases: list[str]) -> str:
         """
         Return the Advisory ID for the given aliases.

vulnerabilities/migrations/0091_advisory_advisory_id.py

@@ -0,0 +1,22 @@
+# Generated by Django 4.2.17 on 2025-03-28 06:22
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("vulnerabilities", "0090_migrate_advisory_aliases"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="advisory",
+            name="advisory_id",
+            field=models.CharField(
+                blank=True,
+                help_text="An advisory id, such as CVE-123-345 when available",
+                max_length=100,
+            ),
+        ),
+    ]

vulnerabilities/models.py

@@ -1323,6 +1323,16 @@ class Advisory(models.Model):
         null=False,
         help_text="A 64 character unique identifier for the content of the advisory since we use sha256 as hex",
     )
+
+    advisory_id = models.CharField(
+        max_length=100,
+        blank=True,
+        help_text="An advisory id, such as CVE-123-345 when available",
+    )
+    url = models.URLField(
+        blank=True,
+        help_text="Link to the advisory on the upstream website",
+    )
     aliases = models.ManyToManyField(
         Alias,
         through="AdvisoryRelatedAlias",
@@ -1354,10 +1364,6 @@ class Advisory(models.Model):
         "module name importing the advisory. Eg:"
         "vulnerabilities.pipeline.nginx_importer.NginxImporterPipeline",
     )
-    url = models.URLField(
-        blank=True,
-        help_text="Link to the advisory on the upstream website",
-    )
 
     objects = AdvisoryQuerySet.as_manager()
 

vulnerabilities/pipelines/__init__.py

@@ -131,14 +131,16 @@ def collect_advisories(self) -> Iterable[AdvisoryData]:
         Populate the `self.collected_advisories_count` field and yield AdvisoryData
         """
         raise NotImplementedError
-    
-    def get_advisory_id(self, aliases: list[str]) -> str:
+
+    @classmethod
+    def get_advisory_id(cls, aliases: list[str]) -> str:
         """
         Return the Advisory ID for the given aliases.
         """
         raise NotImplementedError
-    
-    def get_cve_id(self, aliases: list[str]) -> str:
+
+    @classmethod
+    def get_cve_id(cls, aliases: list[str]) -> str:
         """
         Return the CVE ID for the given aliases.
         """

vulnerabilities/pipelines/add_advisory_id.py

@@ -0,0 +1,53 @@
+# 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.
+#
+
+from aboutcode.pipeline import LoopProgress
+from django.db import transaction
+
+from vulnerabilities.importers import IMPORTERS_REGISTRY
+from vulnerabilities.models import Advisory
+from vulnerabilities.models import Alias
+from vulnerabilities.pipelines import VulnerableCodePipeline
+
+
+class AddAdvisoryID(VulnerableCodePipeline):
+    """
+    Pipeline to map CVEs from VulnerabilitySeverity to corresponding Advisories with CVSS3.1 scores.
+    """
+
+    pipeline_id = "add_advisory_id"
+
+    @classmethod
+    def steps(cls):
+        return (cls.add_advisory_id,)
+
+    def add_advisory_id(self):
+
+        advisories = Advisory.objects.all()
+
+        advisories_to_update = []
+
+        batch_size = 500
+
+        progress = LoopProgress(total_iterations=advisories.count(), logger=self.log)
+
+        for advisory in progress.iter(advisories.iterator(chunk_size=batch_size)):
+            importer_name = advisory.created_by
+            aliases = Alias.objects.filter(advisories=advisory).values_list("alias", flat=True)
+            advisory_id = IMPORTERS_REGISTRY[importer_name].get_advisory_id(aliases=aliases)
+            advisory.advisory_id = advisory_id
+            advisories_to_update.append(advisory)
+            if len(advisories_to_update) >= batch_size:
+                self.do_bulk_update(advisories_to_update)
+                advisories_to_update = []
+        self.do_bulk_update(advisories_to_update)
+        self.log(f"Pipeline [{self.pipeline_name}] completed.")
+
+    def do_bulk_update(self, advisories_to_update):
+        Advisory.objects.bulk_update(advisories_to_update, ["advisory_id"])
+        self.log(f"Updated {len(advisories_to_update)} advisories with advisory_id.")