Host Vulnerability Scanner

#!/usr/bin/env python3
"""
Host Vulnerability Scanner

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/host-vulnerability-scanner

It covers what the script does, whether it is safe to run, the
inventory file and credentials it needs, how to read the output,
and what to do when something fails.

    pip install "paramiko>=3.0" "requests>=2.28"
"""

import argparse, json, os, re, sys, socket
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
# Windows consoles default to a legacy code page (cp1252), which cannot encode
# the arrows, box-drawing and check-mark characters used in the output below.
# Printing one raises UnicodeEncodeError and kills the script mid-scan, so make
# stdout UTF-8 and degrade unencodable characters instead of crashing.
for _stream in (sys.stdout, sys.stderr):
    try:
        _stream.reconfigure(encoding="utf-8", errors="replace")
    except (AttributeError, ValueError):
        pass        # already wrapped, or not a real stream (piped/redirected)

# Optional imports
try:
    import paramiko
    HAS_PARAMIKO = True
except ImportError:
    HAS_PARAMIKO = False

try:
    import requests
    HAS_REQUESTS = True
except ImportError:
    HAS_REQUESTS = False

TIMEOUT = 20

# ─── Known vulnerable package versions (subset for demo) ───
KNOWN_VULNS = [
    {"package": "glibc", "max_version": "2.35-0ubuntu3.3", "cve": "CVE-2023-4911",
     "cvss": 7.8, "severity": "CRITICAL",
     "description": "Looney Tunables — local privilege escalation via GLIBC_TUNABLES",
     "fix": "apt upgrade libc6 → 2.35-0ubuntu3.4+"},
    {"package": "openssh-server", "max_version": "9.2p1", "cve": "CVE-2023-38408",
     "cvss": 9.8, "severity": "CRITICAL",
     "description": "Remote code execution via ssh-agent forwarding (PKCS#11)",
     "fix": "apt/yum upgrade openssh-server → 9.3p1+"},
    {"package": "openssl", "max_version": "3.0.8", "cve": "CVE-2023-0464",
     "cvss": 5.9, "severity": "MEDIUM",
     "description": "DoS via crafted ASN.1 object with excessive memory consumption",
     "fix": "apt/yum upgrade openssl → 3.0.9+"},
    {"package": "curl", "max_version": "7.88.0", "cve": "CVE-2023-38545",
     "cvss": 9.8, "severity": "CRITICAL",
     "description": "SOCKS5 heap overflow — worst curl vulnerability in years",
     "fix": "apt/yum upgrade curl → 8.4.0+"},
    {"package": "sudo", "max_version": "1.9.12", "cve": "CVE-2023-42465",
     "cvss": 7.0, "severity": "HIGH",
     "description": "Rowhammer attack via sudo exploitation",
     "fix": "apt/yum upgrade sudo → 1.9.13+"},
    {"package": "linux-kernel", "max_version": "5.15.99", "cve": "CVE-2023-32233",
     "cvss": 7.8, "severity": "HIGH",
     "description": "Use-after-free in Netfilter nf_tables — local privilege escalation",
     "fix": "Update kernel to 5.15.100+ or 6.2+"},
    {"package": "bash", "max_version": "5.1", "cve": "CVE-2022-3715",
     "cvss": 6.6, "severity": "MEDIUM",
     "description": "Heap buffer write overflow in bash read_mbstring_char",
     "fix": "Update bash to 5.2+"},
    {"package": "python3", "max_version": "3.10.8", "cve": "CVE-2022-45061",
     "cvss": 7.5, "severity": "HIGH",
     "description": "DoS via slow IDNA decoding attack",
     "fix": "Update Python to 3.10.9+"},
]

EOL_OS = {
    "CentOS 8": "2021-12-31",
    "CentOS 6": "2020-11-30",
    "CentOS 7": "2024-06-30",
    "Ubuntu 18.04": "2023-04-30",
    "Ubuntu 20.04": "2025-04-30",
    "Debian 10": "2024-06-30",
    "Windows Server 2008": "2020-01-14",
    "Windows Server 2012": "2023-10-10",
    "Red Hat Enterprise Linux 7": "2024-06-30",
    "RHEL 7": "2024-06-30",
}

def run_ssh(conn, cmd: str) -> str:
    try:
        stdin, stdout, stderr = conn.exec_command(cmd, timeout=TIMEOUT)
        return stdout.read().decode("utf-8", errors="replace").strip()
    except Exception:
        return ""

