Broaden vuln coverage across dependency sources

Merge vulnerability results across NVD, GHAD, and CIRCL so we do
not miss advisories that are absent from a single source.

For top-level dependencies, keep the strongest matcher available for
each dependency and add CIRCL product lookups as an extra source.
NVD remains the primary source for curated CPE-based checks, GHAD is
still used for npm-addressable packages, and CIRCL now fills gaps by
returning product-scoped CVE records that we filter locally by the
resolved dependency version.

For vendored npm package trees, keep the installed-package workflow
and extend it with the same multi-source model. Query NVD first,
merge GHAD matches while preferring CVE identifiers and preserving
GHSA aliases, and fall back to CIRCL product search when upstream
advisory coverage is incomplete. This catches cases like
ip-address@10.2.0 where valid CVEs existed outside GHAD coverage.

Also harden reconciliation behavior by carrying advisory aliases
through the emitted vulnerability records, and extend the local test
harness to cover NVD/GHAD merging, CIRCL fallback behavior, and
deterministic network stubbing for GHAD-only tests.

Signed-off-by: Santiago Gimeno <santiago.gimeno@gmail.com>

Author: santigimeno

dep_checker/main.py

@@ -21,14 +21,17 @@
 from gql import gql, Client
 from gql.transport.aiohttp import AIOHTTPTransport
 from nvdlib import searchCVE  # type: ignore
-from packaging.specifiers import SpecifierSet
+from packaging.specifiers import InvalidSpecifier, SpecifierSet
+from packaging.version import InvalidVersion
 from typing import Optional, List
 from pathlib import Path
 
 import json
 import logging
 import sys
 import traceback
+import urllib.parse
+import urllib.request
 
 
 class Vulnerability:
@@ -96,6 +99,10 @@ def default(self, obj):
           vulnerableVersionRange
           advisory {
             ghsaId
+            identifiers {
+              type
+              value
+            }
             permalink
             withdrawnAt
           }
