fix npm affected package , add npm test

Signed-off-by: ziadhany <ziadhany2016@gmail.com>

Author: ziadhany

requirements.txt

@@ -106,7 +106,7 @@ toml==0.10.2
 tomli==2.0.1
 traitlets==5.1.1
 typing_extensions==4.1.1
-univers==30.7.0
+univers==30.8.0
 urllib3==1.26.9
 wcwidth==0.2.5
 websocket-client==0.59.0

vulnerabilities/importers/__init__.py

@@ -12,8 +12,10 @@
 from vulnerabilities.importers import github
 from vulnerabilities.importers import gitlab
 from vulnerabilities.importers import nginx
+from vulnerabilities.importers import npm
 from vulnerabilities.importers import nvd
 from vulnerabilities.importers import openssl
+from vulnerabilities.importers import pypa
 from vulnerabilities.importers import pysec
 from vulnerabilities.importers import redhat
 
@@ -27,6 +29,8 @@
     pysec.PyPIImporter,
     debian.DebianImporter,
     gitlab.GitLabGitImporter,
+    pypa.PyPaImporter,
+    npm.NpmImporter,
 ]
 
 IMPORTERS_REGISTRY = {x.qualified_name: x for x in IMPORTERS_REGISTRY}

vulnerabilities/importers/npm.py

@@ -9,11 +9,10 @@
 
 # Author: Navonil Das (@NavonilDas)
 import logging
+from pathlib import Path
 from typing import Iterable
 from typing import List
 from typing import Optional
-from typing import Set
-from typing import Tuple
 from urllib.parse import quote
 
 import pytz
@@ -24,13 +23,11 @@
 from univers.versions import SemverVersion
 
 from vulnerabilities.importer import AdvisoryData
-from vulnerabilities.importer import GitConfig
+from vulnerabilities.importer import AffectedPackage
 from vulnerabilities.importer import GitImporter
 from vulnerabilities.importer import Reference
 from vulnerabilities.package_managers import NpmVersionAPI
-from vulnerabilities.package_managers import PackageVersion
 from vulnerabilities.utils import load_json
-from vulnerabilities.utils import nearest_patched_package
 
 NPM_URL = "https://registry.npmjs.org{}"
 logger = logging.getLogger(__name__)
@@ -39,163 +36,73 @@
 class NpmImporter(GitImporter):
     license_url = "https://github.com/nodejs/security-wg/blob/main/LICENSE.md"
     spdx_license_expression = "MIT"
-    config = GitConfig(
-        repository_url="https://github.com/nodejs/security-wg.git",
-        working_directory="npm",
-        branch="main",
-    )
-    cutoff_timestamp = 1
 
     def __init__(self):
-        super().__init__(config=self.config, cutoff_timestamp=self.cutoff_timestamp)
-        self._added_files, self._updated_files = self.file_changes(
-            recursive=True, file_ext="json", subdir="vuln/npm"
-        )
-        self.pkg_manager_api = NpmVersionAPI()
-
-    def parse_advisory_data(self, record) -> Optional[AdvisoryData]:
-        package_name = record["module_name"].strip()
-
-        publish_date = parse(record["updated_at"])
-        publish_date = publish_date.replace(tzinfo=pytz.UTC)
-
-        all_versions = self.pkg_manager_api.fetch(package_name)
-        aff_range = record.get("vulnerable_versions")
-        if not aff_range:
-            aff_range = ""
-        fixed_range = record.get("patched_versions")
-        if not fixed_range:
-            fixed_range = ""
-
-        # if aff_range == "*" or fixed_range == "*":
-        #     return None
-
-        impacted_versions, resolved_versions = categorize_versions(
-            all_versions, aff_range, fixed_range
-        )
-
-        impacted_purls = _versions_to_purls(package_name, impacted_versions)
-        resolved_purls = _versions_to_purls(package_name, resolved_versions)
-
-        vuln_reference = [
-            Reference(
-                url=NPM_URL.format(f'/-/npm/v1/advisories/{record["id"]}'),
-                reference_id=record["id"],
-            )
-        ]
-        cve_id = record.get("cves") or []
-
-        return AdvisoryData(
-            aliases=cve_id,
-            summary=record.get("overview", ""),
-            affected_packages=nearest_patched_package(impacted_purls, resolved_purls),
-            references=vuln_reference,
-            date_published=publish_date,
-        )
+        super().__init__(repo_url="git+https://github.com/nodejs/security-wg.git")
 
     def advisory_data(self) -> Iterable[AdvisoryData]:
