Refactor the code for npm_live importer

Signed-off-by: ziad hany <ziadhany2016@gmail.com>

Author: ziadhany

vulnerabilities/pipelines/v2_importers/npm_live_importer.py

@@ -11,9 +11,10 @@
 from typing import Iterable
 
 from packageurl import PackageURL
+from univers.versions import InvalidVersion
 from univers.versions import SemverVersion
 
-from vulnerabilities.importer import AdvisoryData
+from vulnerabilities.importer import AdvisoryDataV2
 from vulnerabilities.pipelines.v2_importers.npm_importer import NpmImporterPipeline
 from vulnerabilities.utils import load_json
 
@@ -58,31 +59,35 @@ def get_purl_inputs(self):
 
         self.purl = purl
 
-    def collect_advisories(self) -> Iterable[AdvisoryData]:
+    def collect_advisories(self) -> Iterable[AdvisoryDataV2]:
         vuln_directory = Path(self.vcs_response.dest_dir) / "vuln" / "npm"
-        advisory_files = list(vuln_directory.glob("*.json"))
-
         package_name = self.purl.name
         filtered_files = []
-        for advisory_file in advisory_files:
-            try:
-                data = load_json(advisory_file)
-                if data.get("module_name") == package_name:
-                    affected_package = self.get_affected_package(data, package_name)
-                    if not self.purl.version or self._version_is_affected(affected_package):
-                        filtered_files.append(advisory_file)
-            except Exception as e:
-                self.log(f"Error processing advisory file {advisory_file}: {str(e)}")
-        advisory_files = filtered_files
-
-        for advisory in list(advisory_files):
+        for advisory_file in vuln_directory.glob("*.json"):
+            data = load_json(advisory_file)
+            if data.get("module_name") == package_name:
+                affected_package = self.get_affected_package(data, package_name)
+                if not self.purl.version or self._version_is_related(affected_package):
+                    filtered_files.append(advisory_file)
+
+        for advisory in filtered_files:
             result = self.to_advisory_data(advisory)
             if result:
                 yield result
 
-    def _version_is_affected(self, affected_package):
-        if not self.purl.version or not affected_package.affected_version_range:
+    def _version_is_related(self, affected_package):
+        try:
+            package_version = SemverVersion(self.purl.version)
+        except InvalidVersion as e:
+            self.log(f"Invalid PURL version: {self.purl.version!r}: {str(e)}")
+            return False
+
+        if (
+            affected_package.affected_version_range
+            and package_version in affected_package.affected_version_range
+        ) or (
+            affected_package.fixed_version_range
+            and package_version in affected_package.fixed_version_range
+        ):
             return True
-
-        purl_version = SemverVersion(self.purl.version)
-        return purl_version in affected_package.affected_version_range
+        return False

vulnerabilities/tests/pipelines/v2_importers/test_npm_live_importer_pipeline_v2.py

@@ -20,8 +20,9 @@
 
 from vulnerabilities.importer import AffectedPackageV2
 from vulnerabilities.pipelines.v2_importers.npm_live_importer import NpmLiveImporterPipeline
+from vulnerabilities.tests import util_tests
 
-TEST_DATA = Path(__file__).parent.parent.parent / "test_data" / "npm"
+TEST_DATA = Path(__file__).parent.parent.parent / "test_data" / "npm_live"
 
 
 def test_package_first_mode_valid_npm_package(tmp_path):
@@ -37,17 +38,15 @@ def test_package_first_mode_valid_npm_package(tmp_path):
 
     mock_vcs_response = SimpleNamespace(dest_dir=str(tmp_path), delete=lambda: None)
 
-    purl = PackageURL(type="npm", name="npm", version="1.2.0")
+    purl = PackageURL(type="npm", name="decamelize", version="1.1.0")
     pipeline = NpmLiveImporterPipeline(purl=purl)
     pipeline.vcs_response = mock_vcs_response
 
     pipeline.get_purl_inputs()
