npm importer - improver migration

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

Author: ziadhany

vulnerabilities/importer.py

@@ -312,10 +312,28 @@ def advisory_data(self) -> Iterable[AdvisoryData]:
         raise NotImplementedError
 
 
+@dataclasses.dataclass
+class GitConfig:
+    repository_url: str
+    branch: Optional[str] = None
+    create_working_directory: bool = True
+    remove_working_directory: bool = True
+    working_directory: Optional[str] = ""
+    last_run_date: Optional[str] = None
+    cutoff_date: Optional[str] = None
+
+
 # TODO: Needs rewrite
 class GitImporter(Importer):
-    def validate_configuration(self) -> None:
+    def __init__(self, config, cutoff_timestamp):
+        super().__init__()
+        self.config = config
+        self.cutoff_timestamp = cutoff_timestamp
 
+        self._ensure_working_directory()
+        self._ensure_repository()
+
+    def validate_configuration(self) -> None:
         if not self.config.create_working_directory and self.config.working_directory is None:
             self.error(
                 '"create_working_directory" is not set but "working_directory" is set to '
@@ -336,10 +354,6 @@ def validate_configuration(self) -> None:
                 "the default, which calls tempfile.mkdtemp()"
             )
 
-    def __enter__(self):
-        self._ensure_working_directory()
-        self._ensure_repository()
-
     def __exit__(self, exc_type, exc_val, exc_tb):
         if self.config.remove_working_directory:
             shutil.rmtree(self.config.working_directory)
@@ -353,7 +367,6 @@ def file_changes(
         """
         Returns all added and modified files since last_run_date or cutoff_date (whichever is more
         recent).
-
         :param subdir: filter by files in this directory
         :param recursive: whether to include files in subdirectories
         :param file_ext: filter files by this extension
@@ -477,6 +490,12 @@ def _update_from_remote(self, remote, branch) -> None:
         branch.set_reference(remote.refs[branch.name])
         self._repo.head.reset(index=True, working_tree=True)
 
+    def advisory_data(self):
+        raise NotImplementedError
+
+    def error(self, param):
+        pass
+
 
 def _include_file(
     path: str,

vulnerabilities/importers/__init__.py

@@ -12,6 +12,7 @@
 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 pysec
@@ -27,6 +28,7 @@
     pysec.PyPIImporter,
     debian.DebianImporter,
     gitlab.GitLabAPIImporter,
+    npm.NpmImporter,
 ]
 
 IMPORTERS_REGISTRY = {x.qualified_name: x for x in IMPORTERS_REGISTRY}

vulnerabilities/importers/npm.py

@@ -8,108 +8,96 @@
 #
 
 # Author: Navonil Das (@NavonilDas)
-
-import asyncio
+import logging
+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
 from dateutil.parser import parse
 from packageurl import PackageURL
-from univers.version_range import VersionRange
+from univers.version_range import NpmVersionRange
+from univers.versions import InvalidVersion
 from univers.versions import SemverVersion
 
 from vulnerabilities.importer import AdvisoryData
+from vulnerabilities.importer import GitConfig
 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__)
 
 
 class NpmImporter(GitImporter):
-    def __enter__(self):
-        super(NpmImporter, self).__enter__()
-        if not getattr(self, "_added_files", None):
-            self._added_files, self._updated_files = self.file_changes(
-                recursive=True, file_ext="json", subdir="./vuln/npm"
-            )
-
-        self._versions = NpmVersionAPI()
-        self.set_api(self.collect_packages())
-
-    def updated_advisories(self) -> Set[AdvisoryData]:
-        files = self._updated_files.union(self._added_files)
-        advisories = []
-        for f in files:
-            processed_data = self.process_file(f)
-            if processed_data:
-                advisories.extend(processed_data)
-        return self.batch_advisories(advisories)
-
-    def set_api(self, packages):
-        asyncio.run(self._versions.load_api(packages))
-
-    def collect_packages(self):
-        packages = set()
-        files = self._updated_files.union(self._added_files)
-        for f in files:
-            data = load_json(f)
-            packages.add(data["module_name"].strip())
-
-        return packages
-
-    @property
-    def versions(self):  # quick hack to make it patchable
-        return self._versions
-
-    def process_file(self, file) -> List[AdvisoryData]:
+    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()
 
-        record = load_json(file)
-        advisories = []
+    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.versions.get(package_name, until=publish_date).valid_versions
+        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 []
+        # 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,
+        )
 
-        for cve_id in record.get("cves") or [""]:
-            advisories.append(
-                AdvisoryData(
-                    summary=record.get("overview", ""),
-                    vulnerability_id=cve_id,
-                    affected_packages=nearest_patched_package(impacted_purls, resolved_purls),
-                    references=vuln_reference,
-                )
-            )
-        return advisories
+    def advisory_data(self) -> Iterable[AdvisoryData]:
+        files = self._updated_files.union(self._added_files)
+        for file in files:
+            record = load_json(file)
+            yield self.parse_advisory_data(record)
 
 
 def _versions_to_purls(package_name, versions):
@@ -150,10 +138,10 @@ def normalize_ranges(version_range_string):
 
 
 def categorize_versions(
-    all_versions: Set[str],
+    all_versions: Iterable[PackageVersion],
     affected_version_range: str,
     fixed_version_range: str,
-) -> Tuple[Set[str], Set[str]]:
+) -> Tuple[Set[SemverVersion], Set[SemverVersion]]:
     """
     Seperate list of affected versions and unaffected versions from all versions
     using the ranges specified.
@@ -168,31 +156,51 @@ def categorize_versions(
     fix_spec = []
 
     if affected_version_range:
-        aff_specs = normalize_ranges(affected_version_range)
-        aff_spec = [
-            VersionRange.from_scheme_version_spec_string("semver", spec)
-            for spec in aff_specs
-            if len(spec) >= 3
-        ]
+        aff_spec = get_version_range(affected_version_range)
 
     if fixed_version_range:
-        fix_specs = normalize_ranges(fixed_version_range)
-        fix_spec = [
-            VersionRange.from_scheme_version_spec_string("semver", spec)
-            for spec in fix_specs
-            if len(spec) >= 3
-        ]
+        fix_spec = get_version_range(fixed_version_range)
+
     aff_ver, fix_ver = set(), set()
+
     # Unaffected version is that version which is in the fixed_version_range
     # or which is absent in the affected_version_range
-    for ver in all_versions:
-        ver_obj = SemverVersion(ver)
+    for ver in get_all_versions(all_versions):
 
-        if not any([ver_obj in spec for spec in aff_spec]) or any(
-            [ver_obj in spec for spec in fix_spec]
-        ):
+        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)
 
     return aff_ver, fix_ver
+
+
+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}")
+
+    return ver_range_objs
+
+
+def get_all_versions(all_versions) -> List[SemverVersion]:
+    """
+
+    Args:
+        all_versions:
+
+    Returns:
+
+    """
+    ver_objs = []
+    for ver in all_versions:
+        try:
+            ver_objs.append(SemverVersion(ver.value))
+        except InvalidVersion:
+            logger.error(f"InvalidVersion {ver.value}")
+    return ver_objs