Sara/cve_analyzer.py

# Auxiliary module cve_analyzer.py for searching CVE using the NIST NVD database
#
# Copyright (c) 2026 Mahama Bazarov
# Licensed under the Apache 2.0 License
# This project is not affiliated with or endorsed by SIA Mikrotīkls

import json, re, os, requests, time
from datetime import datetime
from packaging.version import Version, InvalidVersion
from colorama import Fore, Style

# NVD v2.0 URL and settings
NVD_URL = "https://services.nvd.nist.gov/rest/json/cves/2.0"
KEYWORD = "routeros"
RESULTS_PER_PAGE = 2000
OUTPUT_FILE = "routeros_cves.json"
TIMEDRIFT = 900  # 900 seconds
GUTTER = 2  # spaces between columns


# Simple role-based color mapping
def paint(role: str, text: str) -> str:
    role_value = (role or "").lower()

    if role_value in ("crit", "fail"):
        return Fore.RED + text + Style.RESET_ALL
    if role_value == "warn":
        return Fore.YELLOW + text + Style.RESET_ALL
    if role_value == "ok":
        return Fore.GREEN + text + Style.RESET_ALL
    if role_value == "info":
        return Fore.CYAN + text + Style.RESET_ALL
    if role_value == "label":
        return Fore.CYAN + text + Style.RESET_ALL
    if role_value == "value":
        return Style.BRIGHT + text + Style.RESET_ALL

    return text


# ANSI and OSC-8 stripping for width calculation
ANSI_RE = re.compile(r"\x1b\[[0-9;]*[A-Za-z]")
OSC8_BEL = re.compile(r"\x1b]8;;.*?\x07")
OSC8_ST = re.compile(r"\x1b]8;;.*?\x1b\\")


def strip_osc8(s: str) -> str:
    tmp = OSC8_BEL.sub("", s)
    tmp = OSC8_ST.sub("", tmp)
    return tmp


def visible_len(s: str) -> int:
    raw = strip_osc8(s)
    no_ansi = ANSI_RE.sub("", raw)
    return len(no_ansi)


def pad_r(s: str, width: int) -> str:
    length = visible_len(s)
    pad_len = width - length
    if pad_len < 0:
        pad_len = 0
    return s + " " * pad_len


def pad_l(s: str, width: int) -> str:
    length = visible_len(s)
    pad_len = width - length
    if pad_len < 0:
        pad_len = 0
    return " " * pad_len + s


# Clickable hyperlink (OSC-8)
def term_link(text: str, url: str) -> str:
    return f"\x1b]8;;{url}\x1b\\{text}\x1b]8;;\x1b\\"


# Convert RouterOS version string to Version object
def normalize_version(v):
    if not v:
        return None
    try:
        return Version(v)
    except InvalidVersion:
        cleaned = re.sub(r"(rc|beta|testing|stable)[\d\-]*", "", v, flags=re.IGNORECASE)
        try:
            return Version(cleaned)
        except InvalidVersion:
            return None


# Extract version ranges from CVE description text
def extract_ranges_from_description(description):
    description = (description or "").lower()
    ranges = []

    matches = re.findall(r"(?:from\s+)?v?(\d+\.\d+(?:\.\d+)?)\s+to\s+v?(\d+\.\d+(?:\.\d+)?)", description)
    for start, end in matches:
        ranges.append({"versionStartIncluding": start, "versionEndIncluding": end})

    matches = re.findall(r"before\s+v?(\d+\.\d+(?:\.\d+)?)", description)
    for end in matches:
        ranges.append({"versionEndExcluding": end})

    matches = re.findall(r"after\s+v?(\d+\.\d+(?:\.\d+)?)", description)
    for start in matches:
        ranges.append({"versionStartExcluding": start})

    matches = re.findall(r"through\s+v?(\d+\.\d+(?:\.\d+)?)", description)
    for end in matches:
        ranges.append({"versionEndIncluding": end})

    matches = re.findall(r"v?(\d+\.\d+)\.x", description)
    for base in matches:
        ranges.append({"versionStartIncluding": f"{base}.0", "versionEndIncluding": f"{base}.999"})

    matches = re.findall(r"(?:up to|and below)\s+v?(\d+\.\d+(?:\.\d+)?)", description)
    for end in matches:
        ranges.append({"versionEndIncluding": end})

    return ranges


