fixup

Author: santigimeno

dep_checker/main.py

@@ -21,17 +21,14 @@
 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
+from packaging.specifiers import SpecifierSet
 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:
@@ -112,9 +109,6 @@ 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"):
@@ -148,194 +142,6 @@ def merge_vulnerabilities(vulnerabilities: list[Vulnerability]) -> list[Vulnerab
     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
 ) -> tuple[dict[str, Dependency], list[str]]:
@@ -618,22 +424,6 @@ 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:
@@ -667,7 +457,7 @@ def main() -> int:
             print(f"Traceback: {traceback.format_exc()}", file=sys.stderr)
 
     merged_vulnerabilities = merge_vulnerabilities(
-        ghad_vulnerabilities + nvd_vulnerabilities + circl_vulnerabilities + npm_vulnerabilities
+        ghad_vulnerabilities + nvd_vulnerabilities + npm_vulnerabilities
     )
     all_vulnerabilities = {
         "vulnerabilities": merged_vulnerabilities,