fixup

Author: santigimeno

dep_checker/npm_audit.py

@@ -497,25 +497,41 @@ def fetch_global_advisories(self, packages: List[Dict[str, str]]) -> List[Dict]:
         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", ",".join(requested)),
-            ("per_page", "100"),
-        ])
-        url = f"https://api.github.com/advisories?{query}"
+
         headers = {
             "Accept": "application/vnd.github+json",
             "X-GitHub-Api-Version": GITHUB_API_VERSION,
             "User-Agent": "nsolid-dependency-vuln-assessments",
         }
-        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[requested] = [item for item in payload if isinstance(item, dict)]
-        return cache[requested]
+        advisories: List[Dict] = []
+        seen_ids: set[str] = set()
+        batch_size = 25
+
+        for index in range(0, len(requested), batch_size):
+            batch = requested[index:index + batch_size]
+            query = urllib.parse.urlencode(
+                [("ecosystem", "npm"), *( ("affects[]", item) for item in batch ), ("per_page", "100")]
+            )
+            url = f"https://api.github.com/advisories?{query}"
+            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 batch {index // batch_size + 1} of {((len(requested) - 1) // batch_size) + 1}: {payload}"
+                )
+                continue
+            for item in payload:
+                if not isinstance(item, dict):
+                    continue
+                advisory_id = item.get("ghsa_id") or item.get("cve_id") or id(item)
+                if advisory_id in seen_ids:
+                    continue
+                seen_ids.add(advisory_id)
+                advisories.append(item)
+
+        cache[requested] = advisories
+        return advisories
 
     def preferred_global_advisory_cve(self, advisory: Dict) -> Optional[str]:
         for identifier in advisory.get("identifiers") or []:

dep_checker/test_npm_audit.py

@@ -393,6 +393,37 @@ def read(self):
         npm_audit.urllib.request.urlopen = original_urlopen
 
 
+
+def test_fetch_global_advisories_uses_repeated_affects_params() -> None:
+    class FakeResponse:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, tb):
+            return False
+
+        def read(self):
+            return b'[]'
+
+    seen = {}
+    original_urlopen = npm_audit.urllib.request.urlopen
+
+    def fake_urlopen(request, timeout=None):
+        seen["url"] = request.full_url
+        return FakeResponse()
+
+    npm_audit.urllib.request.urlopen = fake_urlopen
+    try:
+        checker = NPMAuditChecker(Path("/tmp"), timeout=60)
+        checker.fetch_global_advisories([
+            {"name": "ip-address", "version": "10.2.0"},
+            {"name": "lodash", "version": "4.17.21"},
+        ])
+        assert "affects%5B%5D=ip-address%4010.2.0" in seen["url"], seen
+        assert "affects%5B%5D=lodash%404.17.21" in seen["url"], seen
+    finally:
+        npm_audit.urllib.request.urlopen = original_urlopen
+
 def test_global_advisories_are_merged_with_ghad_results() -> None:
     class FakeClient:
         def execute(self, query, variable_values):
@@ -769,6 +800,7 @@ def test_npm_audit_basic() -> None:
     test_enolock_falls_back_to_install_and_normal_audit()
     test_nvd_primary_merges_ghad_aliases()
     test_fetch_global_advisories_returns_empty_for_non_list_payload()
+    test_fetch_global_advisories_uses_repeated_affects_params()
     test_global_advisories_are_merged_with_ghad_results()
     test_query_failure_is_skipped_per_package()
     test_invalid_advisory_range_or_version_is_skipped()