Migrate Rockylinux Importer to Importer_Pipeline

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

Author: ambuj-1211

vulnerabilities/importers/__init__.py

@@ -29,7 +29,6 @@
 from vulnerabilities.importers import pysec
 from vulnerabilities.importers import redhat
 from vulnerabilities.importers import retiredotnet
-from vulnerabilities.importers import rockylinux
 from vulnerabilities.importers import ruby
 from vulnerabilities.importers import suse_scores
 from vulnerabilities.importers import ubuntu
@@ -43,6 +42,7 @@
 from vulnerabilities.pipelines import npm_importer
 from vulnerabilities.pipelines import nvd_importer
 from vulnerabilities.pipelines import pypa_importer
+from vulnerabilities.pipelines import rockylinux_importer
 
 IMPORTERS_REGISTRY = [
     pysec.PyPIImporter,
@@ -70,7 +70,6 @@
     oss_fuzz.OSSFuzzImporter,
     ruby.RubyImporter,
     github_osv.GithubOSVImporter,
-    rockylinux.RockyLinuxImporter,
     curl.CurlImporter,
     epss.EPSSImporter,
     vulnrichment.VulnrichImporter,
@@ -80,6 +79,7 @@
     gitlab_importer.GitLabImporterPipeline,
     github_importer.GitHubAPIImporterPipeline,
     nvd_importer.NVDImporterPipeline,
+    rockylinux_importer.RockylinuxImporterPipeline,
 ]
 
 IMPORTERS_REGISTRY = {

vulnerabilities/improvers/valid_versions.py

@@ -34,7 +34,6 @@
 from vulnerabilities.importers.github_osv import GithubOSVImporter
 from vulnerabilities.importers.istio import IstioImporter
 from vulnerabilities.importers.oss_fuzz import OSSFuzzImporter
-from vulnerabilities.importers.rockylinux import RockyLinuxImporter
 from vulnerabilities.importers.ruby import RubyImporter
 from vulnerabilities.importers.ubuntu import UbuntuImporter
 from vulnerabilities.improver import MAX_CONFIDENCE
@@ -46,6 +45,7 @@
 from vulnerabilities.pipelines.gitlab_importer import GitLabImporterPipeline
 from vulnerabilities.pipelines.nginx_importer import NginxImporterPipeline
 from vulnerabilities.pipelines.npm_importer import NpmImporterPipeline
+from vulnerabilities.pipelines.rockylinux_importer import RockylinuxImporterPipeline
 from vulnerabilities.utils import AffectedPackage as LegacyAffectedPackage
 from vulnerabilities.utils import clean_nginx_git_tag
 from vulnerabilities.utils import get_affected_packages_by_patched_package
@@ -480,7 +480,7 @@ class GithubOSVImprover(ValidVersionImprover):
 
 
 class RockyLinuxImprover(ValidVersionImprover):
-    importer = RockyLinuxImporter
+    importer = RockylinuxImporterPipeline
     ignorable_versions = []
 
 

vulnerabilities/importers/rockylinux.py …ilities/pipelines/rockylinux_importer.pyvulnerabilities/importers/rockylinux.py renamed to vulnerabilities/pipelines/rockylinux_importer.py vulnerabilities/importers/rockylinux.py renamed to vulnerabilities/pipelines/rockylinux_importer.py

@@ -13,62 +13,70 @@
 from typing import Iterable
 from typing import List
 
-import dateparser
 import requests
 from cwe2.database import Database
+from dateutil import parser as dateparser
 from packageurl import PackageURL
 from univers.version_range import RpmVersionRange
 
 from vulnerabilities import severity_systems
 from vulnerabilities.importer import AdvisoryData
 from vulnerabilities.importer import AffectedPackage
-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.utils import get_cwe_id
-from vulnerabilities.utils import get_item
 from vulnerabilities.utils import requests_with_5xx_retry
 
 logger = logging.getLogger(__name__)
-
-# FIXME: we should use a centralized retry
 requests_session = requests_with_5xx_retry(max_retries=5, backoff_factor=1)
 
 
-def fetch_cves() -> Iterable[List[Dict]]:
-    page_no = 0
-    cve_data_list = []
-    while True:
-        current_url = f"https://errata.rockylinux.org/api/v2/advisories?filters.product=&filters.fetchRelated=true&page={page_no}&limit=100"
-        try:
-            response = requests_session.get(current_url)
-            if response.status_code != requests.codes.ok:
-                logger.error(f"Failed to fetch RedHat CVE results from {current_url}")
-                break
-            cve_data = response.json().get("advisories") or []
-            cve_data_list.extend(cve_data)
-        except Exception as e:
-            logger.error(f"Failed to fetch rockylinux CVE results from {current_url} {e}")
-            break
-        if not cve_data:
-            break
-        page_no += 1
-    return cve_data_list
-
-
-class RockyLinuxImporter(Importer):
+class RockylinuxImporterPipeline(VulnerableCodeBaseImporterPipeline):
+    pipeline_id = "rockylinux_importer"
+
     spdx_license_expression = "CC-BY-4.0"
     license_url = "https://access.redhat.com/security/data"
-    importer_name = "Rocky Importer"
+    importer_name = "Rockylinux Importer"
+
+    @classmethod
+    def steps(cls):
+        return (cls.fetch, cls.collect_and_store_advisories, cls.import_new_advisories)
+
+    def fetch(self) -> Iterable[List[Dict]]:
+        page_no = 0
+        self.advisory_data = []
+
+        while True:
+            current_url = f"https://errata.rockylinux.org/api/v2/advisories?filters.product=&filters.fetchRelated=true&page={page_no}&limit=100"
+            try:
+                response = requests_session.get(current_url)
+                if response.status_code != requests.codes.ok:
+                    logger.error(f"Failed to fetch RedHat CVE results from {current_url}")
+                    break
+                cve_data = response.json().get("advisories") or []
+                self.advisory_data.extend(cve_data)
+            except Exception as e:
+                logger.error(f"Failed to fetch rockylinux CVE results from {current_url} {e}")
+                break
+            if not cve_data:
+                break
+            page_no += 1
+
+    def advisories_count(self):
+        return len(self.advisory_data)
+
+    def collect_advisories(self) -> Iterable[AdvisoryData]:
+        for rl_advisory in self.advisory_data:
+            yield to_advisory(rl_advisory)
 
-    def advisory_data(self) -> Iterable[AdvisoryData]:
 
-        for rockylinux_cve in fetch_cves():
-            yield to_advisory(rockylinux_cve)
+class VersionParsingError(Exception):
+    pass
 
 
-def to_advisory(advisory_data):
+def to_advisory(advisory_data) -> AdvisoryData:
 
     """
     Convert Rockylinux advisory data into an AdvisoryData object.
@@ -169,24 +177,36 @@ def to_advisory(advisory_data):
         for fix in advisory_data["fixes"]
     ]
 
-    for ref in advisory_data.get("cves") or []:
+    for ref in advisory_data.get("cves", []):
 
         name = ref.get("name", "")
         if not isinstance(name, str):
             logger.error(f"Invalid advisory type {name}")
             continue
 
-        if "CVE" in name.upper():
-            severities = VulnerabilitySeverity(
-                system=severity_systems.CVSSV31,
-                value=ref.get("cvss3BaseScore", ""),
-                scoring_elements=ref.get("cvss3ScoringVector", "")
-                if ref.get("cvss3ScoringVector", "") != "UNKNOWN"
-                else "",
-            )
-            references.append(
-                Reference(severities=[severities], url=ref.get("sourceLink", ""), reference_id=name)
-            )
+        if ref.get("sourceLink", ""):
+            name_upper = name.upper()
+            cvss_vector = ref.get("cvss3ScoringVector", "")
+
+            if "CVE" in name_upper and cvss_vector != "UNKNOWN":
+                base_score = ref.get("cvss3BaseScore", "")
+
+                if base_score and cvss_vector:
+                    severities = [
+                        VulnerabilitySeverity(
+                            system=severity_systems.CVSSV31,
+                            value=base_score,
+                            scoring_elements=cvss_vector,
+                        )
+                    ]
+                else:
+                    severities = []
+
+                references.append(
+                    Reference(
+                        severities=severities, url=ref.get("sourceLink", ""), reference_id=name
+                    )
+                )
 
     return AdvisoryData(
         aliases=aliases,
@@ -199,10 +219,6 @@ def to_advisory(advisory_data):
     )
 
 
-class VersionParsingError(Exception):
-    pass
-
-
 def get_cwes_from_rockylinux_advisory(advisory_data) -> [int]:
     """
     Extract CWE IDs from advisory data and validate them against a database.

vulnerabilities/tests/test_rockylinux.py …nes/test_rockylinux_importer_pipeline.pyvulnerabilities/tests/test_rockylinux.py renamed to vulnerabilities/tests/pipelines/test_rockylinux_importer_pipeline.py vulnerabilities/tests/test_rockylinux.py renamed to vulnerabilities/tests/pipelines/test_rockylinux_importer_pipeline.py

@@ -7,40 +7,40 @@
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
 
-import json
 import os
+from pathlib import Path
 from unittest import TestCase
 from unittest.mock import patch
 
 from packageurl import PackageURL
 
-from vulnerabilities.importers import rockylinux
-from vulnerabilities.importers.rockylinux import get_cwes_from_rockylinux_advisory
-from vulnerabilities.importers.rockylinux import to_advisory
+from vulnerabilities.pipelines.rockylinux_importer import get_cwes_from_rockylinux_advisory
+from vulnerabilities.pipelines.rockylinux_importer import to_advisory
 from vulnerabilities.rpm_utils import rpm_to_purl
 from vulnerabilities.utils import load_json
 
-BASE_DIR = os.path.dirname(os.path.abspath(__file__))
-TEST_DATA = os.path.join(BASE_DIR, "test_data", "rockylinux")
+TEST_DATA = Path(__file__).parent.parent / "test_data" / "rockylinux"
 
 
-class TestRockyLinuxImporter(TestCase):
+class TestRockylinuxImporterPipeline(TestCase):
     def test_to_advisory1(self):
         test1 = os.path.join(TEST_DATA, "rockylinux_test1.json")
         mock_response = load_json(test1)
         expected_result = load_json(os.path.join(TEST_DATA, "rockylinux_expected1.json"))
+        # print(f"1st is {to_advisory(mock_response).to_dict()}")
         assert to_advisory(mock_response).to_dict() == expected_result
 
     def test_to_advisory2(self):
         test2 = os.path.join(TEST_DATA, "rockylinux_test2.json")
         mock_response2 = load_json(test2)
         expected_result2 = load_json(os.path.join(TEST_DATA, "rockylinux_expected2.json"))
         assert to_advisory(mock_response2).to_dict() == expected_result2
+        # print(f"2nd is {to_advisory(mock_response2).to_dict()}")
 
     def test_rpm_to_purl(self):
-        assert rockylinux.rpm_to_purl("foobar", "rocky-linux") is None
-        assert rockylinux.rpm_to_purl("foo-bar-devel-0:sys76", "rocky-linux") is None
-        assert rockylinux.rpm_to_purl("cockpit-0:264.1-1.el8.aarch64", "rocky-linux") == PackageURL(
+        assert rpm_to_purl("foobar", "rocky-linux") is None
+        assert rpm_to_purl("foo-bar-devel-0:sys76", "rocky-linux") is None
+        assert rpm_to_purl("cockpit-0:264.1-1.el8.aarch64", "rocky-linux") == PackageURL(
             type="rpm",
             namespace="rocky-linux",
             name="cockpit",

vulnerabilities/tests/test_data/rockylinux/rockylinux_expected2.json

@@ -141,78 +141,6 @@
       "reference_type": "",
       "url": "https://bugzilla.redhat.com/show_bug.cgi?id=2270666",
       "severities": []
-    },
-    {
-      "reference_id": "CVE-2023-5388",
-      "reference_type": "",
-      "url": "https://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2023-5388",
-      "severities": [
-        { "system": "cvssv3.1", "value": "UNKNOWN", "scoring_elements": "" }
-      ]
-    },
-    {
-      "reference_id": "CVE-2024-0743",
-      "reference_type": "",
-      "url": "https://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2024-0743",
-      "severities": [
-        { "system": "cvssv3.1", "value": "UNKNOWN", "scoring_elements": "" }
-      ]
-    },
-    {
-      "reference_id": "CVE-2024-1936",
-      "reference_type": "",
-      "url": "https://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2024-1936",
-      "severities": [
-        { "system": "cvssv3.1", "value": "UNKNOWN", "scoring_elements": "" }
-      ]
-    },
-    {
-      "reference_id": "CVE-2024-2607",
-      "reference_type": "",
-      "url": "https://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2024-2607",
-      "severities": [
-        { "system": "cvssv3.1", "value": "UNKNOWN", "scoring_elements": "" }
-      ]
-    },
-    {
-      "reference_id": "CVE-2024-2608",
-      "reference_type": "",
-      "url": "https://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2024-2608",
-      "severities": [
-        { "system": "cvssv3.1", "value": "UNKNOWN", "scoring_elements": "" }
-      ]
-    },
-    {
-      "reference_id": "CVE-2024-2610",
-      "reference_type": "",
-      "url": "https://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2024-2610",
-      "severities": [
-        { "system": "cvssv3.1", "value": "UNKNOWN", "scoring_elements": "" }
-      ]
-    },
-    {
-      "reference_id": "CVE-2024-2611",
-      "reference_type": "",
-      "url": "https://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2024-2611",
-      "severities": [
-        { "system": "cvssv3.1", "value": "UNKNOWN", "scoring_elements": "" }
-      ]
-    },
-    {
-      "reference_id": "CVE-2024-2612",
-      "reference_type": "",
-      "url": "https://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2024-2612",
-      "severities": [
-        { "system": "cvssv3.1", "value": "UNKNOWN", "scoring_elements": "" }
-      ]
-    },
-    {
-      "reference_id": "CVE-2024-2614",
-      "reference_type": "",
-      "url": "https://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2024-2614",
-      "severities": [
-        { "system": "cvssv3.1", "value": "UNKNOWN", "scoring_elements": "" }
-      ]
     }
   ],
   "date_published": "2024-03-27T04:34:32.999941+00:00",