-        files = self._updated_files.union(self._added_files)
+        self.clone()
+        path = Path(self.vcs_response.dest_dir)
+
+        glob = "vuln/npm/**/*.json"  # subdir="vuln/npm"
+        files = (p for p in path.glob(glob) if p.is_file())
         for file in files:
+            print(file)
             record = load_json(file)
-            yield self.parse_advisory_data(record)
-
-
-def _versions_to_purls(package_name, versions):
-    purls = {f"pkg:npm/{quote(package_name)}@{v}" for v in versions}
-    return [PackageURL.from_string(s) for s in purls]
-
-
-def normalize_ranges(version_range_string):
-    """
-    - Splits version range strings with "||" operator into separate ranges.
-    - Removes spaces between range operator and range operands
-    - Normalizes 'x' ranges
-    Example:
-    >>> z = normalize_ranges(">=6.1.3 < 7.0.0 || >=7.0.3")
-    >>> assert z == [">=6.1.3,<7.0.0", ">=7.0.3"]
-    """
-
-    version_ranges = version_range_string.split("||")
-    version_ranges = list(map(str.strip, version_ranges))
-    for id, version_range in enumerate(version_ranges):
-
-        # TODO: This is cryptic, simplify this if possible
-        version_ranges[id] = ",".join(version_range.split())
-        version_ranges[id] = version_ranges[id].replace(">=,", ">=")
-        version_ranges[id] = version_ranges[id].replace("<=,", "<=")
-        version_ranges[id] = version_ranges[id].replace("<=,", "<=")
-        version_ranges[id] = version_ranges[id].replace("<,", "<")
-        version_ranges[id] = version_ranges[id].replace(">,", ">")
-
-        # "x" is interpretted as wild card character here. These are not part of semver
-        # spec. We replace the "x" with aribitarily large number to simulate the effect.
-        if ".x." in version_ranges[id]:
-            version_ranges[id] = version_ranges[id].replace(".x", ".10000.0")
-        if ".x" in version_ranges[id]:
-            version_ranges[id] = version_ranges[id].replace(".x", ".10000")
-
-    return version_ranges
-
-
-def categorize_versions(
-    all_versions: Iterable[PackageVersion],
-    affected_version_range: str,
-    fixed_version_range: str,
-) -> Tuple[Set[SemverVersion], Set[SemverVersion]]:
-    """
-    Seperate list of affected versions and unaffected versions from all versions
-    using the ranges specified.
-
-    :return: impacted, resolved versions
-    """
-    if not all_versions:
-        # NPM registry has no data regarding this package, we skip these
-        return set(), set()
-
-    aff_spec = []
-    fix_spec = []
-
-    if affected_version_range:
-        aff_spec = get_version_range(affected_version_range)
-
-    if fixed_version_range:
-        fix_spec = get_version_range(fixed_version_range)
-
-    aff_ver, fix_ver = set(), set()
+            yield parse_advisory_data(record)
 
-    # Unaffected version is that version which is in the fixed_version_range
-    # or which is absent in the affected_version_range
-    for ver in get_all_versions(all_versions):
 
-        if not any([ver in spec for spec in aff_spec]) or any([ver in spec for spec in fix_spec]):
-            fix_ver.add(ver)
-        else:
-            aff_ver.add(ver)
+def parse_advisory_data(record) -> Optional[AdvisoryData]:
+    cves = record.get("cves") or []
+    overview = record.get("overview", "")
+    package_name = record["module_name"].strip()
 
-    return aff_ver, fix_ver
+    publish_date = parse(record["updated_at"])
+    publish_date = publish_date.replace(tzinfo=pytz.UTC)
 
+    pkg_manager_api = NpmVersionAPI()
+    all_versions = pkg_manager_api.fetch(package_name)
+    aff_range = record.get("vulnerable_versions") or ""
+    fixed_range = record.get("patched_versions") or ""
 
-def get_version_range(version_range) -> List[NpmVersionRange]:
-    fix_specs = normalize_ranges(version_range)
-    ver_range_objs = []
-    for spec in fix_specs:
-        if len(spec) >= 3:
-            try:
-                ver_range_objs.append(NpmVersionRange.from_string(f"vers:npm/{spec}"))
-            except InvalidVersion:
-                logger.error(f"InvalidVersionRange {spec}")
+    fixed_versions = get_fixed_version(
+        map_all_versions(all_versions), NpmVersionRange.from_native(fixed_range)
+    )
+    # if aff_range == "*" or fixed_range == "*":
+    #    return None
 
