fixup

Author: santigimeno

dep_checker/npm_audit.py

@@ -388,30 +388,31 @@ def merge_vulnerability(vuln) -> None:
                             advisory_aliases=self.advisory_aliases(vuln["advisory"], preferred_id),
                         )
                     )
+        try:
+            global_advisories = self.fetch_global_advisories(packages_for_global_advisories)
+        except Exception as exc:
+            self.failed_packages.append(
+                f"{package_dir}: global advisory query failed: {exc}"
+            )
+            logger.warning(f"Skipping global advisory query for {package_dir}: {exc}")
+            global_advisories = []
+
         for package in packages_for_global_advisories:
-            try:
-                for vuln in self.query_global_advisory_vulnerabilities(package):
-                    merge_vulnerability(
-                        vulnerability_class(
-                            id=vuln["id"],
-                            url=vuln["url"],
-                            dependency=package["name"],
-                            version=package["version"],
-                            source="npm",
-                            severity=vuln.get("severity"),
-                            via=[vuln["summary"]] if vuln.get("summary") else [],
-                            fix_available=vuln.get("fix_available"),
-                            main_dep_name=main_dep_name,
-                            main_dep_path=main_dep_path,
-                            advisory_aliases=vuln.get("aliases", []),
-                        )
+            for vuln in self.match_global_advisories(package, global_advisories):
+                merge_vulnerability(
+                    vulnerability_class(
+                        id=vuln["id"],
+                        url=vuln["url"],
+                        dependency=package["name"],
+                        version=package["version"],
+                        source="npm",
+                        severity=vuln.get("severity"),
+                        via=[vuln["summary"]] if vuln.get("summary") else [],
+                        fix_available=vuln.get("fix_available"),
+                        main_dep_name=main_dep_name,
+                        main_dep_path=main_dep_path,
+                        advisory_aliases=vuln.get("aliases", []),
                     )
-            except Exception as exc:
-                self.failed_packages.append(
-                    f"{package_dir}: global advisory query failed for {package['name']}@{package['version']}: {exc}"
-                )
-                logger.warning(
-                    f"Skipping global advisory query for {package['name']}@{package['version']}: {exc}"
                 )
 
         for package in packages:
@@ -456,8 +457,7 @@ def merge_vulnerability(vuln) -> None:
         logger.info(f"Parsed {len(vulnerabilities)} merged advisory matches from {package_dir}")
         return vulnerabilities
 
-    def query_global_advisory_vulnerabilities(self, package: Dict[str, str]) -> List[Dict[str, object]]:
-        advisories = self.fetch_global_advisories(package["name"], package["version"])
+    def match_global_advisories(self, package: Dict[str, str], advisories: List[Dict]) -> List[Dict[str, object]]:
         results: List[Dict[str, object]] = []
         seen_ids: set[str] = set()
         for advisory in advisories:
@@ -488,16 +488,18 @@ def query_global_advisory_vulnerabilities(self, package: Dict[str, str]) -> List
                 })
         return results
 
-    def fetch_global_advisories(self, package_name: str, version: str) -> List[Dict]:
+    def fetch_global_advisories(self, packages: List[Dict[str, str]]) -> List[Dict]:
+        if not packages:
+            return []
         cache = getattr(self, "_global_advisory_cache", None)
         if cache is None:
             cache = self._global_advisory_cache = {}
-        key = (package_name, version)
-        if key in cache:
-            return cache[key]
+        requested = tuple(sorted({f"{package['name']}@{package['version']}" for package in packages}))
+        if requested in cache:
+            return cache[requested]
         query = urllib.parse.urlencode([
             ("ecosystem", "npm"),
-            ("affects[]", f"{package_name}@{version}"),
+            ("affects", ",".join(requested)),
             ("per_page", "100"),
         ])
         url = f"https://api.github.com/advisories?{query}"
@@ -506,15 +508,14 @@ def fetch_global_advisories(self, package_name: str, version: str) -> List[Dict]
             "X-GitHub-Api-Version": GITHUB_API_VERSION,
             "User-Agent": "nsolid-dependency-vuln-assessments",
         }
-        if self.gh_token is not None:
-            headers["Authorization"] = f"Bearer {self.gh_token}"
         request = urllib.request.Request(url, headers=headers)
         with urllib.request.urlopen(request, timeout=min(self.timeout, 30)) as response:
             payload = json.load(response)
         if not isinstance(payload, list):
+            logger.warning(f"Global advisory query returned non-list payload for {len(requested)} packages: {payload}")
             return []
-        cache[key] = [item for item in payload if isinstance(item, dict)]
-        return cache[key]
+        cache[requested] = [item for item in payload if isinstance(item, dict)]
+        return cache[requested]
 
     def preferred_global_advisory_cve(self, advisory: Dict) -> Optional[str]:
         for identifier in advisory.get("identifiers") or []:

dep_checker/test_npm_audit.py

@@ -388,7 +388,7 @@ def read(self):
     npm_audit.urllib.request.urlopen = lambda *args, **kwargs: FakeResponse()
     try:
         checker = NPMAuditChecker(Path("/tmp"), timeout=60, gh_token="token")
-        assert checker.fetch_global_advisories("example", "1.0.0") == []
+        assert checker.fetch_global_advisories([{"name": "example", "version": "1.0.0"}]) == []
     finally:
         npm_audit.urllib.request.urlopen = original_urlopen