Add NPM Live V2 Importer Pipeline #1936

* Add NPM Live V2 Importer

* Add tests for the NPM Live V2 Importer

* Tested functionally using the Live Evaluation API in #1969

Signed-off-by: Michael Ehab Mikhail <[email protected]>

Author: michaelehab

vulnerabilities/importers/__init__.py

@@ -52,6 +52,7 @@
 from vulnerabilities.pipelines.v2_importers import istio_importer as istio_importer_v2
 from vulnerabilities.pipelines.v2_importers import mozilla_importer as mozilla_importer_v2
 from vulnerabilities.pipelines.v2_importers import npm_importer as npm_importer_v2
+from vulnerabilities.pipelines.v2_importers import npm_live_importer as npm_live_importer_v2
 from vulnerabilities.pipelines.v2_importers import nvd_importer as nvd_importer_v2
 from vulnerabilities.pipelines.v2_importers import oss_fuzz as oss_fuzz_v2
 from vulnerabilities.pipelines.v2_importers import postgresql_importer as postgresql_importer_v2
@@ -117,3 +118,9 @@
         oss_fuzz.OSSFuzzImporter,
     ]
 )
+
+LIVE_IMPORTERS_REGISTRY = create_registry(
+    [
+        npm_live_importer_v2.NpmLiveImporterPipeline,
+    ]
+)

vulnerabilities/pipelines/v2_importers/npm_live_importer.py

@@ -0,0 +1,88 @@
+#
+# 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"]
+
+    @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

vulnerabilities/tests/pipelines/v2_importers/test_npm_live_importer_pipeline_v2.py

@@ -0,0 +1,133 @@
+#
+# 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.
+#
+
+import json
+import os
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+from packageurl import PackageURL
+from univers.version_constraint import VersionConstraint
+from univers.version_range import NpmVersionRange
+from univers.versions import SemverVersion
+
+from vulnerabilities.importer import AffectedPackageV2
+from vulnerabilities.pipelines.v2_importers.npm_live_importer import NpmLiveImporterPipeline
+
+TEST_DATA = Path(__file__).parent.parent.parent / "test_data" / "npm"
+
+
+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)
+
+    advisory_file = vuln_dir / "152.json"
+    advisory_file.write_text(json.dumps(sample_data))
+
+    mock_vcs_response = SimpleNamespace(dest_dir=str(tmp_path), delete=lambda: None)
+
+    purl = PackageURL(type="npm", name="npm", version="1.2.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"
+
+
+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)
+
+    advisory_file = vuln_dir / "152.json"
+    advisory_file.write_text(json.dumps(sample_data))
+
+    mock_vcs_response = SimpleNamespace(dest_dir=str(tmp_path), delete=lambda: None)
+
+    purl = PackageURL(type="npm", name="npm", version="1.4.0")
+    pipeline = NpmLiveImporterPipeline(purl=purl)
+    pipeline.vcs_response = mock_vcs_response
+
+    pipeline.get_purl_inputs()
+    advisories = list(pipeline.collect_advisories())
+
+    assert len(advisories) == 0
+
+
+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 = NpmLiveImporterPipeline(purl=purl)
+    pipeline.vcs_response = mock_vcs_response
+
+    with pytest.raises(ValueError):
+        pipeline.get_purl_inputs()
+
+
+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"
+
+    advisory_file = vuln_dir / "152.json"
+    advisory_file.write_text(json.dumps(sample_data))
+
+    mock_vcs_response = SimpleNamespace(dest_dir=str(tmp_path), delete=lambda: None)
+
+    purl = PackageURL(type="npm", name="nonexistent-package", version="1.0.0")
+    pipeline = NpmLiveImporterPipeline(purl=purl)
+    pipeline.vcs_response = mock_vcs_response
+
+    pipeline.get_purl_inputs()
+    advisories = list(pipeline.collect_advisories())
+
+    assert len(advisories) == 0
+
+
+def test_version_is_affected():
+    purl = PackageURL(type="npm", name="npm", version="1.2.0")
+    pipeline = NpmLiveImporterPipeline(purl=purl)
+    pipeline.get_purl_inputs()
+
+    affected_package = AffectedPackageV2(
+        package=PackageURL(type="npm", name="npm"),
+        affected_version_range=NpmVersionRange(
+            constraints=(VersionConstraint(comparator="<", version=SemverVersion(string="1.3.3")),)
+        ),
+    )
+
+    assert pipeline._version_is_affected(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