Modify NPM importer to support package-first mode using SCM approach #1936

Signed-off-by: Michael Ehab Mikhail <michael.ehab@hotmail.com>

Author: michaelehab

vulnerabilities/pipelines/npm_importer.py

@@ -56,13 +56,6 @@ def __init__(self, *args, purl=None, **kwargs):
 
     @classmethod
     def steps(cls):
-        if not cls.is_batch_run:
-            return [
-                cls.fetch_package_advisories,
-                cls.collect_and_store_advisories,
-                cls.import_new_advisories,
-            ]
-
         return [
             cls.clone,
             cls.collect_and_store_advisories,
@@ -74,58 +67,32 @@ def clone(self):
         self.log(f"Cloning `{self.repo_url}`")
         self.vcs_response = fetch_via_vcs(self.repo_url)
 
-    def fetch_package_advisories(self):
-        if not self.purl or self.purl.type != "npm":
-            return
-
-        self.log(f"Fetching advisories for package {self.purl.name}")
-
-        package_name = self.purl.name
-
-        self.temp_dir = tempfile.mkdtemp()
-        self.package_advisories = []
-
-        api_url = "https://api.github.com/repos/nodejs/security-wg/contents/vuln/npm"
-        response = requests.get(api_url)
+    def advisories_count(self):
+        vuln_directory = Path(self.vcs_response.dest_dir) / "vuln" / "npm"
+        return sum(1 for _ in vuln_directory.glob("*.json"))
 
-        if response.status_code != 200:
-            self.log(f"Failed to fetch advisories directory: {response.status_code}")
-            return
+    def collect_advisories(self) -> Iterable[AdvisoryData]:
+        vuln_directory = Path(self.vcs_response.dest_dir) / "vuln" / "npm"
+        advisory_files = list(vuln_directory.glob("*.json"))
 
-        for item in response.json():
-            if item["type"] == "file" and item["name"].endswith(".json"):
-                file_url = item["download_url"]
+        if not self.is_batch_run:
+            package_name = self.purl.name
+            filtered_files = []
+            for advisory_file in advisory_files:
                 try:
-                    file_content = requests.get(file_url).json()
-
-                    if file_content.get("module_name") == package_name:
-                        file_path = os.path.join(self.temp_dir, item["name"])
-                        with open(file_path, "w") as f:
-                            json.dump(file_content, f)
-                        self.package_advisories.append(file_path)
+                    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 {item['name']}: {str(e)}")
+                    self.log(f"Error processing advisory file {advisory_file}: {str(e)}")
+            advisory_files = filtered_files
 
-        self.log(f"Found {len(self.package_advisories)} advisories for package {package_name}")
-
-    def advisories_count(self):
-        if NpmImporterPipeline.is_batch_run:
-            vuln_directory = Path(self.vcs_response.dest_dir) / "vuln" / "npm"
-            return sum(1 for _ in vuln_directory.glob("*.json"))
-        else:
-            return len(getattr(self, "package_advisories", []))
-
-    def collect_advisories(self) -> Iterable[AdvisoryData]:
-        if NpmImporterPipeline.is_batch_run:
-            vuln_directory = Path(self.vcs_response.dest_dir) / "vuln" / "npm"
-            for advisory in vuln_directory.glob("*.json"):
-                yield from self.to_advisory_data(advisory)
-        else:
-            if not hasattr(self, "package_advisories"):
-                return
-
-            for advisory_path in self.package_advisories:
-                yield from self.to_advisory_data(Path(advisory_path))
+        for advisory in list(advisory_files):
+            for result in self.to_advisory_data(advisory):
+                if result:
+                    yield result
 
     def to_advisory_data(self, file: Path) -> Iterable[AdvisoryData]:
         data = load_json(file)
@@ -241,11 +208,5 @@ def clean_downloads(self):
             self.log(f"Removing cloned repository")
             self.vcs_response.delete()
 
-        if hasattr(self, "temp_dir") and os.path.exists(self.temp_dir):
-            import shutil
-
-            self.log(f"Removing temporary directory")
-            shutil.rmtree(self.temp_dir)
-
     def on_failure(self):
         self.clean_downloads()

vulnerabilities/pipelines/v2_importers/npm_importer.py

@@ -59,12 +59,6 @@ def __init__(self, *args, purl=None, **kwargs):
 
     @classmethod
     def steps(cls):
-        if not cls.is_batch_run:
-            return (
-                cls.fetch_package_advisories,
-                cls.collect_and_store_advisories,
-                cls.clean_downloads,
-            )
         return (
             cls.clone,
             cls.collect_and_store_advisories,
@@ -75,60 +69,32 @@ def clone(self):
         self.log(f"Cloning `{self.repo_url}`")
         self.vcs_response = fetch_via_vcs(self.repo_url)
 
-    def fetch_package_advisories(self):
-        if not self.purl or self.purl.type != "npm":
-            return
-
-        self.log(f"Fetching advisories for package {self.purl.name}")
-
-        package_name = self.purl.name
-
-        self.temp_dir = tempfile.mkdtemp()
-        self.package_advisories = []
-
-        api_url = "https://api.github.com/repos/nodejs/security-wg/contents/vuln/npm"
-        response = requests.get(api_url)
+    def advisories_count(self):
+        vuln_directory = Path(self.vcs_response.dest_dir) / "vuln" / "npm"
+        return sum(1 for _ in vuln_directory.glob("*.json"))
 
-        if response.status_code != 200:
-            self.log(f"Failed to fetch advisories directory: {response.status_code}")
-            return
+    def collect_advisories(self) -> Iterable[AdvisoryData]:
+        vuln_directory = Path(self.vcs_response.dest_dir) / "vuln" / "npm"
+        advisory_files = list(vuln_directory.glob("*.json"))
 
-        for item in response.json():
-            if item["type"] == "file" and item["name"].endswith(".json"):
-                file_url = item["download_url"]
+        if not self.is_batch_run:
+            package_name = self.purl.name
+            filtered_files = []
+            for advisory_file in advisory_files:
                 try:
-                    file_content = requests.get(file_url).json()
-
-                    if file_content.get("module_name") == package_name:
-                        file_path = os.path.join(self.temp_dir, item["name"])
-                        with open(file_path, "w") as f:
-                            json.dump(file_content, f)
-                        self.package_advisories.append(file_path)
+                    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 {item['name']}: {str(e)}")
-
-        self.log(f"Found {len(self.package_advisories)} advisories for package {package_name}")
-
-    def advisories_count(self):
-        if NpmImporterPipeline.is_batch_run:
-            vuln_directory = Path(self.vcs_response.dest_dir) / "vuln" / "npm"
-            return sum(1 for _ in vuln_directory.glob("*.json"))
-        else:
-            return len(getattr(self, "package_advisories", []))
-
-    def collect_advisories(self) -> Iterable[AdvisoryData]:
-        if NpmImporterPipeline.is_batch_run:
-            vuln_directory = Path(self.vcs_response.dest_dir) / "vuln" / "npm"
-            for advisory in vuln_directory.glob("*.json"):
-                yield self.to_advisory_data(advisory)
-        else:
-            if not hasattr(self, "package_advisories"):
-                return
+                    self.log(f"Error processing advisory file {advisory_file}: {str(e)}")
+            advisory_files = filtered_files
 
-            for advisory_path in self.package_advisories:
-                result = self.to_advisory_data(Path(advisory_path))
-                if result:
-                    yield result
+        for advisory in list(advisory_files):
+            result = self.to_advisory_data(advisory)
+            if result:
+                yield result
 
     def to_advisory_data(self, file: Path) -> Iterable[AdvisoryData]:
         if file.name == "index.json":

vulnerabilities/tests/pipelines/test_npm_importer_pipeline.py

@@ -12,7 +12,7 @@
 import json
 import os
 from pathlib import Path
-from unittest.mock import MagicMock
+from types import SimpleNamespace
 from unittest.mock import patch
 
 from packageurl import PackageURL
@@ -80,31 +80,23 @@ def test_npm_improver(mock_response):
     util_tests.check_results_against_json(result, expected_file)
 
 
-@patch("requests.get")
-def test_package_first_mode_valid_npm_package(mock_get):
-    mock_dir_response = MagicMock()
-    mock_dir_response.status_code = 200
-    mock_dir_response.json.return_value = [
-        {
-            "type": "file",
-            "name": "152.json",
-            "download_url": "https://raw.githubusercontent.com/nodejs/security-wg/main/vuln/npm/152.json",
-        }
-    ]
+def test_package_first_mode_valid_npm_package(tmp_path):
+    vuln_dir = tmp_path / "vuln" / "npm"
+    vuln_dir.mkdir(parents=True)
 
     npm_sample_file = os.path.join(TEST_DATA, "npm_sample.json")
     with open(npm_sample_file) as f:
         sample_data = json.load(f)
 
-    mock_file_response = MagicMock()
-    mock_file_response.json.return_value = sample_data
+    advisory_file = vuln_dir / "152.json"
+    advisory_file.write_text(json.dumps(sample_data))
 
-    mock_get.side_effect = [mock_dir_response, mock_file_response]
+    mock_vcs_response = SimpleNamespace(dest_dir=str(tmp_path), delete=lambda: None)
 
     purl = PackageURL(type="npm", name="npm", version="1.2.0")
     pipeline = NpmImporterPipeline(purl=purl)
+    pipeline.vcs_response = mock_vcs_response
 
-    pipeline.fetch_package_advisories()
     advisories = list(pipeline.collect_advisories())
 
     assert len(advisories) == 1
@@ -113,91 +105,74 @@ def test_package_first_mode_valid_npm_package(mock_get):
     assert advisories[0].affected_packages[0].package.name == "npm"
 
 
-@patch("requests.get")
-def test_package_first_mode_unaffected_version(mock_get):
-    mock_dir_response = MagicMock()
-    mock_dir_response.status_code = 200
-    mock_dir_response.json.return_value = [
-        {
-            "type": "file",
-            "name": "152.json",
-            "download_url": "https://raw.githubusercontent.com/nodejs/security-wg/main/vuln/npm/152.json",
-        }
-    ]
+def test_package_first_mode_unaffected_version(tmp_path):
+    vuln_dir = tmp_path / "vuln" / "npm"
+    vuln_dir.mkdir(parents=True)
 
     npm_sample_file = os.path.join(TEST_DATA, "npm_sample.json")
     with open(npm_sample_file) as f:
         sample_data = json.load(f)
 
-    mock_file_response = MagicMock()
-    mock_file_response.json.return_value = sample_data
+    advisory_file = vuln_dir / "152.json"
+    advisory_file.write_text(json.dumps(sample_data))
 
-    mock_get.side_effect = [mock_dir_response, mock_file_response]
+    mock_vcs_response = SimpleNamespace(dest_dir=str(tmp_path), delete=lambda: None)
 
     purl = PackageURL(type="npm", name="npm", version="1.4.0")
     pipeline = NpmImporterPipeline(purl=purl)
+    pipeline.vcs_response = mock_vcs_response
 
-    pipeline.fetch_package_advisories()
     advisories = list(pipeline.collect_advisories())
 
     assert len(advisories) == 0
 
 
-@patch("requests.get")
-def test_package_first_mode_invalid_package_type(mock_get):
+def test_package_first_mode_invalid_package_type(tmp_path):
+    vuln_dir = tmp_path / "vuln" / "npm"
+    vuln_dir.mkdir(parents=True)
+
+    mock_vcs_response = SimpleNamespace(dest_dir=str(tmp_path), delete=lambda: None)
+
     purl = PackageURL(type="pypi", name="django", version="3.0.0")
     pipeline = NpmImporterPipeline(purl=purl)
+    pipeline.vcs_response = mock_vcs_response
 
-    pipeline.fetch_package_advisories()
     advisories = list(pipeline.collect_advisories())
 
     assert len(advisories) == 0
-    mock_get.assert_not_called()
-
-
-@patch("requests.get")
-def test_package_first_mode_package_not_found(mock_get):
-    mock_dir_response = MagicMock()
-    mock_dir_response.status_code = 200
-    mock_dir_response.json.return_value = [
-        {
-            "type": "file",
-            "name": "152.json",
-            "download_url": "https://raw.githubusercontent.com/nodejs/security-wg/main/vuln/npm/152.json",
-        }
-    ]
+
+
+def test_package_first_mode_package_not_found(tmp_path):
+    vuln_dir = tmp_path / "vuln" / "npm"
+    vuln_dir.mkdir(parents=True)
 
     npm_sample_file = os.path.join(TEST_DATA, "npm_sample.json")
     with open(npm_sample_file) as f:
         sample_data = json.load(f)
 
     sample_data["module_name"] = "some-other-package"
 
-    mock_file_response = MagicMock()
-    mock_file_response.json.return_value = sample_data
+    advisory_file = vuln_dir / "152.json"
+    advisory_file.write_text(json.dumps(sample_data))
 
-    mock_get.side_effect = [mock_dir_response, mock_file_response]
+    mock_vcs_response = SimpleNamespace(dest_dir=str(tmp_path), delete=lambda: None)
 
     purl = PackageURL(type="npm", name="nonexistent-package", version="1.0.0")
     pipeline = NpmImporterPipeline(purl=purl)
+    pipeline.vcs_response = mock_vcs_response
 
-    pipeline.fetch_package_advisories()
     advisories = list(pipeline.collect_advisories())
 
     assert len(advisories) == 0
 
 
-@patch("requests.get")
-def test_package_first_mode_api_error(mock_get):
-    mock_error_response = MagicMock()
-    mock_error_response.status_code = 404
-
-    mock_get.return_value = mock_error_response
+def test_package_first_mode_missing_vuln_directory(tmp_path):
+    mock_vcs_response = SimpleNamespace(dest_dir=str(tmp_path), delete=lambda: None)
 
     purl = PackageURL(type="npm", name="npm", version="1.0.0")
     pipeline = NpmImporterPipeline(purl=purl)
+    pipeline.vcs_response = mock_vcs_response
 
-    pipeline.fetch_package_advisories()
     advisories = list(pipeline.collect_advisories())
 
     assert len(advisories) == 0
@@ -228,3 +203,9 @@ def test_version_is_affected():
         fixed_version=SemverVersion(string="1.3.3"),
     )
     assert pipeline._version_is_affected(affected_package_no_range) == True
+    affected_package_no_range = AffectedPackage(
+        package=PackageURL(type="npm", name="npm"),
+        affected_version_range=None,
+        fixed_version=SemverVersion(string="1.3.3"),
+    )
+    assert pipeline._version_is_affected(affected_package_no_range) == True