Integración de Elastic Security | Axur

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.

URL del endpoint: https://YOUR_ELASTIC_CLUSTER/axur-feed-input/_doc?pipeline=axur-feed-input-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_splitted
Las 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_splitted
Consulta 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:
1. Privilegios de índice: añada axur-feed-* (o los índices específicos que creó) con write y create_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] 😊