Augment npm scans with GH advisories and NVD data

Pass the NVD key into npm package scanning, query GitHub's global
advisory API for exact installed npm package versions, and enrich CVE
matches with NVD metadata when available.

Update npm audit tests to cover the extra advisory fetch path, alias
preservation during enrichment, and the added import stubs for local
execution.

Author: santigimeno

dep_checker/main.py

@@ -437,7 +437,7 @@ def main() -> int:
 
             from npm_audit import NPMAuditChecker
             print("Running npm package vulnerability audit...", file=sys.stderr)
-            npm_checker = NPMAuditChecker(repo_path, npm_timeout, gh_token=gh_token)
+            npm_checker = NPMAuditChecker(repo_path, npm_timeout, gh_token=gh_token, nvd_key=nvd_key)
             npm_vulnerabilities = npm_checker.check_npm_vulnerabilities(Vulnerability)
             if npm_checker.failed_packages:
                 scan_complete = False

dep_checker/npm_audit.py

@@ -12,16 +12,22 @@
 import subprocess
 import logging
 import re
+import urllib.error
+import urllib.parse
+import urllib.request
 from pathlib import Path
 from typing import List, Dict, Optional
 
 from gql import gql, Client
 from gql.transport.aiohttp import AIOHTTPTransport
+from nvdlib import searchCVE  # type: ignore
 from packaging.specifiers import InvalidSpecifier, SpecifierSet
 from packaging.version import InvalidVersion
 
 logger = logging.getLogger(__name__)
 
+GITHUB_API_VERSION = "2026-03-10"
+
 
 class AuditParseError(Exception):
     """Raised when npm audit returns output that cannot be parsed reliably."""
@@ -80,10 +86,11 @@ class AuditParseError(Exception):
 class NPMAuditChecker:
     """Handles npm audit vulnerability checking for package.json files."""
 
-    def __init__(self, repo_path: Path, timeout: int = 300, gh_token: Optional[str] = None):
+    def __init__(self, repo_path: Path, timeout: int = 300, gh_token: Optional[str] = None, nvd_key: Optional[str] = None):
         self.repo_path = repo_path
         self.timeout = timeout
         self.gh_token = gh_token
+        self.nvd_key = nvd_key
         self.exclude_paths = EXCLUDE_PATHS
         # A failed package-level audit means the aggregate npm result is partial.
         # The caller uses this to prevent reconciliation from closing valid issues.
@@ -307,70 +314,240 @@ def walk(name: str, node: Dict, path_parts: Optional[list[str]] = None) -> None:
     def query_installed_package_vulnerabilities(
         self, package_dir: Path, packages: List[Dict[str, str]], vulnerability_class
     ) -> List:
-        """Query GitHub advisories for exact installed package versions."""
-        if self.gh_token is None:
-            raise RuntimeError("GitHub token is required to scan installed npm package trees")
-
-        transport = AIOHTTPTransport(
-            url="https://api.github.com/graphql",
-            headers={"Authorization": f"bearer {self.gh_token}"},
-        )
-        client = Client(
-            transport=transport,
-            fetch_schema_from_transport=True,
-            serialize_variables=True,
-            parse_results=True,
-        )
-
-        vulnerabilities = []
+        """Query NVD and GitHub advisories for exact installed package versions."""
+        vulnerabilities_by_id: Dict[str, object] = {}
         main_dep_name = package_dir.name
         main_dep_path = str(package_dir.relative_to(self.repo_path))
 
