Skip to content
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
7 changes: 7 additions & 0 deletions vulnerabilities/importers/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -117,3 +118,9 @@
oss_fuzz.OSSFuzzImporter,
]
)

LIVE_IMPORTERS_REGISTRY = create_registry(
[
npm_live_importer_v2.NpmLiveImporterPipeline,
]
)
88 changes: 88 additions & 0 deletions vulnerabilities/pipelines/v2_importers/npm_live_importer.py
Original file line number Diff line number Diff line change
@@ -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
Original file line number Diff line number Diff line change
@@ -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