cve_search_cli.py

#!/usr/bin/env python3
import requests
import argparse
import sys
import os
import subprocess
from pathlib import Path
import json
import time
import random
from typing import Any, Dict, List, Optional, Tuple
from contextlib import nullcontext
from datetime import date, datetime, timedelta
try:
    from rich import print
    from rich.console import Console
    from rich.panel import Panel
    from rich.text import Text
except ImportError:
    print("[bold red]Rich library not found. Please install with 'pip install rich'.")
    sys.exit(1)

console = Console()
BASE_URL = "https://cve.circl.lu/api"
NVD_BASE_URL = "https://services.nvd.nist.gov/rest/json/cves/2.0"

REQUEST_TIMEOUT = (10, 30)
DEFAULT_KEYWORD_LIMIT = 5
NVD_MAX_RESULTS_PER_PAGE = 2000
session = requests.Session()
session.headers.update(
    {
        "User-Agent": "cvecli (https://github.com/DebaA17/CVE-scanner-cli)",
        "Accept": "application/json",
    }
)

RETRY_STATUSES = {429, 500, 502, 503, 504}


def _sleep_backoff(attempt: int) -> float:
    delay = min(2 ** (attempt - 1), 6) + random.random() * 0.25
    time.sleep(delay)
    return delay


def _get_json(
    url: str,
    *,
    params: Optional[Dict[str, Any]] = None,
    retries: int = 3,
    json_mode: bool = False,
    label: str = "API",
) -> Tuple[Optional[int], Optional[Any], Optional[Exception]]:
    last_exc: Optional[Exception] = None

    status_ctx = nullcontext()
    status_obj = None
    if not json_mode:
        status_ctx = console.status(f"{label}: starting…", spinner="dots")

    with status_ctx as maybe_status:
        status_obj = maybe_status

        for attempt in range(1, retries + 1):
            try:
                if status_obj is not None:
                    status_obj.update(f"{label}: request {attempt}/{retries}…")

                resp = session.get(url, params=params, timeout=REQUEST_TIMEOUT)

                status = resp.status_code
                if status in RETRY_STATUSES and attempt < retries:
                    delay = _sleep_backoff(attempt)
                    if status_obj is not None:
                        status_obj.update(
                            f"{label}: HTTP {status}; retrying in {delay:.1f}s ({attempt}/{retries})…"
                        )
                    continue

                try:
                    data = resp.json()
                except ValueError:
                    data = None

                if status_obj is not None:
                    status_obj.update(f"{label}: received HTTP {status}.")

                return status, data, None
            except requests.RequestException as exc:
                last_exc = exc
                if attempt < retries:
                    delay = _sleep_backoff(attempt)
                    if status_obj is not None:
                        status_obj.update(
                            f"{label}: network error; retrying in {delay:.1f}s ({attempt}/{retries})…"
                        )
                    continue
                break

    return None, None, last_exc


def get_version() -> str:
    version = os.environ.get("SNAP_VERSION")
    if version:
        return version

    version = os.environ.get("CVECLI_VERSION")
    if version:
        return version

    candidate_dirs: List[Path] = []
    meipass = getattr(sys, "_MEIPASS", None)
    if meipass:
        candidate_dirs.append(Path(meipass))
    candidate_dirs.append(Path(__file__).resolve().parent)

    for directory in candidate_dirs:
        try:
            version_file = directory / "VERSION"
            if version_file.is_file():
                file_version = version_file.read_text(encoding="utf-8").strip()
                if file_version:
                    return file_version
        except Exception:
            pass

    try:
        script_dir = Path(__file__).resolve().parent
        result = subprocess.run(
            ["git", "-C", str(script_dir), "describe", "--tags", "--always"],
            capture_output=True,
            text=True,
            check=False,
            timeout=1,
        )
        described = (result.stdout or "").strip()
        if described:
            return described
    except Exception:
        pass

    return "unknown"


def _status(message: str, enabled: bool):
    if not enabled:
        return nullcontext()
    return console.status(message, spinner="dots")


