Inviare i log in streaming a Microsoft Sentinel con Logstash e API basata su DCR

Il plug-in di output per Logstash di Microsoft Sentinel supporta le trasformazioni della pipeline e la configurazione avanzata tramite le Data Collection Rules (DCR). Il plug-in inoltra i log da origini dati esterne in tabelle personalizzate o standard in Log Analytics o Microsoft Sentinel.

Questo articolo illustra come configurare il plug-in Logstash per trasmettere i dati in Log Analytics o Microsoft Sentinel usando dcr, con controllo completo sullo schema di output.

Con il plug-in è possibile:

• Controllare la configurazione dei nomi e dei tipi di colonna.
• Eseguire trasformazioni in fase di inserimento, ad esempio filtro o arricchimento.
• Inserire log personalizzati in una tabella personalizzata o inserire un flusso di input Syslog nella tabella Syslog di Log Analytics.

L'inserimento nelle tabelle standard è limitato solo alle tabelle standard supportate per l'inserimento di log personalizzati.

Per altre informazioni sull'uso del motore di raccolta dati Logstash, vedere Introduzione a Logstash.

Panoramica dell'architettura

Il motore Logstash è composto da tre componenti:

• Plug-in di input: raccolta personalizzata di dati da varie origini.
• Plug-in filtro: manipolazione e normalizzazione dei dati in base ai criteri specificati.
• Plug-in di output: invio personalizzato di dati raccolti ed elaborati a varie destinazioni.

Il plug-in invia dati in formato JSON all'area di lavoro Log Analytics usando l'API di inserimento log. I dati vengono inseriti in log personalizzati o in una tabella standard.

• Altre informazioni sull'API di inserimento log.

Distribuire il plug-in di output di Microsoft Sentinel su Logstash

Per configurare il plug-in, seguire questa procedura:

• Rivedi i prerequisiti del plugin Logstash
• Installare il plug-in
• Creare un file di esempio
• Creare le risorse necessarie relative a DCR
• Configurare il file di configurazione Logstash
• Riavviare Logstash
• Visualizza i log in arrivo in Microsoft Sentinel
• Monitorare i log di controllo del plug-in di output

Prerequisiti del plug-in Logstash

• Installare una versione supportata di Logstash. Il plug-in supporta le versioni logstash seguenti:

  • 7.0 - 7.17.13

  • 8.0 - 8.9 (queste versioni richiedono un aggiornamento di sicurezza, secondo Logstash)

  • 8.11 - 8.15 (queste versioni richiedono un aggiornamento di sicurezza, secondo Logstash)

  • 8.19.2 (questa versione richiede un aggiornamento di sicurezza, secondo Logstash)

  • 9.0.8 (questa versione richiede un aggiornamento di sicurezza, secondo Logstash)

  • 9.1.10 (questa versione richiede un aggiornamento di sicurezza, secondo Logstash)

  • 9.2.4 - 9.2.5 (queste versioni richiedono un aggiornamento di sicurezza, secondo Logstash)

  • 9.3.3

  • 9.4.0

    Nota

    Se si usa Logstash 8, è consigliabile disabilitare ECS nella pipeline.

• Verificare di avere un'area di lavoro Log Analytics con almeno i diritti di collaboratore.

• Verificare di disporre delle autorizzazioni per creare oggetti DCR nell'area di lavoro.

Installare il plug-in

Il plug-in di output Microsoft Sentinel è disponibile nella raccolta Logstash in RubyGems.

• Seguire le istruzioni nel documento Logstash Working with plugins per installare il plug-in microsoft-sentinel-log-analytics-logstash-output-plugin . Per eseguire l'installazione in un'installazione di Logstash esistente, eseguire il comando seguente:

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

• Se il sistema Logstash non ha accesso a Internet, seguire le istruzioni nel documento Gestione plug-in offline di Logstash per preparare e usare un pacchetto di plug-in offline. Ciò richiede la creazione di un altro sistema Logstash con accesso a Internet.

Creare un file di esempio

In questa sezione viene creato un file di esempio in uno di questi scenari:

• Creare un file di esempio per i log personalizzati
• Creare un file di esempio per inserire i log nella tabella Syslog

Creare un file di esempio per i log personalizzati

In questo scenario si configura il plug-in di input Logstash per inviare eventi a Microsoft Sentinel. Questo esempio usa il plug-in di input del generatore per simulare gli eventi. È possibile usare qualsiasi altro plug-in di input.

In questo esempio il file di configurazione Logstash è simile al seguente:

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

Per creare il file di esempio, seguire questa procedura:

1. Copiare la configurazione del plug-in di output seguente nel file di configurazione 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. Assicurarsi che il percorso del file a cui si fa riferimento esista già, quindi avviare Logstash.

   Il plug-in scrive dieci record in un file di esempio denominato sampleFile<epoch seconds>.json nel percorso configurato quando sono presenti 10 eventi da campionare o quando il processo Logstash viene chiuso normalmente. Ad esempio: c:\temp\sampleFile1648453501.json. Di seguito è riportato un file di esempio creato dal plug-in:

   [
           {
               "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
       ...
   
       ]    
   

   Il plug-in aggiunge automaticamente queste proprietà a ogni record:

   • ls_timestamp: ora in cui il record viene ricevuto dal plug-in di input
   • ls_version: la versione della pipeline di Logstash.

   È possibile rimuovere questi campi quando si crea il DCR.

Creare un file di esempio per inserire i log nella tabella Syslog

In questo scenario si configura il plug-in di input Logstash per inviare eventi syslog a Microsoft Sentinel.

1. Se i messaggi syslog non sono già stati inoltrati nel computer Logstash, è possibile usare il comando logger per generare messaggi. Ad esempio (per 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
   

   Ecco un esempio per il plug-in di input Logstash:

   input {
        syslog {
            port => 514
       }
   }
   

2. Copiare la configurazione del plug-in di output seguente nel file di configurazione 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. Assicurarsi che il percorso del file esista già, quindi avviare Logstash.

   Il plug-in scrive dieci record in un file di esempio denominato sampleFile<epoch seconds>.json nel percorso configurato quando sono presenti 10 eventi da campionare o quando il processo Logstash viene chiuso normalmente. Ad esempio: c:\temp\sampleFile1648453501.json. Di seguito è riportato un file di esempio creato dal plug-in:

   [
           {
               "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"
           }
   ]    
   
   

   Il plug-in aggiunge automaticamente queste proprietà a ogni record:

   • ls_timestamp: ora in cui il record viene ricevuto dal plug-in di input
   • ls_version: la versione della pipeline di Logstash.

   È possibile rimuovere questi campi quando si crea il DCR.

Creare le risorse DCR necessarie

Per configurare il plug-in Logstash basato su DCR Microsoft Sentinel, creare prima di tutto le risorse correlate a DCR.

In questa sezione vengono create risorse da usare per il DCR, in uno di questi scenari:

• Creare risorse DCR per l'inserimento in una tabella personalizzata
• Creare risorse DCR per l'inserimento in una tabella standard

Creare risorse DCR per l'inserimento in una tabella personalizzata

Per inserire i dati in una tabella personalizzata, seguire questa procedura (in base all'esercitazione Inviare dati a Azure monitorare i log usando l'API REST (portale di Azure):

1. Esaminare i prerequisiti.

2. Configurare l'applicazione.

3. Aggiungere una tabella di log personalizzata.

4. Analizzare e filtrare i dati di esempio usando il file di esempio creato nella sezione precedente.

5. Raccogliere informazioni dal DCR.

6. Assegnare le autorizzazioni al DCR.

   Ignorare il passaggio Invia dati di esempio.

Se incontri problemi, consulta i passaggi di risoluzione dei problemi dell'API di Ingestione dei Logs.

Creare risorse DCR per l'inserimento in una tabella standard

Per inserire i dati in una tabella standard, ad esempio Syslog o CommonSecurityLog, si usa un processo basato sull'esercitazione Inviare dati ai log di Monitoraggio di Azure usando l'API REST (modelli di Resource Manager). Mentre l'esercitazione illustra come inserire dati in una tabella personalizzata, è possibile modificare facilmente il processo per inserire i dati in una tabella standard. I passaggi seguenti indicano le modifiche rilevanti nei passaggi.

1. Esaminare i prerequisiti.

2. Raccogliere i dettagli dell'area di lavoro.

3. Configurare un'applicazione.

   Ignorare il passaggio Crea nuova tabella nell'area di lavoro Log Analytics. Questo passaggio non è rilevante quando si inserisce dati in una tabella standard, perché la tabella è già definita in Log Analytics.

4. Creare il DCR. In questo passaggio:

   • Fornisci il file di esempio che hai creato in Crea un file di esempio.
   • Utilizzare il file di esempio creato per definire la streamDeclarations proprietà . Ognuno dei campi nel file di esempio deve avere una colonna corrispondente con lo stesso nome e il tipo appropriato (vedere l'esempio seguente).
   • Configurare il valore della outputStream proprietà con il nome della tabella standard anziché la tabella personalizzata. A differenza delle tabelle personalizzate, i nomi di tabella standard non hanno il _CL suffisso .
   • Il prefisso del nome della tabella deve essere Microsoft- anziché Custom-. In questo esempio il valore della outputStream proprietà è Microsoft-Syslog.

5. Assegnare le autorizzazioni a un DCR.

   Ignorare il passaggio Invia dati di esempio.

Se incontri problemi, consulta i passaggi di risoluzione dei problemi dell'API di Ingestione dei Logs.

Esempio: DCR che inserisce i dati nella tabella Syslog

Tenere presenti questi punti:

• I streamDeclarations nomi e i tipi di colonna devono corrispondere ai campi di file di esempio, ma non è necessario specificarli tutti. Ad esempio, nel DCR seguente, i campi PRI, type e ls_version sono omessi nella colonna streamDeclarations.
• La proprietà dataflows trasforma l'input nel formato della tabella Syslog e imposta outputStream su 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'))]"
    }
  }
}

Configurare il file di configurazione Logstash

Il plug-in supporta due metodi di autenticazione: entità servizio (credenziali client) e identità gestita (senza password). Scegliere il metodo più adatto all'ambiente.

Autenticazione del service principal

Per configurare il file di configurazione Logstash affinché integri i log in una tabella personalizzata usando l'autenticazione del principale di servizio, recupera i seguenti valori: client_id, client_secret, tenant_id, data_collection_endpoint, dcr_id, e stream_name.

Dopo aver recuperato i valori necessari:

1. Sostituire la sezione di output del file di configurazione Logstash creato nel passaggio precedente con l'esempio seguente.
2. Sostituire le stringhe segnaposto nell'esempio seguente con i valori recuperati.
3. Assicurarsi di modificare l'attributo create_sample_file in false.

Esempio: Configurazione del plug-in di output 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"
    }
}

Autenticazione dell'identità gestita (senza password)

Quando non fornisci le credenziali del principale servizio (client_id, , e tenant_id), il plugin si autentica utilizzando DefaultAzureCredential dall'client_secretAzure SDK. DefaultAzureCredential prova una sequenza di metodi di autenticazione e utilizza il primo che riesce. In un ambiente server, i metodi rilevanti vengono tentati in quest'ordine:

1. Variabili di ambiente: Legge le credenziali da variabili di ambiente come AZURE_CLIENT_ID, AZURE_TENANT_ID, e AZURE_CLIENT_SECRET per autenticarsi come principale di servizio.
2. Identità del carico di lavoro: Se il plugin viene eseguito su un host Azure con l'identità del carico di lavoro abilitata (ad esempio, AKS con la variabile AZURE_FEDERATED_TOKEN_FILE di ambiente impostata), il plugin effettua uno scambio di token OIDC.
3. Identità gestita: Se l'host ha un'identità gestita abilitata, il plugin si autentica usando quell'identità. Questo metodo copre Azure VM, set di scalabilità di macchine virtuali e server abilitati ad Azure Arc.

Per la sequenza completa delle credenziali che DefaultAzureCredential tentano, vedi Catene di credenziali nella libreria Azure Identity for Java.

Configurazione necessaria per l'identità gestita:

Esempio: Identità gestita

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>"
    }
}

Configurazione facoltativa

Riavviare Logstash

Riavviare Logstash con la configurazione del plug-in di output aggiornata. Verificare che i dati siano inseriti nella tabella corretta in base alla configurazione DCR.

Visualizza i log in arrivo in Microsoft Sentinel

Per verificare che i dati di log raggiungano l'area di lavoro, seguire questa procedura:

1. Verificare che i messaggi vengano inviati al plug-in di output.

2. Dal menu di navigazione di Microsoft Sentinel, selezionare Logs. Sotto l'intestazione Tabelle espandere la categoria Log personalizzati . Trovare e selezionare il nome della tabella specificata (con un _CL suffisso) nella configurazione.

   

3. Per visualizzare i record nella tabella, eseguire una query sulla tabella usando il nome della tabella come schema.

   

Monitorare i log di controllo del plug-in di output

Per monitorare la connettività e l'attività del plug-in di output Microsoft Sentinel, abilitare il file di log Logstash appropriato. Vedere il documento Logstash Directory Layout per il percorso del file di log.

Se non vengono visualizzati dati in questo file di log, generare e inviare alcuni eventi in locale tramite i plug-in di input e filtro per assicurarsi che il plug-in di output stia ricevendo dati. Microsoft Sentinel supporta solo i problemi relativi al plug-in di output.

Sicurezza di rete

Definire le impostazioni di rete e abilitare l'isolamento della rete per il plug-in di output Microsoft Sentinel Logstash.

Tag del servizio di rete virtuale

Il plug-in di output di Microsoft Sentinel supporta i tag di servizio della rete virtuale di Azure. Sono necessari tag AzureMonitor e AzureActiveDirectory .

I tag di servizio di Rete virtuale di Azure possono essere usati per definire i controlli di accesso alla rete in gruppi di sicurezza di rete, Firewall di Azure e route definite dall'utente. Usare i tag del servizio anziché indirizzi IP specifici quando si creano regole di sicurezza e route. Negli scenari in cui non è possibile utilizzare i tag di servizio di Rete virtuale di Azure, i requisiti del firewall sono indicati di seguito.

Requisiti del firewall

Nella tabella seguente sono elencati i requisiti del firewall per gli scenari in cui non è possibile usare Azure tag del servizio di rete virtuale.

Cronologia delle versioni del plug-in

2.5.0

• Aggiunta una configurazione proxy opzionale per ciascun plugin per l'autenticazione e il traffico di acquisizione usando proxy, proxy_aad e proxy_endpoint.
• Aggiornato il gestore Netty, i componenti HTTP, HTTP/2 e DNS da 4.1.133.Final a 4.1.136.Final.
• Aggiornato Jackson Databind e Jackson Core dalla 2.18.6 alla 2.18.8.

2.4.0

• I thread lavoratori vengono ora eseguiti come passaggi limitati pianificati dall'executor: le eccezioni ripristinabili vengono registrate e il lavoratore riprende nel ciclo successivo; gli errori JVM fatali vengono registrati e rilanciati.
• Corretto l’arresto normale in modo che i batch in corso vengano svuotati (batcher, poi unificatore, poi mittente) prima che i lavoratori si arrestino, entro il limite di max_graceful_shutdown_time_seconds.
• Sono stati aggiunti timeout di caricamento configurabili connect_timeout_seconds (valore predefinito: 15) e write_timeout_seconds (valore predefinito: 60); i tentativi di connessione e scrittura vengono ripetuti in caso di timeout.
• Aggiunti l'ID del thread, il tipo di eccezione, la dimensione del batch e il flusso DCR ai log degli errori del batch.

2.3.3

• Risolta la perdita di fedeltà dei tipi numerici e booleani: i campi basati sui tipi JRuby interni di Logstash (ad esempio, porte e conteggi di byte) vengono ora mantenuti come numeri e valori booleani JSON nativi invece di essere convertiti in stringhe, garantendo un'acquisizione affidabile nei DCR con colonne tipizzate.

2.3.2

• Risolta la terminazione silenziosa del thread lavoratore causata da eccezioni non gestite nel ciclo del processo di lavoro.
• Corretta una NullPointerException in SenderWorker quando Azure restituisce una LogsUploadException con una risposta HTTP nulla.
• Aggiunta una gestione resiliente degli errori con tracciamento consecutivo degli errori per ridurre i fallimenti permanenti dei lavoratori.
• Aggiunta di un valore di configurazione opzionale id per la telemetria.
• Aggiunto il flusso DCR al log dei batch inviati.

2.3.0

• Funzionalità abilitate con Logstash 9.4.
• Versioni di dipendenza aggiornate per librerie esterne (azure-sdk-bom, logback, slf4j, Netty).

2.2.1

• Aggiunge una linea di logging a livello informativo quando i lotti vengono inviati con successo.

2.2.0

• Aggiunge la possibilità di usare sia valori di configurazione nuovi che vecchi.

2.1.2

• Aggiornamenti della documentazione.

2.1.0

• Correzione della normalizzazione degli eventi.

2.0.0

• Il plug-in è stato sottoposto a refactoring, passando da Ruby a Java.
• Aggiunta l'autenticazione ManagedIdentity.
• È stata spostata la codebase da GitHub a Azure DevOps.
• Base di codice chiusa.

1.2.0

• Aggiunge il supporto per l'autenticazione con identità gestite per le VM/VMSS di Azure (assegnate dal sistema e dall'utente tramite IMDS).
• Aggiunge il supporto per l'identità del carico di lavoro di AKS tramite lo scambio di token OIDC.
• Aggiunge il supporto per l'identità gestita di Azure Arc per server ibridi e locali.
• Rileva automaticamente il metodo di autenticazione in fase di esecuzione in base all'ambiente (identità del carico di lavoro env vars, agente Arc o fallback IMDS).
• Esegue la migrazione del client HTTP da excon a rest-client per migliorare la compatibilità dell'ecosistema dei plug-in JRuby e Logstash.
• Rinomina i riferimenti ad Azure Active Directory in Microsoft Entra ID.

1.1.4

• Limita excon la versione della libreria a una versione inferiore a 1.0.0 per garantire che la porta venga sempre usata quando si usa un proxy.

1.1.3

• Sostituisce la rest-client libreria usata per la connessione a Azure con la excon libreria.

1.1.1

• Aggiunge il supporto per Azure cloud us government e Microsoft Azure gestito da 21Vianet in Cina.

1.1.0

• Consente di impostare valori proxy diversi per le connessioni API.
• Aggiorna la versione per l'API di inserimento log a 2023-01-01.
• Rinomina il plug-in in microsoft-sentinel-log-analytics-logstash-output-plugin.

1.0.0

• Versione iniziale per il plug-in di output Logstash per Microsoft Sentinel. Questo plug-in usa le regole di raccolta dati (DCR) con l'API di inserimento dei log di Monitoraggio di Azure.

Problemi noti

Quando si usa Logstash installato in un'immagine Docker di Lite Ubuntu, potrebbe essere visualizzato l'avviso seguente:

java.lang.RuntimeException: getprotobyname_r failed

Per risolvere questo errore, installare il pacchetto netbase nel dockerfile:

USER root
RUN apt install netbase -y

Per altre informazioni, vedere Regressione JNR in Logstash 7.17.0 (Docker).For more information, see JNR regression in Logstash 7.17.0 (Docker).

Se il tasso di eventi del tuo ambiente è basso, aumenta il valore di max_waiting_time_for_batch_seconds e max_waiting_for_unifier_time_seconds a 60 o più. È possibile monitorare il payload di acquisizione utilizzando le metriche DCR. Per maggiori informazioni sulle variabili di tempo di attesa, consulta la tabella di configurazione Opzionale .

Limitazioni

• L'inserimento nelle tabelle standard è limitato solo alle tabelle standard supportate per l'inserimento di log personalizzati.

• Le colonne del flusso di input nella streamDeclarations proprietà devono iniziare con una lettera. Se si avvia una colonna con altri caratteri (ad esempio @ o _), l'operazione ha esito negativo.

• Il TimeGenerated campo datetime è obbligatorio. È necessario includere questo campo nella trasformazione KQL.

• Per ulteriori possibili problemi, consulta i passaggi di risoluzione dei problemi dell'API di Ingestione dei Logs.