Transmitir registros a Microsoft Sentinel con Logstash y la API basada en DCR

El complemento de salida Logstash de Microsoft Sentinel admite transformaciones de canalización y configuración avanzada a través de reglas de recopilación de datos (DCR). El complemento reenvía registros de orígenes de datos externos a tablas personalizadas o estándar en Log Analytics o Microsoft Sentinel.

En este artículo, aprenderá a configurar el complemento Logstash para transmitir datos a Log Analytics o Microsoft Sentinel mediante DCR, con control total sobre el esquema de salida.

Con el complemento, puede:

• Controlar la configuración de los nombres y tipos de columna.
• Realice transformaciones en tiempo de ingesta, como filtrado o enriquecimiento.
• Ingiera registros personalizados en una tabla personalizada o ingiera un flujo de entrada de Syslog en la tabla syslog de Log Analytics.

La ingesta en tablas estándar solo se limita a las tablas estándar admitidas para la ingesta de registros personalizados.

Para más información sobre cómo trabajar con el motor de recopilación de datos de Logstash, consulte Introducción a Logstash.

Introducción a la arquitectura

El motor de Logstash se compone de tres componentes:

• Complementos de entrada: colección personalizada de datos de varios orígenes.
• Filtrar complementos: Manipulación y normalización de datos según los criterios especificados.
• Plugins de salida: envío personalizado de los datos recopilados y procesados a varios destinos.

El complemento envía datos con formato JSON al área de trabajo de Log Analytics mediante la API de ingesta de registros. Los datos se ingieren en registros personalizados o en una tabla estándar.

• Obtenga más información sobre la API de ingesta de registros.

Implementación del complemento de salida Microsoft Sentinel en Logstash

Para configurar el complemento, siga estos pasos:

• Revisa los requisitos previos del plugin Logstash
• Instalación del complemento
• Creación de un archivo de ejemplo
• Cree los recursos necesarios relacionados con DCR
• Configuración del archivo de configuración de Logstash
• Reinicio de Logstash
• Visualización de los registros entrantes en Microsoft Sentinel
• Supervisar los registros de auditoría del plugin de salida

Requisitos previos del complemento Logstash

• Instale una versión compatible de Logstash. El complemento admite las siguientes versiones de Logstash:

  • 7.0 - 7.17.13

  • 8.0 - 8.9 (estas versiones requieren una actualización de seguridad, según Logstash)

  • 8.11 - 8.15 (estas versiones requieren una actualización de seguridad, según Logstash)

  • 8.19.2 (esta versión requiere una actualización de seguridad, según Logstash)

  • 9.0.8 (esta versión requiere una actualización de seguridad, según Logstash)

  • 9.1.10 (esta versión requiere una actualización de seguridad, según Logstash)

  • 9.2.4 - 9.2.5 (estas versiones requieren una actualización de seguridad, según Logstash)

  • 9.3.3

  • 9.4.0

    Nota:

    Si usa Logstash 8, se recomienda deshabilitar ECS en la canalización.

• Compruebe que tiene un área de trabajo de Log Analytics con al menos derechos de colaborador.

• Compruebe que tiene permisos para crear objetos DCR en el área de trabajo.

Instalación del complemento

El complemento de salida Microsoft Sentinel está disponible en la colección Logstash en RubyGems.

• Siga las instrucciones del documento Logstash Working with plugins (Trabajo con complementos de Logstash) para instalar el complemento microsoft-sentinel-log-analytics-logstash-output-plugin . Para instalar en una instalación de Logstash existente, ejecute el siguiente comando:

  logstash-plugin install microsoft-sentinel-log-analytics-logstash-output-plugin
  

• Si el sistema Logstash no tiene acceso a Internet, siga las instrucciones del documento Administración de complementos sin conexión de Logstash para preparar y usar un paquete de complementos sin conexión. (Esto requiere la creación de otro sistema Logstash con acceso a Internet).

Creación de un archivo de ejemplo

En esta sección, creará un archivo de ejemplo en uno de estos escenarios:

• Creación de un archivo de ejemplo para registros personalizados
• Creación de un archivo de ejemplo para ingerir registros en la tabla syslog

Creación de un archivo de ejemplo para registros personalizados

