467 lines
16 KiB
Python
Executable File
467 lines
16 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
import time
|
|
import json
|
|
import yaml
|
|
import requests
|
|
import re
|
|
from datetime import datetime, timedelta, timezone
|
|
from packaging.version import Version, InvalidVersion
|
|
|
|
BASE_URL = "https://euvdservices.enisa.europa.eu/api/search"
|
|
HEADERS = {
|
|
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) EUVD-Auditor/1.0"
|
|
}
|
|
|
|
|
|
def evaluar_parcheo_cvss(vector_str):
|
|
"""
|
|
Evalúa el vector CVSS v3.x y devuelve la calificación de Parcheo:
|
|
'Urgente', 'Diferible' o 'Descartable'.
|
|
|
|
Reglas aplicadas:
|
|
- Descartable: AV:L / AV:P o PR:H
|
|
- Urgente: AV:N + AC:L + PR:N + UI:N y además (C:H o I:H o A:H o S:C)
|
|
- Diferible: Resto de casos
|
|
"""
|
|
if not vector_str or vector_str == "N/A":
|
|
return "Diferible"
|
|
|
|
# Convertir el vector en un diccionario de componentes (ej: {'AV': 'N', 'AC': 'L', ...})
|
|
metrics = {}
|
|
parts = vector_str.split('/')
|
|
for part in parts:
|
|
if ':' in part:
|
|
k, v = part.split(':', 1)
|
|
metrics[k.upper()] = v.upper()
|
|
|
|
av = metrics.get('AV', '')
|
|
ac = metrics.get('AC', '')
|
|
pr = metrics.get('PR', '')
|
|
ui = metrics.get('UI', '')
|
|
s = metrics.get('S', '')
|
|
c = metrics.get('C', '')
|
|
i = metrics.get('I', '')
|
|
a = metrics.get('A', '')
|
|
|
|
# 1. Reglas de Descartable
|
|
if av in ['L', 'P'] or pr == 'H':
|
|
return "Descartable"
|
|
|
|
# 2. Reglas de Urgente
|
|
es_remoto_facil = (av == 'N' and ac == 'L' and pr == 'N' and ui == 'N')
|
|
impacto_alto = (c == 'H' or i == 'H' or a == 'H' or s == 'C')
|
|
|
|
if es_remoto_facil and impacto_alto:
|
|
return "Urgente"
|
|
|
|
# 3. Resto de escenarios -> Diferible
|
|
return "Diferible"
|
|
|
|
|
|
def calcular_parcheo_global(lista_parcheo_vulnerabilidades):
|
|
"""
|
|
Determina la calificación de Parcheo del producto según la máxima criticidad encontrada:
|
|
Si hay al menos una Urgente -> Urgente
|
|
Sino, si hay al menos una Diferible -> Diferible
|
|
Sino, si todas son Descartables -> Descartable
|
|
Sin vulnerabilidades aplicables -> Sin Acción
|
|
"""
|
|
if not lista_parcheo_vulnerabilidades:
|
|
return "Sin Acción"
|
|
|
|
if "Urgente" in lista_parcheo_vulnerabilidades:
|
|
return "Urgente"
|
|
elif "Diferible" in lista_parcheo_vulnerabilidades:
|
|
return "Diferible"
|
|
else:
|
|
return "Descartable"
|
|
|
|
|
|
def parse_single_version(v_str):
|
|
"""
|
|
Limpia y extrae la versión semántica limpia de una cadena.
|
|
Ejemplos: 'v1.8.0' -> '1.8.0', 'openjdk 1.8.0_20' -> '1.8.0'
|
|
"""
|
|
if not v_str:
|
|
return None
|
|
cleaned = re.sub(r'^(openjdk|v|version)\s*', '', v_str.strip(), flags=re.IGNORECASE)
|
|
match = re.search(r'\b\d+(\.\d+)+', cleaned)
|
|
if match:
|
|
try:
|
|
return Version(match.group(0))
|
|
except InvalidVersion:
|
|
return None
|
|
return None
|
|
|
|
|
|
def matches_product_version(target_version_str, product_version_str):
|
|
"""
|
|
Evalúa matemáticamente si target_version_str se ve afectada por product_version_str.
|
|
"""
|
|
if not product_version_str or product_version_str.lower() in ["n/a", "unknown", "*"]:
|
|
return True, "Versión indeterminada en la fuente (Revisar manualmente)"
|
|
|
|
t_ver = parse_single_version(target_version_str)
|
|
|
|
# Procesar si aparece como "parche" en ENISA
|
|
if product_version_str.lower().startswith("patch:"):
|
|
patch_version = product_version_str.split(":", 1)[1].strip()
|
|
|
|
target_version = parse_single_version(target_version_str)
|
|
fixed_version = parse_single_version(patch_version)
|
|
|
|
if target_version and fixed_version:
|
|
if target_version >= fixed_version:
|
|
return (
|
|
False,
|
|
f"Versión ya corregida mediante parche {fixed_version}"
|
|
)
|
|
else:
|
|
return (
|
|
True,
|
|
f"{target_version} anterior al parche {fixed_version}"
|
|
)
|
|
|
|
return (
|
|
False,
|
|
f"Versión marcada como corregida mediante parche {patch_version}"
|
|
)
|
|
|
|
if not t_ver:
|
|
if target_version_str.strip().lower() in product_version_str.lower():
|
|
return True, f"Coincidencia directa por cadena '{product_version_str}'"
|
|
return False, f"No coincide con '{product_version_str}'"
|
|
|
|
p_str = product_version_str.strip()
|
|
|
|
# 1. Manejo de rangos con palabra "to" (ej. "1.7.0 to 1.8.2")
|
|
if " to " in p_str.lower():
|
|
parts = re.split(r'\s+to\s+', p_str, flags=re.IGNORECASE)
|
|
if len(parts) == 2:
|
|
min_v = parse_single_version(parts[0])
|
|
max_v = parse_single_version(parts[1])
|
|
if min_v and max_v:
|
|
if min_v <= t_ver <= max_v:
|
|
return True, f"Afecta al rango {min_v} <= {t_ver} <= {max_v}"
|
|
else:
|
|
return False, f"Fuera del rango ({min_v} a {max_v})"
|
|
|
|
# 2. Extraer condiciones/operadores de la cadena product_version
|
|
tokens = re.findall(r'(<=|>=|<|>)?\s*(\d+(?:\.\d+)+[a-zA-Z0-9_\.-]*)', p_str)
|
|
|
|
if not tokens:
|
|
if t_ver.public in p_str:
|
|
return True, f"Coincidencia textual directa con '{product_version_str}'"
|
|
return False, f"No coincide con '{product_version_str}'"
|
|
|
|
all_conditions_met = True
|
|
reasons = []
|
|
|
|
for op, ver_str in tokens:
|
|
cond_ver = parse_single_version(ver_str)
|
|
if not cond_ver:
|
|
continue
|
|
|
|
if op == "<":
|
|
met = t_ver < cond_ver
|
|
reasons.append(f"{t_ver} < {cond_ver} ({'Sí' if met else 'No'})")
|
|
elif op == "<=":
|
|
met = t_ver <= cond_ver
|
|
reasons.append(f"{t_ver} <= {cond_ver} ({'Sí' if met else 'No'})")
|
|
elif op == ">":
|
|
met = t_ver > cond_ver
|
|
reasons.append(f"{t_ver} > {cond_ver} ({'Sí' if met else 'No'})")
|
|
elif op == ">=":
|
|
met = t_ver >= cond_ver
|
|
reasons.append(f"{t_ver} >= {cond_ver} ({'Sí' if met else 'No'})")
|
|
else:
|
|
met = (t_ver == cond_ver) or (t_ver.public.startswith(cond_ver.public))
|
|
reasons.append(f"Rama/Coincidencia {cond_ver} ({'Sí' if met else 'No'})")
|
|
|
|
if not met:
|
|
all_conditions_met = False
|
|
|
|
rule_msg = " | ".join(reasons)
|
|
if all_conditions_met:
|
|
return True, f"Cumple condiciones: {rule_msg}"
|
|
else:
|
|
return False, f"No cumple condiciones: {rule_msg}"
|
|
|
|
|
|
def fetch_fast_enisa_vulnerabilities(vendor, product, from_score=7.0, days_back=90):
|
|
vulnerabilities = []
|
|
from_date = (datetime.now(timezone.utc) - timedelta(days=days_back)).strftime("%Y-%m-%d")
|
|
|
|
params = {
|
|
"vendor": vendor,
|
|
"product": product,
|
|
"fromScore": from_score,
|
|
"fromDate": from_date,
|
|
"page": 0,
|
|
"size": 100
|
|
}
|
|
|
|
max_retries = 3
|
|
for attempt in range(max_retries):
|
|
try:
|
|
response = requests.get(BASE_URL, headers=HEADERS, params=params, timeout=15)
|
|
|
|
if response.status_code == 429:
|
|
wait_time = (attempt + 1) * 5
|
|
time.sleep(wait_time)
|
|
continue
|
|
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
items = data.get("items", data.get("content", [])) if isinstance(data, dict) else data
|
|
|
|
if not items and product:
|
|
params_fallback = {
|
|
"text": product,
|
|
"fromScore": from_score,
|
|
"fromDate": from_date,
|
|
"page": 0,
|
|
"size": 100
|
|
}
|
|
res_fb = requests.get(BASE_URL, headers=HEADERS, params=params_fallback, timeout=15)
|
|
if res_fb.status_code == 200:
|
|
items = res_fb.json().get("items", [])
|
|
|
|
vulnerabilities.extend(items)
|
|
break
|
|
|
|
except Exception as e:
|
|
if attempt == max_retries - 1:
|
|
print(f" Error final para {product}: {e}")
|
|
time.sleep(2)
|
|
|
|
return vulnerabilities
|
|
|
|
|
|
def generate_markdown_report(total_consolidados, total_analizadas, total_aplicables,
|
|
total_no_aplicables, total_revision, resumen_objetivos,
|
|
aplicables_severidad, output_file="informe_vulnerabilidades.md"):
|
|
"""
|
|
Genera el informe ejecutivo en Markdown compatible con Pandoc/XeLaTeX.
|
|
"""
|
|
now_iso = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
|
|
|
|
md_content = f"""# Informe Ejecutivo de Vulnerabilidades
|
|
|
|
**Generado en:** {now_iso} | **Fuentes:** ENISA
|
|
|
|
* **Informes consolidados:** {total_consolidados}
|
|
* **Vulnerabilidades analizadas:** {total_analizadas}
|
|
* **Aplicables:** {total_aplicables}
|
|
* **No aplicables:** {total_no_aplicables}
|
|
* **Revisión manual:** {total_revision}
|
|
|
|
---
|
|
|
|
## Resumen por objetivo
|
|
|
|
| Fuente | Producto | Versión | Analizadas | Aplicables | No aplicables | Revisión manual | Necesidad Parcheo |
|
|
|---|---|---|---|---|---|---|---|
|
|
"""
|
|
|
|
for obj in resumen_objetivos:
|
|
parcheo_val = obj.get('Parcheo', 'Sin Acción')
|
|
# Formato visual destacado según urgencia
|
|
if parcheo_val == "Urgente":
|
|
parcheo_str = "**URGENTE**"
|
|
elif parcheo_val == "Diferible":
|
|
parcheo_str = "Diferible"
|
|
elif parcheo_val == "Descartable":
|
|
parcheo_str = "Descartable"
|
|
else:
|
|
parcheo_str = "Sin Acción"
|
|
|
|
md_content += f"| {obj['Fuente']} | {obj['Producto']} | {obj['Versión']} | {obj['Analizadas']} | {obj['Aplicables']} | {obj['No aplicables']} | {obj['Revisión manual']} | {parcheo_str} |\n"
|
|
|
|
md_content += """
|
|
---
|
|
|
|
## Detalle de Vulnerabilidades Aplicables
|
|
"""
|
|
|
|
if not aplicables_severidad:
|
|
md_content += "\n*No se encontraron vulnerabilidades aplicables con los criterios seleccionados en el período indicado.*\n"
|
|
else:
|
|
aplicables_ordenadas = sorted(
|
|
aplicables_severidad,
|
|
key=lambda x: float(x["BaseScore"].split()[0]) if x.get("BaseScore") and x["BaseScore"].split()[0].replace('.', '', 1).isdigit() else 0.0,
|
|
reverse=True
|
|
)
|
|
|
|
for app in aplicables_ordenadas:
|
|
score_val = app['BaseScore'].split()[0] if app.get('BaseScore') else "0.0"
|
|
try:
|
|
score_num = float(score_val)
|
|
except ValueError:
|
|
score_num = 0.0
|
|
|
|
if score_num >= 9.0:
|
|
badge = "[**CRÍTICA**]"
|
|
elif score_num >= 7.0:
|
|
badge = "[**ALTA**]"
|
|
elif score_num >= 4.0:
|
|
badge = "[**MEDIA**]"
|
|
else:
|
|
badge = "[**BAJA**]"
|
|
|
|
just = app['Justificación'].strip()
|
|
score_vector = app.get('BaseScoreVector', 'N/A')
|
|
parcheo_ind = app.get('Parcheo', 'Diferible')
|
|
|
|
md_content += f"""
|
|
### {badge} - {app['ID']} - {app['CVE']}
|
|
|
|
* **Producto / Versión:** `{app['Producto']}` (v`{app['Versión']}`) | **Fuente:** {app['Fuente']}
|
|
* **BaseScore:** {app['BaseScore']} | **Parcheo Vulnerabilidad:** **{parcheo_ind}**
|
|
* **Vector CVSS:** `{score_vector}`
|
|
* **Regla versión:** {app['Regla versión']}
|
|
* **Justificación / Descripción:**
|
|
> {just}
|
|
|
|
---
|
|
"""
|
|
|
|
with open(output_file, "w", encoding="utf-8") as f:
|
|
f.write(md_content)
|
|
|
|
print(f" Informe Markdown generado exitosamente en '{output_file}'")
|
|
|
|
|
|
def process_targets(config_path="targets.yaml"):
|
|
try:
|
|
with open(config_path, "r", encoding="utf-8") as f:
|
|
config = yaml.safe_load(f)
|
|
except FileNotFoundError:
|
|
print(f" Archivo '{config_path}' no encontrado.")
|
|
return
|
|
|
|
resumen_objetivos = []
|
|
aplicables_severidad = []
|
|
|
|
total_consolidados = 0
|
|
total_analizadas = 0
|
|
total_aplicables = 0
|
|
total_no_aplicables = 0
|
|
total_revision = 0
|
|
|
|
print(" Iniciando análisis de precisión en ENISA EUVD...\n")
|
|
|
|
for target in config.get("targets", []):
|
|
vendor = target.get("vendor", "")
|
|
product = target.get("product", "")
|
|
version = target.get("version", "")
|
|
|
|
total_consolidados += 1
|
|
print(f" Analizando: {vendor}/{product} (v{version})")
|
|
|
|
items = fetch_fast_enisa_vulnerabilities(vendor=vendor, product=product, from_score=0.0, days_back=90)
|
|
|
|
num_analizadas = len(items)
|
|
num_aplicables = 0
|
|
num_no_aplicables = 0
|
|
num_revision = 0
|
|
parcheo_vulnerabilidades_target = []
|
|
|
|
for item in items:
|
|
score = item.get("baseScore", 0.0) or 0.0
|
|
score_ver = item.get("baseScoreVersion", "3.1")
|
|
score_vector = item.get("baseScoreVector", "N/A")
|
|
euvd_id = item.get("id", "N/A")
|
|
|
|
aliases = item.get("aliases", "")
|
|
aliases_list = aliases.strip().split("\n") if isinstance(aliases, str) else []
|
|
cve_id = aliases_list[0] if aliases_list and aliases_list[0] else euvd_id
|
|
|
|
desc = item.get("description", "Sin descripción disponible.")
|
|
desc_short = desc[:180].strip() + "..." if len(desc) > 180 else desc
|
|
|
|
enisa_products = item.get("enisaIdProduct", [])
|
|
is_affected = False
|
|
match_rule = "Sin información de versión en la API"
|
|
|
|
if enisa_products:
|
|
for prod_entry in enisa_products:
|
|
p_version = prod_entry.get("product_version", "")
|
|
affected, rule_msg = matches_product_version(version, p_version)
|
|
if affected:
|
|
is_affected = True
|
|
match_rule = f"`product_version`: \"{p_version}\" -> ({rule_msg})"
|
|
break
|
|
else:
|
|
is_affected = True
|
|
match_rule = "Evaluación por coincidencia general de producto"
|
|
|
|
if is_affected:
|
|
num_aplicables += 1
|
|
parcheo_vuln = evaluar_parcheo_cvss(score_vector)
|
|
parcheo_vulnerabilidades_target.append(parcheo_vuln)
|
|
|
|
aplicables_severidad.append({
|
|
"Fuente": "ENISA",
|
|
"ID": euvd_id,
|
|
"CVE": cve_id,
|
|
"Producto": product,
|
|
"Versión": version,
|
|
"BaseScore": f"{score} {score_ver}".strip(),
|
|
"BaseScoreVector": score_vector,
|
|
"Regla versión": match_rule,
|
|
"Justificación": desc_short,
|
|
"Parcheo": parcheo_vuln
|
|
})
|
|
else:
|
|
num_no_aplicables += 1
|
|
|
|
total_analizadas += num_analizadas
|
|
total_aplicables += num_aplicables
|
|
total_no_aplicables += num_no_aplicables
|
|
total_revision += num_revision
|
|
|
|
# Evaluar el parcheo acumulado para el producto
|
|
parcheo_global_target = calcular_parcheo_global(parcheo_vulnerabilidades_target)
|
|
|
|
resumen_objetivos.append({
|
|
"Fuente": "ENISA",
|
|
"Producto": product,
|
|
"Versión": version,
|
|
"Analizadas": num_analizadas,
|
|
"Aplicables": num_aplicables,
|
|
"No aplicables": num_no_aplicables,
|
|
"Revisión manual": num_revision,
|
|
"Parcheo": parcheo_global_target
|
|
})
|
|
|
|
print(f" Analizadas: {num_analizadas} | Aplicables: {num_aplicables} | Descartadas por versión: {num_no_aplicables} | Parcheo: {parcheo_global_target}\n")
|
|
time.sleep(0.5)
|
|
|
|
reporte_json = {
|
|
"generado_en": datetime.now(timezone.utc).isoformat(),
|
|
"fuentes": ["ENISA"],
|
|
"resumen_por_objetivo": resumen_objetivos,
|
|
"aplicables_por_severidad": aplicables_severidad
|
|
}
|
|
with open("reporte_vulnerabilidades.json", "w", encoding="utf-8") as f:
|
|
json.dump(reporte_json, f, indent=2, ensure_ascii=False)
|
|
|
|
generate_markdown_report(
|
|
total_consolidados=total_consolidados,
|
|
total_analizadas=total_analizadas,
|
|
total_aplicables=total_aplicables,
|
|
total_no_aplicables=total_no_aplicables,
|
|
total_revision=total_revision,
|
|
resumen_objetivos=resumen_objetivos,
|
|
aplicables_severidad=aplicables_severidad,
|
|
output_file="informe_vulnerabilidades.md"
|
|
)
|
|
|
|
print(" Análisis finalizado y archivo 'informe_vulnerabilidades.md' actualizado con éxito.")
|
|
|
|
if __name__ == "__main__":
|
|
process_targets()
|