Add NPM live importer #1936

* Add NPM live pipeline importer to filter advisories affecting a single PURL

* Add tests for NPM live importer

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

Author: michaelehab

vulnerabilities/pipelines/npm_importer.py

@@ -9,19 +9,14 @@
 
 # Author: Navonil Das (@NavonilDas)
 
-import json
-import os
-import tempfile
 from pathlib import Path
 from typing import Iterable
 
 import pytz
-import requests
 from dateutil.parser import parse
 from fetchcode.vcs import fetch_via_vcs
 from packageurl import PackageURL
 from univers.version_range import NpmVersionRange
-from univers.versions import SemverVersion
 
 from vulnerabilities.importer import AdvisoryData
 from vulnerabilities.importer import AffectedPackage
@@ -44,24 +39,14 @@ class NpmImporterPipeline(VulnerableCodeBaseImporterPipeline):
     repo_url = "git+https://github.com/nodejs/security-wg"
     importer_name = "Npm Importer"
 
-    is_batch_run = True
-
-    def __init__(self, *args, purl=None, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.purl = purl
-        if self.purl:
-            NpmImporterPipeline.is_batch_run = False
-            if self.purl.type != "npm":
-                print(f"Warning: This importer handles NPM packages. Current PURL: {self.purl!s}")
-
     @classmethod
     def steps(cls):
-        return [
+        return (
             cls.clone,
             cls.collect_and_store_advisories,
             cls.import_new_advisories,
             cls.clean_downloads,
-        ]
+        )
 
     def clone(self):
         self.log(f"Cloning `{self.repo_url}`")
@@ -73,26 +58,9 @@ def advisories_count(self):
 
     def collect_advisories(self) -> Iterable[AdvisoryData]:
         vuln_directory = Path(self.vcs_response.dest_dir) / "vuln" / "npm"
-        advisory_files = list(vuln_directory.glob("*.json"))
-
-        if not self.is_batch_run:
-            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 result in self.to_advisory_data(advisory):
-                if result:
-                    yield result
+
+        for advisory in vuln_directory.glob("*.json"):
+            yield from self.to_advisory_data(advisory)
 
     def to_advisory_data(self, file: Path) -> Iterable[AdvisoryData]:
         data = load_json(file)
@@ -144,11 +112,6 @@ def to_advisory_data(self, file: Path) -> Iterable[AdvisoryData]:
             affected_packages.append(self.get_affected_package(data, package_name))
         advsisory_aliases = data.get("cves") or []
 
-        if self.purl and self.purl.version:
-            affected_package = affected_packages[0] if affected_packages else None
-            if affected_package and not self._version_is_affected(affected_package):
-                return
-
         for alias in advsisory_aliases:
             yield AdvisoryData(
                 summary=build_description(summary=summary, description=description),
@@ -159,13 +122,6 @@ def to_advisory_data(self, file: Path) -> Iterable[AdvisoryData]:
                 url=f"https://github.com/nodejs/security-wg/blob/main/vuln/npm/{id}.json",
             )
 
-    def _version_is_affected(self, affected_package):
-        if not self.purl.version or not affected_package.affected_version_range:
-            return True
-
-        purl_version = SemverVersion(self.purl.version)
-        return purl_version in affected_package.affected_version_range
-
     def get_affected_package(self, data, package_name):
         affected_version_range = None
         unaffected_version_range = None
@@ -209,4 +165,4 @@ def clean_downloads(self):
             self.vcs_response.delete()
 
     def on_failure(self):
-        self.clean_downloads()
+        self.clean_downloads()

vulnerabilities/pipelines/v2_importers/npm_importer.py

@@ -10,18 +10,14 @@
 # Author: Navonil Das (@NavonilDas)
 
 import json
-import os
-import tempfile
 from pathlib import Path
 from typing import Iterable
 
 import pytz
-import requests
 from dateutil.parser import parse
 from fetchcode.vcs import fetch_via_vcs
 from packageurl import PackageURL
 from univers.version_range import NpmVersionRange
-from univers.versions import SemverVersion
 
 from vulnerabilities.importer import AdvisoryData
 from vulnerabilities.importer import AffectedPackage
@@ -47,16 +43,6 @@ class NpmImporterPipeline(VulnerableCodeBaseImporterPipelineV2):
     repo_url = "git+https://github.com/nodejs/security-wg"
     unfurl_version_ranges = True
 
-    is_batch_run = True
-
-    def __init__(self, *args, purl=None, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.purl = purl
-        if self.purl:
-            NpmImporterPipeline.is_batch_run = False
-            if self.purl.type != "npm":
-                print(f"Warning: This importer handles NPM packages. Current PURL: {self.purl!s}")
-
     @classmethod
     def steps(cls):
         return (
@@ -75,32 +61,18 @@ def advisories_count(self):
 
     def collect_advisories(self) -> Iterable[AdvisoryData]:
         vuln_directory = Path(self.vcs_response.dest_dir) / "vuln" / "npm"
-        advisory_files = list(vuln_directory.glob("*.json"))
-
-        if not self.is_batch_run:
-            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):
-            result = self.to_advisory_data(advisory)
-            if result:
-                yield result
+
+        for advisory in vuln_directory.glob("*.json"):
+            yield self.to_advisory_data(advisory)
 
     def to_advisory_data(self, file: Path) -> Iterable[AdvisoryData]:
         if file.name == "index.json":
             self.log(f"Skipping {file.name} file")
             return
         data = load_json(file)
+        advisory_text = None
+        with open(file) as f:
+            advisory_text = f.read()
         id = data.get("id")
         description = data.get("overview") or ""
         summary = data.get("title") or ""
@@ -153,11 +125,6 @@ def to_advisory_data(self, file: Path) -> Iterable[AdvisoryData]:
             affected_packages.append(self.get_affected_package(data, package_name))
         advsisory_aliases = data.get("cves") or []
 
-        if self.purl and self.purl.version:
-            affected_package = affected_packages[0] if affected_packages else None
-            if affected_package and not self._version_is_affected(affected_package):
-                return
-
         return AdvisoryData(
             advisory_id=f"npm-{id}",
             aliases=advsisory_aliases,
@@ -167,15 +134,9 @@ def to_advisory_data(self, file: Path) -> Iterable[AdvisoryData]:
             references_v2=references,
             severities=severities,
             url=f"https://github.com/nodejs/security-wg/blob/main/vuln/npm/{id}.json",
+            original_advisory_text=advisory_text or json.dumps(data, indent=2, ensure_ascii=False),
         )
 
-    def _version_is_affected(self, affected_package):
-        if not self.purl.version or not affected_package.affected_version_range:
-            return True
-
-        purl_version = SemverVersion(self.purl.version)
-        return purl_version in affected_package.affected_version_range
-
     def get_affected_package(self, data, package_name):
         affected_version_range = None
         unaffected_version_range = None
@@ -218,11 +179,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()
+        self.clean_downloads()

vulnerabilities/pipelines/v2_importers/npm_live_importer.py

@@ -0,0 +1,92 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/aboutcode-org/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+from pathlib import Path
+from typing import Iterable
+
+from packageurl import PackageURL
+from univers.versions import SemverVersion
+
+from vulnerabilities.importer import AdvisoryData
+from vulnerabilities.pipelines.v2_importers.npm_importer import NpmImporterPipeline
+from vulnerabilities.utils import load_json
+
+
+class NpmLiveImporterPipeline(NpmImporterPipeline):
+    """
+    Node.js Security Working Group importer pipeline
+
+    Import advisories from nodejs security working group including node proper advisories and npm advisories for a single PURL.
+    """
+
+    pipeline_id = "nodejs_security_wg_live_importer"
+    supported_types = ["npm"]
+    spdx_license_expression = "MIT"
+    license_url = "https://github.com/nodejs/security-wg/blob/main/LICENSE.md"
+    repo_url = "git+https://github.com/nodejs/security-wg"
+    unfurl_version_ranges = True
+
+    @classmethod
+    def steps(cls):
+        return (
+            cls.get_purl_inputs,
+            cls.clone,
+            cls.collect_and_store_advisories,
+            cls.clean_downloads,
+        )
+    
+    def get_purl_inputs(self):
+        purl = self.inputs["purl"]
+        if not purl:
+            raise ValueError("PURL is required for NpmLiveImporterPipeline")
+
+        if isinstance(purl, str):
+            purl = PackageURL.from_string(purl)
+
+        if not isinstance(purl, PackageURL):
+            raise ValueError(f"Object of type {type(purl)} {purl!r} is not a PackageURL instance")
+
+        if purl.type not in self.supported_types:
+            raise ValueError(
+                f"PURL: {purl!s} is not among the supported package types {self.supported_types!r}"
+            )
+
+        if not purl.version:
+            raise ValueError(f"PURL: {purl!s} is expected to have a version")
+
+        self.purl = purl
+
+    def collect_advisories(self) -> Iterable[AdvisoryData]:
+        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):
+            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:
+            return True
+
+        purl_version = SemverVersion(self.purl.version)
+        return purl_version in affected_package.affected_version_range