-        for package in packages:
-            try:
-                result = client.execute(
-                    github_vulnerabilities_query,
-                    variable_values={"package_name": package["name"]},
-                )
-            except Exception as exc:
-                logger.warning(
-                    f"Skipping GitHub advisory query for {package['name']}@{package['version']}: {exc}"
-                )
-                continue
-            for vuln in result["securityVulnerabilities"]["nodes"]:
-                if vuln["advisory"]["withdrawnAt"] is not None:
-                    continue
+        def merge_vulnerability(vuln) -> None:
+            candidate_ids = [vuln.id, *(getattr(vuln, "advisory_aliases", []) or [])]
+            for candidate_id in candidate_ids:
+                existing = vulnerabilities_by_id.get(candidate_id)
+                if existing is not None:
+                    aliases = list(getattr(existing, "advisory_aliases", []) or [])
+                    for alias in getattr(vuln, "advisory_aliases", []) or []:
+                        if alias != existing.id and alias not in aliases:
+                            aliases.append(alias)
+                    setattr(existing, "advisory_aliases", aliases)
+                    return
+            vulnerabilities_by_id[vuln.id] = vuln
+
+        packages_for_global_advisories = list(packages)
+        if self.gh_token is not None:
+            transport = AIOHTTPTransport(
+                url="https://api.github.com/graphql",
+                headers={"Authorization": f"bearer {self.gh_token}"},
+            )
+            client = Client(
+                transport=transport,
+                fetch_schema_from_transport=True,
+                serialize_variables=True,
+                parse_results=True,
+            )
+
+            for package in packages:
                 try:
