#!/usr/bin/env python3
# SPDX-License-Identifier: Apache-2.0
# Copyright 2024-2026 Lukasz Krzemien (biuro@softspark.eu)
# Source: https://github.com/softspark/ai-toolkit

"""CVE dependency scanner — detect ecosystems and run native audit tools.

Stdlib only. No external dependencies.
"""

import json
import os
import shutil
import subprocess
import sys
from pathlib import Path

ECOSYSTEMS = {
    "npm": {
        "manifests": ["package.json"],
        "locks": ["package-lock.json", "yarn.lock", "pnpm-lock.yaml"],
        "tool": "npm",
        "audit_cmd": ["npm", "audit", "--json"],
        "fix_cmd": ["npm", "audit", "fix"],
        "install_hint": "npm is bundled with Node.js",
    },
    "pip": {
        "manifests": ["requirements.txt", "pyproject.toml", "setup.py", "setup.cfg"],
        "locks": ["requirements.txt"],
        "tool": "pip-audit",
        "audit_cmd": ["pip-audit", "--format=json", "--output=-"],
        "fix_cmd": ["pip-audit", "--fix"],
        "install_hint": "pip install pip-audit",
    },
    "composer": {
        "manifests": ["composer.json"],
        "locks": ["composer.lock"],
        "tool": "composer",
        "audit_cmd": ["composer", "audit", "--format=json"],
        "fix_cmd": ["composer", "update"],
        "install_hint": "https://getcomposer.org/download/",
    },
    "cargo": {
        "manifests": ["Cargo.toml"],
        "locks": ["Cargo.lock"],
        "tool": "cargo-audit",
        "audit_cmd": ["cargo", "audit", "--json"],
        "fix_cmd": ["cargo", "audit", "fix"],
        "install_hint": "cargo install cargo-audit",
    },
    "go": {
        "manifests": ["go.mod"],
        "locks": ["go.sum"],
        "tool": "govulncheck",
        "audit_cmd": ["govulncheck", "-json", "./..."],
        "fix_cmd": ["go", "get", "-u", "./..."],
        "install_hint": "go install golang.org/x/vuln/cmd/govulncheck@latest",
    },
    "ruby": {
        "manifests": ["Gemfile"],
        "locks": ["Gemfile.lock"],
        "tool": "bundle-audit",
        "audit_cmd": ["bundle-audit", "check", "--format=json"],
        "fix_cmd": ["bundle-audit", "update"],
        "install_hint": "gem install bundler-audit",
    },
    "dart": {
        "manifests": ["pubspec.yaml"],
        "locks": ["pubspec.lock"],
        "tool": "dart",
        "audit_cmd": ["dart", "pub", "outdated", "--json"],
        "fix_cmd": ["dart", "pub", "upgrade"],
        "install_hint": "https://dart.dev/get-dart",
    },
}


def detect_ecosystems(root: Path) -> list[dict]:
    """Detect which package ecosystems are present in the project."""
    found = []
    for name, eco in ECOSYSTEMS.items():
        detected_files = []
        has_lock = False
        for m in eco["manifests"]:
            if (root / m).exists():
                detected_files.append(m)
        for m in eco["locks"]:
            if (root / m).exists():
                detected_files.append(m)
                has_lock = True
        # Deduplicate (requirements.txt is both manifest and lock)
        detected_files = list(dict.fromkeys(detected_files))
        if detected_files:
            tool_path = shutil.which(eco["tool"])
            found.append({
                "name": name,
                "files": detected_files,
                "has_lock": has_lock,
                "tool": eco["tool"],
                "tool_available": tool_path is not None,
                "install_hint": eco["install_hint"],
                "audit_cmd": eco["audit_cmd"],
                "fix_cmd": eco["fix_cmd"],
            })
    return found


def run_audit(eco: dict, root: Path, fix: bool = False) -> dict:
    """Run the native audit command for an ecosystem."""
    cmd = eco["fix_cmd"] if fix else eco["audit_cmd"]
    result = {
        "ecosystem": eco["name"],
        "tool": eco["tool"],
        "command": " ".join(cmd),
        "success": False,
        "raw_output": "",
        "error": "",
    }

    if not eco["tool_available"]:
        result["error"] = f"Tool '{eco['tool']}' not installed. Install: {eco['install_hint']}"
        return result

    if not eco.get("has_lock", True) and eco["name"] in ("npm",):
        result["warning"] = "No lock file — run `npm install` first"
        result["success"] = True
        return result

    try:
        proc = subprocess.run(
            cmd,
            cwd=str(root),
            capture_output=True,
            text=True,
            timeout=120,
        )
        result["raw_output"] = proc.stdout
        result["error"] = proc.stderr if proc.returncode not in (0, 1) else ""
        # npm audit returns 1 when vulnerabilities found — that's not an error
        result["success"] = True
    except FileNotFoundError:
        result["error"] = f"Tool '{eco['tool']}' not found in PATH"
    except subprocess.TimeoutExpired:
        result["error"] = f"Audit command timed out after 120s"

    return result


