Nell'articolo precedente ci siamo conosciuti con lo stack ELK, da quali prodotti software è composto. E il primo compito che un ingegnere affronta lavorando con lo stack ELK è l'invio dei log per la memorizzazione in Elasticsearch per un'analisi successiva. Tuttavia, questo è solo a parole, Elasticsearch memorizza i log sotto forma di documenti con campi e valori specifici, il che significa che l'ingegnere deve utilizzare vari strumenti per analizzare i messaggi che vengono inviati dai sistemi finali. Ciò può essere fatto in diversi modi: scrivere un programma che aggiunga documenti al database tramite API o utilizzare soluzioni già pronte. Nell'ambito di questo corso, tratteremo la soluzione Logstash, che fa parte dello stack ELK. Vedremo come inviare i log dai sistemi finali a Logstash, e poi andremo a configurare il file di configurazione per il parsing e l'inoltro al database Elasticsearch. A questo scopo, utilizzeremo i log del firewall Check Point come sistema di ingresso.
Nel corso non viene trattata l'installazione dello stack ELK, poiché esistono un'enorme quantità di articoli su questo argomento; ci concentreremo sugli aspetti di configurazione.
Elaboriamo un piano d'azione per la configurazione di Logstash:
- Controllare che Elasticsearch accetti i log (verificare il funzionamento e l'apertura della porta).
- Esaminiamo come possiamo inviare eventi a Logstash, scegliamo un metodo e implementiamolo.
- Configuriamo l'input nel file di configurazione di Logstash.
- Configuriamo l'output nel file di configurazione di Logstash in modalità debug, per capire come appare il messaggio di log.
- Configuriamo il filtro.
- Configuriamo l'output corretto in Elasticsearch.
- Avviamo Logstash.
- Controlliamo i log in Kibana.
Esaminiamo più in dettaglio ogni punto:
Controllare che Elasticsearch accetti i log
Per fare ciò, puoi controllare la disponibilità di Elasticsearch dalla macchina su cui è stato distribuito Logstash utilizzando il comando curl. Se hai configurato l'autenticazione, passa anche il nome utente/password tramite curl, specificando la porta 9200 se non l'hai modificata. Se ricevi una risposta come indicato di seguito, tutto è a posto.
[elastic@elasticsearch ~]$ curl -u <> : <> -sS -XGET "<>: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" : "Lo sai, per la ricerca"
}
[elastic@elasticsearch ~]$
Se la risposta non arriva, potrebbero esserci diversi tipi di errore: il processo elasticsearch non è in esecuzione, è stata indicata una porta errata oppure la porta è bloccata dal firewall sul server dove è installato elasticsearch.
Consideriamo come inviare i log a Logstash tramite il firewall Check Point
Con il server di gestione Check Point è possibile inviare i log a Logstash tramite syslog, utilizzando l'utilità log_exporter, puoi trovare ulteriori dettagli su di essa al , qui lasceremo solo il comando che crea il flusso:
cp_log_export add name check_point_syslog target-server <> target-port 5555 protocol tcp format generic read-mode semi-unified
<> — indirizzo del server su cui è in esecuzione Logstash, target-port 5555 — porta a cui invieremo i log, l'invio dei log tramite tcp potrebbe sovraccaricare il server, pertanto in alcuni casi è più corretto utilizzare udp.
Configuriamo l'INPUT nel file di configurazione di 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 INPUT indichiamo da dove il sistema prenderà i log, in FILTER analizziamo il log — impostiamo come dividere il messaggio in campi e valori, in OUTPUT configuriamo il flusso in uscita — dove i log analizzati saranno inviati.
Iniziamo a configurare INPUT, consideriamo alcuni tipi che possono essere — file, tcp e exe.
Tcp:
input {
tcp {
port => 5555
host => “10.10.1.205”
type => "checkpoint"
mode => "server"
}
}
mode => «server»
Indica che Logstash accetta connessioni.
port => 5555
host => “10.10.1.205”
Accettiamo connessioni all'indirizzo IP 10.10.1.205 (Logstash), la porta 5555 — la porta deve essere consentita dalla politica del firewall.
type => «checkpoint»
Etichettiamo il documento, molto utile nel caso in cui tu abbia più connessioni in ingresso. In seguito, per ogni connessione, puoi scrivere un filtro specifico utilizzando la struttura logica if.
File:
input {
file {
path => " /var/log/openvas_report/*"
type => "openvas"
start_position => "beginning"
}
}
Descrizione delle impostazioni:
path => «/var/log/openvas_report/*»
Indichiamo la directory da cui devono essere letti i file.
type => "openvas"
Tipo di evento.
start_position => "beginning"
Quando si modifica il file, viene letto completamente, se impostato su "end", il sistema attende l'arrivo di nuove registrazioni alla fine del file.
Exec:
input {
exec {
command => "ls -alh"
interval => 30
}
}
Con questo input, viene eseguita (solo!) un comando shell e la sua uscita è racchiusa in un messaggio di log.
command => "ls -alh"
Il comando di cui ci interessa l'uscita.
interval => 30
Intervallo di chiamata del comando in secondi.
Per ricevere i log dal firewall, definiamo un filtro tcp o udp, a seconda di come i log vengono inviati a Logstash.
Configuriamo l'Output nel file di configurazione di Logstash in modalità debug, per capire come appare il messaggio di log.
Dopo aver configurato INPUT, è necessario capire come apparirà il messaggio di log, quali metodi usare per configurare il filtro (parser) dei log.
A tale scopo, utilizzeremo un filtro che restituisce il risultato in stdout per visualizzare il messaggio originale; l'intero file di configurazione attualmente apparirà così:
input
{
tcp
{
port => 5555
type => "checkpoint"
mode => "server"
host => "10.10.1.205"
}
}
output
{
if [type] == "checkpoint"
{
stdout { codec => json }
}
}
Eseguiamo il comando di verifica:
sudo /usr/share/logstash/bin//logstash -f /etc/logstash/conf.d/checkpoint.conf
Possiamo vedere il risultato, l'immagine è cliccabile:
Se copiamo questo, 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="ts-spb-dc-01@tssolution.local" 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,
Guardando questi dati, comprendiamo che i log hanno il formato: campo = valore o chiave = valore, e quindi si adatta il filtro chiamato kv. Per scegliere correttamente il filtro per ogni caso specifico, sarebbe utile familiarizzare con essi nella documentazione tecnica, oppure chiedere a un amico.
Impostiamo il filtro
Nella fase precedente abbiamo selezionato kv, di seguito è presentata la configurazione di questo filtro:
filter {
if [type] == "checkpoint"{
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
Scegliamo il simbolo con cui divideremo il campo e il valore — “=”. Se abbiamo nel log voci identiche, salviamo nel database solo un'istanza, altrimenti otterremo un array di valori identici, cioè se abbiamo il messaggio “foo = some foo=some” salviamo solo foo = some.
Impostiamo l'Output corretto in ElasticSearch
Dopo aver impostato il filtro, possiamo esportare i log 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 default. Ogni documento viene salvato in un indice specifico, in questo caso salviamo nell'indice «checkpoint-» + la data temporale corrente. Ogni indice può avere un certo insieme di campi, oppure viene creato automaticamente con l'apparizione di un nuovo campo nel messaggio, le impostazioni dei campi e il loro tipo possono essere visualizzati nella mappatura.
Se hai impostato l'autenticazione (ne parleremo più avanti), devono essere necessariamente indicati le credenziali per la scrittura in un indice specifico, in questo esempio è «tssolution» con la password «cool». È possibile delimitare i diritti degli utenti per la scrittura dei log solo in un indice specifico e non in altri.
Avviamo Logstash.
File di configurazione di 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
Avviamo il processo Logstash:
sudo systemctl start logstash
Controlliamo che il processo sia avviato:
sudo systemctl status logstash
Controlliamo se il socket è attivo:
netstat -nat |grep 5555
Controlliamo i log in Kibana.
Dopo aver avviato tutto, accediamo a Kibana — Discover, assicuriamoci che sia tutto configurato correttamente, l'immagine è cliccabile!
Tutti i log sono al loro posto e possiamo vedere tutti i campi e i loro valori!
Conclusione
Abbiamo esaminato come scrivere il file di configurazione di Logstash, ottenendo così un parser per tutti i campi e valori. Ora possiamo lavorare con la ricerca e la creazione di grafici per determinati campi. Nel prosieguo del corso esamineremo la visualizzazione in Kibana e creeremo un dashboard semplice. Vale la pena menzionare che il file di configurazione di Logstash deve 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 continueremo a farlo.
Quindi, seguite gli aggiornamenti (, , , ), .
Fonte: habr.com
