NVD Vulnerability Scanner for Homelab - Python implementation

vuln-scanner.py

#!/usr/bin/env python3
"""
NVD Vulnerability Scanner for Homelab
Scans installed packages against National Vulnerability Database
"""

import requests
import json
import subprocess
from packaging import version
from datetime import datetime, timedelta

class VulnerabilityScanner:
    def __init__(self, api_key=None):
        self.nvd_url = "https://services.nvd.nist.gov/rest/json/cves/2.0"
        self.headers = {
            "Accept": "application/json",
            "apiKey": api_key if api_key else ""
        }
        self.cache = {}
        
    def get_installed_packages(self, host):
        """Get list of installed packages from remote host via SSH."""
        try:
            result = subprocess.run(
                ["ssh", host, "dpkg -l"],
                capture_output=True,
                text=True,
                timeout=30
            )
            packages = []
            for line in result.stdout.split('\n'):
                if line.startswith('ii'):
                    parts = line.split()
                    if len(parts) >= 3:
                        packages.append({
                            'name': parts[1],
                            'version': parts[2],
                            'host': host
                        })
            return packages
        except subprocess.TimeoutExpired:
            print(f"Timeout scanning {host}")
            return []
            
    def query_nvd(self, package_name, days=30):
        """Query NVD for CVEs affecting package."""
        if package_name in self.cache:
            return self.cache[package_name]
            
        params = {
            "keyword": package_name,
            "resultsPerPage": 20,
            "pubStartDate": (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d"),
            "pubEndDate": datetime.now().strftime("%Y-%m-%d")
        }
        
        try:
            response = requests.get(
                self.nvd_url,
                headers=self.headers,
                params=params,
                timeout=10
            )
            response.raise_for_status()
            cves = response.json().get('vulnerabilities', [])
            self.cache[package_name] = cves
            return cves
        except requests.RequestException as e:
            print(f"NVD query failed for {package_name}: {e}")
            return []
            
    def version_in_range(self, installed, vuln_start, vuln_end=None):
        """Check if installed version falls in vulnerable range."""
        try:
            installed_ver = version.parse(installed)
            start_ver = version.parse(vuln_start)
            
            if vuln_end:
                end_ver = version.parse(vuln_end)
                return start_ver <= installed_ver <= end_ver
            else:
                return start_ver <= installed_ver
        except version.InvalidVersion:
            return False
            
    def scan_package(self, package):
        """Scan single package for vulnerabilities."""
        cves = self.query_nvd(package['name'])
        vulnerabilities = []
        
        for cve_item in cves:
            cve = cve_item.get('cve', {})
            cve_id = cve.get('id')
            
            # Extract CVSS score
            metrics = cve.get('metrics', {})
            cvss_v3 = metrics.get('cvssMetricV31', [{}])[0]
            severity = cvss_v3.get('cvssData', {}).get('baseSeverity', 'UNKNOWN')
            score = cvss_v3.get('cvssData', {}).get('baseScore', 0.0)
            
            # Check version ranges
            configs = cve.get('configurations', [])
            for config in configs:
                nodes = config.get('nodes', [])
                for node in nodes:
                    cpe_matches = node.get('cpeMatch', [])
                    for cpe in cpe_matches:
                        if cpe.get('vulnerable'):
                            version_start = cpe.get('versionStartIncluding')
                            version_end = cpe.get('versionEndExcluding')
                            
                            if self.version_in_range(
                                package['version'],
                                version_start or '0',
                                version_end
                            ):
                                vulnerabilities.append({
                                    'cve_id': cve_id,
                                    'severity': severity,
                                    'score': score,
                                    'package': package['name'],
                                    'installed_version': package['version'],
                                    'host': package['host']
                                })
                                
        return vulnerabilities
        
    def scan_homelab(self, hosts):
        """Scan all homelab hosts for vulnerabilities."""
        all_vulns = []
        
        for host in hosts:
            print(f"Scanning {host}...")
            packages = self.get_installed_packages(host)
            
            for pkg in packages:
                vulns = self.scan_package(pkg)
                all_vulns.extend(vulns)
                
        return all_vulns
        
    def generate_report(self, vulnerabilities):
        """Generate vulnerability report grouped by severity."""
        report = {
            'CRITICAL': [],
            'HIGH': [],
            'MEDIUM': [],
            'LOW': []
        }
        
        for vuln in vulnerabilities:
            severity = vuln['severity']
            report[severity].append(vuln)
            
        return report

if __name__ == "__main__":
    # Initialize scanner with NVD API key
    scanner = VulnerabilityScanner(api_key="your_nvd_api_key_here")
    
    # Define homelab hosts
    hosts = [
        "homelab-server-01",
        "homelab-server-02",
        "docker-host",
        "proxmox-node-01"
    ]
    
    # Scan all hosts
    vulnerabilities = scanner.scan_homelab(hosts)
    
    # Generate report
    report = scanner.generate_report(vulnerabilities)
    
    # Print summary
    print(f"\n=== Vulnerability Scan Report ===")
    print(f"Total vulnerabilities: {len(vulnerabilities)}")
    print(f"Critical: {len(report['CRITICAL'])}")
    print(f"High: {len(report['HIGH'])}")
    print(f"Medium: {len(report['MEDIUM'])}")
    print(f"Low: {len(report['LOW'])}")
    
    # Print critical vulnerabilities
    if report['CRITICAL']:
        print(f"\n=== CRITICAL Vulnerabilities ===")
        for vuln in report['CRITICAL']:
            print(f"{vuln['cve_id']}: {vuln['package']} {vuln['installed_version']} on {vuln['host']}")