def connect_ssh(host: str, username: str = "root", key_file: str = None,
                password: str = None, port: int = 22):
    if not HAS_PARAMIKO:
        return None
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        if key_file:
            client.connect(host, port=port, username=username,
                           key_filename=key_file, timeout=TIMEOUT)
        else:
            client.connect(host, port=port, username=username,
                           password=password, timeout=TIMEOUT)
        return client
    except Exception as e:
        print(f"    [!] SSH connection failed: {e}")
        return None

def get_os_info(conn) -> dict:
    os_info = {}
    os_info["os"] = run_ssh(conn, "cat /etc/os-release 2>/dev/null | grep PRETTY_NAME | cut -d= -f2 | tr -d '\"'") or \
                    run_ssh(conn, "cat /etc/redhat-release 2>/dev/null") or "Unknown"
    os_info["kernel"] = run_ssh(conn, "uname -r")
    os_info["hostname"] = run_ssh(conn, "hostname")
    os_info["arch"] = run_ssh(conn, "uname -m")
    os_info["uptime"] = run_ssh(conn, "uptime -p 2>/dev/null || uptime")
    return os_info

def list_packages_linux(conn) -> dict:
    packages = {}
    # Try dpkg (Debian/Ubuntu)
    dpkg_out = run_ssh(conn, "dpkg-query -W -f='${Package}\\t${Version}\\n' 2>/dev/null | head -500")
    if dpkg_out:
        for line in dpkg_out.splitlines():
            parts = line.split("\t")
            if len(parts) == 2:
                packages[parts[0]] = parts[1]
        return packages
    # Try rpm (RHEL/CentOS)
    rpm_out = run_ssh(conn, "rpm -qa --queryformat='%{NAME}\\t%{VERSION}-%{RELEASE}\\n' 2>/dev/null | head -500")
    if rpm_out:
        for line in rpm_out.splitlines():
            parts = line.split("\t")
            if len(parts) == 2:
                packages[parts[0]] = parts[1]
    return packages

def get_listening_services(conn) -> list:
    out = run_ssh(conn, "ss -tlnp 2>/dev/null || netstat -tlnp 2>/dev/null")
    services = []
    for line in out.splitlines()[1:]:
        parts = line.split()
        if len(parts) >= 4:
            addr = parts[3] if len(parts) > 3 else ""
            process = parts[-1] if "pid" in parts[-1].lower() or "/" in parts[-1] else ""
            services.append({"address": addr, "process": process})
    return services

def check_cis_ssh(conn) -> list:
    """CIS Benchmark SSH hardening checks."""
    issues = []
    checks = {
        "PermitRootLogin": ("no", "Root login should be disabled"),
        "PasswordAuthentication": ("no", "Password auth should be disabled — use keys only"),
        "Protocol": ("2", "SSH Protocol 1 is insecure"),
        "PermitEmptyPasswords": ("no", "Empty passwords must be forbidden"),
        "MaxAuthTries": ("4", "Limit auth attempts to reduce brute-force"),
        "X11Forwarding": ("no", "X11 forwarding exposes display — disable unless needed"),
        "AllowAgentForwarding": ("no", "Agent forwarding enables pivoting"),
    }
    ssh_conf = run_ssh(conn, "cat /etc/ssh/sshd_config 2>/dev/null")
    for param, (safe_val, desc) in checks.items():
        m = re.search(rf"^\s*{param}\s+(\S+)", ssh_conf, re.MULTILINE | re.IGNORECASE)
        current = m.group(1).lower() if m else "default"
        if current != safe_val.lower() and current != "default":
            issues.append({"check": f"SSH {param}", "current": current,
                           "expected": safe_val, "description": desc})
        elif not m:
            issues.append({"check": f"SSH {param}", "current": "not set (using default)",
                           "expected": safe_val, "description": desc})
    return issues

def check_cis_linux(conn) -> list:
    """Basic CIS Linux checks: world-writable files, SUID binaries, cron."""
    issues = []
    ww = run_ssh(conn, "find /tmp /var /home -maxdepth 3 -perm -002 -not -type d 2>/dev/null | head -10")
    if ww:
        for f in ww.splitlines():
            issues.append({"check": "World-writable file", "path": f,
                           "description": "World-writable file can be modified by any user — tampering risk"})
    suid = run_ssh(conn, "find / -perm -4000 -type f 2>/dev/null | grep -v snap | head -10")
    known_suid = {"/usr/bin/sudo", "/usr/bin/passwd", "/bin/su", "/usr/bin/pkexec"}
    if suid:
        for f in suid.splitlines():
            if f not in known_suid:
                issues.append({"check": "Unexpected SUID binary", "path": f,
                               "description": "Unexpected SUID binary — could be used for privilege escalation"})
    return issues