En este escenario, configurará el complemento de entrada Logstash para enviar eventos a Microsoft Sentinel. En este ejemplo se usa el complemento de entrada del generador para simular eventos. Puede usar cualquier otro complemento de entrada.

En este ejemplo, el archivo de configuración de Logstash tiene el siguiente aspecto:

input {
      generator {
            lines => [
                 "This is a test log message"
            ]
           count => 10
      }
}

Para crear el archivo de ejemplo, siga estos pasos:

1. Copie la configuración del complemento de salida siguiente en el archivo de configuración de Logstash.

   output {
       microsoft-sentinel-log-analytics-logstash-output-plugin {
         create_sample_file => true
         sample_file_path => "<enter the path to the file in which the sample data will be written>" #for example: "c:\\temp" (for windows) or "/tmp" for Linux. 
       }
   }
   

2. Asegúrese de que la ruta de acceso del archivo a la que se hace referencia ya existe y, a continuación, inicie Logstash.

   El complemento escribe diez registros en un archivo de ejemplo denominado sampleFile<epoch seconds>.json en la ruta de acceso configurada una vez que hay 10 eventos para muestrear o cuando el proceso de Logstash se cierra correctamente. Por ejemplo: c:\temp\sampleFile1648453501.json. Esta es parte de un archivo de ejemplo que el complemento crea:

   [
           {
               "host": "logstashMachine",
               "sequence": 0,
               "message": "This is a test log message",
               "ls_timestamp": "2022-03-28T17:45:01.690Z",
               "ls_version": "1"
           },
           {
               "host": "logstashMachine",
               "sequence": 1
       ...
   
       ]    
   

   El complemento agrega automáticamente estas propiedades a cada registro:

   • ls_timestamp: la hora en que se recibe el registro desde el complemento de entrada
   • ls_version: la versión de canalización de Logstash.

   Puede quitar estos campos al crear la DCR.

Creación de un archivo de ejemplo para ingerir registros en la tabla syslog

En este escenario, configurará el complemento de entrada Logstash para enviar eventos de Syslog a Microsoft Sentinel.

1. Si aún no tiene mensajes de Syslog reenviados a la máquina de Logstash, puede usar el comando logger para generar mensajes. Por ejemplo (para Linux):

   logger -p local4.warn --rfc3164 --tcp -t CEF "0|Microsoft|Device|cef-test|example|data|1|here is some more data for the example" -P 514 -d -n 127.0.0.1
   

   Este es un ejemplo del complemento de entrada Logstash:

   input {
        syslog {
            port => 514
       }
   }
   

2. Copie la configuración del complemento de salida siguiente en el archivo de configuración de Logstash.

   output {
       microsoft-sentinel-log-analytics-logstash-output-plugin {
         create_sample_file => true
         sample_file_path => "<enter the path to the file in which the sample data will be written>" #for example: "c:\\temp" (for windows) or "/tmp" for Linux. 
       }
   }
   

3. Asegúrese de que la ruta de acceso del archivo ya existe y, a continuación, inicie Logstash.

   El complemento escribe diez registros en un archivo de ejemplo denominado sampleFile<epoch seconds>.json en la ruta de acceso configurada una vez que hay 10 eventos para muestrear o cuando el proceso de Logstash se cierra correctamente. Por ejemplo: c:\temp\sampleFile1648453501.json. Esta es parte de un archivo de ejemplo que el complemento crea:

   [
           {
               "logsource": "logstashMachine",
               "facility": 20,
               "severity_label": "Warning",
               "severity": 4,
               "timestamp": "Apr  7 08:26:04",
               "program": "CEF:",
               "host": "127.0.0.1",
               "facility_label": "local4",
               "priority": 164,
               "message": "0|Microsoft|Device|cef-test|example|data|1|here is some more data for the example",
               "ls_timestamp": "2022-04-07T08:26:04.000Z",
               "ls_version": "1"
           }
   ]    
   
   

   El complemento agrega automáticamente estas propiedades a cada registro:

   • ls_timestamp: la hora en que se recibe el registro desde el complemento de entrada
   • ls_version: la versión de canalización de Logstash.

   Puede quitar estos campos al crear la DCR.

Creación de los recursos de DCR necesarios

Para configurar el complemento Logstash basado en DCR Microsoft Sentinel, cree primero los recursos relacionados con DCR.

En esta sección, crea los recursos que usará con su DCR, en uno de estos escenarios:

• Crear recursos DCR para la ingesta en una tabla personalizada
• Cree recursos DCR para la ingesta en una tabla estándar

Crear recursos DCR para la ingesta en una tabla personalizada

Para ingerir los datos en una tabla personalizada, siga estos pasos (en función del tutorial Enviar datos a Azure Supervisar registros mediante la API REST (Azure Portal):

1. Revise los requisitos previos.

2. Configure la aplicación.

3. Agregue una tabla de registro personalizada.

4. Analice y filtre los datos de ejemplo mediante el archivo de ejemplo que creó en la sección anterior.

5. Recopile información del DCR.

6. Asigne permisos al DCR.

   Omita el paso Enviar datos de ejemplo.

Si encuentras algún problema, consulta los pasos de solución de problemas de la API de Ingestión de Logs.

Cree recursos DCR para la ingesta en una tabla estándar

Para ingerir datos en una tabla estándar, como Syslog o CommonSecurityLog, se usa un proceso basado en el tutorial Envío de datos a los registros de Azure Monitor mediante la API REST (plantillas de Azure Resource Manager). Aunque en el tutorial se explica cómo ingerir datos en una tabla personalizada, puede ajustar fácilmente el proceso para ingerir datos en una tabla estándar. Los pasos siguientes indican los cambios pertinentes en los pasos.

1. Revise los requisitos previos.

2. Recopilar detalles del área de trabajo.

3. Configurar una aplicación.

   Omita el paso Crear nueva tabla en el área de trabajo de Log Analytics. Este paso no es relevante al ingerir datos en una tabla estándar, ya que la tabla ya está definida en Log Analytics.

4. Cree el DCR. En este paso:

   • Proporciona el archivo de ejemplo que creaste en Crear un archivo de ejemplo.
   • Use el archivo de ejemplo que creó para definir la streamDeclarations propiedad . Cada uno de los campos del archivo de ejemplo debe tener una columna correspondiente con el mismo nombre y el tipo adecuado (vea el ejemplo siguiente).
   • Configure el valor de la outputStream propiedad con el nombre de la tabla estándar en lugar de la tabla personalizada. A diferencia de las tablas personalizadas, los nombres de tabla estándar no tienen el _CL sufijo .
   • El prefijo del nombre de tabla debe ser Microsoft- en lugar de Custom-. En este ejemplo, el valor de la outputStream propiedad es Microsoft-Syslog.

5. Asignar permisos a una DCR.

   Omita el paso Enviar datos de ejemplo.

Si encuentras algún problema, consulta los pasos de solución de problemas de la API de Ingestión de Logs.

Ejemplo: DCR que ingiere datos en la tabla syslog

Tenga en cuenta estos puntos:

• Los streamDeclarations nombres y tipos de columna deben ser los mismos que los campos de archivo de ejemplo, pero no es necesario especificar todos ellos. Por ejemplo, en la DCR siguiente, los campos PRI, type y ls_version se omiten de la columna streamDeclarations.
• La propiedad dataflows transforma la entrada al formato de tabla Syslog y establece outputStream en Microsoft-Syslog.

{
  "$schema": "https://schema.management.azure.com/schemas/2019-04-01/deploymentTemplate.json#",
  "contentVersion": "1.0.0.0",
  "parameters": {
    "dataCollectionRuleName": {
      "type": "String",
      "metadata": {
        "description": "Specifies the name of the Data Collection Rule to create."
      }
    },
    "location": {
      "defaultValue": "[resourceGroup().location]",
      "type": "String",
      "metadata": {
        "description": "Specifies the location in which to create the Data Collection Rule."
      }
    },
    "workspaceResourceId": {
      "type": "String",
      "metadata": {
        "description": "Specifies the Azure resource ID of the Log Analytics workspace to use."
      }
    }
  },
  "resources": [
    {
      "type": "Microsoft.Insights/dataCollectionRules",
      "apiVersion": "2021-09-01-preview",
      "name": "[parameters('dataCollectionRuleName')]",
      "location": "[parameters('location')]",
      "properties": {
        "streamDeclarations": {
          "Custom-SyslogStream": {
            "columns": [
              { "name": "ls_timestamp", "type": "datetime" },
              { "name": "timestamp", "type": "datetime" },
              { "name": "message", "type": "string" },
              { "name": "facility_label", "type": "string" },
              { "name": "severity_label", "type": "string" },
              { "name": "host", "type": "string" },
              { "name": "logsource", "type": "string" }
            ]
          }
        },
        "destinations": {
          "logAnalytics": [
            {
              "workspaceResourceId": "[parameters('workspaceResourceId')]",
              "name": "clv2ws1"
            }
          ]
        },
        "dataFlows": [
          {
            "streams": ["Custom-SyslogStream"],
            "destinations": ["clv2ws1"],
            "transformKql": "source | project TimeGenerated = ls_timestamp, EventTime = todatetime(timestamp), Computer = logsource, HostName = logsource, HostIP = host, SyslogMessage = message, Facility = facility_label, SeverityLevel = severity_label",
            "outputStream": "Microsoft-Syslog"
          }
        ]
      }
    }
  ],
  "outputs": {
    "dataCollectionRuleId": {
      "type": "String",
      "value": "[resourceId('Microsoft.Insights/dataCollectionRules', parameters('dataCollectionRuleName'))]"
    }
  }
}

Configuración del archivo de configuración de Logstash

El complemento admite dos métodos de autenticación: entidad de servicio (credenciales de cliente) e identidad administrada (sin contraseña). Elija el método que se adapte a su entorno.

Autenticación de entidad de servicio

Para configurar el archivo de configuración de Logstash para que ingiera los registros en una tabla personalizada mediante autenticación de entidad de servicio, recupera los siguientes valores: client_id, client_secret, tenant_id, data_collection_endpoint, dcr_id y stream_name.

Después de recuperar los valores necesarios:

1. Reemplace la sección de salida del archivo de configuración de Logstash que creó en el paso anterior por el ejemplo siguiente.
2. Reemplace las cadenas de marcador de posición del ejemplo siguiente por los valores recuperados.
3. Asegúrese de cambiar el create_sample_file atributo a false.

Ejemplo: Configuración del plugin de salida del service principal

output {
    microsoft-sentinel-log-analytics-logstash-output-plugin {
      client_id => "<enter your client_id value here>"
      client_secret => "<enter your client_secret value here>"
      tenant_id => "<enter your tenant id here>"
      data_collection_endpoint => "<enter your logsIngestion URI here>"
      dcr_id => "<enter your DCR immutableId here>"
      stream_name => "<enter your stream name here>"
      create_sample_file=> false
      sample_file_path => "c:\\temp"
    }
}

Autenticación de identidad administrada (sin contraseña)

Cuando no proporcionas las credenciales principales del servicio (client_id, , y tenant_id), el plugin se autentica usando DefaultAzureCredential desde el client_secretSDK de Azure. DefaultAzureCredential prueba una secuencia de métodos de autenticación y utiliza el primero que tiene éxito. En un entorno de servidor, se intentan los métodos relevantes en este orden:

1. Variables de entorno: Lee credenciales de variables de entorno como AZURE_CLIENT_ID, AZURE_TENANT_ID, y AZURE_CLIENT_SECRET para autenticar como un principal de servicio.
2. Identidad de carga de trabajo: Si el plugin se ejecuta en un host de Azure con la identidad de carga de trabajo activada (por ejemplo, AKS con la AZURE_FEDERATED_TOKEN_FILE variable de entorno activada), el plugin realiza un intercambio de tokens OIDC.
3. Identidad gestionada: Si el host tiene una identidad gestionada activada, el plugin se autentica usando esa identidad. Este método cubre máquinas virtuales de Azure, Virtual Machine Scale Sets y servidores habilitados para Azure Arc.

Para la secuencia completa de credenciales que DefaultAzureCredential intenta, véase Cadenas de credenciales en la biblioteca Azure Identity for Java.

Configuración necesaria para la identidad administrada:

Ejemplo: Identidad gestionada

output {
    microsoft-sentinel-log-analytics-logstash-output-plugin {
      data_collection_endpoint => "<enter your DCE logsIngestion URI here>"
      dcr_id => "<enter your DCR immutableId here>"
      stream_name => "<enter your stream name here>"
    }
}

Configuración opcional

Reinicio de Logstash

Reinicie Logstash con la configuración actualizada del complemento de salida. Compruebe que los datos se incorporan a la tabla correcta según la configuración del DCR.

Visualización de los registros entrantes en Microsoft Sentinel

Para comprobar que los datos de registro llegan al área de trabajo, siga estos pasos:

1. Compruebe que los mensajes se envían al complemento de salida.

2. En el menú de navegación Microsoft Sentinel, seleccione Registros. En el encabezado Tablas , expanda la categoría Registros personalizados . Busque y seleccione el nombre de la tabla que especificó (con un _CL sufijo) en la configuración.

   

3. Para ver los registros de la tabla, consulte la tabla mediante el nombre de la tabla como esquema.

   

Supervisar los registros de auditoría del plugin de salida

Para supervisar la conectividad y la actividad del complemento de salida Microsoft Sentinel, habilite el archivo de registro de Logstash adecuado. Consulte el documento Diseño de directorio de Logstash para obtener la ubicación del archivo de registro.

Si no ve ningún dato en este archivo de registro, genere y envíe algunos eventos localmente a través de los complementos de entrada y filtro para asegurarse de que el complemento de salida recibe datos. Microsoft Sentinel solo admite problemas relacionados con el complemento de salida.

Seguridad de red

Defina la configuración de red y habilite el aislamiento de red para el complemento de salida Microsoft Sentinel Logstash.

Etiquetas de servicio de red virtual

El complemento de salida de Microsoft Sentinel admite etiquetas de servicio de red virtual de Azure. Se requieren las etiquetas AzureMonitor y AzureActiveDirectory .

Azure Virtual Network etiquetas de servicio se pueden usar para definir controles de acceso de red en grupos de seguridad de red, Azure Firewall y rutas definidas por el usuario. Use etiquetas de servicio en lugar de direcciones IP específicas al crear reglas de seguridad y rutas. En escenarios en los que no se pueden usar etiquetas de servicio Azure Virtual Network, a continuación se indican los requisitos de firewall.

Requisitos de firewall

En la tabla siguiente se enumeran los requisitos de firewall para escenarios en los que no se pueden usar Azure etiquetas de servicio de red virtual.

Historial de versiones del complemento

2.5.0

• Se añadió una configuración opcional de proxy por plugin para el tráfico de autenticación e ingesta usando proxy, proxy_aad, y proxy_endpoint.
• Actualizado el manejador Netty, HTTP, HTTP/2 y componentes DNS de 4.1.133.Final a 4.1.136.Final.
• Actualizado Jackson Databind y Jackson Core de la versión 2.18.6 a la 2.18.8.

2.4.0

• Los subprocesos de los trabajadores ahora se ejecutan como pasadas limitadas y programadas por el ejecutor: las excepciones recuperables se registran y el trabajador se reanuda en el siguiente ciclo; los errores fatales de la JVM se registran y se vuelven a lanzar.
• Se ha corregido el apagado ordenado para que los lotes en curso se vacíen (primero los procesadores de lotes, luego los unificadores y, por último, los emisores) antes de que los trabajadores se detengan, con un límite de max_graceful_shutdown_time_seconds.
• Se añadieron tiempos connect_timeout_seconds de espera configurables para la subida (por defecto 15) y write_timeout_seconds (por defecto 60); se vuelven a intentar los tiempos de espera de conexión y escritura.
• Se añadió ID de hilo, tipo de excepción, tamaño de lote y flujo DCR a los registros de fallo por lotes.

2.3.3

• Se corrigió la pérdida de fidelidad de los tipos numéricos y booleanos: los campos basados en los tipos JRuby internos de Logstash (por ejemplo, puertos y recuentos de bytes) ahora se conservan como números y valores booleanos JSON nativos, en lugar de convertirse en cadenas, lo que garantiza una ingesta fiable en los DCR con columnas tipadas.

2.3.2

• Se ha corregido la muerte silenciosa de los hilos de los trabajadores causada por excepciones no capturadas en el bucle de procesamiento de los trabajadores.
• Se corrige NullPointerException en SenderWorker cuando Azure devuelve una LogsUploadException con una respuesta HTTP nula.
• Se añadió una gestión resiliente de errores con seguimiento consecutivo de errores para reducir fallos permanentes de los trabajadores.
• Se añadió un valor de configuración opcional id para la telemetría.
• Se ha añadido el flujo DCR al registro de los lotes enviados.

2.3.0

• Funcionalidad habilitada con Logstash 9.4.
• Se han actualizado las versiones de las dependencias de las bibliotecas externas (azure-sdk-bom, logback, slf4j, Netty).

2.2.1

• Añade una línea de registro a nivel de información cuando se envían con éxito los lotes.

2.2.0

• Añade la capacidad de usar valores de configuración nuevos o antiguos.

2.1.2

• Actualizaciones de documentación.

2.1.0

• Se ha corregido la normalización de eventos.

2.0.0

• Refactorizar el complemento de Ruby a Java.
• Se agregó la autenticación ManagedIdentity.
• Se ha movido código base de GitHub a Azure DevOps.
• Código base cerrado.

1.2.0

• Añade compatibilidad con la autenticación de identidad administrada para VM/VMSS de Azure (asignadas por el sistema y asignadas por el usuario a través de IMDS).
• Añade compatibilidad con la identidad de carga de trabajo de AKS a través del intercambio de tokens OIDC.
• Agrega compatibilidad con identidades administradas de Azure Arc para servidores híbridos y locales.
• Detecta automáticamente el método de autenticación durante la ejecución basada en el entorno (variables de entorno de identidad de carga de trabajo, agente de Arc o, como recurso alternativo, IMDS).
• Migra el cliente HTTP de excon a rest-client para mejorar la compatibilidad con el ecosistema de plugins de JRuby y Logstash.
• Cambia el nombre de las referencias a Azure Active Directory a Microsoft Entra ID.

1.1.4

• Limita excon la versión de la biblioteca a una versión inferior a 1.0.0 para asegurarse de que el puerto siempre se usa cuando se usa un proxy.

1.1.3

• Reemplaza la rest-client biblioteca que se usa para conectarse a Azure por la excon biblioteca.

1.1.1

• Agrega compatibilidad con Azure nube del Gobierno de EE. UU. y Microsoft Azure operados por 21Vianet en China.

1.1.0

• Permite establecer valores de proxy diferentes para las conexiones de API.
• Actualiza la versión de la API de ingesta de registros a 2023-01-01.
• Cambia el nombre del complemento a microsoft-sentinel-log-analytics-logstash-output-plugin.

1.0.0

• La versión inicial del complemento de salida de Logstash para Microsoft Sentinel. Este complemento usa reglas de recopilación de datos (DCR) con la API de ingesta de registros de Azure Monitor.

Problemas conocidos

Al usar Logstash instalado en una imagen de Docker de Lite Ubuntu, puede aparecer la siguiente advertencia:

java.lang.RuntimeException: getprotobyname_r failed

Para resolver este error, instale el paquete netbase en el dockerfile:

USER root
RUN apt install netbase -y

Para obtener más información, consulte Regresión de JNR en Logstash 7.17.0 (Docker).

Si la tasa de eventos de tu entorno es baja, aumenta el valor de max_waiting_time_for_batch_seconds y max_waiting_for_unifier_time_seconds a 60 o más. Puede supervisar la carga de ingesta mediante métricas de DCR. Para más información sobre las variables de tiempo de espera, consulte la tabla de configuración opcional .

Limitaciones

• La ingesta en tablas estándar solo se limita a las tablas estándar admitidas para la ingesta de registros personalizados.

• Las columnas del flujo de entrada de la streamDeclarations propiedad deben comenzar con una letra. Si inicia una columna con otros caracteres (por ejemplo @ , o _), se produce un error en la operación.

• El campo TimeGenerated datetime es obligatorio. Debe incluir este campo en la transformación KQL.

• Para posibles problemas adicionales, revisa los pasos de solución de problemas de la API de Ingestión de Logs.