Construcción de un Pipeline de Enriquecimiento de IOC con OpenCTI
Descripción general
OpenCTI es una plataforma de código abierto para gestionar inteligencia de ciberamenazas, construida sobre STIX 2.1 como modelo de datos nativo. Este skill cubre la construcción de un pipeline automatizado de enriquecimiento de IOC utilizando el ecosistema de conectores de OpenCTI para enriquecer indicadores con contexto de VirusTotal, Shodan, AbuseIPDB, GreyNoise y otras fuentes. El pipeline enriquece automáticamente los indicadores recién ingeridos, los correlaciona con actores de amenazas y campañas conocidos, y los puntúa para la priorización de los analistas.
Cuándo usarlo
- Al implementar o configurar capacidades de construcción de pipeline de enriquecimiento de IOC en tu entorno
- Al establecer controles de seguridad alineados con requisitos de cumplimiento normativo
- Al construir o mejorar la arquitectura de seguridad para este dominio
- Al realizar evaluaciones de seguridad que requieran esta implementación
Requisitos previos
- Docker y Docker Compose para el despliegue de OpenCTI
- Python 3.9+ con la biblioteca
pycti - Claves API para servicios de enriquecimiento: VirusTotal, Shodan, AbuseIPDB, GreyNoise
- Comprensión del modelo de datos STIX 2.1 y sus relaciones
- ElasticSearch o OpenSearch como backend de OpenCTI
- RabbitMQ o Redis para la mensajería de conectores
Conceptos clave
Arquitectura de OpenCTI
OpenCTI utiliza un frontend API GraphQL respaldado por ElasticSearch para almacenamiento y Redis/RabbitMQ para comunicación de conectores. Los datos se almacenan de forma nativa como objetos STIX 2.1 con relaciones. Los conectores se clasifican en: Importación Externa (ingesta de feeds), Importación Interna (análisis de archivos), Enriquecimiento Interno (adición de contexto) y Stream (exportación en tiempo real).
Modelo de conector de enriquecimiento
Los conectores de enriquecimiento interno se activan automáticamente cuando se crean nuevos observables o manualmente por los analistas. Cada conector recibe objetos STIX, consulta servicios externos y devuelve paquetes STIX 2.1 que aumentan el observable original con contexto adicional, etiquetas y relaciones.
Puntuación de confianza
OpenCTI utiliza una escala de confianza de 0 a 100 para los indicadores. Los conectores de enriquecimiento pueden actualizar las puntuaciones de confianza basándose en validación externa: tasas de detección de VirusTotal, datos de exposición de Shodan, recuentos de informes de AbuseIPDB y resultados de clasificación de GreyNoise.
Flujo de trabajo
Paso 1: Desplegar OpenCTI con Docker Compose
# docker-compose.yml (servicios clave)
version: '3'
services:
opencti:
image: opencti/platform:6.4.4
environment:
- APP__PORT=8080
- [email protected]
- APP__ADMIN__PASSWORD=ChangeMeNow
- APP__ADMIN__TOKEN=your-admin-token-uuid
- ELASTICSEARCH__URL=http://elasticsearch:9200
- MINIO__ENDPOINT=minio
- RABBITMQ__HOSTNAME=rabbitmq
ports:
- "8080:8080"
depends_on:
- elasticsearch
- minio
- rabbitmq
- redis
connector-virustotal:
image: opencti/connector-virustotal:6.4.4
environment:
- OPENCTI_URL=http://opencti:8080
- OPENCTI_TOKEN=your-admin-token-uuid
- CONNECTOR_ID=connector-virustotal-id
- CONNECTOR_NAME=VirusTotal
- CONNECTOR_SCOPE=StixFile,Artifact,IPv4-Addr,Domain-Name,Url
- CONNECTOR_AUTO=true
- VIRUSTOTAL_TOKEN=your-vt-api-key
- VIRUSTOTAL_MAX_TLP=TLP:AMBER
connector-shodan:
image: opencti/connector-shodan:6.4.4
environment:
- OPENCTI_URL=http://opencti:8080
- OPENCTI_TOKEN=your-admin-token-uuid
- CONNECTOR_ID=connector-shodan-id
- CONNECTOR_NAME=Shodan
- CONNECTOR_SCOPE=IPv4-Addr
- CONNECTOR_AUTO=true
- SHODAN_TOKEN=your-shodan-api-key
- SHODAN_MAX_TLP=TLP:AMBER
connector-abuseipdb:
image: opencti/connector-abuseipdb:6.4.4
environment:
- OPENCTI_URL=http://opencti:8080
- OPENCTI_TOKEN=your-admin-token-uuid
- CONNECTOR_ID=connector-abuseipdb-id
- CONNECTOR_NAME=AbuseIPDB
- CONNECTOR_SCOPE=IPv4-Addr
- CONNECTOR_AUTO=true
- ABUSEIPDB_API_KEY=your-abuseipdb-key
Paso 2: Construir un conector de enriquecimiento personalizado
import os
from pycti import OpenCTIConnectorHelper, get_config_variable
from stix2 import (
Bundle, Indicator, Note, Relationship,
IPv4Address, DomainName
)
import requests
class CustomEnrichmentConnector:
def __init__(self):
config = {
"opencti": {
"url": os.environ.get("OPENCTI_URL"),
"token": os.environ.get("OPENCTI_TOKEN"),
},
"connector": {
"id": os.environ.get("CONNECTOR_ID"),
"name": "CustomEnrichment",
"scope": "IPv4-Addr,Domain-Name,Url",
"auto": True,
"type": "INTERNAL_ENRICHMENT",
},
}
self.helper = OpenCTIConnectorHelper(config)
self.helper.listen(self._process_message)
def _process_message(self, data):
entity_id = data["entity_id"]
stix_object = self.helper.api.stix_cyber_observable.read(id=entity_id)
if not stix_object:
return "Observable no encontrado"
observable_type = stix_object["entity_type"]
observable_value = stix_object.get("value", "")
enrichment_results = []
if observable_type == "IPv4-Addr":
enrichment_results = self._enrich_ip(observable_value, entity_id)
elif observable_type == "Domain-Name":
enrichment_results = self._enrich_domain(observable_value, entity_id)
if enrichment_results:
bundle = Bundle(objects=enrichment_results, allow_custom=True)
self.helper.send_stix2_bundle(bundle.serialize())
return "Enriquecimiento completado"
def _enrich_ip(self, ip_address, entity_id):
"""Enriquecer dirección IP con contexto de GreyNoise, AbuseIPDB."""
objects = []
# API comunitaria de GreyNoise
try:
gn_response = requests.get(
f"https://api.greynoise.io/v3/community/{ip_address}",
headers={"key": os.environ.get("GREYNOISE_API_KEY")},
timeout=30,
)
if gn_response.status_code == 200:
gn_data = gn_response.json()
classification = gn_data.get("classification", "unknown")
noise = gn_data.get("noise", False)
riot = gn_data.get("riot", False)
note_content = (
f"## Enriquecimiento GreyNoise\n"
f"- Clasificación: {classification}\n"
f"- Ruido en Internet: {noise}\n"
f"- RIOT (Servicio Benigno): {riot}\n"
f"- Nombre: {gn_data.get('name', 'N/A')}\n"
f"- Última vez visto: {gn_data.get('last_seen', 'N/A')}"
)
note = Note(
content=note_content,
object_refs=[entity_id],
abstract=f"GreyNoise: {classification}",
allow_custom=True,
)
objects.append(note)
# Añadir etiquetas según clasificación
if classification == "malicious":
self.helper.api.stix_cyber_observable.add_label(
id=entity_id, label_name="greynoise:malicioso"
)
elif riot:
self.helper.api.stix_cyber_observable.add_label(
id=entity_id, label_name="greynoise:servicio-benigno"
)
except Exception as e:
self.helper.log_error(f"Enriquecimiento GreyNoise falló: {e}")
return objects
def _enrich_domain(self, domain, entity_id):
"""Enriquecer dominio con contexto WHOIS y DNS."""
objects = []
try:
# Usar API de SecurityTrails para enriquecimiento de dominio
st_response = requests.get(
f"https://api.securitytrails.com/v1/domain/{domain}",
headers={"APIKEY": os.environ.get("SECURITYTRAILS_API_KEY")},
timeout=30,
)
if st_response.status_code == 200:
st_data = st_response.json()
current_dns = st_data.get("current_dns", {})
a_records = [
r.get("ip") for r in current_dns.get("a", {}).get("values", [])
]
note_content = (
f"## Enriquecimiento SecurityTrails\n"
f"- Registros A: {', '.join(a_records)}\n"
f"- Ranking Alexa: {st_data.get('alexa_rank', 'N/A')}\n"
f"- Nombre de host: {st_data.get('hostname', 'N/A')}"
)
note = Note(
content=note_content,
object_refs=[entity_id],
abstract=f"SecurityTrails: {domain}",
allow_custom=True,
)
objects.append(note)
except Exception as e:
self.helper.log_error(f"Enriquecimiento SecurityTrails falló: {e}")
return objects
if __name__ == "__main__":
connector = CustomEnrichmentConnector()