def parse_npm_audit(raw: str) -> list[dict]:
    """Parse npm audit JSON output into unified findings."""
    findings = []
    try:
        data = json.loads(raw)
    except (json.JSONDecodeError, ValueError):
        return findings

    vulns = data.get("vulnerabilities", {})
    for pkg_name, info in vulns.items():
        for via in info.get("via", []):
            if isinstance(via, dict):
                severity = via.get("severity", "unknown").upper()
                findings.append({
                    "severity": severity,
                    "package": pkg_name,
                    "installed": info.get("range", "unknown"),
                    "cve": via.get("cve", [via.get("url", "N/A")]),
                    "title": via.get("title", "Unknown"),
                    "url": via.get("url", ""),
                    "fixed_in": info.get("fixAvailable", {}).get("version", "unknown") if isinstance(info.get("fixAvailable"), dict) else "unknown",
                })
    return findings


def parse_pip_audit(raw: str) -> list[dict]:
    """Parse pip-audit JSON output into unified findings."""
    findings = []
    try:
        data = json.loads(raw)
    except (json.JSONDecodeError, ValueError):
        return findings

    for dep in data.get("dependencies", []):
        for vuln in dep.get("vulns", []):
            vuln_id = vuln.get("id", "N/A")
            aliases = vuln.get("aliases", [])
            # Use GHSA link if available, otherwise OSV
            url = ""
            for alias in aliases:
                if alias.startswith("GHSA-"):
                    url = f"https://github.com/advisories/{alias}"
                    break
            if not url:
                url = f"https://osv.dev/vulnerability/{vuln_id}"

            fix_versions = vuln.get("fix_versions", [])
            # Determine severity from CVE ID prefix heuristic — pip-audit
            # doesn't provide severity directly, so we mark as HIGH by default
            severity = "HIGH"

            findings.append({
                "severity": severity,
                "package": f"{dep['name']}@{dep['version']}",
                "installed": dep.get("version", "unknown"),
                "cve": vuln_id,
                "title": vuln.get("description", "")[:120] + ("..." if len(vuln.get("description", "")) > 120 else ""),
                "url": url,
                "fixed_in": ", ".join(fix_versions) if fix_versions else "unknown",
            })
    return findings


def parse_cargo_audit(raw: str) -> list[dict]:
    """Parse cargo audit JSON output into unified findings."""
    findings = []
    try:
        data = json.loads(raw)
    except (json.JSONDecodeError, ValueError):
        return findings

    for vuln in data.get("vulnerabilities", {}).get("list", []):
        advisory = vuln.get("advisory", {})
        pkg = vuln.get("package", {})
        findings.append({
            "severity": "HIGH",
            "package": f"{pkg.get('name', 'unknown')}@{pkg.get('version', 'unknown')}",
            "installed": pkg.get("version", "unknown"),
            "cve": advisory.get("id", "N/A"),
            "title": advisory.get("title", "Unknown"),
            "url": advisory.get("url", ""),
            "fixed_in": ", ".join(vuln.get("versions", {}).get("patched", [])) or "unknown",
        })
    return findings


def _table(headers: list[str], rows: list[list[str]]) -> list[str]:
    """Render an aligned plain-text table with box-drawing borders."""
    widths = [len(h) for h in headers]
    for row in rows:
        for i, cell in enumerate(row):
            widths[i] = max(widths[i], len(cell))

    sep = "├" + "┼".join("─" * (w + 2) for w in widths) + "┤"
    top = "┌" + "┬".join("─" * (w + 2) for w in widths) + "┐"
    bot = "└" + "┴".join("─" * (w + 2) for w in widths) + "┘"

    def fmt_row(cells: list[str]) -> str:
        parts = []
        for i, cell in enumerate(cells):
            parts.append(f" {cell:<{widths[i]}} ")
        return "│" + "│".join(parts) + "│"

    lines = [top, fmt_row(headers), sep]
    for row in rows:
        lines.append(fmt_row(row))
    lines.append(bot)
    return lines


