fixup

Author: santigimeno

dep_checker/npm_audit.py

@@ -27,6 +27,7 @@
 logger = logging.getLogger(__name__)
 
 GITHUB_API_VERSION = "2026-03-10"
+TRACE_PACKAGES = {"ip-address"}
 
 
 class AuditParseError(Exception):
@@ -310,7 +311,16 @@ def walk(name: str, node: Dict, path_parts: Optional[list[str]] = None) -> None:
             if name in roots and isinstance(node, dict):
                 walk(name, node, [])
 
-        return list(seen.values())
+        packages = list(seen.values())
+        logger.info(
+            f"Collected {len(packages)} installed bundled npm packages from {package_dir} (roots={len(roots)})"
+        )
+        for traced in packages:
+            if traced["name"] in TRACE_PACKAGES:
+                logger.info(
+                    f"Trace package present in installed npm tree: {traced['name']}@{traced['version']} path={traced['path']}"
+                )
+        return packages
 
     def query_installed_package_vulnerabilities(
         self, package_dir: Path, packages: List[Dict[str, str]], vulnerability_class
@@ -334,6 +344,14 @@ def merge_vulnerability(vuln) -> None:
             vulnerabilities_by_id[vuln.id] = vuln
 
         packages_for_global_advisories = list(packages)
+        logger.info(
+            f"Querying npm advisories for {len(packages_for_global_advisories)} installed packages from {package_dir}"
+        )
+        for traced in packages_for_global_advisories:
+            if traced["name"] in TRACE_PACKAGES:
+                logger.info(
+                    f"Trace package entering advisory queries: {traced['name']}@{traced['version']} path={traced.get('path', '<unknown>')}"
+                )
         if self.gh_token is not None:
             transport = AIOHTTPTransport(
                 url="https://api.github.com/graphql",
@@ -357,6 +375,10 @@ def merge_vulnerability(vuln) -> None:
                         f"Skipping GitHub advisory query for {package['name']}@{package['version']}: {exc}"
                     )
                     continue
+                if package["name"] in TRACE_PACKAGES:
+                    logger.info(
+                        f"Trace package GHAD results for {package['name']}@{package['version']}: {len(result['securityVulnerabilities']['nodes'])} candidates"
+                    )
                 for vuln in result["securityVulnerabilities"]["nodes"]:
                     if vuln["advisory"]["withdrawnAt"] is not None:
                         continue
@@ -373,6 +395,10 @@ def merge_vulnerability(vuln) -> None:
                         )
                         continue
                     preferred_id = self.preferred_advisory_id(vuln["advisory"])
+                    if package["name"] in TRACE_PACKAGES:
+                        logger.info(
+                            f"Trace package GHAD match: {package['name']}@{package['version']} matched {preferred_id} range={vuln['vulnerableVersionRange']}"
+                        )
                     merge_vulnerability(
                         vulnerability_class(
                             id=preferred_id,
@@ -390,6 +416,9 @@ def merge_vulnerability(vuln) -> None:
                     )
         try:
             global_advisories = self.fetch_global_advisories(packages_for_global_advisories)
+            logger.info(
+                f"Fetched {len(global_advisories)} global advisories for {len(packages_for_global_advisories)} installed packages from {package_dir}"
+            )
         except Exception as exc:
             self.failed_packages.append(
                 f"{package_dir}: global advisory query failed: {exc}"
@@ -398,7 +427,12 @@ def merge_vulnerability(vuln) -> None:
             global_advisories = []
 
         for package in packages_for_global_advisories:
-            for vuln in self.match_global_advisories(package, global_advisories):
+            matched_global = self.match_global_advisories(package, global_advisories)
+            if package["name"] in TRACE_PACKAGES:
+                logger.info(
+                    f"Trace package global advisory matches for {package['name']}@{package['version']}: {len(matched_global)}"
+                )
+            for vuln in matched_global:
                 merge_vulnerability(
                     vulnerability_class(
                         id=vuln["id"],
@@ -460,6 +494,10 @@ def merge_vulnerability(vuln) -> None:
     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()
+        if package["name"] in TRACE_PACKAGES:
+            logger.info(
+                f"Trace package evaluating {len(advisories)} global advisories for {package['name']}@{package['version']}"
+            )
         for advisory in advisories:
             if advisory.get("withdrawn_at") is not None:
                 continue
@@ -473,9 +511,18 @@ def match_global_advisories(self, package: Dict[str, str], advisories: List[Dict
                     continue
                 try:
                     vulnerable_range = self.normalize_version_range(vuln.get("vulnerable_version_range") or "")
-                    if not vulnerable_range or not SpecifierSet(vulnerable_range).contains(package["version"], prereleases=True):
+                    matched = bool(vulnerable_range) and SpecifierSet(vulnerable_range).contains(package["version"], prereleases=True)
+                    if package["name"] in TRACE_PACKAGES:
+                        logger.info(
+                            f"Trace package global advisory candidate: {package['name']}@{package['version']} advisory={preferred_id} range={vuln.get('vulnerable_version_range')} normalized={vulnerable_range!r} matched={matched}"
+                        )
+                    if not matched:
                         continue
-                except (InvalidSpecifier, InvalidVersion):
+                except (InvalidSpecifier, InvalidVersion) as exc:
+                    if package["name"] in TRACE_PACKAGES:
+                        logger.warning(
+                            f"Trace package global advisory parse failure for {package['name']}@{package['version']} advisory={preferred_id}: {exc}"
+                        )
                     continue
                 seen_ids.add(preferred_id)
                 results.append({
@@ -509,6 +556,12 @@ def fetch_global_advisories(self, packages: List[Dict[str, str]]) -> List[Dict]:
 
         for index in range(0, len(requested), batch_size):
             batch = requested[index:index + batch_size]
+            logger.info(
+                f"Fetching global advisories batch {index // batch_size + 1} of {((len(requested) - 1) // batch_size) + 1} with {len(batch)} packages"
+            )
+            traced_batch = [item for item in batch if item.split('@', 1)[0] in TRACE_PACKAGES]
+            if traced_batch:
+                logger.info(f"Trace package present in global advisory batch: {traced_batch}")
             query = urllib.parse.urlencode(
                 [("ecosystem", "npm"), *( ("affects[]", item) for item in batch ), ("per_page", "100")]
             )
@@ -521,6 +574,15 @@ def fetch_global_advisories(self, packages: List[Dict[str, str]]) -> List[Dict]:
                     f"Global advisory query returned non-list payload for batch {index // batch_size + 1} of {((len(requested) - 1) // batch_size) + 1}: {payload}"
                 )
                 continue
+            logger.info(
+                f"Global advisory query returned {len(payload)} advisory records for batch {index // batch_size + 1}"
+            )
+            if traced_batch:
+                for item in payload:
+                    if isinstance(item, dict):
+                        logger.info(
+                            f"Trace package batch advisory seen: ghsa={item.get('ghsa_id')} cve={item.get('cve_id')} summary={item.get('summary', '')[:120]}"
+                        )
             for item in payload:
                 if not isinstance(item, dict):
                     continue