def check_cve_match(packages: dict) -> list:
    matches = []
    for vuln in KNOWN_VULNS:
        pkg = vuln["package"]
        if pkg in packages:
            matches.append({
                "cve": vuln["cve"], "package": pkg, "installed_version": packages[pkg],
                "cvss": vuln["cvss"], "severity": vuln["severity"],
                "description": vuln["description"], "fix": vuln["fix"]
            })
    return matches

def check_eol(os_str: str) -> str:
    for name, eol_date in EOL_OS.items():
        if name.lower() in os_str.lower():
            return eol_date
    return None

def query_nvd_cve(cve_id: str) -> dict:
    """Query NVD API v2 for CVE details — free, no key required."""
    if not HAS_REQUESTS:
        return {}
    try:
        r = requests.get(f"https://services.nvd.nist.gov/rest/json/cves/2.0?cveId={cve_id}",
                         timeout=15, headers={"User-Agent": "VulnScanner/1.0"})
        if r.status_code == 200:
            items = r.json().get("vulnerabilities", [])
            if items:
                cve_data = items[0].get("cve", {})
                metrics = cve_data.get("metrics", {})
                cvss_v3 = metrics.get("cvssMetricV31", []) or metrics.get("cvssMetricV30", [])
                score = cvss_v3[0]["cvssData"]["baseScore"] if cvss_v3 else None
                return {"id": cve_id, "score": score}
    except Exception:
        pass
    return {}

def scan_host(host: str, username: str, key_file: str, password: str) -> dict:
    """
    Scan one host and return its findings.

    Output is COLLECTED rather than printed, because several hosts are
    scanned concurrently and interleaved prints would be unreadable. The
    caller prints result['_lines'] in input order.
    """
    lines: list = []
    emit = lines.append
    emit(f"\n[+] {host}")
    result = {"host": host, "reachable": False, "os_info": {}, "packages": {},
              "cve_findings": [], "cis_ssh": [], "cis_linux": [], "risk_score": "UNKNOWN"}
    conn = connect_ssh(host, username, key_file, password)
    if not conn:
        emit(f"    [!] Cannot connect — check SSH credentials and access")
        result['_lines'] = lines
        return result

    result["reachable"] = True
    os_info = get_os_info(conn)
    result["os_info"] = os_info
    emit(f"    OS: {os_info.get('os','')} | Kernel: {os_info.get('kernel','')}")

    eol = check_eol(os_info.get("os", ""))
    if eol:
        emit(f"    ✗ CRITICAL: OS is End-of-Life since {eol} — no security patches available!")
        result["eol"] = eol

    packages = list_packages_linux(conn)
    result["package_count"] = len(packages)
    emit(f"    Packages: {len(packages)} installed")

    cve_matches = check_cve_match(packages)
    result["cve_findings"] = cve_matches
    if cve_matches:
        emit(f"    CVE Findings ({len(cve_matches)}):")
        for m in cve_matches:
            icon = {"CRITICAL": "✗ CRITICAL", "HIGH": "✗ HIGH",
                    "MEDIUM": "⚠ MEDIUM", "LOW": "ℹ LOW"}.get(m["severity"], m["severity"])
            emit(f"      {icon}  {m['cve']}  {m['package']} {m['installed_version']}  CVSS {m['cvss']}")
            emit(f"             {m['description']}")
            emit(f"             Fix: {m['fix']}")

    cis_ssh = check_cis_ssh(conn)
    result["cis_ssh"] = cis_ssh
    if cis_ssh:
        emit(f"    CIS SSH Benchmark ({len(cis_ssh)} issues):")
        for issue in cis_ssh:
            emit(f"      ✗ {issue['check']}: {issue['current']}  → should be '{issue['expected']}'")

    cis_linux = check_cis_linux(conn)
    result["cis_linux"] = cis_linux
    if cis_linux:
        for issue in cis_linux:
            emit(f"      ⚠ {issue['check']}: {issue.get('path','')}")

    critical = sum(1 for c in cve_matches if c["severity"] == "CRITICAL")
    high = sum(1 for c in cve_matches if c["severity"] == "HIGH")
    eol_risk = 2 if eol else 0
    if critical + eol_risk >= 1:
        result["risk_score"] = "CRITICAL"
    elif high >= 2:
        result["risk_score"] = "HIGH"
    elif high >= 1 or len(cis_ssh) >= 3:
        result["risk_score"] = "MEDIUM"
    else:
        result["risk_score"] = "LOW"
    emit(f"    Risk Score: {result['risk_score']}")
    conn.close()
    result['_lines'] = lines
    return result