def format_text_report(ecosystems: list[dict], results: list[dict], all_findings: list[dict]) -> str:
    """Format findings as a terminal-friendly report."""
    lines = ["", "═══ CVE Scan Report ═══", ""]

    # Ecosystems detected
    eco_rows = []
    for eco in ecosystems:
        status = "✓" if eco["tool_available"] else "⚠ not installed"
        files = ", ".join(eco["files"])
        eco_rows.append([eco["name"], files, eco["tool"], status])
    lines.extend(_table(["Ecosystem", "Files", "Tool", "Status"], eco_rows))
    lines.append("")

    # Summary
    severity_counts = {"CRITICAL": 0, "HIGH": 0, "MODERATE": 0, "MEDIUM": 0, "LOW": 0}
    for f in all_findings:
        sev = f["severity"].upper()
        if sev in severity_counts:
            severity_counts[sev] += 1
        else:
            severity_counts[sev] = severity_counts.get(sev, 0) + 1

    # Merge MODERATE into MEDIUM for display
    severity_counts["MEDIUM"] += severity_counts.pop("MODERATE", 0)

    total = sum(severity_counts.values())
    summary_rows = []
    for sev in ["CRITICAL", "HIGH", "MEDIUM", "LOW"]:
        if severity_counts.get(sev, 0) > 0:
            summary_rows.append([sev, str(severity_counts[sev])])
    summary_rows.append(["TOTAL", str(total)])
    lines.extend(_table(["Severity", "Count"], summary_rows))
    lines.append("")

    if not all_findings:
        lines.append("✓ No known vulnerabilities found.")
        lines.append("")

    # Findings by severity
    if all_findings:
        severity_order = {"CRITICAL": 0, "HIGH": 1, "MODERATE": 2, "MEDIUM": 2, "LOW": 3}
        sorted_findings = sorted(all_findings, key=lambda f: severity_order.get(f["severity"].upper(), 9))

        finding_rows = []
        for f in sorted_findings:
            sev = f["severity"].upper()
            if sev == "MODERATE":
                sev = "MEDIUM"
            cve = f["cve"] if isinstance(f["cve"], str) else ", ".join(f["cve"]) if f["cve"] else "N/A"
            title = f["title"][:60] + ("..." if len(f["title"]) > 60 else "")
            finding_rows.append([sev, f["package"], cve, f["fixed_in"], title])

        lines.extend(_table(["Severity", "Package", "CVE", "Fix", "Title"], finding_rows))
        lines.append("")

        # Detailed advisory links
        lines.append("Advisory links:")
        for f in sorted_findings:
            if f.get("url"):
                cve = f["cve"] if isinstance(f["cve"], str) else ", ".join(f["cve"]) if f["cve"] else "N/A"
                lines.append(f"  {cve}: {f['url']}")
        lines.append("")

    # Install hints for missing tools
    missing = [e for e in ecosystems if not e["tool_available"]]
    if missing:
        lines.append("Missing tools:")
        for eco in missing:
            lines.append(f"  {eco['name']}: install with: {eco['install_hint']}")
        lines.append("")

    # Warnings (non-fatal issues like missing lock files)
    warnings = [r for r in results if r.get("warning")]
    if warnings:
        for r in warnings:
            lines.append(f"⚠ {r['ecosystem']}: {r['warning']}")
        lines.append("")

    # Errors (fatal issues)
    errors = [r for r in results if r.get("error")]
    if errors:
        for r in errors:
            lines.append(f"✗ {r['ecosystem']}: {r['error']}")
        lines.append("")

    return "\n".join(lines)


def main():
    import argparse

    parser = argparse.ArgumentParser(description="CVE dependency scanner")
    parser.add_argument("--json", action="store_true", help="JSON output")
    parser.add_argument("--fix", action="store_true", help="Auto-fix vulnerabilities")
    parser.add_argument("--ecosystem", type=str, help="Scan specific ecosystem only")
    parser.add_argument("path", nargs="?", default=".", help="Project root path")
    args = parser.parse_args()

    root = Path(args.path).resolve()
    if not root.is_dir():
        print(f"Error: {root} is not a directory", file=sys.stderr)
        sys.exit(1)

    # Detect
    ecosystems = detect_ecosystems(root)
    if args.ecosystem:
        ecosystems = [e for e in ecosystems if e["name"] == args.ecosystem]

    if not ecosystems:
        msg = "No supported package ecosystems detected."
        if args.json:
            print(json.dumps({"ecosystems": [], "findings": [], "message": msg}))
        else:
            print(msg)
        sys.exit(0)

    # Run audits
    results = []
    all_findings = []
    for eco in ecosystems:
        result = run_audit(eco, root, fix=args.fix)
        results.append(result)

        # Parse results with ecosystem-specific parsers
        if result["success"] and result["raw_output"]:
            parsers = {
                "npm": parse_npm_audit,
                "pip": parse_pip_audit,
                "cargo": parse_cargo_audit,
            }
            parser = parsers.get(eco["name"])
            if parser:
                findings = parser(result["raw_output"])
                for f in findings:
                    f["ecosystem"] = eco["name"]
                all_findings.extend(findings)

    # Output
    if args.json:
        output = {
            "ecosystems": [{"name": e["name"], "files": e["files"], "tool_available": e["tool_available"]} for e in ecosystems],
            "results": results,
            "findings": all_findings,
            "total_findings": len(all_findings),
        }
        print(json.dumps(output, indent=2))
    else:
        report = format_text_report(ecosystems, results, all_findings)
        print(report)

    # Exit code: non-zero if CRITICAL or HIGH found
    high_or_critical = [f for f in all_findings if f["severity"].upper() in ("CRITICAL", "HIGH")]
    if high_or_critical:
        sys.exit(1)


if __name__ == "__main__":
    main()