Create amazon linux importer pipeline

Signed-off-by: ambuj <kulshreshthaak.12@gmail.com>

Author: ambuj-1211

vulnerabilities/importers/__init__.py

@@ -7,8 +7,6 @@
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
 
-from vulnerabilities.importers import alpine_linux
-from vulnerabilities.importers import amazon_linux
 from vulnerabilities.importers import apache_httpd
 from vulnerabilities.importers import apache_kafka
 from vulnerabilities.importers import apache_tomcat
@@ -37,6 +35,7 @@
 from vulnerabilities.importers import xen
 from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipeline
 from vulnerabilities.pipelines import alpine_linux_importer
+from vulnerabilities.pipelines import amazon_linux_importer
 from vulnerabilities.pipelines import github_importer
 from vulnerabilities.pipelines import gitlab_importer
 from vulnerabilities.pipelines import nginx_importer
@@ -72,7 +71,7 @@
     curl.CurlImporter,
     epss.EPSSImporter,
     vulnrichment.VulnrichImporter,
-    amazon_linux.AmazonLinuxImporter,
+    amazon_linux_importer.AmazonLinuxImporterPipeline,
     pypa_importer.PyPaImporterPipeline,
     npm_importer.NpmImporterPipeline,
     nginx_importer.NginxImporterPipeline,

vulnerabilities/improvers/valid_versions.py

@@ -24,7 +24,6 @@
 from vulnerabilities.importer import AffectedPackage
 from vulnerabilities.importer import Importer
 from vulnerabilities.importer import UnMergeablePackageError
-from vulnerabilities.importers.amazon_linux import AmazonLinuxImporter
 from vulnerabilities.importers.apache_httpd import ApacheHTTPDImporter
 from vulnerabilities.importers.apache_kafka import ApacheKafkaImporter
 from vulnerabilities.importers.apache_tomcat import ApacheTomcatImporter
@@ -42,6 +41,7 @@
 from vulnerabilities.improver import Inference
 from vulnerabilities.models import Advisory
 from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipeline
+from vulnerabilities.pipelines.amazon_linux_importer import AmazonLinuxImporterPipeline
 from vulnerabilities.pipelines.github_importer import GitHubAPIImporterPipeline
 from vulnerabilities.pipelines.gitlab_importer import GitLabImporterPipeline
 from vulnerabilities.pipelines.nginx_importer import NginxImporterPipeline
@@ -480,9 +480,10 @@ class GithubOSVImprover(ValidVersionImprover):
 
 
 class AmazonLinuxImprover(ValidVersionImprover):
-    importer = AmazonLinuxImporter
+    importer = AmazonLinuxImporterPipeline
     ignorable_versions = []
 
+
 class CurlImprover(ValidVersionImprover):
     importer = CurlImporter
     ignorable_versions = []

vulnerabilities/importers/amazon_linux.py …ities/pipelines/amazon_linux_importer.pyvulnerabilities/importers/amazon_linux.py renamed to vulnerabilities/pipelines/amazon_linux_importer.py vulnerabilities/importers/amazon_linux.py renamed to vulnerabilities/pipelines/amazon_linux_importer.py

@@ -16,6 +16,7 @@
 from typing import Optional
 
 import pytz
+import requests
 from bs4 import BeautifulSoup
 from packageurl import PackageURL
 from univers.versions import RpmVersion
@@ -25,25 +26,37 @@
 from vulnerabilities.importer import Importer
 from vulnerabilities.importer import Reference
 from vulnerabilities.importer import VulnerabilitySeverity
+from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipeline
 from vulnerabilities.rpm_utils import rpm_to_purl
 from vulnerabilities.severity_systems import SCORING_SYSTEMS
 from vulnerabilities.utils import fetch_response
 from vulnerabilities.utils import is_cve
 
 LOGGER = logging.getLogger(__name__)
-BASE_URL = "https://alas.aws.amazon.com/"
 
 
-class AmazonLinuxImporter(Importer):
-    spdx_license_expression = "CC BY 4.0"
-    license_url = " "  # TODO
+class AmazonLinuxImporterPipeline(VulnerableCodeBaseImporterPipeline):
+    """Imports Amazon Linux security advisories"""
 
+    pipeline_id = "amazon_linux_importer"
+    BASE_URL = "https://alas.aws.amazon.com/"
+    spdx_license_expression = "CC BY 4.0"
+    license_url = "Unknown"
     importer_name = "Amazon Linux Importer"
 
