Use node js git security advisories instead of using npm registry

Signed-off-by: Shivam Sandbhor <shivam.sandbhor@gmail.com>

Author: sbs2001

vulnerabilities/importers/npm.py

@@ -36,103 +36,71 @@
 from packageurl import PackageURL
 
 from vulnerabilities.data_source import Advisory
-from vulnerabilities.data_source import DataSource
+from vulnerabilities.data_source import GitDataSource
 
 NPM_URL = 'https://registry.npmjs.org{}'
-PAGE = '/-/npm/v1/security/advisories?perPage=100&page=0'
 
 
-class NpmDataSource(DataSource):
+class NpmDataSource(GitDataSource):
 
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self._api_response = None
+    def __enter__(self):
+        super(NpmDataSource, self).__enter__()
         self._versions = VersionAPI()
-        self._added_records, self._updated_records = [], []
-        self._added_advisories, self._updated_advisories = [], []
+        if not getattr(self, '_added_files', None):
+            self._added_files, self._updated_files = self.file_changes(
+                recursive=True, file_ext='json', subdir='./vuln/npm')
+            
 
-    def __enter__(self):
-        self._api_response = self._fetch()
-        self._categorize_records()
+
+    def updated_advisories(self) -> Set[Advisory]:
+        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)
+        print(advisories)
+        return self.batch_advisories(advisories)
 
     @property
     def versions(self):  # quick hack to make it patchable
         return self._versions
 
-    def _fetch(self) -> Mapping[str, Any]:
-        data = None
-        nextpage = PAGE
-        while nextpage:
-            try:
-                with urlopen(NPM_URL.format(nextpage)) as response:
-                    response = json.load(response)
-
-                    if data is None:
-                        data = response
-                    else:
-                        data['objects'].extend(response.get('objects', []))
-
-                nextpage = response.get('urls', {}).get('next')
-
-            except HTTPError as error:
-                if error.code == 404:
-                    return data
-                else:
-                    raise
-
-        return data
-
-    def _categorize_records(self) -> None:
-        for advisory in self._api_response['objects']:
-            created = parse(advisory['created']).timestamp()
-            updated = parse(advisory['updated']).timestamp()
-
-            if created > self.cutoff_timestamp:
-                self._added_records.append(advisory)
-            elif updated > self.cutoff_timestamp:
-                self._updated_records.append(advisory)
-
-    def _parse(self, records: List[Mapping[str, Any]]) -> List[Advisory]:
-        advisories = []
-
-        for record in records:
-            package_name = record['module_name']
-            all_versions = self.versions.get(package_name)
-            aff_range = record.get('vulnerable_versions', '')
-            fixed_range = record.get('patched_versions', '')
-
-            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)
-
-            for cve_id in record.get('cves') or ['']:
-                advisories.append(Advisory(
-                    summary=record.get('overview', ''),
-                    cve_id=cve_id,
-                    impacted_package_urls=impacted_purls,
-                    resolved_package_urls=resolved_purls,
-                    reference_urls=[NPM_URL.format(f'/-/npm/v1/advisories/{record["id"]}')],
-                ))
+    def process_file(self, file) -> List[Advisory]:
+        print(file)
+        with open(file) as f:
+            record = json.load(f)
+        print(record)
+        advisories = []        
+        package_name = record['module_name']
+        all_versions = self.versions.get(package_name)
+        aff_range = record.get('vulnerable_versions', '')
+        fixed_range = record.get('patched_versions', '')
+
+        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)
+
+        for cve_id in record.get('cves') or ['']:
+            advisories.append(Advisory(
+                summary=record.get('overview', ''),
+                cve_id=cve_id,
+                impacted_package_urls=impacted_purls,
+                resolved_package_urls=resolved_purls,
+                reference_urls=[NPM_URL.format(f'/-/npm/v1/advisories/{record["id"]}')],
+            ))
 
         return advisories
 
-    def added_advisories(self) -> Set[Advisory]:
-        return self.batch_advisories(self._parse(self._added_records))
-
-    def updated_advisories(self) -> Set[Advisory]:
-        return self.batch_advisories(self._parse(self._updated_records))
-
-
 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 categorize_versions(
         all_versions: Set[str],
         aff_version_range: str,

vulnerabilities/tests/test_data/npm.zip

@@ -22,7 +22,10 @@
 #  Visit https://github.com/nexB/vulnerablecode/ for support and download.
 import json
 import os
+import shutil
+import tempfile
 from unittest.mock import patch
+import zipfile
 
 from django.test import TestCase
 
@@ -42,34 +45,43 @@
 })
 
 
-@patch('vulnerabilities.importers.NpmDataSource.versions', new=MOCK_VERSION_API)
+@patch('vulnerabilities.importers.NpmDataSource._update_from_remote')
 class NpmImportTest(TestCase):
+
+    tempdir = None
+
     @classmethod
     def setUpClass(cls) -> None:
-        fixture_path = os.path.join(TEST_DATA, 'npm_test.json')
-        with open(fixture_path) as f:
-            cls.mock_response = json.load(f)
+        cls.tempdir = tempfile.mkdtemp()
+        zip_path = os.path.join(TEST_DATA, 'npm.zip')
+
+        with zipfile.ZipFile(zip_path, "r") as zip_ref:
+            zip_ref.extractall(cls.tempdir)
 
         cls.importer = models.Importer.objects.create(
             name='npm_unittests',
             license='',
             last_run=None,
             data_source='NpmDataSource',
-            data_source_cfg={},
+            data_source_cfg={
+                'repository_url' : 'https://example.git',
+                'working_directory': os.path.join(cls.tempdir, 'npm_test'),
+                'create_working_directory': False,
+                'remove_working_directory': False
+                }
         )
 
     @classmethod
     def tearDownClass(cls) -> None:
         # Make sure no requests for unexpected package names have been made during the tests.
+        shutil.rmtree(cls.tempdir)
         assert len(MOCK_VERSION_API.cache) == 3, MOCK_VERSION_API.cache
 
-    def test_import(self):
+    def test_import(self, _):
         runner = ImportRunner(self.importer, 5)
 
-        with patch(
-                'vulnerabilities.importers.NpmDataSource._fetch',
-                return_value=self.mock_response
-        ):
+
+        with patch('vulnerabilities.importers.NpmDataSource.versions', new=MOCK_VERSION_API):
             runner.run()
 
         assert models.Vulnerability.objects.count() == 3