def _extract_cve_summary(data: dict) -> dict:
    cve_id = data.get("cveMetadata", {}).get("cveId", data.get("id"))
    published = format_date(data.get("cveMetadata", {}).get("datePublished", data.get("Published")))
    modified = format_date(data.get("cveMetadata", {}).get("dateUpdated", data.get("Modified")))

    cvss = None
    severity = None
    metrics = data.get("containers", {}).get("cna", {}).get("metrics", [])
    if metrics:
        cvss = metrics[0].get("cvssV3_1", {}).get("baseScore")
        severity = metrics[0].get("cvssV3_1", {}).get("baseSeverity")
    if not cvss:
        cvss = data.get("cvss")

    summary = data.get("summary")
    if not summary:
        cna = data.get("containers", {}).get("cna", {})
        descriptions = cna.get("descriptions", [])
        if descriptions:
            summary = descriptions[0].get("value")

    references: List[str] = []
    for ref in data.get("containers", {}).get("cna", {}).get("references", []):
        url = ref.get("url")
        if url:
            references.append(url)

    return {
        "id": cve_id,
        "published": published,
        "modified": modified,
        "cvss": cvss,
        "severity": severity,
        "description": summary,
        "references": references,
    }


def _extract_nvd_cve_summary(cve: dict) -> dict:
    cve_id = cve.get("id")
    published = format_date(cve.get("published"))
    modified = format_date(cve.get("lastModified"))

    cvss = None
    severity = None
    metrics = cve.get("metrics") or {}
    for key in ("cvssMetricV31", "cvssMetricV30", "cvssMetricV2"):
        metric_list = metrics.get(key) or []
        if metric_list:
            cvss_data = (metric_list[0] or {}).get("cvssData") or {}
            cvss = cvss_data.get("baseScore")
            severity = cvss_data.get("baseSeverity") or (metric_list[0] or {}).get("baseSeverity")
            if cvss:
                break

    description = None
    for desc in cve.get("descriptions") or []:
        if desc.get("lang") == "en" and desc.get("value"):
            description = desc.get("value")
            break
    if not description and (cve.get("descriptions") or []):
        description = (cve.get("descriptions") or [{}])[0].get("value")

    references = [r.get("url") for r in (cve.get("references") or []) if r.get("url")]

    return {
        "id": cve_id,
        "published": published,
        "modified": modified,
        "cvss": cvss,
        "severity": severity,
        "description": description,
        "references": references,
    }


def _print_summary_panel(summary_obj: dict) -> None:
    cve_id = summary_obj.get("id")
    published = summary_obj.get("published")
    modified = summary_obj.get("modified")
    cvss = summary_obj.get("cvss")
    severity = summary_obj.get("severity")
    summary = summary_obj.get("description")
    references = summary_obj.get("references") or []

    panel_text = Text()
    panel_text.append("CVE ID: ", style="bold yellow")
    panel_text.append(f"{cve_id}\n", style="white")
    panel_text.append("Published: ", style="bold green")
    panel_text.append(f"{published}\n", style="white")
    panel_text.append("Modified: ", style="bold green")
    panel_text.append(f"{modified}\n", style="white")
    panel_text.append("CVSS Score: ", style="bold magenta")
    panel_text.append(f"{cvss if cvss else 'N/A'}\n", style="white")
    panel_text.append("Severity: ", style="bold magenta")
    panel_text.append(f"{severity if severity else 'N/A'}\n\n", style="white")
    panel_text.append("Description:\n", style="bold cyan")
    panel_text.append(f"{summary if summary else 'No description available.'}\n\n", style="white")
    if references:
        panel_text.append("References:\n", style="bold blue")
        for ref in references:
            panel_text.append(f"  {ref}\n", style="blue underline")
    console.print(Panel(panel_text, title=f"{cve_id}", expand=False, border_style="red"))


def _write_json(obj: object) -> None:
    sys.stdout.write(json.dumps(obj, ensure_ascii=False) + "\n")

def format_date(date_str):
    if not date_str:
        return "N/A"
    try:
        return datetime.fromisoformat(date_str.replace('Z', '+00:00')).strftime('%Y-%m-%d %H:%M')
    except Exception:
        return date_str


def _parse_year(year) -> Optional[int]:
    if year is None or year == "":
        return None
    try:
        parsed = int(str(year).strip())
    except (TypeError, ValueError):
        raise argparse.ArgumentTypeError("year must be a 4-digit number, e.g., 2026")

    if parsed < 1999 or parsed > 9999:
        raise argparse.ArgumentTypeError("year must be between 1999 and 9999")
    return parsed


def _nvd_datetime(day: date, end_of_day: bool = False) -> str:
    if end_of_day:
        return f"{day.isoformat()}T23:59:59.999Z"
    return f"{day.isoformat()}T00:00:00.000Z"


def _year_date_ranges(year: int) -> List[Tuple[str, str]]:
    ranges: List[Tuple[str, str]] = []
    current = date(year, 1, 1)
    final = date(year, 12, 31)

    while current <= final:
        chunk_end = min(current + timedelta(days=119), final)
        ranges.append((_nvd_datetime(current), _nvd_datetime(chunk_end, end_of_day=True)))
        current = chunk_end + timedelta(days=1)

    return ranges