-                    vulnerable_range = self.normalize_version_range(vuln["vulnerableVersionRange"])
-                    if not SpecifierSet(vulnerable_range).contains(package["version"], prereleases=True):
-                        continue
-                except (InvalidSpecifier, InvalidVersion) as exc:
-                    self.failed_packages.append(
-                        f"{package_dir}: invalid advisory match for {package['name']}@{package['version']}: {exc}"
+                    result = client.execute(
+                        github_vulnerabilities_query,
+                        variable_values={"package_name": package["name"]},
                     )
+                except Exception as exc:
                     logger.warning(
-                        f"Skipping advisory match for {package['name']}@{package['version']}: {exc}"
+                        f"Skipping GitHub advisory query for {package['name']}@{package['version']}: {exc}"
                     )
                     continue
-                preferred_id = self.preferred_advisory_id(vuln["advisory"])
-                vulnerabilities.append(
+                for vuln in result["securityVulnerabilities"]["nodes"]:
+                    if vuln["advisory"]["withdrawnAt"] is not None:
+                        continue
+                    try:
+                        vulnerable_range = self.normalize_version_range(vuln["vulnerableVersionRange"])
+                        if not SpecifierSet(vulnerable_range).contains(package["version"], prereleases=True):
+                            continue
+                    except (InvalidSpecifier, InvalidVersion) as exc:
+                        self.failed_packages.append(
+                            f"{package_dir}: invalid advisory match for {package['name']}@{package['version']}: {exc}"
+                        )
+                        logger.warning(
+                            f"Skipping advisory match for {package['name']}@{package['version']}: {exc}"
+                        )
+                        continue
+                    preferred_id = self.preferred_advisory_id(vuln["advisory"])
+                    merge_vulnerability(
+                        vulnerability_class(
+                            id=preferred_id,
+                            url=vuln["advisory"]["permalink"],
+                            dependency=package["name"],
+                            version=package["version"],
+                            source="npm",
+                            severity=vuln.get("severity"),
+                            via=[vuln["advisory"]["summary"]] if vuln["advisory"].get("summary") else [],
+                            fix_available=vuln.get("firstPatchedVersion") is not None,
+                            main_dep_name=main_dep_name,
+                            main_dep_path=main_dep_path,
+                            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:
+            matched_global = self.match_global_advisories(package, global_advisories)
+            for vuln in matched_global:
+                merge_vulnerability(
                     vulnerability_class(
-                        id=preferred_id,
-                        url=vuln["advisory"]["permalink"],
+                        id=vuln["id"],
+                        url=vuln["url"],
                         dependency=package["name"],
                         version=package["version"],
                         source="npm",
                         severity=vuln.get("severity"),
-                        via=[vuln["advisory"]["summary"]] if vuln["advisory"].get("summary") else [],
-                        fix_available=vuln.get("firstPatchedVersion") is not None,
+                        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=self.advisory_aliases(vuln["advisory"], preferred_id),
+                        advisory_aliases=vuln.get("aliases", []),
                     )
                 )
 
-        logger.info(f"Parsed {len(vulnerabilities)} GitHub advisory matches from {package_dir}")
-        return vulnerabilities
+        for package in packages:
+            for vuln in list(vulnerabilities_by_id.values()):
+                if vuln.dependency != package["name"] or not str(vuln.id).startswith("CVE-"):
+                    continue
+                try:
+                    matches = searchCVE(
+                        cveId=vuln.id,
+                        key=self.nvd_key,
+                        delay=6 if self.nvd_key else False,
+                    )
+                except Exception as exc:
+                    self.failed_packages.append(
+                        f"{package_dir}: NVD enrichment failed for {package['name']}@{package['version']} {vuln.id}: {exc}"
+                    )
+                    logger.warning(
+                        f"Skipping NVD enrichment for {package['name']}@{package['version']} {vuln.id}: {exc}"
+                    )
+                    continue
+                if not matches:
+                    continue
+                cve = matches[0]
+                try:
+                    severity = None
+                    if hasattr(cve, "metrics") and cve.metrics:
+                        if hasattr(cve.metrics, 'cvssMetricV31') and cve.metrics.cvssMetricV31:
+                            severity = cve.metrics.cvssMetricV31[0].cvssData.baseSeverity
+                        elif hasattr(cve.metrics, 'cvssMetricV30') and cve.metrics.cvssMetricV30:
+                            severity = cve.metrics.cvssMetricV30[0].cvssData.baseSeverity
+                        elif hasattr(cve.metrics, 'cvssMetricV2') and cve.metrics.cvssMetricV2:
+                            base_score = cve.metrics.cvssMetricV2[0].cvssData.baseScore
+                            severity = "HIGH" if base_score >= 7.0 else "MEDIUM" if base_score >= 4.0 else "LOW"
+                except (AttributeError, IndexError, TypeError):
+                    severity = None
+                if severity is not None:
+                    vuln.severity = severity
+                if getattr(cve, 'url', None):
+                    vuln.url = cve.url
+
+        return list(vulnerabilities_by_id.values())
+
+    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:
+            if advisory.get("withdrawn_at") is not None:
+                continue
+            preferred_id = advisory.get("cve_id") or self.preferred_global_advisory_cve(advisory)
+            if not preferred_id or preferred_id in seen_ids:
+                continue
+            aliases = self.global_advisory_aliases(advisory, preferred_id)
+            for vuln in advisory.get("vulnerabilities") or []:
+                package_info = vuln.get("package") or {}
+                if package_info.get("ecosystem") != "npm" or package_info.get("name") != package["name"]:
+                    continue
+                try:
+                    vulnerable_range = self.normalize_version_range(vuln.get("vulnerable_version_range") or "")
+                    matched = bool(vulnerable_range) and SpecifierSet(vulnerable_range).contains(package["version"], prereleases=True)
+                    if not matched:
+                        continue
+                except (InvalidSpecifier, InvalidVersion):
+                    continue
+                seen_ids.add(preferred_id)
+                results.append({
+                    "id": preferred_id,
+                    "url": advisory.get("html_url") or advisory.get("url") or f"https://github.com/advisories/{advisory.get('ghsa_id', preferred_id)}",
+                    "severity": str(advisory.get("severity") or "").upper() or None,
+                    "summary": advisory.get("summary") or "",
+                    "aliases": aliases,
+                    "fix_available": bool(vuln.get("first_patched_version") or vuln.get("patched_versions")),
+                })
+        return results
+
+    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 = {}
+        requested = tuple(sorted({f"{package['name']}@{package['version']}" for package in packages}))
+        if requested in cache:
+            return cache[requested]
+
+        headers = {
+            "Accept": "application/vnd.github+json",
+            "X-GitHub-Api-Version": GITHUB_API_VERSION,
+            "User-Agent": "nsolid-dependency-vuln-assessments",
+        }
+        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 []:
+            if identifier.get("type") == "CVE" and identifier.get("value"):
+                return identifier["value"]
+        return None
+
+    def global_advisory_aliases(self, advisory: Dict, preferred_id: str) -> list[str]:
+        aliases: list[str] = []
+        ghsa_id = advisory.get("ghsa_id")
+        if isinstance(ghsa_id, str) and ghsa_id and ghsa_id != preferred_id:
+            aliases.append(ghsa_id)
+        for identifier in advisory.get("identifiers") or []:
+            value = identifier.get("value")
+            if value and value != preferred_id and value not in aliases:
+                aliases.append(value)
+        return aliases
 
     def normalize_version_range(self, version_range: str) -> str:
         """Normalize GitHub advisory version syntax to packaging-compatible specifiers."""