# Check if current version falls into a vulnerable range
def is_version_affected(current_v, version_info):
    def get(v_key):
        return normalize_version(version_info.get(v_key))

    criteria_raw = version_info.get("criteria", "") or ""
    criteria = criteria_raw.lower()
    end_excl_raw = version_info.get("versionEndExcluding", "")

    if criteria and ("mikrotik" not in criteria or "routeros" not in criteria):
        return False

    if isinstance(end_excl_raw, str) and end_excl_raw.startswith("7") and str(current_v).startswith("6."):
        return False
    if isinstance(end_excl_raw, str) and end_excl_raw.startswith("6") and str(current_v).startswith("7."):
        return False

    start_incl = get("versionStartIncluding")
    start_excl = get("versionStartExcluding")
    end_incl = get("versionEndIncluding")
    end_excl = get("versionEndExcluding")

    if not any([start_incl, start_excl, end_incl, end_excl]):
        version_match = re.search(r"routeros:([\w.\-]+)", criteria_raw)
        if version_match:
            version_exact = normalize_version(version_match.group(1))
            return version_exact is not None and current_v == version_exact
        return False

    for raw_key, normed in zip(
        ["versionStartIncluding", "versionStartExcluding", "versionEndIncluding", "versionEndExcluding"],
        [start_incl, start_excl, end_incl, end_excl],
    ):
        if version_info.get(raw_key) and normed is None:
            return False

    if start_incl and current_v < start_incl:
        return False
    if start_excl and current_v <= start_excl:
        return False
    if end_incl and current_v > end_incl:
        return False
    if end_excl and current_v >= end_excl:
        return False

    return True


# Download all RouterOS CVEs from NVD and save locally
def fetch_all_cves():
    all_cves = []
    start_index = 0

    print(Fore.CYAN + "[*] Fetching CVEs from NVD...")
    while True:
        params = {
            "keywordSearch": KEYWORD,
            "startIndex": start_index,
            "resultsPerPage": RESULTS_PER_PAGE,
        }
        try:
            response = requests.get(NVD_URL, params=params, timeout=30)
            response.raise_for_status()
            data = response.json()
        except requests.exceptions.RequestException as e:
            print(Fore.RED + f"[-] HTTP Error: {e}")
            break
        except json.JSONDecodeError:
            print(Fore.RED + "[-] Failed to parse JSON from NVD.")
            break

        cve_items = data.get("vulnerabilities", [])
        total_results = data.get("totalResults", 0)

        for item in cve_items:
            cve = item.get("cve", {})
            cve_id = cve.get("id")
            description = next((d["value"] for d in cve.get("descriptions", []) if d.get("lang") == "en"), "")
            severity = "UNKNOWN"
            score = "N/A"
            published = cve.get("published", "")

            metrics = cve.get("metrics", {})
            if "cvssMetricV31" in metrics:
                cvss = metrics["cvssMetricV31"][0]["cvssData"]
                severity = cvss.get("baseSeverity", "UNKNOWN")
                score = cvss.get("baseScore", "N/A")
            elif "cvssMetricV30" in metrics:
                cvss = metrics["cvssMetricV30"][0]["cvssData"]
                severity = cvss.get("baseSeverity", "UNKNOWN")
                score = cvss.get("baseScore", "N/A")

            affected_versions = []
            for config in cve.get("configurations", []):
                for node in config.get("nodes", []):
                    for match in node.get("cpeMatch", []):
                        if not match.get("vulnerable", False):
                            continue
                        criteria = match.get("criteria", "") or ""
                        crit_l = criteria.lower()
                        if "mikrotik" not in crit_l or "routeros" not in crit_l:
                            continue
                        affected_versions.append(
                            {
                                "criteria": criteria,
                                "versionStartIncluding": match.get("versionStartIncluding"),
                                "versionStartExcluding": match.get("versionStartExcluding"),
                                "versionEndIncluding": match.get("versionEndIncluding"),
                                "versionEndExcluding": match.get("versionEndExcluding"),
                            }
                        )

            all_cves.append(
                {
                    "cve_id": cve_id,
                    "description": description,
                    "severity": severity,
                    "cvss_score": score,
                    "published": published,
                    "affected_versions": affected_versions,
                }
            )

        start_index += RESULTS_PER_PAGE
        if start_index >= total_results:
            break
        time.sleep(1.5)

    with open(OUTPUT_FILE, "w", encoding="utf-8") as f:
        json.dump(all_cves, f, indent=2, ensure_ascii=False)
    print(Fore.GREEN + f"[+] Saved {len(all_cves)} CVEs to {OUTPUT_FILE}")