def _parse_limit(limit) -> int:
    if limit is None or limit == "":
        return DEFAULT_KEYWORD_LIMIT
    try:
        parsed = int(str(limit).strip())
    except (TypeError, ValueError):
        raise argparse.ArgumentTypeError("limit must be a number, e.g., 10")
    if parsed < 1 or parsed > 200:
        raise argparse.ArgumentTypeError("limit must be between 1 and 200")
    return parsed

def _fetch_cve_by_id_nvd(cve_id: str, *, json_mode: bool = False) -> Optional[Dict[str, Any]]:
    status, data, err = _get_json(
        NVD_BASE_URL,
        params={"cveId": cve_id},
        retries=3,
        json_mode=json_mode,
        label=f"NVD ({cve_id})",
    )
    if err is not None:
        return None
    if status != 200 or not isinstance(data, dict):
        return None

    vulns = data.get("vulnerabilities") or []
    if not vulns:
        return None
    cve = (vulns[0] or {}).get("cve") or {}
    if not cve:
        return None
    return _extract_nvd_cve_summary(cve)


def search_cve_by_id(cve_id, *, json_mode: bool = False) -> bool:
    cve_id = (cve_id or "").strip().upper()
    if not cve_id:
        console.print("[bold red]Error: CVE ID is empty.[/]")
        return False

    circl_url = f"{BASE_URL}/cve/{cve_id}"
    status, data, err = _get_json(
        circl_url,
        retries=3,
        json_mode=json_mode,
        label=f"CIRCL ({cve_id})",
    )

    if err is not None:
        # Network error; try NVD fallback.
        if not json_mode:
            console.print("[dim]CIRCL network error; falling back to NVD…[/]")
        summary_obj = _fetch_cve_by_id_nvd(cve_id, json_mode=json_mode)
        if summary_obj:
            if json_mode:
                _write_json(summary_obj)
            else:
                _print_summary_panel(summary_obj)
            return True

        console.print(
            f"[bold red]Network error while fetching CVE data.[/]\n{err}\n"
            "[dim]If you installed via Snap, ensure the 'network' plug is connected: "
            "'snap connections cvecli'.[/]"
        )
        return False

    if status == 200 and isinstance(data, dict):
        summary_obj = _extract_cve_summary(data)
        if json_mode:
            _write_json(summary_obj)
        else:
            _print_summary_panel(summary_obj)
        return True

    if status == 404:
        console.print(f"[bold red]Error: CVE {cve_id} not found.[/]")
        return False

    # CIRCL is returning errors (often temporary). Try NVD fallback.
    if not json_mode:
        console.print(f"[dim]CIRCL returned HTTP {status}; falling back to NVD…[/]")
    summary_obj = _fetch_cve_by_id_nvd(cve_id, json_mode=json_mode)
    if summary_obj:
        if json_mode:
            _write_json(summary_obj)
        else:
            _print_summary_panel(summary_obj)
        return True

    console.print(
        f"[bold red]Error: Failed to fetch CVE {cve_id} (HTTP {status}).[/]"
    )
    return False

