În trecut, ne-am familiarizat cu stiva ELK, din ce produse software este formată. Prima sarcină cu care se confruntă un inginer în lucrul cu stiva ELK este trimiterea jurnalelor pentru stocare în Elasticsearch pentru analiza ulterioară. Totuși, aceasta este doar o explicație simplă; Elasticsearch stochează jurnalele sub formă de documente cu câmpuri și valori definite, ceea ce înseamnă că inginerul trebuie să utilizeze diferite instrumente pentru a analiza mesajul care este trimis din sistemele finale. Acest lucru se poate face în mai multe moduri — fie prin scrierea unui program care va adăuga documente în baza de date prin API, fie folosind soluții deja existente. În cadrul acestui curs, vom lua în considerare soluția Logstash, care este parte din stiva ELK. Vom vedea cum putem trimite jurnalele din sistemele finale către Logstash și apoi vom configura fișierul de configurare pentru analiză și redirecționare către baza de date Elasticsearch. În acest scop, vom lua ca sistem de intrare jurnalele de la firewall-ul Check Point.
În cadrul cursului, nu vom aborda instalarea stivei ELK, deoarece există o mulțime de articole pe această temă, ne vom concentra pe componenta de configurare.
Vom elabora un plan de acțiune pentru configurarea Logstash:
- Verificarea că Elasticsearch va accepta jurnale (verificarea funcționalității și deschiderii portului).
- Analizăm modul în care putem trimite evenimente în Logstash, alegem o metodă și o implementăm.
- Configurăm Input în fișierul de configurare Logstash.
- Configurăm Output în fișierul de configurare Logstash în modul de depanare, pentru a înțelege cum arată mesajul de jurnal.
- Configurăm Filter.
- Configurăm Output-ul corect în Elasticsearch.
- Pornim Logstash.
- Verificăm jurnalele în Kibana.
Vom analiza fiecare punct în detaliu:
Verificarea că Elasticsearch va accepta jurnalele
Pentru aceasta, se poate verifica accesibilitatea din Elasticsearch din sistemul pe care este instalat Logstash prin comanda curl. Dacă aveți configurați autentificarea, transmiteți utilizator/parolă prin curl, specificând portul 9200 dacă nu l-ați schimbat. Dacă primim un răspuns similar cu cel indicat mai jos, înseamnă că totul este în regulă.
[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" : "Știi, pentru căutare"
}
[elastic@elasticsearch ~]$
Dacă nu a venit un răspuns, pot exista mai multe tipuri de erori: procesul elasticsearch nu funcționează, portul specificat este greșit sau portul este blocat de firewall-ul de pe serverul care rulează elasticsearch.
Să vedem cum putem trimite jurnalele în Logstash prin intermediul firewall-ului Check Point.
De la serverul de management Check Point, putem trimite jurnalele în Logstash prin syslog, utilizând utilitarul log_exporter; mai multe detalii pot fi găsite aici. , aici vom lăsa doar comanda care creează fluxul:
cp_log_export add name check_point_syslog target-server <> target-port 5555 protocol tcp format generic read-mode semi-unified
<> — adresa serverului pe care rulează Logstash, target-port 5555 — portul către care vom trimite jurnalele; trimiterea jurnale pe tcp poate solicita serverul, prin urmare în unele cazuri este mai corect să folosim udp.
Configurăm INPUT în fișierul de configurare Logstash.
În mod implicit, fișierul de configurare se află în directorul /etc/logstash/conf.d/. Fișierul de configurare este compus din 3 părți semnificative: INPUT, FILTER, OUTPUT. INPUT specificăm de unde va lua sistemul jurnalele, în FILTER parsing logs — configurăm cum să împărțim mesajul în câmpuri și valori, în OUTPUT configurăm fluxul de ieșire — unde vor fi trimise jurnalele parsate.
Mai întâi, să configurăm INPUT, vom examina câteva tipuri care pot exista — file, tcp și exe.
Tcp:
input {
tcp {
port => 5555
host => “10.10.1.205”
type => "checkpoint"
mode => "server"
}
}
mode => «server»
Indică faptul că Logstash acceptă conexiuni.
port => 5555
host => “10.10.1.205”
Acceptăm conexiuni pe adresa IP 10.10.1.205 (Logstash), portul 5555 — portul trebuie să fie permis de politica firewall-ului.
type => «checkpoint»
Etichetează documentul, foarte util în cazul în care aveți mai multe conexiuni de intrare. Ulterior, pentru fiecare conexiune, se poate scrie un filtru specific folosind construcția logică if.
File:
input {
file {
path => "\/var\/log\/openvas_report\/*"
type => "openvas"
start_position => "beginning"
}
}
Descrierea setărilor:
path => «\/var\/log\/openvas_report\/*»
Specificăm directorul în care trebuie să citim fișierele.
type => "openvas"
Tipul evenimentului.
start_position => "beginning"
Când se modifică fișierul, se citește fișierul în întregime; dacă se setează „end”, sistemul așteaptă apariția unor noi înregistrări la sfârșitul fișierului.
Exec:
input {
exec {
command => "ls -alh"
interval => 30
}
}
Prin acest input se lansează (numai!) comanda shell și outputul său este înfășurat într-un mesaj de log.
command => "ls -alh"
Comanda al cărei output ne interesează.
interval => 30
Intervalul de apelare a comenzii în secunde.
Pentru a primi loguri de la firewall, se configurează filtrul tcp sau udp, în funcție de modul în care sunt trimise logurile în Logstash.
Configurăm Output în fișierul de configurare Logstash în modul debug, pentru a înțelege cum arată mesajul de log.
După ce am configurat INPUT, trebuie să înțelegem cum va arăta mesajul de log și ce metode trebuie utilizate pentru a configura filtrul (parserul) logurilor.
Pentru aceasta, vom folosi un filtru care dă rezultatul în stdout pentru a vizualiza mesajul inițial; întreaga configurație a fișierului până în acest moment va arăta astfel:
input
{
tcp
{
port => 5555
type => "checkpoint"
mode => "server"
host => “10.10.1.205”
}
}
output
{
if [type] == "checkpoint"
{
stdout { codec=> json }
}
}
Executăm comanda pentru verificare:
sudo /usr/share/logstash/bin//logstash -f /etc/logstash/conf.d/checkpoint.conf
Vedem rezultatul, imaginea este clicabilă:
Dacă copiem, va arăta astfel:
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,
Privind la aceste mesaje, înțelegem că logurile au forma: câmp = valoare sau key = value, prin urmare se potrivește filtrul numit kv. Pentru a alege corect filtrul pentru fiecare situație specifică, ar fi bine să ne familiarizăm cu ele în documentația tehnică sau să întrebăm un prieten.
Configurăm Filtrul
În etapa anterioară am ales kv; mai jos este configurarea acestui filtru:
filter {
if [type] == "checkpoint"{
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
Alegem simbolul după care vom împărți câmpul și valoarea — “=”. Dacă avem în log aceleași înregistrări, păstrăm în bază doar un singur exemplat, altfel veți obține un array de valori identice; astfel, dacă avem mesajul “foo = some foo=some” scriem doar foo = some.
Configurăm Output-ul corect în ElasticSearch
După ce filtrul a fost configurat, putem exporta logurile în bază elasticsearch:
output
{
if [type] == "checkpoint"
{
elasticsearch
{
hosts => ["10.10.1.200:9200"]
index => "checkpoint-%{+YYYY.MM.dd}"
user => "tssolution"
password => "cool"
}
}
}
Dacă documentul este semnat cu tipul checkpoint, salvăm evenimentul în baza elasticsearch, care acceptă conexiuni pe 10.10.1.200 la portul 9200 în mod implicit. Fiecare document este salvat într-un index specific, în acest caz salvăm în indexul „checkpoint-” + data curentă. Fiecare index poate avea un set specific de câmpuri sau este creat automat la apariția unui nou câmp în mesaj, setările câmpurilor și tipurile lor pot fi văzute în mappings.
Dacă aveți configurată autentificarea (pe care o vom discuta mai târziu), trebuie să fie specifice credențiale pentru a scrie în indexul specific; în acest exemplu, acesta este „tssolution” cu parola „cool”. Se pot limita drepturile utilizatorilor pentru a scrie loguri doar într-un anumit index și nu în alte indexuri.
Lansăm Logstash.
Fișierul de configurare 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"
}
}
}
Verificăm fișierul de configurare pentru corectitudinea sa:
/usr/share/logstash/bin//logstash -f checkpoint.conf
Lansăm procesul Logstash:
sudo systemctl start logstash
Verificăm dacă procesul a fost lansat:
sudo systemctl status logstash
Verificăm dacă socketul s-a ridicat:
netstat -nat |grep 5555
Verificăm jurnalele în Kibana.
După ce totul este pornit, intrăm în Kibana — Discover, ne asigurăm că totul este configurat corect, imaginea este clicabilă!
Toate log-urile sunt în ordine și putem vedea toate câmpurile și valorile lor!
Concluzie
Am discutat despre cum să scriem un fișier de configurare Logstash, rezultând un parser pentru toate câmpurile și valorile. Acum putem lucra cu căutarea și crearea de grafice pe câteva câmpuri specifice. În următoarele lecții, vom explora vizualizarea în Kibana și vom crea un dashboard simplu. Merită menționat că fișierul de configurare Logstash trebuie completat constant în anumite situații, de exemplu, atunci când dorim să înlocuim o valoare a câmpului dintr-un număr în cuvânt. În articolele următoare, vom face asta în mod constant.
Așadar, urmăriți actualizările (, , , ), .
Sursa: habr.com
