Nel passato ci siamo familiarizzati con il stack ELK, di quali prodotti software è composto. La prima sfida che incontra un ingegnere lavorando con lo stack ELK è l'invio dei log per la loro memorizzazione in Elasticsearch per un'ulteriore analisi. Tuttavia, è solo semplice a parole, Elasticsearch memorizza i log come documenti con determinati campi e valori, il che significa che l'ingegnere deve, usando vari strumenti, analizzare il messaggio che viene inviato dai sistemi finali. Questo può essere fatto in diversi modi: scrivendo un programma da zero che aggiunge documenti al database tramite API oppure utilizzando soluzioni già pronte. Nell'ambito di questo corso analizzeremo una soluzione Logstash, che fa parte dello stack ELK. Vedremo come inviare i log dai sistemi finali a Logstash e poi configureremo il file di configurazione per l'analisi e il reindirizzamento nel database Elasticsearch. A tal fine, utilizziamo i log del firewall Check Point come sistema di ingresso.
Nel corso non verrà trattata l'installazione dello stack ELK, poiché esistono un enorme numero di articoli su questo argomento, ci concentreremo sugli aspetti di configurazione.
Definiamo un piano d'azione per la configurazione di Logstash:
- Verifica che Elasticsearch stia ricevendo i log (controllo della funzionalità e apertura della porta).
- Analizziamo come possiamo inviare eventi a Logstash, scegliamo il 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 Filter.
- Configuriamo un Output corretto in Elasticsearch.
- Avviamo Logstash.
- Controlliamo i log in Kibana.
Esaminiamo più nel dettaglio ogni punto:
Verifica che Elasticsearch stia ricevendo i log
A tal fine, puoi controllare il collegamento a Elasticsearch dal sistema in cui è installato Logstash utilizzando il comando curl. Se hai configurato l'autenticazione, trasmetti anche nome utente/password tramite curl, specificando la porta 9200 se non l'hai modificata. Se ricevi una risposta simile a quella indicata di seguito, significa che 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" : "You Know, for Search"
}
[elastic@elasticsearch ~]$
Se la risposta non arriva, potrebbero esserci diversi tipi di errori: il processo elasticsearch potrebbe non essere attivo, potrebbe essere specificata una porta errata o la porta potrebbe essere bloccata dal firewall sul server dove è installato elasticsearch.
Vediamo 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; ulteriori dettagli possono essere trovati qui. , qui indichiamo 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 gira Logstash, target-port 5555 — la porta su cui invieremo i log. L'invio di log tramite tcp può sovraccaricare il server, quindi in alcuni casi è preferibile utilizzare udp.
Configuriamo l'INPUT nel file di configurazione di Logstash
Di default, 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 — configurando come suddividere il messaggio in campi e valori, in OUTPUT configuriamo il flusso in uscita — dove saranno inviati i log analizzati.
Iniziamo a configurare l'INPUT, esaminiamo alcuni tipi possibili — file, tcp ed 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 autorizzata dalla policy del firewall.
type => «checkpoint»
Etichettiamo il documento, molto utile nel caso in cui abbiate più connessioni in ingresso. Successivamente, per ogni connessione, è possibile scrivere il proprio filtro utilizzando una struttura logica if.
File:
input {
file {
path => "/var/log/openvas_report/*"
type => "openvas"
start_position => "beginning"
}
}
Descrizione delle impostazioni:
path => "/var/log/openvas_report/*"
Specifica la directory da cui devono essere letti i file.
type => "openvas"
Tipo di evento.
start_position => "beginning"
Se il file viene modificato, legge l'intero file; se impostato su "end", il sistema attende l'aggiunta di nuove voci alla fine del file.
Exec:
input {
exec {
command => "ls -alh"
interval => 30
}
}
Con questo input viene eseguita (solo!) un comando shell e la sua output viene racchiuso in un messaggio di log.
command => "ls -alh"
Comando di cui ci interessa l'output.
interval => 30
Intervallo di chiamata del comando in secondi.
Per ricevere i log dal firewall, specifichiamo il filtro tcp o udp, a seconda di come vengono inviati i log 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 l'INPUT, è necessario capire come apparirà il messaggio di log, quali metodi utilizzare per configurare il filtro (parser) dei log.
A tal fine, utilizzeremo un filtro che restituisce il risultato in stdout per visualizzare il messaggio originale. L'attuale file di configurazione 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 per la verifica:
sudo /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/checkpoint.conf
Vediamo il risultato, l'immagine è cliccabile:
Se copiamo, 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
Dall'analisi di questi messaggi, comprendiamo che i log sono formati da: campo = valore oppure key = value, il che significa che è adatto il filtro chiamato kv. Per scegliere il filtro appropriato per ogni caso specifico, sarebbe utile rivolgersi alla documentazione tecnica o chiedere a un amico.
Configuriamo il Filter
Nella fase precedente abbiamo scelto kv; di seguito è riportata la configurazione di questo filtro:
filter {
if [type] == "checkpoint"{
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
Selezioniamo il simbolo in base al quale divideremo il campo e il valore - "=". Se nel registro ci sono voci duplicate, salviamo nel database solo un'istanza; altrimenti, risulterà un array di valori identici, quindi se abbiamo il messaggio "foo = some foo=some", salviamo solo foo = some.
Configuriamo correttamente l'Output in ElasticSearch
Dopo aver configurato il Filter, 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 come 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, lo salviamo nell'indice "checkpoint-" + la data temporale corrente. Ogni indice può avere un insieme specifico di campi, oppure viene creato automaticamente all'apparire di un nuovo campo nel messaggio; le impostazioni dei campi e il loro tipo possono essere visualizzate nei mappings.
Se hai configurato l'autenticazione (che discuteremo più avanti), è necessario fornire le credenziali per la scrittura in un indice specifico, in questo esempio "tssolution" con la password "cool". È possibile delimitare i diritti degli utenti per scrivere i log solo in un determinato indice 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 il file di configurazione per verificarne la correttezza:
/usr/share/logstash/bin//logstash -f checkpoint.conf
Avviamo il processo Logstash:
sudo systemctl start logstash
Controlliamo che il processo sia stato 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 un file di configurazione Logstash, ottenendo un parser per tutti i campi e valori. Ora possiamo lavorare con la ricerca e la creazione di grafici su specifici campi. Nel proseguo del corso, analizzeremo la visualizzazione in Kibana e creeremo un semplice dashboard. È importante notare che il file di configurazione di Logstash deve essere aggiornato continuamente in determinate situazioni, ad esempio quando vogliamo sostituire il valore di un campo da un numero a una parola. Negli articoli successivi faremo questo frequentemente.
Quindi segui gli aggiornamenti (, , , ), .
Fonte: habr.com