# Load local CVE cache and optionally refresh it
def load_cve_data():
    if not os.path.isfile(OUTPUT_FILE):
        print(Fore.YELLOW + f"[!] {OUTPUT_FILE} not found.")
        fetch_all_cves()
    else:
        print(Fore.YELLOW + f"[?] {OUTPUT_FILE} already exists.")
        # Get file last change date
        file_mtime = os.path.getmtime(OUTPUT_FILE)
        current_time = time.time()
        time_diff = current_time - file_mtime
        #answer = input(Fore.YELLOW + "    Overwrite it with fresh CVE data? [yes/no]: ").strip().lower()
        #if answer == "no":
        if time_diff < TIMEDRIFT:
            print(Fore.GREEN + f"    File is recent (modified {time_diff:.0f}s ago). Skipping download.")
        else:
        #if answer == "yes":
            print(Fore.YELLOW + f"    File is old (modified {time_diff:.0f}s ago). Downloading fresh data...")
            fetch_all_cves()

    try:
        with open(OUTPUT_FILE, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception as e:
        print(Fore.RED + f"[-] Failed to load {OUTPUT_FILE}: {e}")
        return None


# Format CVSS score as 0.1 or N/A
def fmt_cvss(x):
    try:
        value = float(x)
        return f"{value:0.1f}"
    except Exception:
        return "N/A"


# Severity rank for sorting
sev_rank = {
    "CRITICAL": 0,
    "HIGH": 1,
    "MEDIUM": 2,
    "LOW": 3,
    "UNKNOWN": 4,
}


def key_tuple(m):
    sev = m.get("severity", "UNKNOWN") or "UNKNOWN"
    sev_u = sev.upper()
    rank = sev_rank.get(sev_u, 4)
    pub = m.get("published") or ""
    return (rank, pub)


# Severity tag (CRIT/HIGH/MED/LOW/UNK)
def sev_tag(sev: str):
    sev_u = (sev or "UNKNOWN").upper()
    if sev_u == "CRITICAL":
        return paint("crit", "CRIT")
    if sev_u == "HIGH":
        return paint("fail", "HIGH")
    if sev_u == "MEDIUM":
        return paint("warn", "MED")
    if sev_u == "LOW":
        return paint("info", "LOW")
    return paint("value", "UNK")


def count_seg(label: str, n: int):
    role_map = {
        "CRIT": "crit",
        "HIGH": "fail",
        "MED": "warn",
        "LOW": "info",
        "UNK": "value",
    }
    role = role_map[label]
    return paint(role, label) + ": " + paint("value", str(n))


# Summary header
def render_summary(version: str, matches, counters):
    c_crit = counters.get("CRITICAL", 0)
    c_high = counters.get("HIGH", 0)
    c_med = counters.get("MEDIUM", 0)
    c_low = counters.get("LOW", 0)
    c_unk = counters.get("UNKNOWN", 0)

    line1 = (
        paint("label", "Target RouterOS Version:") + " " + paint("value", version)
        + "    "
        + paint("label", "Matched CVEs:") + " " + paint("value", str(len(matches)))
    )

    parts = [
        count_seg("CRIT", c_crit),
        count_seg("HIGH", c_high),
        count_seg("MED", c_med),
        count_seg("LOW", c_low),
        count_seg("UNK", c_unk),
    ]
    line2 = " | ".join(parts)

    print(line1)
    print(line2)
    print()


# Table rendering
def render_cve(rows):
    max_id_len = 14
    for m in rows:
        cve_id = m.get("cve_id", "") or ""
        length = len(cve_id)
        if length > max_id_len:
            max_id_len = length

    id_w = max_id_len + 2
    if id_w < 16:
        id_w = 16
    if id_w > 22:
        id_w = 22

    sev_w = 5
    cvss_w = 4
    pub_w = 10
    sep = " " * GUTTER

    head = (
        pad_r(paint("label", "CVE ID"), id_w)
        + sep
        + pad_r(paint("label", "SEV"), sev_w)
        + sep
        + pad_r(paint("label", "CVSS"), cvss_w)
        + sep
        + pad_r(paint("label", "PUBLISHED"), pub_w)
    )
    print(head)

    for m in rows:
        cve_id = m.get("cve_id", "") or ""
        link = term_link(cve_id, "https://nvd.nist.gov/vuln/detail/" + cve_id)

        sev_t = sev_tag(m.get("severity", "UNKNOWN"))
        cvss = fmt_cvss(m.get("cvss_score", "N/A"))

        published = m.get("published") or "-"
        published = str(published)[:10]

        line = (
            pad_r(paint("info", link), id_w)
            + sep
            + pad_r(sev_t, sev_w)
            + sep
            + pad_l(paint("value", cvss), cvss_w)
            + sep
            + pad_r(paint("value", published), pub_w)
        )
        print(line)


# Core CVE matching logic for a given RouterOS version
def run_cve_match_for_version(current_v, current_version: str):
    cve_data = load_cve_data()
    if not cve_data:
        return

    counters = {"CRITICAL": 0, "HIGH": 0, "MEDIUM": 0, "LOW": 0, "UNKNOWN": 0}
    matches = []

    for cve in cve_data:
        matched = False
        affected_versions = cve.get("affected_versions", [])

        if not affected_versions:
            affected_versions = extract_ranges_from_description(cve.get("description", ""))

        for version_info in affected_versions:
            if is_version_affected(current_v, version_info):
                matched = True
                break

        if not matched:
            continue

        sev = (cve.get("severity", "UNKNOWN") or "UNKNOWN").upper()
        counters[sev] = counters.get(sev, 0) + 1

        matches.append(
            {
                "cve_id": cve.get("cve_id", ""),
                "severity": sev,
                "cvss_score": cve.get("cvss_score", "N/A"),
                "published": cve.get("published", ""),
            }
        )

    print()
    render_summary(current_version, matches, counters)

    if not matches:
        print(paint("ok", "[*] No known CVEs found for this RouterOS version"))
        print()
        return

    rows = sorted(matches, key=key_tuple)
    render_cve(rows)


# Live CVE audit based on device version (kept for potential reuse)
def run_cve_audit(connection):
    print(paint("label", "[+] Search for CVEs for a specific version"))
    output = connection.send_command("/system resource print")
    match = re.search(r"version:\s*([\w.\-]+)", output)
    if not match:
        print(Fore.RED + "[-] ERROR: Could not determine RouterOS version.")
        return

    current_version = match.group(1)
    current_v = normalize_version(current_version)
    if not current_v:
        print(Fore.RED + f"[-] ERROR: RouterOS version '{current_version}' is invalid.")
        return

    run_cve_match_for_version(current_v, current_version)



# CVE audit for manually provided RouterOS version string (used by Sara)
def run_cve_audit_for_version(version_str: str):
    print(paint("label", "[+] Search for CVEs for a specific version"))
    current_version = version_str.strip()
    current_v = normalize_version(current_version)
    if not current_v:
        print(Fore.RED + f"[-] ERROR: RouterOS version '{current_version}' is invalid.")
        return

    run_cve_match_for_version(current_v, current_version)