Vulners Software Audit

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Audit software versions against the Vulners Software Audit API (BYOK)."""

from __future__ import annotations

import json

from kittysploit import *
from lib.vulners.client import (
    VulnersClient,
    VulnersError,
    parse_software_entries,
    resolve_api_key,
    summarize_audit,
)


class Module(Scanner):
    __info__ = {
        "name": "Vulners Software Audit",
        "description": (
            "Query the Vulners Software Audit API for known vulnerabilities matching "
            "one or more software products/versions (CPE string or vendor:product:version). "
            "Bring-your-own-key (BYOK): set api_key, plugin config, or VULNERS_API_KEY. "
            "Prefer `plugin run vulners …` as the hub; this scanner is a thin wrapper."
        ),
        "author": ["KittySploit Team"],
        "references": [
            "https://docs.vulners.com/docs/api/audit/",
            "https://vulners.com/",
        ],
        "tags": [
            "vulners",
            "audit",
            "cpe",
            "cve",
            "vulnerability-intelligence",
            "scanner",
        ],
        "severity": "info",
        "agent": {
            "risk": "passive",
            "effects": [],
            "expected_requests": 1,
            "reversible": True,
            "approval_required": False,
            "produces": ["risk_signals", "exploit_paths"],
        },
    }

    software = OptString(
        "",
        "Software to audit: CPE, vendor:product:version, or product:version "
        "(comma/newline separated for multiple)",
        True,
    )
    api_key = OptString(
        "",
        "Vulners API key (or plugin config / env VULNERS_API_KEY)",
        False,
    )
    match = OptString(
        "partial",
        "Match mode: partial (default) or full",
        False,
    )
    catalog = OptString(
        "official",
        "CPE catalog: official or extended",
        False,
    )
    max_vulns = OptInteger(
        50,
        "Max vulnerabilities to print/store per software entry",
        False,
    )
    timeout = OptInteger(
        30,
        "HTTP timeout in seconds",
        False,
    )
    output_file = OptString(
        "",
        "Optional path to write full JSON response",
        False,
    )

    def _load_plugin_api_key(self) -> str:
        try:
            path = data_dir() / "vulners" / "config.json"
            if not path.is_file():
                return ""
            with open(path, encoding="utf-8") as fh:
                data = json.load(fh)
            if isinstance(data, dict):
                return str(data.get("api_key") or "").strip()
        except (OSError, json.JSONDecodeError, TypeError):
            return ""
        return ""

    def run(self):
        api_key = resolve_api_key(
            str(self.api_key or ""),
            config_key=self._load_plugin_api_key(),
        )
        if not api_key:
            print_error(
                "Vulners API key required: set option api_key, "
                "`plugin run vulners config set --api-key KEY`, or env VULNERS_API_KEY"
            )
            return False

        raw_software = str(self.software or "").strip()
        if not raw_software:
            print_error("software is required (CPE or vendor:product:version)")
            return False

        entries = parse_software_entries(raw_software)
        if not entries:
            print_error("No valid software entries parsed")
            return False

        try:
            timeout = int(self.timeout or 30)
        except (TypeError, ValueError):
            timeout = 30
        try:
            limit = int(self.max_vulns or 50)
        except (TypeError, ValueError):
            limit = 50

        print_status(f"Auditing {len(entries)} software entr(y/ies) via Vulners…")
        try:
            client = VulnersClient(api_key, timeout=timeout)
            results = client.audit_software(
                entries,
                match=str(self.match or "partial"),
                catalog=str(self.catalog or "official"),
            )
        except VulnersError as exc:
            print_error(str(exc))
            return False
        except Exception as exc:
            print_error(str(exc))
            return False

        out_path = str(self.output_file or "").strip()
        if out_path:
            try:
                with open(out_path, "w", encoding="utf-8") as fh:
                    json.dump(results, fh, indent=2, ensure_ascii=False)
                print_success(f"Wrote full response to {out_path}")
            except OSError as exc:
                print_warning(f"Could not write output_file: {exc}")

        total, findings, top_severity = summarize_audit(results, limit)

        if total == 0:
            print_success("No vulnerabilities matched for the submitted software")
            self.set_info(
                severity="info",
                reason="No vulnerabilities matched",
                software=entries,
                vulnerability_count=0,
                findings=[],
            )
            return True

        print_warning(f"Found {total} matching vulnerabilit(y/ies) (showing up to {limit})")
        for item in findings[: min(limit, 25)]:
            cve_s = ",".join(item["cvelist"][:3]) or "-"
            wild = " [KEV/wild]" if item.get("wild_exploited") else ""
            print_info(
                f"[{item['severity']}] {item.get('id')} | {cve_s} | "
                f"{(item.get('title') or '')[:80]}{wild}"
            )

        self.report_finding(
            f"Vulners software audit: {total} matching advisory/ies",
            severity=top_severity if top_severity != "info" else "medium",
            evidence={
                "software": entries,
                "vulnerability_count": total,
                "findings": findings[:limit],
                "api": VulnersClient.AUDIT_URL,
            },
            impact={
                "summary": (
                    "Submitted software versions match known advisories in Vulners."
                ),
            },
            remediation={
                "summary": "Review matched advisories and upgrade/patch affected components.",
                "actions": [
                    "Prioritize critical/high and wild-exploited items",
                    "Cross-check CVE IDs in cvelist against your inventory",
                    "Validate fixed versions from vendor advisories",
                ],
            },
        )
        self.set_info(
            severity=top_severity,
            reason=f"{total} Vulners matches",
            software=entries,
            vulnerability_count=total,
            findings=findings[:limit],
        )
        print_success("Vulners software audit completed")
        return True