-    advisories = list(pipeline.collect_advisories())
 
-    assert len(advisories) == 1
-    assert advisories[0].aliases == ["CVE-2013-4116"]
-    assert len(advisories[0].affected_packages) == 1
-    assert advisories[0].affected_packages[0].package.name == "npm"
+    result = [adv.to_dict() for adv in pipeline.collect_advisories()]
+    expected_file = Path(TEST_DATA / "parse-advisory-npm-expected.json")
+    util_tests.check_results_against_json(result, expected_file)
 
 
 def test_package_first_mode_unaffected_version(tmp_path):
@@ -124,10 +123,7 @@ def test_version_is_affected():
         ),
     )
 
-    assert pipeline._version_is_affected(affected_package) == True
+    assert pipeline._version_is_related(affected_package) == True
 
     pipeline.purl = PackageURL(type="npm", name="npm", version="1.4.0")
-    assert pipeline._version_is_affected(affected_package) == False
-
-    pipeline.purl = PackageURL(type="npm", name="npm")
-    assert pipeline._version_is_affected(affected_package) == True
+    assert pipeline._version_is_related(affected_package) == False

vulnerabilities/tests/test_data/npm_live/npm_sample.json

@@ -0,0 +1,24 @@
+{
+  "id": 308,
+  "created_at": "2017-01-26",
+  "updated_at": "2017-04-14",
+  "title": "Regular Expression Denial of Service",
+  "author": {
+    "name": "saurik",
+    "website": null,
+    "username": null
+  },
+  "module_name": "decamelize",
+  "publish_date": "2017-04-14",
+  "cves": [],
+  "vulnerable_versions": ">=1.1.0 <=1.1.1",
+  "patched_versions": ">=1.1.2",
+  "overview": "Decamelize is used to convert a dash/dot/underscore/space separated string to camelCase. \n\nDecamelize uses regular expressions to evaluate a string and takes unescaped separator values, which can be used to create a denial of service attack.",
+  "recommendation": "Upgrade to version 1.1.2 or later.",
+  "references": [
+    "https://github.com/sindresorhus/decamelize/issues/5"
+  ],
+  "cvss_vector": null,
+  "cvss_score": null,
+  "coordinating_vendor": "^Lift Security"
+}

vulnerabilities/tests/test_data/npm_live/parse-advisory-npm-expected.json

@@ -0,0 +1,40 @@
+[
+  {
+    "advisory_id": "npm-308",
+    "aliases": [],
+    "summary": "Regular Expression Denial of Service\nDecamelize is used to convert a dash/dot/underscore/space separated string to camelCase. \n\nDecamelize uses regular expressions to evaluate a string and takes unescaped separator values, which can be used to create a denial of service attack.",
+    "affected_packages": [
+      {
+        "package": {
+          "type": "npm",
+          "namespace": "",
+          "name": "decamelize",
+          "version": "",
+          "qualifiers": "",
+          "subpath": ""
+        },
+        "affected_version_range": "vers:npm/>=1.1.0|<=1.1.1",
+        "fixed_version_range": "vers:npm/>=1.1.2",
+        "introduced_by_commit_patches": [],
+        "fixed_by_commit_patches": []
+      }
+    ],
+    "references": [
+      {
+        "reference_id": "",
+        "reference_type": "",
+        "url": "https://github.com/sindresorhus/decamelize/issues/5"
+      },
+      {
+        "reference_id": "308",
+        "reference_type": "",
+        "url": "https://github.com/nodejs/security-wg/blob/main/vuln/npm/308.json"
+      }
+    ],
+    "patches": [],
+    "severities": [],
+    "date_published": "2017-01-26T00:00:00+00:00",
+    "weaknesses": [],
+    "url": "https://github.com/nodejs/security-wg/blob/main/vuln/npm/308.json"
+  }
+]