Automatización del Enriquecimiento de IOC
Cuándo Usar
Usa esta habilidad cuando:
- Estés construyendo un playbook SOAR que enriquezca automáticamente las alertas SIEM con contexto de inteligencia de amenazas antes de enrutarlas a los analistas.
- Estés creando un pipeline Python para el enriquecimiento masivo de IOCs a partir de envíos de correos electrónicos de phishing.
- Quieras reducir el tiempo medio de triaje (MTTT) del analista precargando el contexto de la alerta con datos de VT, Shodan y MISP.
No uses esta habilidad para decisiones de bloqueo completamente automatizadas sin revisión humana; la automatización del enriquecimiento debe informar las decisiones, no ejecutar bloqueos de forma autónoma para acciones de alto impacto.
Prerrequisitos
- Plataforma SOAR (Cortex XSOAR, Splunk SOAR, Tines o n8n) o entorno Python 3.9+
- Claves API: VirusTotal, AbuseIPDB, Shodan, y al menos un TIP (MISP o OpenCTI)
- Endpoint de integración SIEM para la ingesta de alertas
- Presupuestos de límite de tasa documentados por API (VT: 4/min gratuita, 500/min empresarial)
Flujo de Trabajo
Paso 1: Diseñar la Arquitectura del Pipeline de Enriquecimiento
Define el flujo de enriquecimiento para cada tipo de IOC:
Alerta SIEM → Extraer IOCs → Clasificar Tipo → Enrutar a funciones de enriquecimiento
Dirección IP → AbuseIPDB + Shodan + VirusTotal IP + MISP
Dominio → VirusTotal Dominio + PassiveTotal + Shodan + MISP
URL → URLScan.io + VirusTotal URL + Google Safe Browsing
Hash de archivo → VirusTotal Archivos + MalwareBazaar + MISP
→ Agregar resultados → Calcular puntuación de confianza → Actualizar alerta → Notificar al analista
Paso 2: Implementar Funciones Python de Enriquecimiento
import requests
import time
from dataclasses import dataclass, field
from typing import Optional
RETARDO_LIMITE_TASA = 0.25 # 4 solicitudes/segundo para VT nivel gratuito
@dataclass
class ResultadoEnriquecimiento:
valor_ioc: str
tipo_ioc: str
vt_malicioso: int = 0
vt_total: int = 0
confianza_abuso: int = 0
puertos_shodan: list = field(default_factory=list)
eventos_misp: list = field(default_factory=list)
puntuacion_confianza: int = 0
def enriquecer_ip(ip: str, clave_vt: str, clave_abuso: str, clave_shodan: str) -> ResultadoEnriquecimiento:
resultado = ResultadoEnriquecimiento(ip, "ip")
# Consulta IP en VirusTotal
resp_vt = requests.get(
f"https://www.virustotal.com/api/v3/ip_addresses/{ip}",
headers={"x-apikey": clave_vt}
)
if resp_vt.status_code == 200:
stats = resp_vt.json()["data"]["attributes"]["last_analysis_stats"]
resultado.vt_malicioso = stats.get("malicious", 0)
resultado.vt_total = sum(stats.values())
time.sleep(RETARDO_LIMITE_TASA)
# AbuseIPDB
resp_abuso = requests.get(
"https://api.abuseipdb.com/api/v2/check",
headers={"Key": clave_abuso, "Accept": "application/json"},
params={"ipAddress": ip, "maxAgeInDays": 90}
)
if resp_abuso.status_code == 200:
resultado.confianza_abuso = resp_abuso.json()["data"]["abuseConfidenceScore"]
# Calcular puntuación de confianza compuesta
resultado.puntuacion_confianza = min(
(resultado.vt_malicioso / max(resultado.vt_total, 1)) * 60 +
(resultado.confianza_abuso / 100) * 40, 100
)
return resultado
def enriquecer_hash(sha256: str, clave_vt: str) -> ResultadoEnriquecimiento:
resultado = ResultadoEnriquecimiento(sha256, "sha256")
resp_vt = requests.get(
f"https://www.virustotal.com/api/v3/files/{sha256}",
headers={"x-apikey": clave_vt}
)
if resp_vt.status_code == 200:
stats = resp_vt.json()["data"]["attributes"]["last_analysis_stats"]
resultado.vt_malicioso = stats.get("malicious", 0)
resultado.vt_total = sum(stats.values())
resultado.puntuacion_confianza = int((resultado.vt_malicioso / max(resultado.vt_total, 1)) * 100)
return resultado
Paso 3: Construir Playbook SOAR (Cortex XSOAR)
En Cortex XSOAR, crea un playbook de enriquecimiento:
- Disparador: Alerta creada en SIEM (mediante webhook o sondeo)
- Extraer IOCs: Usa la tarea "Extraer Indicadores" con patrones regex para IP, dominio, URL, hash
- Enriquecimiento en paralelo: Distribuye a múltiples tareas de enriquecimiento simultáneamente
- Enriquecimiento VT: Ejecuta comandos
!vt-file-scano!vt-ip-scan - Verificación AbuseIPDB: Ejecuta el comando
!abuseipdb-check-ip - Búsqueda MISP: Ejecuta
!misp-searchpara referenciar cruzadamente - Agregación de puntuación: Tarea de transformación Python que calcula la puntuación compuesta
- Enrutamiento condicional: Si puntuación ≥70 → Cola de Alta Prioridad; si 40–69 → Media; <40 → Cierre automático con nota
- Enriquecimiento de alerta: Escribe los resultados del enriquecimiento en el contexto de la alerta para la vista del analista
Paso 4: Manejar Límites de Tasa y Fallos
import time
from functools import wraps
def limitado_por_tasa(max_por_segundo):
intervalo_min = 1.0 / max_por_segundo
def decorador(func):
ultima_llamada = [0.0]
@wraps(func)
def envoltura(*args, **kwargs):
tiempo_transcurrido = time.time() - ultima_llamada[0]
espera = intervalo_min - tiempo_transcurrido
if espera > 0:
time.sleep(espera)
resultado = func(*args, **kwargs)
ultima_llamada[0] = time.time()
return resultado
return envoltura
return decorador
def reintentar_en_429(max_reintentos=3):
def decorador(func):
@wraps(func)
def envoltura(*args, **kwargs):
for intento in range(max_reintentos):
respuesta = func(*args, **kwargs)
if respuesta.status_code == 429:
esperar_por = int(respuesta.headers.get("Retry-After", 60))
time.sleep(esperar_por)
else:
return respuesta
return envoltura
return decorador
Paso 5: Métricas y Ajuste
Realiza seguimiento semanal del rendimiento del pipeline:
- Latencia de enriquecimiento: Objetivo <30 segundos desde que se dispara la alerta hasta la salida enriquecida
- Tasa de éxito de API: Objetivo >99% (identificar eventos de límite de tasa o interrupción)
- Tasa de verdaderos positivos: Realizar seguimiento de las anulaciones del analista sobre las puntuaciones de confianza automatizadas
- Costo: Realizar seguimiento del volumen de llamadas API contra el presupuesto (VT Enterprise: $X por 1M consultas)
Conceptos Clave
| Término | Definición |
|---|---|
| SOAR | Orquestación, Automatización y Respuesta de Seguridad — plataforma para automatizar flujos de trabajo de seguridad e integrar herramientas dispares |
| Playbook de Enriquecimiento | Secuencia de flujo de trabajo automatizada que añade inteligencia contextual a eventos de seguridad sin procesar |
| Límite de Tasa | Restricciones del proveedor de API en la frecuencia de solicitudes (p. ej., VT gratuito: 4 solicitudes/minuto); los pipelines deben respetar estos límites |
| Puntuación de Confianza Compuesta | Puntuación única que agrega señales de múltiples fuentes de enriquecimiento usando una fórmula ponderada |
| Patrón de Abanico | Ejecución paralela de múltiples consultas de enriquecimiento simultáneamente para minimizar la latencia total de enriquecimiento |
Herramientas y Sistemas
- Cortex XSOAR (Palo Alto): SOAR empresarial con más de 700 integraciones en marketplace, incluyendo VT, MISP, Shodan y AbuseIPDB
- Splunk SOAR (Phantom): Plataforma SOAR con playbooks basados en Python; integración nativa con Splunk SIEM
- Tines: Plataforma SOAR sin código con automatización basada en webhook; rentable para equipos pequeños
- TheHive + Cortex: Plataforma de enriquecimiento/IR de código abierto con enriquecimiento de observables mediante analizadores Cortex
Errores Comunes
- Bloquearse por la latencia del enriquecimiento: Si el enriquecimiento tarda más de 5 minutos, los analistas empiezan a trabajar con alertas no enriquecidas, lo que anula el propósito. Establece límites de tiempo de espera y proporciona resultados parciales.
- Sin caché: Consultar el mismo IOC 50 veces genera costos de API innecesarios. Almacena en caché los resultados del enriquecimiento durante 24 horas por defecto.
- Ignorar fallos de API en silencio: Las llamadas de enriquecimiento fallidas deben registrarse y activar lógica de respaldo, no producir silenciosamente resultados vacíos que aparezcan como IOCs limpios.
- Automatizar bloqueos basándose únicamente en la puntuación de enriquecimiento: Las puntuaciones compuestas contienen falsos positivos; se requiere confirmación humana para decisiones de bloqueo contra infraestructura compartida.