-    return ver_range_objs
+    vuln_reference = [
+        Reference(
+            url=NPM_URL.format(f'/-/npm/v1/advisories/{record["id"]}'),
+            reference_id=record["id"],
+        )
+    ]
+
+    return AdvisoryData(
+        aliases=cves,
+        summary=overview,
+        affected_packages=[
+            AffectedPackage(
+                package=PackageURL.from_string(f"pkg:npm/{quote(package_name)}"),
+                affected_version_range=NpmVersionRange.from_native(aff_range),
+                fixed_version=fixed_version,
+            )
+            for fixed_version in fixed_versions
+        ],
+        references=vuln_reference,
+        date_published=publish_date,
+    )
 
 
-def get_all_versions(all_versions) -> List[SemverVersion]:
+def map_all_versions(all_versions) -> List[SemverVersion]:
     """
+    map all versions from PackageVersion to SemverVersion
 
-    Args:
-        all_versions:
+    Parameters:
+    all_versions (PackageVersion): List of PackageVersion
 
     Returns:
-
+     List[SemverVersion]: return a list of SemverVersion
     """
     ver_objs = []
     for ver in all_versions:
@@ -204,3 +111,22 @@ def get_all_versions(all_versions) -> List[SemverVersion]:
         except InvalidVersion:
             logger.error(f"InvalidVersion {ver.value}")
     return ver_objs
+
+
+def get_fixed_version(
+    all_versions: List[SemverVersion], aff_range: NpmVersionRange
+) -> List[SemverVersion]:
+    """
+    return a list of SemverVersion fixed versions
+    """
+    try:
+        fixed_versions = []
+        if not all_versions or not all_versions:
+            return fixed_versions
+        for v in all_versions:
+            if v in aff_range:
+                fixed_versions.append(v)
+        return fixed_versions
+    except Exception as e:
+        logger.error(e)
+        return []

vulnerabilities/tests/test_data/npm.zip

@@ -0,0 +1,62 @@
+{
+  "aliases": [],
+  "summary": "Versions of `@hapi/subtext` prior to 6.1.3 or 7.0.3 are vulnerable to Denial of Service. The Content-Encoding HTTP header parser has a vulnerability which will cause the function to throw a system error if the header contains some invalid values. Because hapi rethrows system errors (as opposed to catching expected application errors), the error is thrown all the way up the stack. If no unhandled exception handler is available, the application will exist, allowing an attacker to shut down services.",
+  "affected_packages": [
+    {
+      "package": {
+        "type": "npm",
+        "namespace": "@hapi",
+        "name": "subtext",
+        "version": null,
+        "qualifiers": null,
+        "subpath": null
+      },
+      "affected_version_range": "vers:npm/>=4.1.0|<6.1.3|>=7.0.0|<7.0.3",
+      "fixed_version": "6.1.3"
+    },
+    {
+      "package": {
+        "type": "npm",
+        "namespace": "@hapi",
+        "name": "subtext",
+        "version": null,
+        "qualifiers": null,
+        "subpath": null
+      },
+      "affected_version_range": "vers:npm/>=4.1.0|<6.1.3|>=7.0.0|<7.0.3",
+      "fixed_version": "7.0.3"
+    },
+    {
+      "package": {
+        "type": "npm",
+        "namespace": "@hapi",
+        "name": "subtext",
+        "version": null,
+        "qualifiers": null,
+        "subpath": null
+      },
+      "affected_version_range": "vers:npm/>=4.1.0|<6.1.3|>=7.0.0|<7.0.3",
+      "fixed_version": "7.0.4"
+    },
+    {
+      "package": {
+        "type": "npm",
+        "namespace": "@hapi",
+        "name": "subtext",
+        "version": null,
+        "qualifiers": null,
+        "subpath": null
+      },
+      "affected_version_range": "vers:npm/>=4.1.0|<6.1.3|>=7.0.0|<7.0.3",
+      "fixed_version": "8.0.0"
+    }
+  ],
+  "references": [
+    {
+      "reference_id": 1476,
+      "url": "https://registry.npmjs.org/-/npm/v1/advisories/1476",
+      "severities": []
+    }
+  ],
+  "date_published": "2020-02-18T18:00:29.843000+00:00"
+}