Esta guía explica cómo ingerir feeds de Axur Platform en Elastic Security utilizando la entrega Push (Webhook 2.0) sin infraestructura externa. Utilizaremos solo características nativas de Elastic Stack: índices, pipelines de ingesta y Watcher para dividir los datos en documentos a nivel de ítem.
Importante: Este es solo un posible enfoque de integración. El diseño del índice y las transformaciones de datos pueden personalizarse para que coincidan con su entorno y preferencias. El enfoque a continuación fue diseñado para ingerir datos directamente del Axur Feed y dividirlos en índices específicos por entidad (por ejemplo, tickets y credenciales).
Nota: Este tutorial asume que ya tiene acceso a Axur Platform y la API Key requerida.
Requisitos previos
Elastic Stack (Elasticsearch + Kibana) versión 8.x
Acceso de administrador a Kibana
Acceso a la red desde la infraestructura de Axur a su clúster de Elastic
Elastic API Key con permisos de escritura para los índices de destino
Consideraciones de costo y facturación
Importante: Esta integración utiliza la API REST y Watcher integrados de Elastic. No hay facturación por parte de Axur para esto.
Principales impulsores de costos:
Licencias/características de Elastic utilizadas (Watcher requiere una licencia apropiada)
Almacenamiento para eventos ingeridos e índices derivados
Egreso/ingreso de red según su alojamiento
Consejos para controlar el costo:
Comience con filtros de feed más estrictos para limitar el volumen inicialmente
Aplique políticas de ILM a los índices creados aquí
Monitoree la ingesta y ajuste los filtros de feed según sea necesario
Sugerencia: Un glosario de términos de Elastic utilizados en este tutorial está disponible al final del documento.
1) Crear la Axur API Key
Cree una API Key. Tenga el valor a mano para los pasos posteriores.
Generar una API Key
En Axur Platform, vaya a API Keys configuration.
Cree una nueva API key y copie el valor de la API Key de forma segura.
Nota: El feed solo devuelve datos a los que el usuario de la API key tiene permiso de acceso.
Capturas de pantalla:
A partir de este punto (Pasos 2–6), toda la configuración se realiza en la consola de Elastic/Kibana.
2) En Elastic: Crear el índice de entrada
Cree un índice para recibir datos de Axur directamente a través de HTTP POST. Utilice flattened para manejar campos dinámicos.
PUT axur-feed-input
{
"mappings": {
"dynamic": true,
"dynamic_templates": [
{
"collectionData_flattened": {
"path_match": "collectionData.*",
"mapping": { "type": "flattened" }
}
}
]
}
}
Capturas de pantalla:
3) En Elastic: Crear el pipeline de ingesta
Agregue una marca de tiempo de ingesta a cada documento recibido en el índice de entrada.
PUT _ingest/pipeline/axur-feed-input-pipeline
{
"processors": [
{ "set": { "field": "ingested_at", "value": "{{_ingest.timestamp}}" } }
]
}
Importante: La Axur Feed Push URL debe incluir ?pipeline=axur-feed-input-pipeline para adjuntar este pipeline.
Capturas de pantalla:
4) En Elastic: Crear los índices de salida
Estos índices almacenarán documentos a nivel de ítem, ya divididos por tipo.
Índice de tickets:
PUT axur-feed-ticket_splitted
{
"mappings": {
"dynamic": true,
"properties": {
"item": { "type": "flattened" },
"feedData": { "type": "flattened" }
}
}
}
Índice de credenciales (detecciones):
PUT axur-feed-credential_splitted
{
"mappings": {
"dynamic": true,
"properties": {
"item": { "type": "flattened" },
"feedData": { "type": "flattened" }
}
}
}
Capturas de pantalla:
5) En Elastic: Crear el índice de punto de control
Almacene la última marca de tiempo procesada para evitar el reprocesamiento.
PUT axur-feed-checkpoint
{
"mappings": {
"properties": {
"last_ts": { "type": "date" }
}
}
}
Capturas de pantalla:
6) En Elastic: Crear el Watcher para dividir ítems
Este Watcher procesa solo los nuevos documentos de entrada e indexa los ítems divididos en los índices de salida, actualizando el punto de control.
PUT _watcher/watch/axur_split_batch
{
"trigger": { "schedule": { "interval": "1m" } },
"input": {
"chain": {
"inputs": [
{
"cp": {
"search": {
"request": {
"indices": ["axur-feed-checkpoint"],
"body": {
"size": 1,
"query": { "ids": { "values": ["checkpoint"] } }
}
}
}
}
},
{
"src": {
"search": {
"request": {
"indices": ["axur-feed-input"],
"body": {
"size": 200,
"query": {
"range": {
"ingested_at": {
"gt": "{{#ctx.payload.cp.hits.hits}}{{ctx.payload.cp.hits.hits.0._source.last_ts}}{{/ctx.payload.cp.hits.hits}}{{^ctx.payload.cp.hits.hits}}1970-01-01T00:00:00Z{{/ctx.payload.cp.hits.hits}}"
}
}
},
"sort": [{ "ingested_at": "asc" }]
}
}
}
}
}
]
}
},
"condition": {
"script": {
"lang": "painless",
"source": "def h = ctx?.payload?.src?.hits; if (h == null) return false; def t = h.total; if (t == null) return false; if (t instanceof Map) return (t.value != null && t.value > 0); return (t > 0);"
}
},
"transform": {
"script": {
"lang": "painless",
"source": """
def tics = new ArrayList();
def creds = new ArrayList();
def maxTs = null;
for (def hit : ctx.payload.src.hits.hits) {
def src = hit._source;
if (src == null) continue;
def ts = src.containsKey('ingested_at') ? src.ingested_at : null;
if (ts != null) {
if (maxTs == null) { maxTs = ts; }
else if (java.time.ZonedDateTime.parse(ts).isAfter(java.time.ZonedDateTime.parse(maxTs))) { maxTs = ts; }
}
def feed = src.containsKey('feedData') ? src.feedData : null;
if (!src.containsKey('collectionData')) continue;
def cd = src.collectionData;
if (cd != null && cd.containsKey('tickets') && cd.tickets != null) {
for (def it : cd.tickets) {
if (it == null) continue;
tics.add(['feedData': feed, 'item': it]);
}
}
if (cd != null && cd.containsKey('detections') && cd.detections != null) {
for (def it2 : cd.detections) {
if (it2 == null) continue;
creds.add(['feedData': feed, 'item': it2]);
}
}
}
return ['tics': tics, 'creds': creds, 'next_checkpoint': maxTs];
"""
}
},
"actions": {
"index_tics": {
"foreach": "ctx.payload.tics",
"max_iterations": 500,
"transform": { "script": { "lang": "painless", "source": "return ctx.payload._value;" } },
"index": { "index": "axur-feed-ticket_splitted", "execution_time_field": "indexed_at", "refresh": "wait_for" }
},
"index_creds": {
"foreach": "ctx.payload.creds",
"max_iterations": 500,
"transform": { "script": { "lang": "painless", "source": "return ctx.payload._value;" } },
"index": { "index": "axur-feed-credential_splitted", "execution_time_field": "indexed_at", "refresh": "wait_for" }
},
"update_checkpoint": {
"condition": { "script": { "lang": "painless", "source": "return ctx.payload.next_checkpoint != null;" } },
"transform": { "script": { "lang": "painless", "source": "return ['last_ts': ctx.payload.next_checkpoint];" } },
"index": { "index": "axur-feed-checkpoint", "doc_id": "checkpoint", "execution_time_field": "updated_at", "refresh": "wait_for" }
}
}
}
Capturas de pantalla:
7) Configurar el endpoint del Axur PUSH Feed
Apunte el Axur Feed a su índice de entrada de Elastic y adjunte el pipeline.
Encabezados a añadir:
Authorization: ApiKey <YOUR_ENCODED_ELASTIC_API_KEY>(copie el formato codificado de la UI de Elastic API Keys)
Secret Key: configure un secreto seguro en el feed; Axur firma las solicitudes con HMAC. Opcionalmente, verifique en el lado de Elastic.
Capturas de pantalla:
Validación
Después de la configuración, envíe un documento de prueba al índice de entrada (Kibana → Dev Tools):
POST axur-feed-input/_doc?pipeline=axur-feed-input-pipeline
{
"feedData": { "feedType": "ticket", "customerKey": "TEST" },
"collectionData": {
"tickets": [ { "ticket": { "id": "TCK123", "type": "leak" } } ],
"detections": [ { "credential": { "user": "alice" } } ]
}
}
Espere ~1 minuto y verifique:
GET axur-feed-ticket_splitted/_search
GET axur-feed-credential_splitted/_search
GET axur-feed-checkpoint/_doc/checkpoint
Debería ver documentos a nivel de ítem en cada índice de salida y una last_ts actualizada en el punto de control. Las ejecuciones subsiguientes no deberían reprocesar los mismos documentos de entrada.
Capturas de pantalla:
Resolución de problemas
No hay documentos en los índices de salida: asegúrese de que Watcher esté activo y no esté fallando; revise
Watcher > History.Errores de indexación 401/403: valide los permisos de Elastic API Key (
write,create_index).Duplicados: confirme que el documento de punto de control se actualice; valide la sincronización horaria del clúster.
Nada ingerido: verifique la accesibilidad de la Push URL desde Axur y el parámetro de pipeline en la ruta.
Ha terminado: dónde encontrar datos y un ejemplo de alerta
Los tickets se indexan en:
axur-feed-ticket_splittedLas credenciales (detecciones) se indexan en:
axur-feed-credential_splitted
Consultas de ejemplo (Kibana → Discover):
GET axur-feed-ticket_splitted/_search
{
"query": { "term": { "item.detection.status": "open" } },
"size": 20,
"sort": [{ "indexed_at": { "order": "desc" } }]
}
GET axur-feed-credential_splitted/_search
{
"query": { "exists": { "field": "item.credential.user" } },
"size": 20,
"sort": [{ "indexed_at": { "order": "desc" } }]
}
Cree una regla de detección (Elastic Security → Rules → Create rule → Custom query):
Patrón de índice:
axur-feed-credential_splittedConsulta personalizada (KQL):
item.credential.user:* and item.access.domain:"example.com"
Programación: Cada 5 minutos
Gravedad: Media
Acciones: añada sus notificaciones (correo electrónico, Slack, etc.)
Cómo generar la Elastic API Key
Cree una API Key en Elastic para autenticar las solicitudes Push de Axur.
En Kibana:
Vaya a
Stack Management→Security→API Keys→Create API key.Nombre:
axur-feed-ingest.Privilegios:
Privilegios de índice: añada
axur-feed-*(o los índices específicos que creó) conwriteycreate_index.
Cree y copie la clave en formato
Encoded. Guárdela de forma segura.
Capturas de pantalla:
Glosario
Elasticsearch: Motor de búsqueda y análisis distribuido de Elastic Stack.
Kibana: Interfaz de usuario web para gestionar y visualizar datos de Elasticsearch.
Índice: Un espacio de nombres lógico para documentos en Elasticsearch.
Pipeline de ingesta: Procesadores ejecutados antes de indexar un documento.
Watcher: Característica de alerta/automatización utilizada aquí para implementar la división periódica.
ILM (Index Lifecycle Management): Política para gestionar la retención de datos y el rendimiento.
API Key: Mecanismo de autenticación para Elasticsearch.
Push (Webhook 2.0): Modo de entrega de Axur que envía eventos directamente a su endpoint.
Si tienes cualquier duda, puedes contactarnos en [email protected] 😊