Skip to content

Modify NPM importer to support package-first mode #1941

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 6 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion vulnerabilities/pipelines/npm_importer.py
Original file line number Diff line number Diff line change
Expand Up @@ -165,4 +165,4 @@ def clean_downloads(self):
self.vcs_response.delete()

def on_failure(self):
self.clean_downloads()
self.clean_downloads()
2 changes: 1 addition & 1 deletion vulnerabilities/pipelines/v2_importers/npm_importer.py
Original file line number Diff line number Diff line change
Expand Up @@ -180,4 +180,4 @@ def clean_downloads(self):
self.vcs_response.delete()

def on_failure(self):
self.clean_downloads()
self.clean_downloads()
92 changes: 92 additions & 0 deletions vulnerabilities/pipelines/v2_importers/npm_live_importer.py
Original file line number Diff line number Diff line change
@@ -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
Original file line number Diff line number Diff line change
Expand Up @@ -76,4 +76,4 @@ def test_npm_improver(mock_response):
inference = [data.to_dict() for data in improver.get_inferences(advisory)]
result.extend(inference)
expected_file = os.path.join(TEST_DATA, f"npm-improver-expected.json")
util_tests.check_results_against_json(result, expected_file)
util_tests.check_results_against_json(result, expected_file)
123 changes: 122 additions & 1 deletion vulnerabilities/tests/pipelines/test_npm_importer_pipeline_v2.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,18 +8,26 @@
#

import json
import os
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import MagicMock
from unittest.mock import patch

import pytz
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 AdvisoryData
from vulnerabilities.importer import AffectedPackage
from vulnerabilities.pipelines.v2_importers.npm_importer import NpmImporterPipeline
from vulnerabilities.severity_systems import CVSSV2
from vulnerabilities.severity_systems import CVSSV3

TEST_DATA = Path(__file__).parent.parent / "test_data" / "npm"


def test_clone(monkeypatch):
import vulnerabilities.pipelines.v2_importers.npm_importer as npm_mod
Expand Down Expand Up @@ -58,8 +66,8 @@ def test_advisories_count_and_collect(tmp_path):
(vuln_dir / "001.json").write_text(json.dumps({"id": "001"}))
p = NpmImporterPipeline()
p.vcs_response = SimpleNamespace(dest_dir=str(base), delete=lambda: None)
assert p.advisories_count() == 2
advisories = list(p.collect_advisories())
assert p.advisories_count() == 2
# Should yield None for index.json and one AdvisoryData
real = [a for a in advisories if isinstance(a, AdvisoryData)]
assert len(real) == 1
Expand Down Expand Up @@ -126,3 +134,116 @@ def test_get_affected_package_special_and_standard():
pkg2 = p.get_affected_package(data2, "pkg2")
assert isinstance(pkg2.affected_version_range, NpmVersionRange)
assert pkg2.fixed_version == SemverVersion("2.0.1")


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 = NpmImporterPipeline(purl=purl)
pipeline.vcs_response = mock_vcs_response

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 = NpmImporterPipeline(purl=purl)
pipeline.vcs_response = mock_vcs_response

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 = NpmImporterPipeline(purl=purl)
pipeline.vcs_response = mock_vcs_response

advisories = list(pipeline.collect_advisories())

assert len(advisories) == 0


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 = NpmImporterPipeline(purl=purl)
pipeline.vcs_response = mock_vcs_response

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 = NpmImporterPipeline(purl=purl)

affected_package = AffectedPackage(
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

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
Loading
Loading