Add advisory ID

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

Author: TG1999

vulnerabilities/importer.py

@@ -376,6 +376,7 @@ class Importer:
     vcs_response: VCSResponse = None
     # It needs to be unique and immutable
     importer_name = ""
+    requires_reference_for_advisory_id = False
 
     @classmethod
     def get_advisory_id(cls, aliases: list[str]) -> str:

vulnerabilities/importers/archlinux.py

@@ -29,12 +29,16 @@ class ArchlinuxImporter(Importer):
     spdx_license_expression = "MIT"
     license_url = "https://github.com/archlinux/arch-security-tracker/blob/master/LICENSE"
     importer_name = "Arch Linux Importer"
+    requires_reference_for_advisory_id = True
 
     @classmethod
-    def get_advisory_id(cls, aliases: list[str]) -> str:
+    def get_advisory_id(cls, aliases: list[str], references) -> str:
         """
         Return the Advisory ID for the given aliases.
         """
+        for ref in references:
+            if ref.get("reference_id").startswith("AVG-"):
+                return ref.get("reference_id")
         return cls.get_cve_id(aliases)
 
     def fetch(self) -> Iterable[Mapping]:

vulnerabilities/importers/fireeye.py

@@ -40,6 +40,9 @@ def get_advisory_id(cls, aliases: list[str]) -> str:
         """
         Return the Advisory ID for the given aliases.
         """
+        for alias in aliases:
+            if alias.startswith("MNDT-"):
+                return alias
         return cls.get_cve_id(aliases)
 
     def advisory_data(self) -> Iterable[AdvisoryData]:

vulnerabilities/importers/gentoo.py

@@ -32,12 +32,17 @@ class GentooImporter(Importer):
     # under the [CC-BY-SA-4.0](https://creativecommons.org/licenses/by-sa/4.0/) license.
     license_url = "https://creativecommons.org/licenses/by-sa/4.0/"
     importer_name = "Gentoo Importer"
+    requires_reference_for_advisory_id = True
 
     @classmethod
-    def get_advisory_id(cls, aliases: list[str]) -> str:
+    def get_advisory_id(cls, aliases: list[str], references) -> str:
         """
         Return the Advisory ID for the given aliases.
         """
+        for ref in references:
+            ref_id = ref.get("reference_id")
+            if ref_id and ref_id.startswith("GLSA-"):
+                return ref_id
         return cls.get_cve_id(aliases)
 
     def advisory_data(self) -> Iterable[AdvisoryData]:

vulnerabilities/importers/istio.py

@@ -43,12 +43,17 @@ class IstioImporter(Importer):
     license_url = "https://github.com/istio/istio.io/blob/master/LICENSE"
     repo_url = "git+https://github.com/istio/istio.io/"
     importer_name = "Istio Importer"
+    requires_reference_for_advisory_id = True
 
     @classmethod
-    def get_advisory_id(cls, aliases: list[str]) -> str:
+    def get_advisory_id(cls, aliases: list[str], references) -> str:
         """
         Return the Advisory ID for the given aliases.
         """
+        for ref in references:
+            ref_id = ref.get("reference_id")
+            if ref_id and ref_id.startswith("ISTIO-"):
+                return ref_id
         return cls.get_cve_id(aliases)
 
     def advisory_data(self) -> Set[AdvisoryData]:

vulnerabilities/importers/mozilla.py

@@ -38,12 +38,17 @@ class MozillaImporter(Importer):
     license_url = "https://github.com/mozilla/foundation-security-advisories/blob/master/LICENSE"
     repo_url = "git+https://github.com/mozilla/foundation-security-advisories/"
     importer_name = "Mozilla Importer"
+    requires_reference_for_advisory_id = True
 
     @classmethod
-    def get_advisory_id(cls, aliases: list[str]) -> str:
+    def get_advisory_id(cls, aliases: list[str], references) -> str:
         """
         Return the Advisory ID for the given aliases.
         """
+        for ref in references:
+            ref_id = ref.get("reference_id")
+            if ref_id and ref_id.lower().startswith("mfsa"):
+                return ref_id
         return cls.get_cve_id(aliases)
 
     def advisory_data(self) -> Iterable[AdvisoryData]:

vulnerabilities/importers/redhat.py

@@ -66,12 +66,17 @@ class RedhatImporter(Importer):
     spdx_license_expression = "CC-BY-4.0"
     license_url = "https://access.redhat.com/documentation/en-us/red_hat_security_data_api/1.0/html/red_hat_security_data_api/legal-notice"
     importer_name = "RedHat Importer"
+    requires_reference_for_advisory_id = True
 
     @classmethod
-    def get_advisory_id(cls, aliases: list[str]) -> str:
+    def get_advisory_id(cls, aliases: list[str], references) -> str:
         """
         Return the Advisory ID for the given aliases.
         """