-    def advisory_data(self) -> Iterable[AdvisoryData]:
-        amazon_linux_1_url = BASE_URL + "/index.html"
-        amazon_linux_2_url = BASE_URL + "/alas2.html"
-        amazon_linux_2023_url = BASE_URL + "/alas2023.html"
+    @classmethod
+    def steps(cls):
+        return (
+            cls.fetch,
+            cls.collect_and_store_advisories,
+            cls.import_new_advisories,
+        )
+
+    def fetch(self):
+        self.log(f"Fetch `{self.BASE_URL}`")
+        amazon_linux_1_url = self.BASE_URL + "/index.html"
+        amazon_linux_2_url = self.BASE_URL + "/alas2.html"
+        amazon_linux_2023_url = self.BASE_URL + "/alas2023.html"
         amazonlinux_advisories_pages = [
             amazon_linux_1_url,
             amazon_linux_2_url,
@@ -52,18 +65,40 @@ def advisory_data(self) -> Iterable[AdvisoryData]:
         alas_dict = {}
         for amazonlinux_advisories_page in amazonlinux_advisories_pages:
             alas_dict.update(fetch_alas_id_and_advisory_links(amazonlinux_advisories_page))
+        self.advisory_data = alas_dict
+        # self.advisory_data = requests.get(self.url).text
+
+    def advisories_count(self):
+        return len(self.advisory_data)
+
+    def collect_advisories(self) -> Iterable[AdvisoryData]:
+        """
+        Yield AdvisoryData from nginx security advisories HTML
+        web page.
+        """
 
-        for alas_id, alas_url in alas_dict.items():
+        for alas_id, alas_url in self.advisory_data.items():
             # It iterates through alas_dict to get alas ids and alas url
-            if alas_id and alas_url:
-                alas_advisory_page_content = fetch_response(alas_url).content
-                yield process_advisory_data(alas_id, alas_advisory_page_content, alas_url)
+            if not (alas_id and alas_url):
+                continue
+            try:
+                # Fetch the advisory page content
+                response = fetch_response(alas_url)
+                alas_advisory_page_content = response.content
+
+            except Exception as e:
+                # Log the error and continue to the next item
+                LOGGER.error(f"Failed to fetch advisory {alas_id} from {alas_url}: {str(e)}")
+                continue
+
+            # Process and yield data if successful
+            yield process_advisory_data(alas_id, alas_advisory_page_content, alas_url)
 
 
 def fetch_alas_id_and_advisory_links(page_url: str) -> dict[str, str]:
     """
     Return a dictionary where 'ALAS' entries are the keys and
-    their corresponding advisory page links are the values.
+    their corresponding advisory page link strings are the values.
     """
 
     page_response_content = fetch_response(page_url).content
@@ -253,7 +288,6 @@ def get_date_published(release_date_string):
     # Parse the date and time
     if release_date_string:
         date_part = release_date_string[:16]
-        time_zone = release_date_string[17:]
     else:
         return None
 

…lnerabilities/tests/test_amazon_linux.py …s/test_amazon_linux_importer_pipeline.pyvulnerabilities/tests/test_amazon_linux.py renamed to vulnerabilities/tests/pipelines/test_amazon_linux_importer_pipeline.py vulnerabilities/tests/test_amazon_linux.py renamed to vulnerabilities/tests/pipelines/test_amazon_linux_importer_pipeline.py

@@ -7,17 +7,14 @@
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
 
-import json
 import os
+from pathlib import Path
 from unittest import TestCase
 
-from bs4 import BeautifulSoup
-
-from vulnerabilities.importers.amazon_linux import process_advisory_data
+from vulnerabilities.pipelines.amazon_linux_importer import process_advisory_data
 from vulnerabilities.tests import util_tests
 
-BASE_DIR = os.path.dirname(os.path.abspath(__file__))
-TEST_DATA = os.path.join(BASE_DIR, "test_data/amazon_linux")
+TEST_DATA = Path(__file__).parent.parent / "test_data" / "amazon_linux"
 
 
 class TestAmazonLinuxImporter(TestCase):
@@ -30,6 +27,7 @@ def test_process_advisory_data1(self):
             "ALAS-2024-1943", html_content, "https://alas.aws.amazon.com/ALAS-2024-1943.html"
         ).to_dict()
         expected_file = os.path.join(TEST_DATA, "amazon_linux_expected1.json")
+        # print(f"The result is {result}")
         util_tests.check_results_against_json(result, expected_file)
 
     def test_process_advisory_data2(self):