def main():
    parser = argparse.ArgumentParser(description="Host Vulnerability Scanner")
    parser.add_argument("--hosts", required=True, help="Comma-separated host IPs or hostnames")
    parser.add_argument("--ssh-key", default=None, help="SSH private key file (e.g. ~/.ssh/id_rsa)")
    parser.add_argument("--username", default="root", help="SSH username (default: root)")
    parser.add_argument("--output", default=None, help="JSON output file")
    parser.add_argument("--workers", type=int, default=10,
                        help="How many hosts to scan at once (default: 10). Lower it if the jump host or the network cannot take the concurrency.")
    args = parser.parse_args()

    if not HAS_PARAMIKO:
        print("[!] paramiko not installed: pip install paramiko")
        sys.exit(1)

    # Password auth, when used, comes from the environment - never from an
    # argument. An argument is visible to every user on the machine in
    # `ps auxww` while the scan runs, and it is written to shell history.
    ssh_password = os.environ.get("SSH_PASS")
    if not args.ssh_key and not ssh_password:
        print("[!] No credentials. Use --ssh-key, or set SSH_PASS in the environment.")
        sys.exit(2)

    hosts = [h.strip() for h in args.hosts.split(",")]
    print(f"[*] Host Vulnerability Scanner | {len(hosts)} host(s)")
    print("─" * 40)

    # Hosts are scanned concurrently - the work is almost entirely waiting on
    # SSH, so this is the difference between seconds and minutes on a large
    # list. Results are collected into a dict keyed by host rather than
    # appended from the workers: list.append happens to be thread-safe in
    # CPython, but relying on that is how races get written, and the index
    # order would be completion order rather than input order.
    by_host: dict = {}
    done = 0
    progress_lock = threading.Lock()
    workers = min(args.workers, len(hosts))

    with ThreadPoolExecutor(max_workers=workers) as pool:
        futures = {pool.submit(scan_host, h, args.username,
                               args.ssh_key, ssh_password): h for h in hosts}
        for fut in as_completed(futures):
            host = futures[fut]
            try:
                by_host[host] = fut.result()
            except Exception as exc:        # one host must not abort the scan
                by_host[host] = {
                    "host": host, "reachable": False, "risk_score": "UNKNOWN",
                    "error": f"{type(exc).__name__}: {exc}",
                    "_lines": [f"\n[+] {host}",
                               f"    [!] Scan failed: {type(exc).__name__}: {exc}"],
                }
            with progress_lock:
                done += 1
                print(f"\r[*] Scanned {done}/{len(hosts)} host(s)...",
                      end="", flush=True)
    print("\r" + " " * 40 + "\r", end="")

    # Print in INPUT order so the output is deterministic and diffable,
    # not in whichever order the threads happened to finish.
    all_results = [by_host[h] for h in hosts]
    for r in all_results:
        for line in r.pop("_lines", []):
            print(line)

    critical = sum(1 for r in all_results if r.get("risk_score") == "CRITICAL")
    high = sum(1 for r in all_results if r.get("risk_score") == "HIGH")
    medium = sum(1 for r in all_results if r.get("risk_score") == "MEDIUM")
    total_cves = sum(len(r.get("cve_findings", [])) for r in all_results)
    total_cis = sum(len(r.get("cis_ssh", [])) + len(r.get("cis_linux", [])) for r in all_results)
    print(f"\n[*] Summary: {len(hosts)} hosts | {critical} CRITICAL | {high} HIGH | {medium} MEDIUM | {total_cves} CVEs | {total_cis} CIS violations")

    output = args.output or f"vuln_scan_{datetime.now().strftime('%Y-%m-%d')}.json"
    with open(output, "w") as fh:
        json.dump({"scan_date": datetime.now().isoformat(), "hosts_scanned": len(hosts),
                   "results": all_results}, fh, indent=2)
    print(f"[*] Report saved: {output}")
    return 2 if critical else (1 if high or medium else 0)

if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        print("\nInterrupted.", file=sys.stderr)
        sys.exit(130)