+        for ref in references:
+            ref_id = ref.get("reference_id")
+            if ref_id and ref_id.lower().startswith("RHSA-"):
+                return ref_id
         return cls.get_cve_id(aliases)
 
     def advisory_data(self) -> Iterable[AdvisoryData]:

vulnerabilities/importers/ubuntu_usn.py

@@ -63,12 +63,17 @@ class UbuntuUSNImporter(Importer):
     Thanks
     """
     importer_name = "Ubuntu USN Importer"
+    requires_reference_for_advisory_id = True
 
     @classmethod
-    def get_advisory_id(cls, aliases: list[str]) -> str:
+    def get_advisory_id(cls, aliases: list[str], references) -> str:
         """
         Return the Advisory ID for the given aliases.
         """
+        for ref in references:
+            reference_id = ref.get("reference_id")
+            if reference_id and reference_id.startswith("USN-"):
+                return reference_id
         return cls.get_cve_id(aliases)
 
     def advisory_data(self):

vulnerabilities/importers/xen.py

@@ -45,12 +45,16 @@ class XenImporter(Importer):
      -George
     """
     importer_name = "Xen Importer"
+    requires_reference_for_advisory_id = True
 
     @classmethod
-    def get_advisory_id(cls, aliases: list[str]) -> str:
+    def get_advisory_id(cls, aliases: list[str], references: list[dict]) -> str:
         """
         Return the Advisory ID for the given aliases.
         """
+        for ref in references:
+            if ref.get("reference_id").startswith("XSA-"):
+                return ref.get("reference_id")
         return cls.get_cve_id(aliases)
 
     def advisory_data(self):

vulnerabilities/pipelines/__init__.py

@@ -13,6 +13,7 @@
 from timeit import default_timer as timer
 from traceback import format_exc as traceback_format_exc
 from typing import Iterable
+from typing import Optional
 
 from aboutcode.pipeline import BasePipeline
 from aboutcode.pipeline import LoopProgress
@@ -114,6 +115,7 @@ class VulnerableCodeBaseImporterPipeline(VulnerableCodePipeline):
     repo_url = None
     importer_name = None
     advisory_confidence = MAX_CONFIDENCE
+    requires_reference_for_advisory_id = False
 
     @classmethod
     def steps(cls):
@@ -132,6 +134,13 @@ def collect_advisories(self) -> Iterable[AdvisoryData]:
         """
         raise NotImplementedError
 
+    @classmethod
+    def get_advisory_id(cls, aliases: list[str], references: Optional[list[dict]] = None) -> str:
+        """
+        Return the Advisory ID for the given aliases.
+        """
+        raise NotImplementedError
+
     @classmethod
     def get_advisory_id(cls, aliases: list[str]) -> str:
         """

vulnerabilities/pipelines/add_advisory_id.py

@@ -39,10 +39,17 @@ def add_advisory_id(self):
         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)
+            references = advisory.references
+            importer = IMPORTERS_REGISTRY[importer_name]
+            if not importer.requires_reference_for_advisory_id:
+                advisory_id = importer.get_advisory_id(aliases=aliases)
+            else:
+                advisory_id = importer.get_advisory_id(aliases=aliases, references=references)
             if advisory_id is None:
                 continue
             advisory.advisory_id = advisory_id
+            aliases = Alias.objects.filter(advisories=advisory).exclude(alias=advisory_id)
+            advisory.aliases.set(aliases)
             advisories_to_update.append(advisory)
             if len(advisories_to_update) >= batch_size:
                 self.do_bulk_update(advisories_to_update)

vulnerabilities/pipelines/pypa_importer.py

@@ -42,6 +42,9 @@ def get_advisory_id(cls, aliases: list[str]) -> str:
         """
         Return the Advisory ID for the given aliases.
         """
+        for alias in aliases:
+            if alias.lower().startswith("pysec-"):
+                return alias
         return cls.get_cve_id(aliases)
 
     def clone(self):

vulnerabilities/pipelines/pysec_importer.py

@@ -41,6 +41,9 @@ def get_advisory_id(cls, aliases: list[str]) -> str:
         """
         Return the Advisory ID for the given aliases.
         """
+        for alias in aliases:
+            if alias.startswith("PYSEC-"):
+                return alias
         return cls.get_cve_id(aliases)
 
     def fetch_zip(self):

vulnerabilities/tests/test_add_advisory_pipeline.py

@@ -31,3 +31,4 @@ def test_add_advisory_id(self):
             add_advisory_id.AddAdvisoryID().add_advisory_id()
             advisory.refresh_from_db()
             assert advisory.advisory_id == "CVE-2021-1234"
+            assert advisory.aliases.count() == 0