Add pipeline for adding advisory ID and tests

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

Author: TG1999

vulnerabilities/importer.py

@@ -377,6 +377,21 @@ class Importer:
     # It needs to be unique and immutable
     importer_name = ""
 
+    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.
+        """
+        for alias in aliases:
+            if alias.startswith("CVE-"):
+                return alias
+        return None
+
     def __init__(self):
         if not self.spdx_license_expression:
             raise Exception(f"Cannot run importer {self!r} without a license")

vulnerabilities/importers/apache_httpd.py

@@ -38,6 +38,12 @@ class ApacheHTTPDImporter(Importer):
     license_url = "https://www.apache.org/licenses/LICENSE-2.0"
     importer_name = "Apache HTTPD Importer"
 
+    def get_advisory_id(self, aliases: list[str]) -> str:
+        """
+        Return the Advisory ID for the given aliases.
+        """
+        return self.get_cve_id(aliases)
+
     def advisory_data(self):
         links = fetch_links(self.base_url)
         for link in links:

vulnerabilities/importers/apache_kafka.py

@@ -102,6 +102,12 @@ 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.
+        """
+        return self.get_cve_id(aliases)
+
     def advisory_data(self):
         advisory_page = self.fetch_advisory_page(self)
 

vulnerabilities/importers/apache_tomcat.py

@@ -120,6 +120,12 @@ class ApacheTomcatImporter(Importer):
     license_url = "https://www.apache.org/licenses/LICENSE-2.0"
     importer_name = "Apache Tomcat Importer"
 
+    def get_advisory_id(self, aliases: list[str]) -> str:
+        """
+        Return the Advisory ID for the given aliases.
+        """
+        return self.get_cve_id(aliases)
+
     def fetch_advisory_pages(self):
         """
         Yield the content of each HTML page containing version-related security data.

vulnerabilities/importers/archlinux.py

@@ -30,6 +30,12 @@ class ArchlinuxImporter(Importer):
     license_url = "https://github.com/archlinux/arch-security-tracker/blob/master/LICENSE"
     importer_name = "Arch Linux Importer"
 
+    def get_advisory_id(self, aliases: list[str]) -> str:
+        """
+        Return the Advisory ID for the given aliases.
+        """
+        return self.get_cve_id(aliases)
+
     def fetch(self) -> Iterable[Mapping]:
         response = fetch_response(self.url)
         return response.json()

vulnerabilities/importers/curl.py

@@ -39,6 +39,12 @@ class CurlImporter(Importer):
     importer_name = "Curl Importer"
     api_url = "https://curl.se/docs/vuln.json"
 
+    def get_advisory_id(self, aliases: list[str]) -> str:
+        """
+        Return the Advisory ID for the given aliases.
+        """
+        return self.get_cve_id(aliases)
+
     def fetch(self) -> Iterable[Mapping]:
         response = fetch_response(self.api_url)
         return response.json()

vulnerabilities/importers/debian.py

@@ -82,6 +82,12 @@ class DebianImporter(Importer):
     api_url = "https://security-tracker.debian.org/tracker/data/json"
     importer_name = "Debian Importer"
 
+    def get_advisory_id(self, aliases: list[str]) -> str:
+        """
+        Return the Advisory ID for the given aliases.
+        """
+        return self.get_cve_id(aliases)
+
     def get_response(self):
         response = requests.get(self.api_url)
         if response.status_code == 200:

vulnerabilities/importers/debian_oval.py

@@ -56,6 +56,12 @@ class DebianOvalImporter(OvalImporter):
     """
     importer_name = "Debian Oval Importer"
 
+    def get_advisory_id(self, aliases: list[str]) -> str:
+        """
+        Return the Advisory ID for the given aliases.
+        """
+        return self.get_cve_id(aliases)
+
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         # we could avoid setting translations, and have it

vulnerabilities/importers/elixir_security.py

@@ -41,6 +41,12 @@ def advisory_data(self) -> Set[AdvisoryData]:
             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.
+        """
+        return self.get_cve_id(aliases)
+
     def process_file(self, file, base_path):
         relative_path = str(file.relative_to(base_path)).strip("/")
         advisory_url = (

vulnerabilities/importers/epss.py

@@ -29,6 +29,12 @@ class EPSSImporter(Importer):
     spdx_license_expression = "unknown"
     importer_name = "EPSS Importer"
 
+    def get_advisory_id(self, aliases: list[str]) -> str:
+        """
+        Return the Advisory ID for the given aliases.
+        """
+        return self.get_cve_id(aliases)
+
     def advisory_data(self) -> Iterable[AdvisoryData]:
         response = urllib.request.urlopen(self.advisory_url)
         with gzip.open(response, "rb") as f: