Nel passato ci siamo incontrati Pila ELK, in quali prodotti software è composto. E il primo compito che un ingegnere deve affrontare quando lavora con lo stack ELK è inviare i log per l'archiviazione in elasticsearch per la successiva analisi. Tuttavia, si tratta solo di parole, elasticsearch memorizza i log sotto forma di documenti con determinati campi e valori, il che significa che l'ingegnere deve utilizzare vari strumenti per analizzare il messaggio inviato dai sistemi finali. Questo può essere fatto in diversi modi: scrivi tu stesso un programma che aggiungerà documenti al database utilizzando l'API o utilizza soluzioni già pronte. In questo corso considereremo la soluzione Logstash, che fa parte dello stack ELK. Vedremo come inviare i log dai sistemi endpoint a Logstash, quindi imposteremo un file di configurazione per analizzare e reindirizzare al database Elasticsearch. Per fare ciò, prendiamo i log dal firewall Check Point come sistema in entrata.
Il corso non copre l'installazione dello stack ELK, poiché esistono numerosi articoli su questo argomento; considereremo la componente di configurazione.
Elaboriamo un piano d'azione per la configurazione di Logstash:
- Verifica che elasticsearch accetti i log (verifica della funzionalità e dell'apertura della porta).
- Consideriamo come inviare eventi a Logstash, scegliere un metodo e implementarlo.
- Configuriamo Input nel file di configurazione Logstash.
- Configuriamo l'output nel file di configurazione Logstash in modalità debug per capire come appare il messaggio di log.
- Impostazione del filtro.
- Impostazione dell'output corretto in ElasticSearch.
- Viene avviato Logstash.
- Controllo dei registri a Kibana.
Esaminiamo ogni punto in modo più dettagliato:
Verifica che elasticsearch accetti i log
Per fare ciò, puoi utilizzare il comando curl per verificare l'accesso a Elasticsearch dal sistema su cui è distribuito Logstash. Se hai configurato l'autenticazione, trasferiamo anche l'utente/password tramite curl, specificando la porta 9200 se non l'hai modificata. Se ricevi una risposta simile a quella qui sotto, allora è tutto in ordine.
[elastic@elasticsearch ~]$ curl -u <<user_name>> : <<password>> -sS -XGET "<<ip_address_elasticsearch>>:9200"
{
"name" : "elastic-1",
"cluster_name" : "project",
"cluster_uuid" : "sQzjTTuCR8q4ZO6DrEis0A",
"version" : {
"number" : "7.4.1",
"build_flavor" : "default",
"build_type" : "rpm",
"build_hash" : "fc0eeb6e2c25915d63d871d344e3d0b45ea0ea1e",
"build_date" : "2019-10-22T17:16:35.176724Z",
"build_snapshot" : false,
"lucene_version" : "8.2.0",
"minimum_wire_compatibility_version" : "6.8.0",
"minimum_index_compatibility_version" : "6.0.0-beta1"
},
"tagline" : "You Know, for Search"
}
[elastic@elasticsearch ~]$
Se la risposta non viene ricevuta, potrebbero esserci diversi tipi di errori: il processo elasticsearch non è in esecuzione, è specificata la porta sbagliata o la porta è bloccata da un firewall sul server su cui è installato elasticsearch.
Diamo un'occhiata a come inviare log a Logstash da un checkpoint firewall
Dal server di gestione Check Point puoi inviare log a Logstash tramite syslog utilizzando l'utilità log_exporter, puoi leggere ulteriori informazioni qui , qui lasceremo solo il comando che crea lo stream:
cp_log_export aggiungi nome check_point_syslog server di destinazione < > protocollo target-port 5555 formato tcp modalità lettura generica semi-unificata
< > - indirizzo del server su cui gira Logstash, target-port 5555 - porta a cui invieremo i log, l'invio di log tramite tcp può caricare il server, quindi in alcuni casi è più corretto utilizzare udp.
Impostazione di INPUT nel file di configurazione Logstash
Per impostazione predefinita, il file di configurazione si trova nella directory /etc/logstash/conf.d/. Il file di configurazione è composto da 3 parti significative: INPUT, FILTER, OUTPUT. IN INGRESSO indichiamo da dove il sistema prenderà i log, in FILTRO analizzare il registro: imposta come dividere il messaggio in campi e valori, in USCITA configuriamo il flusso di output, dove verranno inviati i log analizzati.
Innanzitutto, configuriamo INPUT, consideriamo alcuni dei tipi che possono essere: file, tcp ed exe.
TCP:
input {
tcp {
port => 5555
host => “10.10.1.205”
type => "checkpoint"
mode => "server"
}
}
modalità => "server"
Indica che Logstash sta accettando connessioni.
porta => 5555
ospite => “10.10.1.205”
Accettiamo connessioni tramite indirizzo IP 10.10.1.205 (Logstash), porta 5555: la porta deve essere consentita dalla politica del firewall.
digitare => "punto di controllo"
Contrassegniamo il documento, molto comodo se hai più connessioni in entrata. Successivamente per ogni connessione è possibile scrivere il proprio filtro utilizzando il costrutto logico if.
File:
input {
file {
path => "/var/log/openvas_report/*"
type => "openvas"
start_position => "beginning"
}
}
Descrizione delle impostazioni:
percorso => "/var/log/openvas_report/*"
Indichiamo la directory in cui devono essere letti i file.
digitare => "openvas"
Tipo di evento.
start_position => "inizio"
Quando si modifica un file, legge l'intero file; se si imposta "fine", il sistema attende che appaiano nuovi record alla fine del file.
Esegui:
input {
exec {
command => "ls -alh"
interval => 30
}
}
Usando questo input, viene lanciato un (solo!) comando di shell e il suo output viene trasformato in un messaggio di log.
comando => "ls -alh"
Il comando di cui siamo interessati.
intervallo => 30
Intervallo di invocazione del comando in secondi.
Per ricevere i log dal firewall, registriamo un filtro tcp o udp, a seconda di come i log vengono inviati a Logstash.
Configuriamo l'output nel file di configurazione Logstash in modalità debug per capire come appare il messaggio di log
Dopo aver configurato INPUT, dobbiamo capire come apparirà il messaggio di log e quali metodi devono essere utilizzati per configurare il filtro di log (parser).
Per fare ciò, utilizzeremo un filtro che invia il risultato a stdout per visualizzare il messaggio originale; il file di configurazione completo al momento sarà simile a questo:
input
{
tcp
{
port => 5555
type => "checkpoint"
mode => "server"
host => “10.10.1.205”
}
}
output
{
if [type] == "checkpoint"
{
stdout { codec=> json }
}
}
Esegui il comando per verificare:
sudo /usr/share/logstash/bin//logstash -f /etc/logstash/conf.d/checkpoint.conf
Vediamo il risultato, l'immagine è cliccabile:
Se lo copi apparirà così:
action="Accept" conn_direction="Internal" contextnum="1" ifdir="outbound" ifname="bond1.101" logid="0" loguid="{0x5dfb8c13,0x5,0xfe0a0a0a,0xc0000000}" origin="10.10.10.254" originsicname="CN=ts-spb-cpgw-01,O=cp-spb-mgmt-01.tssolution.local.kncafb" sequencenum="8" time="1576766483" version="5" context_num="1" dst="10.10.10.10" dst_machine_name="[email protected]" layer_name="TSS-Standard Security" layer_name="TSS-Standard Application" layer_uuid="dae7f01c-4c98-4c3a-a643-bfbb8fcf40f0" layer_uuid="dbee3718-cf2f-4de0-8681-529cb75be9a6" match_id="8" match_id="33554431" parent_rule="0" parent_rule="0" rule_action="Accept" rule_action="Accept" rule_name="Implicit Cleanup" rule_uid="6dc2396f-9644-4546-8f32-95d98a3344e6" product="VPN-1 & FireWall-1" proto="17" s_port="37317" service="53" service_id="domain-udp" src="10.10.1.180" ","type":"qqqqq","host":"10.10.10.250","@version":"1","port":50620}{"@timestamp":"2019-12-19T14:50:12.153Z","message":"time="1576766483" action="Accept" conn_direction="Internal" contextnum="1" ifdir="outbound" ifname="bond1.101" logid="0" loguid="{0x5dfb8c13,
Osservando questi messaggi, capiamo che i log hanno il seguente aspetto: campo = valore o chiave = valore, il che significa che è adatto un filtro chiamato kv. Per scegliere il filtro giusto per ogni caso specifico, sarebbe una buona idea familiarizzare con loro nella documentazione tecnica o chiedere a un amico.
Impostazione del filtro
Nell'ultima fase abbiamo selezionato kv, la configurazione di questo filtro è presentata di seguito:
filter {
if [type] == "checkpoint"{
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
Selezioniamo il simbolo con cui divideremo il campo e il valore - "=". Se abbiamo voci identiche nel log, salviamo una sola istanza nel database, altrimenti ti ritroverai con un array di valori identici, cioè se abbiamo il messaggio “foo = some foo=some” scriviamo solo foo = alcuni.
Impostazione dell'output corretto in ElasticSearch
Una volta configurato il filtro, è possibile caricare i registri nel database elasticsearch:
output
{
if [type] == "checkpoint"
{
elasticsearch
{
hosts => ["10.10.1.200:9200"]
index => "checkpoint-%{+YYYY.MM.dd}"
user => "tssolution"
password => "cool"
}
}
}
Se il documento è firmato con il tipo checkpoint, salviamo l'evento nel database elasticsearch, che accetta connessioni su 10.10.1.200 sulla porta 9200 per impostazione predefinita. Ogni documento viene salvato in un indice specifico, in questo caso salviamo nell'indice “checkpoint-” + data e ora corrente. Ciascun indice può avere un insieme specifico di campi oppure viene creato automaticamente quando un nuovo campo viene visualizzato in un messaggio; le impostazioni dei campi e il relativo tipo possono essere visualizzati nelle mappature.
Se avete configurato l'autenticazione (ne parleremo più avanti), occorre specificare le credenziali per scrivere su un determinato indice, in questo esempio è “tssolution” con la password “cool”. È possibile differenziare i diritti utente per scrivere i log solo su un indice specifico e non oltre.
Avvia Logstash.
File di configurazione Logstash:
input
{
tcp
{
port => 5555
type => "checkpoint"
mode => "server"
host => “10.10.1.205”
}
}
filter {
if [type] == "checkpoint"{
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
output
{
if [type] == "checkpoint"
{
elasticsearch
{
hosts => ["10.10.1.200:9200"]
index => "checkpoint-%{+YYYY.MM.dd}"
user => "tssolution"
password => "cool"
}
}
}
Controlliamo la correttezza del file di configurazione:
/usr/share/logstash/bin//logstash -f checkpoint.conf
Avvia il processo Logstash:
sudo systemctl avvia logstash
Controlliamo che il processo sia iniziato:
sudo systemctl stato logstash
Controlliamo se il socket è attivo:
netstat -nat |grep 5555
Controllo dei registri a Kibana.
Dopo che tutto è in esecuzione, vai su Kibana - Scopri, assicurati che tutto sia configurato correttamente, l'immagine è cliccabile!
Tutti i registri sono a posto e possiamo vedere tutti i campi e i loro valori!
conclusione
Abbiamo esaminato come scrivere un file di configurazione Logstash e di conseguenza abbiamo ottenuto un parser di tutti i campi e valori. Ora possiamo lavorare con la ricerca e il tracciamento di campi specifici. Successivamente nel corso esamineremo la visualizzazione in Kibana e creeremo una semplice dashboard. Vale la pena ricordare che il file di configurazione Logstash necessita di essere costantemente aggiornato in determinate situazioni, ad esempio quando vogliamo sostituire il valore di un campo da un numero a una parola. Negli articoli successivi lo faremo costantemente.
Quindi rimanete sintonizzati (, , , ), .
Fonte: habr.com