#!/usr/bin/env python3 """ Priorizador de CVE — Ciberkit =================================== Responde a «¿qué parcheo primero?». Extrae los CVE de informes de escáner (Nessus, OpenVAS, Trivy, Grype…), de cualquier texto o de la línea de órdenes y los ordena combinando tres fuentes públicas: - CISA KEV: explotación confirmada (y uso en ransomware). - EPSS (FIRST): probabilidad de explotación en los próximos 30 días. - CVSS (CVE.org): gravedad técnica, más la valoración SSVC de CISA. Usa los mismos criterios que la página «¿Qué parcheo primero?» de la suite. Solo se envían a Internet los identificadores CVE, que son públicos; el contenido de tus informes no sale del equipo. Sin dependencias: Python 3.8+ en Windows, Linux o macOS. Ejemplos -------- python priorizar_cves.py informe_nessus.csv python priorizar_cves.py trivy.json grype.json --expuesto --csv prioridad.csv python priorizar_cves.py CVE-2021-44228 CVE-2024-3400 --json salida.json trivy image -f json miapp:latest | python priorizar_cves.py - python priorizar_cves.py informe.csv --sin-detalle # solo KEV y EPSS (más rápido) Código de salida: 0 nada urgente, 1 hay algo para esta semana, 2 hay algo para hoy, 3 no se pudieron consultar las fuentes (útil en CI). """ import argparse import csv import json import os import re import sys import time import urllib.error import urllib.request from concurrent.futures import ThreadPoolExecutor FORMATO = "suite-redes.cve-prioridad/1" RE_CVE = re.compile(r"CVE-(?:1999|2\d{3})-\d{4,7}", re.I) MAX_FICHERO = 200 * 1024 * 1024 # Los mismos umbrales que assets/js/core/vulnprio.js (un test lo comprueba). UMBRALES = {"epssHoy": 0.5, "epssSemana": 0.1, "epssMes": 0.01, "cvssCritico": 9, "cvssAlto": 7} NIVELES = { "hoy": (1, "Hoy"), "semana": (2, "Esta semana"), "mes": (3, "Este mes"), "normal": (4, "Ciclo normal"), "revisar": (5, "Sin datos"), } URL_EPSS = "https://api.first.org/data/v1/epss?cve={ids}" URL_KEV = [ "https://raw.githubusercontent.com/cisagov/kev-data/develop/known_exploited_vulnerabilities.json", "https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json", ] URL_CVE = "https://cveawg.mitre.org/api/cve/{id}" URL_NVD = "https://services.nvd.nist.gov/rest/json/cves/2.0?cveId={id}" PAUSA_NVD = 6.5 # NVD admite 5 consultas cada 30 s sin clave # ───────────────────────── Entrada ───────────────────────── def clave_cve(cve): _, anio, num = cve.split("-") return (int(anio), int(num)) def extraer_cves(texto): """CVE distintos de un texto con el número de apariciones, del más reciente al más antiguo.""" cuenta = {} for m in RE_CVE.finditer(texto or ""): cve = m.group(0).upper() cuenta[cve] = cuenta.get(cve, 0) + 1 return sorted(cuenta.items(), key=lambda kv: clave_cve(kv[0]), reverse=True) def leer_entradas(entradas, stdin=None): """Junta el texto de ficheros, «-» (entrada estándar) e identificadores sueltos.""" trozos = [] for e in entradas: if e == "-": trozos.append((stdin or sys.stdin).read()) elif os.path.isfile(e): if os.path.getsize(e) > MAX_FICHERO: raise ValueError(f"{e} pesa más de 200 MB") with open(e, encoding="utf-8", errors="replace") as f: trozos.append(f.read()) elif RE_CVE.fullmatch(e.strip()): trozos.append(e) else: raise ValueError(f"«{e}» no es un fichero ni un identificador CVE") return "\n".join(trozos) # ───────────────────────── Fuentes ───────────────────────── def descargar_json(url, timeout=30, intentos=2): """Devuelve el JSON, None si el recurso no existe (404), o lanza la última excepción.""" error = None for i in range(intentos): try: req = urllib.request.Request(url, headers={"user-agent": "suite-redes-cve/1", "accept": "application/json"}) with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read().decode("utf-8")) except urllib.error.HTTPError as e: if e.code == 404: return None error = e except Exception as e: # red o JSON error = e time.sleep(0.5 * (i + 1)) raise error def parse_kev(datos): kev = {} for x in (datos or {}).get("vulnerabilities", []): kev[str(x.get("cveID", "")).upper()] = { "desde": x.get("dateAdded"), "limite": x.get("dueDate"), "ransomware": str(x.get("knownRansomwareCampaignUse", "")).lower() == "known", "fabricante": x.get("vendorProject") or "", "producto": x.get("product") or "", "nombre": x.get("vulnerabilityName") or "", } return kev def parse_epss(datos): epss = {} for x in (datos or {}).get("data", []): try: epss[str(x["cve"]).upper()] = {"epss": float(x["epss"]), "percentil": float(x.get("percentile") or 0), "fecha": x.get("date")} except (KeyError, TypeError, ValueError): continue return epss def _cvss_de(metricas, origen): for clave, version in (("cvssV3_1", "3.1"), ("cvssV4_0", "4.0"), ("cvssV3_0", "3.0")): for m in metricas or []: c = (m or {}).get(clave) if not c or not c.get("vectorString"): continue try: return {"version": version, "nota": float(c["baseScore"]), "vector": c["vectorString"], "origen": origen} except (KeyError, TypeError, ValueError): continue return None def parse_registro(datos): """Registro de CVE.org (formato 5.x) → título, producto, CVSS (fabricante antes que CISA) y SSVC.""" meta = (datos or {}).get("cveMetadata", {}) cont = (datos or {}).get("containers", {}) cna = cont.get("cna", {}) adps = cont.get("adp", []) cisa = next((a for a in adps if "CISA" in (a.get("title") or "")), None) afectados = cna.get("affected") or [] afectado = next((a for a in afectados if a.get("product") and a.get("product") != "n/a"), afectados[0] if afectados else None) ssvc = None for m in (cisa or {}).get("metrics", []): otro = (m or {}).get("other") or {} if otro.get("type") == "ssvc": op = {} for o in (otro.get("content") or {}).get("options", []): op.update(o) ssvc = {"explotacion": str(op.get("Exploitation", "")).lower() or None, "automatizable": str(op.get("Automatable", "")).lower() == "yes"} cvss = _cvss_de(cna.get("metrics"), "CNA") or _cvss_de((cisa or {}).get("metrics"), "CISA") if not cvss: cvss = next((c for c in (_cvss_de(a.get("metrics"), "ADP") for a in adps) if c), None) limpio = lambda v: v if v and v != "n/a" else None return { "rechazado": meta.get("state") == "REJECTED", "publicado": (meta.get("datePublished") or "")[:10] or None, "titulo": cna.get("title"), "fabricante": limpio((afectado or {}).get("vendor")), "producto": limpio((afectado or {}).get("product")), "cvss": cvss, "ssvc": ssvc, } def parse_nvd(datos): """Respuesta de NVD (API 2.0) → CVSS, para cuando CVE.org no trae nota.""" vulns = (datos or {}).get("vulnerabilities") or [] metricas = ((vulns[0] if vulns else {}).get("cve") or {}).get("metrics") or {} for clave, version in (("cvssMetricV31", "3.1"), ("cvssMetricV40", "4.0"), ("cvssMetricV30", "3.0")): lista = metricas.get(clave) or [] m = next((x for x in lista if x.get("type") == "Primary"), lista[0] if lista else None) c = (m or {}).get("cvssData") or {} try: if c.get("vectorString"): return {"version": version, "nota": float(c["baseScore"]), "vector": c["vectorString"], "origen": "NVD"} except (KeyError, TypeError, ValueError): continue return None # ───────────────────────── Priorización ───────────────────────── def _pct(x): """0.9986 → «99,9 %», 0.03 → «3 %», 0.0056 → «0,6 %» (un decimal solo en los extremos).""" if 0 < x < 0.0005: return "<0,1 %" v = x * 100 texto = f"{v:.1f}".rstrip("0").rstrip(".") if x < 0.1 or x >= 0.99 else f"{v:.0f}" return texto.replace(".", ",") + " %" def _gravedad(nota): return "ninguna" if nota == 0 else "baja" if nota < 4 else "media" if nota < 7 else "alta" if nota < 9 else "crítica" def priorizar(item, expuesto=False): """item: {kev, epss, cvss, ssvc, rechazado, no_existe} → (nivel, [motivos]).""" U = UMBRALES motivos = [] if item.get("rechazado"): return "normal", ["CVE rechazado: se retiró y no es una vulnerabilidad válida."] kev = item.get("kev") epss = (item.get("epss") or {}).get("epss") cvss = (item.get("cvss") or {}).get("nota") ssvc = item.get("ssvc") or {} activa = ssvc.get("explotacion") == "active" if kev: motivos.append("Explotación confirmada: está en el catálogo KEV de CISA" + (f" desde el {kev['desde']}" if kev.get("desde") else "") + ".") if kev.get("ransomware"): motivos.append("Se usa en campañas de ransomware.") elif activa: motivos.append("CISA la marca como explotada activamente (SSVC).") if epss is not None: motivos.append(f"EPSS {_pct(epss)}: probabilidad de explotación en 30 días.") if ssvc.get("explotacion") == "poc": motivos.append("Hay prueba de concepto pública (SSVC de CISA).") if ssvc.get("automatizable"): motivos.append("Explotación automatizable: puede usarse en ataques masivos.") if cvss is not None: nota = f"{cvss:.1f}".replace(".", ",") motivos.append(f"CVSS {item['cvss']['version']}: {nota} ({_gravedad(cvss)}).") if kev or activa or (epss is not None and epss >= U["epssHoy"]): nivel = "hoy" elif epss is None and cvss is None: motivos.append("No aparece en CVE.org: comprueba que el identificador sea correcto." if item.get("no_existe") else "Sin nota CVSS ni EPSS todavía (puede ser muy reciente).") return "revisar", motivos elif (epss is not None and epss >= U["epssSemana"]) or (cvss is not None and cvss >= U["cvssCritico"] and epss is not None and epss >= U["epssMes"]): nivel = "semana" elif (cvss is not None and cvss >= U["cvssAlto"]) or (epss is not None and epss >= U["epssMes"]): nivel = "mes" else: nivel = "normal" if epss is None and nivel != "hoy": motivos.append("Sin EPSS: la prioridad se basa solo en la gravedad.") if cvss is None and nivel != "hoy": motivos.append("Sin CVSS publicado: la prioridad se basa solo en la probabilidad.") if expuesto and nivel in ("semana", "mes", "normal"): nivel = {"semana": "hoy", "mes": "semana", "normal": "mes"}[nivel] motivos.append("Sube un nivel porque el sistema está expuesto a Internet.") return nivel, motivos def ordenar(items): return sorted(items, key=lambda it: ( NIVELES[it["nivel"]][0], 0 if it.get("kev") else 1, -((it.get("epss") or {}).get("epss") or -1), -((it.get("cvss") or {}).get("nota") or -1), tuple(-x for x in clave_cve(it["cve"])))) def analizar(cves, expuesto=False, detalle=True, descargar=descargar_json, hilos=8, progreso=None, max_nvd=20, pausa=time.sleep): """cves: [(id, apariciones)] → (items ordenados, fuentes, avisos).""" avisos = [] fuentes = {"kev_version": None, "kev_total": None, "epss_fecha": None, "consultado": time.strftime("%Y-%m-%dT%H:%M:%S%z")} ids = [c for c, _ in cves] kev = None for url in URL_KEV: try: datos = descargar(url, timeout=60) if datos and datos.get("vulnerabilities"): kev = parse_kev(datos) fuentes["kev_version"], fuentes["kev_total"] = datos.get("catalogVersion"), datos.get("count", len(kev)) break except Exception: continue if kev is None: avisos.append("No se pudo descargar el catálogo KEV de CISA: la prioridad puede estar subestimada.") epss, fallo_epss = {}, False for i in range(0, len(ids), 80): try: epss.update(parse_epss(descargar(URL_EPSS.format(ids=",".join(ids[i:i + 80]))))) except Exception: fallo_epss = True if fallo_epss: avisos.append("No se pudo consultar EPSS para todos los CVE: donde falta, la prioridad se basa solo en la gravedad.") fuentes["epss_fecha"] = next((v["fecha"] for v in epss.values() if v.get("fecha")), None) items = [{"cve": c, "apariciones": n, "kev": (kev or {}).get(c), "epss": epss.get(c), "cvss": None, "ssvc": None} for c, n in cves] if detalle: fallos = [0] hechos = [0] def uno(it): try: datos = descargar(URL_CVE.format(id=it["cve"])) if datos is None: it["no_existe"] = True else: it.update(parse_registro(datos)) except Exception: fallos[0] += 1 hechos[0] += 1 if progreso: progreso(hechos[0], len(items), "CVE.org") with ThreadPoolExecutor(max_workers=max(1, min(hilos, 16))) as ex: list(ex.map(uno, items)) if fallos[0]: avisos.append(f"No se pudo consultar el detalle de {fallos[0]} CVE en CVE.org: les falta la nota CVSS.") # NVD como respaldo para los que no traen CVSS (despacio: tiene límite de consultas) sin_cvss = [it for it in items if not it.get("cvss") and not it.get("rechazado") and not it.get("no_existe")] for n, it in enumerate(sin_cvss[:max_nvd]): if n >= 4: pausa(PAUSA_NVD) try: it["cvss"] = parse_nvd(descargar(URL_NVD.format(id=it["cve"]))) except Exception: pass if progreso: progreso(n + 1, min(len(sin_cvss), max_nvd), "NVD") quedan = sum(1 for it in sin_cvss if not it.get("cvss")) if quedan: avisos.append(f"{quedan} CVE no tienen nota CVSS publicada" + (" (o no se consultó NVD: sube --max-nvd)" if len(sin_cvss) > max_nvd else "") + ": se priorizan solo por EPSS y KEV.") else: avisos.append("Sin detalle de CVE.org (--sin-detalle): no hay notas CVSS y la prioridad se basa solo en KEV y EPSS.") for it in items: it["nivel"], it["motivos"] = priorizar(it, expuesto) it["prioridad"], it["plazo"] = NIVELES[it["nivel"]] fuentes["sin_fuentes"] = kev is None and not epss return ordenar(items), fuentes, avisos # ───────────────────────── Salida ───────────────────────── COLUMNAS = ["prioridad", "plazo", "cve", "cvss", "cvss_version", "epss", "epss_percentil", "kev", "kev_fecha_limite", "ransomware", "fabricante", "producto", "apariciones", "motivos"] def _celda(v): s = "" if v is None else str(v) return "'" + s if s[:1] in ("=", "+", "-", "@") else s # sin inyección de fórmulas en Excel def fila_csv(it): kev, epss, cvss = it.get("kev") or {}, it.get("epss") or {}, it.get("cvss") or {} return { "prioridad": it["prioridad"], "plazo": it["plazo"], "cve": it["cve"], "cvss": cvss.get("nota", ""), "cvss_version": cvss.get("version", ""), "epss": epss.get("epss", ""), "epss_percentil": epss.get("percentil", ""), "kev": "sí" if kev else "no", "kev_fecha_limite": kev.get("limite") or "", "ransomware": "sí" if kev.get("ransomware") else "no", "fabricante": _celda(it.get("fabricante") or kev.get("fabricante")), "producto": _celda(it.get("producto") or kev.get("producto")), "apariciones": it["apariciones"], "motivos": " ".join(it["motivos"]), } def codigo_salida(items, fuentes): if fuentes.get("sin_fuentes"): return 3 niveles = {it["nivel"] for it in items} return 2 if "hoy" in niveles else 1 if "semana" in niveles else 0 def imprimir(items, fuentes, avisos, color, limite): def c(cod, s): return f"\033[{cod}m{s}\033[0m" if color else s tono = {"hoy": "31", "semana": "33", "mes": "36", "normal": "32", "revisar": "90"} cuenta = {n: sum(1 for it in items if it["nivel"] == n) for n in NIVELES} print(f"\n{len(items)} CVE · " + " · ".join(f"{c(tono[n], str(cuenta[n]))} {NIVELES[n][1].lower()}" for n in NIVELES if cuenta[n] or n == "hoy")) print(f"{'':2}{'PLAZO':<13}{'CVE':<18}{'CVSS':>5} {'EPSS':>7} {'KEV':<4} PRODUCTO") for it in items[:limite]: kev, epss, cvss = it.get("kev"), it.get("epss"), it.get("cvss") producto = " ".join(x for x in [it.get("fabricante") or (kev or {}).get("fabricante"), it.get("producto") or (kev or {}).get("producto")] if x)[:44] marca = ("KEV" + ("*" if kev.get("ransomware") else "")) if kev else "" t = tono[it["nivel"]] nota = f"{cvss['nota']:.1f}".replace(".", ",") if cvss else "—" prob = _pct(epss["epss"]) if epss else "—" print(f"{c(t, '●')} {c(t, format(it['plazo'], '<13'))}{it['cve']:<18}{nota:>5} {prob:>7} {marca:<4} {producto}") if len(items) > limite: print(f" … y {len(items) - limite} más (usa --csv o --todos)") if any((it.get("kev") or {}).get("ransomware") for it in items): print(" KEV* = usado en campañas de ransomware") for a in avisos: print(c("33", f" ▲ {a}")) pie = [f"KEV {fuentes['kev_version']} ({fuentes['kev_total']} vulnerabilidades)" if fuentes.get("kev_version") else None, f"EPSS del {fuentes['epss_fecha']}" if fuentes.get("epss_fecha") else None] print(" Fuentes: " + " · ".join(x for x in pie if x) if any(pie) else " Fuentes: no disponibles") def preparar_salida(): """Evita que el script falle en consolas o redirecciones sin UTF-8 (p. ej. cp1252 en Windows).""" for flujo in (sys.stdout, sys.stderr): try: if not flujo.isatty(): flujo.reconfigure(encoding="utf-8", errors="replace") else: flujo.reconfigure(errors="replace") except (AttributeError, ValueError): pass def main(argv=None, descargar=descargar_json, stdin=None, pausa=time.sleep): preparar_salida() ap = argparse.ArgumentParser(description="Ordena CVE por urgencia real: CISA KEV, EPSS y CVSS.") ap.add_argument("entradas", nargs="+", help="ficheros de informe, identificadores CVE o «-» para la entrada estándar") ap.add_argument("--expuesto", action="store_true", help="los sistemas están expuestos a Internet (sube un nivel)") ap.add_argument("--sin-detalle", action="store_true", help="no consultar CVE.org (sin CVSS; solo KEV y EPSS)") ap.add_argument("--csv", metavar="FICHERO", help="guarda la lista priorizada en CSV") ap.add_argument("--json", metavar="FICHERO", help="guarda el informe completo en JSON") ap.add_argument("--todos", action="store_true", help="muestra todos los CVE en pantalla (por defecto, 40)") ap.add_argument("--max-nvd", type=int, default=20, metavar="N", help="máximo de CVE sin nota que se buscan en NVD, que es lento (por defecto 20)") ap.add_argument("--hilos", type=int, default=8, help="consultas en paralelo a CVE.org (por defecto 8)") a = ap.parse_args(argv) try: cves = extraer_cves(leer_entradas(a.entradas, stdin)) except (ValueError, OSError) as e: print(f"Error: {e}", file=sys.stderr) return 3 if not cves: print("No hay ningún identificador CVE en la entrada (tienen la forma CVE-2024-12345).", file=sys.stderr) return 3 tty = sys.stderr.isatty() print(f"{len(cves)} CVE distintos. Consultando KEV, EPSS{'' if a.sin_detalle else ' y CVE.org'}…", file=sys.stderr) progreso = (lambda h, t, que: print(f"\r {que}: {h}/{t} ", end="" if h < t else "\n", file=sys.stderr, flush=True)) if tty else None items, fuentes, avisos = analizar(cves, a.expuesto, not a.sin_detalle, descargar, a.hilos, progreso, a.max_nvd, pausa) imprimir(items, fuentes, avisos, sys.stdout.isatty(), len(items) if a.todos else 40) if a.csv: with open(a.csv, "w", newline="", encoding="utf-8-sig") as f: w = csv.DictWriter(f, fieldnames=COLUMNAS) w.writeheader() w.writerows(fila_csv(it) for it in items) print(f"CSV guardado en {a.csv}") if a.json: with open(a.json, "w", encoding="utf-8") as f: json.dump({"formato": FORMATO, "expuesto": a.expuesto, "umbrales": UMBRALES, "fuentes": fuentes, "avisos": avisos, "cves": items}, f, ensure_ascii=False, indent=2) print(f"JSON guardado en {a.json}") return codigo_salida(items, fuentes) if __name__ == "__main__": sys.exit(main())