def search_cve_by_keyword(
    keyword, *, year: Optional[int] = None, limit: int = DEFAULT_KEYWORD_LIMIT, json_mode: bool = False
) -> bool:
    keyword = (keyword or "").strip()
    if not keyword:
        console.print("[bold red]Error: Keyword is empty.[/]")
        return False

    year = _parse_year(year)
    limit = _parse_limit(limit)
    summaries: List[Dict[str, Any]] = []
    seen_ids = set()
    date_ranges = _year_date_ranges(year) if year else [(None, None)]
    cve_year_prefix = f"CVE-{year}-" if year else None
    results_per_page = NVD_MAX_RESULTS_PER_PAGE if year else limit

    for start_date, end_date in date_ranges:
        start_index = 0

        while True:
            params = {
                "keywordSearch": keyword,
                "resultsPerPage": results_per_page,
                "startIndex": start_index,
            }
            if start_date and end_date:
                params.update({"pubStartDate": start_date, "pubEndDate": end_date})

            label = f"NVD keyword ({keyword})"
            if year:
                label += f" CVE year {year}"

            status, data, err = _get_json(
                NVD_BASE_URL,
                params=params,
                retries=3,
                json_mode=json_mode,
                label=label,
            )
            if err is not None:
                console.print(
                    f"[bold red]Network error while searching NVD.[/]\n{err}\n"
                    "[dim]If you installed via Snap, ensure the 'network' plug is connected: "
                    "'snap connections cvecli'.[/]"
                )
                return False

            if status != 200 or not isinstance(data, dict):
                console.print(
                    f"[bold red]Error: Could not fetch CVEs for keyword '{keyword}' (HTTP {status}).[/]"
                )
                return False

            results = data.get("vulnerabilities", [])
            for item in results:
                cve = (item or {}).get("cve") or {}
                cve_id = cve.get("id")
                if not cve_id or cve_id in seen_ids:
                    continue
                if cve_year_prefix and not cve_id.startswith(cve_year_prefix):
                    continue

                summaries.append(_extract_nvd_cve_summary(cve))
                seen_ids.add(cve_id)

                if len(summaries) >= limit:
                    break

            if len(summaries) >= limit:
                break

            total_results = data.get("totalResults") or 0
            start_index += len(results)
            if not year or not results or start_index >= total_results:
                break

        if len(summaries) >= limit:
            break

    if not summaries:
        suffix = f" in {year}" if year else ""
        console.print(f"[bold red]No CVEs found for keyword: {keyword}{suffix}")
        return False

    if json_mode:
        out = {
            "keyword": keyword,
            "year": year,
            "limit": limit,
            "count": len(summaries),
            "results": summaries,
        }
        _write_json(out)
        return True

    any_printed = False
    for summary_obj in summaries:
        if summary_obj.get("id"):
            _print_summary_panel(summary_obj)
            any_printed = True

    return any_printed

def interactive_menu():
    console.print("[bold green]\n==============================\n[bold yellow]Made by DEBASIS - CVE Checker CLI\n==============================\n", style="bold green")
    while True:
        console.print("[bold cyan]\nChoose an option:")
        console.print("[bold magenta]1.[/] Search by CVE ID")
        console.print("[bold magenta]2.[/] Search by keyword in description")
        console.print("[bold magenta]3.[/] Exit")
        choice = input("Enter choice (1/2/3): ").strip()
        if choice == '1':
            cve_id = input("Enter CVE ID (e.g., CVE-2025-55184): ").strip()
            search_cve_by_id(cve_id)
        elif choice == '2':
            keyword = input("Enter keyword (e.g., wordpress, plugin, vuln): ").strip()
            year_text = input("Enter year filter (optional, e.g., 2026): ").strip()
            limit_text = input(f"Enter max results (optional, default {DEFAULT_KEYWORD_LIMIT}): ").strip()
            try:
                year = _parse_year(year_text)
                limit = _parse_limit(limit_text)
            except argparse.ArgumentTypeError as exc:
                console.print(f"[bold red]Error: {exc}[/]")
                continue
            search_cve_by_keyword(keyword, year=year, limit=limit)
        elif choice == '3':
            console.print("[bold green]Goodbye!")
            break
        else:
            console.print("[bold red]Invalid choice. Please try again.")

def main():
    version = get_version()
    parser = argparse.ArgumentParser(description=f"CVE Search CLI Tool (version {version})")
    parser.add_argument("--id", help="Search CVE by ID, e.g., CVE-2024-12345")
    parser.add_argument("--keyword", help="Search CVEs by keyword in description")
    parser.add_argument("--year", type=_parse_year, help="Filter keyword searches by CVE ID year, e.g., 2026")
    parser.add_argument("--limit", type=_parse_limit, help=f"Max results for keyword search (default {DEFAULT_KEYWORD_LIMIT})")
    parser.add_argument("--version", action="version", version=f"cvecli {version}")
    parser.add_argument("--json", action="store_true", help="Output results as JSON")
    args = parser.parse_args()

    if args.year and args.id:
        console.print("[bold red]Error: --year can only be used with --keyword.[/]")
        sys.exit(2)

    if args.year and not args.keyword:
        console.print("[bold red]Error: --year requires --keyword.[/]")
        sys.exit(2)

    if args.limit and not args.keyword:
        console.print("[bold red]Error: --limit requires --keyword.[/]")
        sys.exit(2)

    if args.id:
        ok = search_cve_by_id(args.id, json_mode=args.json)
        if not ok:
            sys.exit(1)
    elif args.keyword:
        ok = search_cve_by_keyword(args.keyword, year=args.year, limit=args.limit, json_mode=args.json)
        if not ok:
            sys.exit(1)
    else:
        if args.json:
            console.print("[bold red]Error: --json requires --id or --keyword.[/]")
            sys.exit(2)
        interactive_menu()

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        console.print("\n[bold yellow]Interrupted by user. Exiting.[/]")
        sys.exit(130)