@@ -105,6 +112,229 @@ def default(self, obj):
 """
 )
 
+CIRCL_PRODUCT_URL = "https://vulnerability.circl.lu/api/vulnerability/"
+
+
+def preferred_advisory_id(advisory: dict) -> str:
+    for identifier in advisory.get("identifiers") or []:
+        if identifier.get("type") == "CVE" and identifier.get("value"):
+            return identifier["value"]
+    return advisory["ghsaId"]
+
+
+def advisory_aliases(advisory: dict, preferred_id: str) -> list[str]:
+    aliases: list[str] = []
+    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 merge_vulnerabilities(vulnerabilities: list[Vulnerability]) -> list[Vulnerability]:
+    merged: dict[str, Vulnerability] = {}
+    for vuln in vulnerabilities:
+        candidate_ids = [vuln.id, *(vuln.advisory_aliases or [])]
+        for candidate_id in candidate_ids:
+            existing = merged.get(candidate_id)
+            if existing is None:
+                continue
+            for alias in vuln.advisory_aliases or []:
+                if alias != existing.id and alias not in existing.advisory_aliases:
+                    existing.advisory_aliases.append(alias)
+            break
+        else:
+            merged[vuln.id] = vuln
+    return list(merged.values())
+
+
+def circl_lookup_name(name: str, dep: Dependency) -> str:
+    if dep.cpe is not None:
+        return dep.cpe.product
+    if dep.npm_name is not None:
+        return dep.npm_name
+    if dep.keyword is not None:
+        return dep.keyword
+    return name.lower().replace(" ", "-")
+
+
+def extract_circl_candidates(payload) -> list[dict]:
+    if isinstance(payload, list):
+        return [item for item in payload if isinstance(item, dict)]
+    if not isinstance(payload, dict):
+        return []
+    for key in ("data", "results", "items", "vulnerabilities"):
+        value = payload.get(key)
+        if isinstance(value, list):
+            return [item for item in value if isinstance(item, dict)]
+    return []
+
+
+def extract_circl_candidate_id(item: dict) -> Optional[str]:
+    for key in ("id", "vuln_id", "vulnerability_id", "cve"):
+        value = item.get(key)
+        if isinstance(value, str) and value.startswith(("CVE-", "GHSA-")):
+            return value
+    metadata = item.get("cveMetadata")
+    if isinstance(metadata, dict):
+        value = metadata.get("cveId")
+        if isinstance(value, str) and value.startswith("CVE-"):
+            return value
+    source = (((item.get("containers") or {}).get("cna") or {}).get("source") or {}).get("advisory")
+    if isinstance(source, str) and source.startswith("GHSA-"):
+        return source
+    return None
+
+
+def extract_circl_candidate_url(item: dict, vuln_id: str) -> str:
+    for key in ("url", "href", "permalink"):
+        value = item.get(key)
+        if isinstance(value, str) and value:
+            return value
+    references = (((item.get("containers") or {}).get("cna") or {}).get("references") or [])
+    for reference in references:
+        if isinstance(reference, dict):
+            value = reference.get("url")
+            if isinstance(value, str) and value:
+                return value
+    if vuln_id.startswith("CVE-"):
+        return f"https://www.cve.org/CVERecord?id={vuln_id}"
+    return f"https://github.com/advisories/{vuln_id}"
+
+
+def extract_circl_candidate_summary(item: dict) -> str:
+    cna = ((item.get("containers") or {}).get("cna") or {})
+    for key in ("title",):
+        value = cna.get(key)
+        if isinstance(value, str) and value:
+            return value
+    descriptions = cna.get("descriptions") or []
+    for description in descriptions:
+        if isinstance(description, dict):
+            value = description.get("value")
+            if isinstance(value, str) and value:
+                return value
+    for key in ("summary", "description", "title"):
+        value = item.get(key)
+        if isinstance(value, str) and value:
+            return value
+    return ""
+
+
+def extract_circl_candidate_severity(item: dict) -> Optional[str]:
+    severity = item.get("severity")
+    if isinstance(severity, str) and severity:
+        return severity.upper()
+    metrics = ((item.get("containers") or {}).get("cna") or {}).get("metrics") or []
+    for metric in metrics:
+        if not isinstance(metric, dict):
+            continue
+        for key in ("cvssV4_0", "cvssV3_1", "cvssV3_0"):
+            cvss = metric.get(key)
+            if isinstance(cvss, dict):
+                value = cvss.get("baseSeverity")
+                if isinstance(value, str) and value:
+                    return value.upper()
+    return None
+
+
+def extract_circl_candidate_aliases(item: dict, preferred_id: str) -> list[str]:
+    aliases: list[str] = []
+    for value in item.get("aliases") or []:
+        if isinstance(value, str) and value and value != preferred_id and value not in aliases:
+            aliases.append(value)
+    source = (((item.get("containers") or {}).get("cna") or {}).get("source") or {}).get("advisory")
+    if isinstance(source, str) and source and source != preferred_id and source not in aliases:
+        aliases.append(source)
+    return aliases
+
+
+def circl_version_matches(package_version: str, version_entry: dict) -> bool:
+    status = version_entry.get("status")
+    if status != "affected":
+        return False
+    version = version_entry.get("version")
+    less_than = version_entry.get("lessThan")
+    less_than_or_equal = version_entry.get("lessThanOrEqual")
+    if not isinstance(version, str):
+        return False
+    normalized = version.strip()
+    if normalized.startswith((">", "<", "=")):
+        try:
+            return SpecifierSet(normalized.replace(",", ", ")).contains(package_version, prereleases=True)
+        except InvalidSpecifier:
+            return False
+    clauses = []
+    if normalized not in ("", "*"):
+        clauses.append(f">={normalized}")
+    if isinstance(less_than, str) and less_than not in ("", "*"):
+        clauses.append(f"<{less_than}")
+    if isinstance(less_than_or_equal, str) and less_than_or_equal not in ("", "*"):
+        clauses.append(f"<={less_than_or_equal}")
+    if clauses:
+        try:
+            return SpecifierSet(",".join(clauses)).contains(package_version, prereleases=True)
+        except InvalidSpecifier:
+            return False
+    try:
+        return package_version == normalized
+    except InvalidVersion:
+        return False
+
+
+def circl_candidate_affects_version(item: dict, lookup_name: str, package_version: str) -> bool:
+    affected = (((item.get("containers") or {}).get("cna") or {}).get("affected") or [])
+    lookup_name = lookup_name.lower()
+    for entry in affected:
+        if not isinstance(entry, dict):
+            continue
+        product = str(entry.get("product") or "").lower()
+        package_entry = str(entry.get("packageName") or "").lower()
+        if product != lookup_name and package_entry != lookup_name:
+            continue
+        for version in entry.get("versions") or []:
+            if isinstance(version, dict) and circl_version_matches(package_version, version):
+                return True
+    return False
+
+
+def query_circl(
+    dependencies: dict[str, Dependency], repo_path: Path
+) -> tuple[list[Vulnerability], list[str]]:
+    found_vulnerabilities: list[Vulnerability] = []
+    failures: list[str] = []
+    for name, dep in dependencies.items():
+        version = dep.version_parser(repo_path)
+        lookup_name = circl_lookup_name(name, dep)
+        query = urllib.parse.urlencode({"product": lookup_name, "per_page": 100})
+        request = urllib.request.Request(
+            f"{CIRCL_PRODUCT_URL}?{query}",
+            headers={"Accept": "application/json", "User-Agent": "nsolid-dependency-vuln-assessments"},
+        )
+        try:
+            with urllib.request.urlopen(request, timeout=30) as response:
+                payload = json.load(response)
+        except Exception as exc:
+            failures.append(f"{name}: {exc}")
+            continue
+        for item in extract_circl_candidates(payload):
+            vuln_id = extract_circl_candidate_id(item)
+            if not vuln_id or vuln_id in ignore_list:
+                continue
+            if not circl_candidate_affects_version(item, lookup_name, version):
+                continue
+            found_vulnerabilities.append(
+                Vulnerability(
+                    id=vuln_id,
+                    url=extract_circl_candidate_url(item, vuln_id),
+                    dependency=name,
+                    version=version,
+                    severity=extract_circl_candidate_severity(item),
+                    advisory_aliases=extract_circl_candidate_aliases(item, vuln_id),
+                )
+            )
+    return found_vulnerabilities, failures
+
 
 def resolve_dependencies(
     repo_path: Path, repo_branch: str
@@ -202,16 +432,19 @@ def query_ghad(
             for v in result["securityVulnerabilities"]["nodes"]
             if v["advisory"]["withdrawnAt"] is None
             and dep_version in SpecifierSet(v["vulnerableVersionRange"])
-            and v["advisory"]["ghsaId"] not in ignore_list
+            and preferred_advisory_id(v["advisory"]) not in ignore_list
         ]
         if matching_vulns:
             found_vulnerabilities.extend(
                 [
                     Vulnerability(
-                        id=vuln["advisory"]["ghsaId"],
+                        id=preferred_advisory_id(vuln["advisory"]),
                         url=vuln["advisory"]["permalink"],
                         dependency=name,
                         version=dep_version,
+                        advisory_aliases=advisory_aliases(
+                            vuln["advisory"], preferred_advisory_id(vuln["advisory"])
+                        ),
                     )
                     for vuln in matching_vulns
                 ]
@@ -385,6 +618,22 @@ def main() -> int:
         print(f"Warning: NVD query failed: {e}", file=sys.stderr)
         print(traceback.format_exc(), file=sys.stderr)
 
+    circl_vulnerabilities: list[Vulnerability] = []
+    try:
+        circl_vulnerabilities, circl_failures = query_circl(dependencies, repo_path)
+        if circl_failures:
+            scan_complete = False
+            print(
+                f"Warning: CIRCL query was incomplete; {len(circl_failures)} dependenc(ies) failed:",
+                file=sys.stderr,
+            )
+            for failure in circl_failures:
+                print(f"  - {failure}", file=sys.stderr)
+    except Exception as e:
+        scan_complete = False
+        print(f"Warning: CIRCL query failed: {e}", file=sys.stderr)
+        print(traceback.format_exc(), file=sys.stderr)
+
     # NPM package vulnerability checking
     npm_vulnerabilities: list[Vulnerability] = []
     if include_npm:
@@ -398,7 +647,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
@@ -417,11 +666,14 @@ def main() -> int:
             print(f"Warning: npm vulnerability checking failed: {e}", file=sys.stderr)
             print(f"Traceback: {traceback.format_exc()}", file=sys.stderr)
 
+    merged_vulnerabilities = merge_vulnerabilities(
+        ghad_vulnerabilities + nvd_vulnerabilities + circl_vulnerabilities + npm_vulnerabilities
+    )
     all_vulnerabilities = {
-        "vulnerabilities": ghad_vulnerabilities + nvd_vulnerabilities + npm_vulnerabilities,
+        "vulnerabilities": merged_vulnerabilities,
         "scan_complete": scan_complete,
     }
-    no_vulnerabilities_found = not ghad_vulnerabilities and not nvd_vulnerabilities and not npm_vulnerabilities
+    no_vulnerabilities_found = not merged_vulnerabilities
 
     if scan_file is not None:
         with open(scan_file, "w") as f: