U prošloj Upoznali smo se s ELK stakom, iz kojih se sastoji. Prvi izazov s kojim se inženjer suočava prilikom rada s ELK stakom je slanje logova na pohranu u Elasticsearch za naknadnu analizu. Međutim, to je jednostavno u teoriji, Elasticsearch pohranjuje logove u obliku dokumenata s određenim poljima i vrijednostima, što znači da inženjer mora koristiti razne alate za parsiranje poruke koja se šalje iz krajnjih sustava. To se može učiniti na nekoliko načina: sami napisati program koji će putem API-ja dodavati dokumente u bazu ili koristiti već gotova rješenja. U okviru ovog tečaja razmatrat ćemo rješenje Logstash, koje je dio ELK staka. Pogledat ćemo kako možemo slati logove iz krajnjih sustava u Logstash, a zatim ćemo konfigurirati konfiguracijski datoteku za parsiranje i preusmjeravanje u Elasticsearch bazu podataka. Za to uzimamo logove s Check Point vatrozida kao ulazni sustav.
Ovaj tečaj ne pokriva instalaciju ELK staka, jer postoji ogromna količina članaka na tu temu, a mi ćemo razmatrati konfiguracijsku komponentu.
Sastavit ćemo plan akcije za konfiguraciju Logstasha:
- Provjera da li će Elasticsearch primati logove (provjera funkcionalnosti i dostupnosti porta).
- Razmatramo kako možemo slati događaje u Logstash, biramo metodu, realiziramo.
- Konfiguriramo Input u konfiguracijskoj datoteci Logstasha.
- Konfiguriramo Output u konfiguracijskoj datoteci Logstasha u debug modu, kako bismo shvatili kako izgleda log poruka.
- Konfiguriramo Filter.
- Konfiguriramo ispravni Output u Elasticsearch.
- Pokrećemo Logstash.
- Provjeravamo logove u Kibani.
Detaljnije razmatramo svaku stavku:
Provjera da li će Elasticsearch primati logove
Za to možemo provjeriti dostupnost u Elasticsearchu iz sustava na kojem je postavljen Logstash pomoću naredbe curl. Ako ste postavili autentifikaciju, tada putem curla također šaljemo korisničko/lozinku i navodimo port 9200 ako ga niste mijenjali. Ako stigne odgovor poput onog koji je naveden u nastavku, znači da je sve u redu.
[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" : "Znate, za pretraživanje"
}
[elastic@elasticsearch ~]$
Ako odgovor nije stigao, može biti nekoliko vrsta grešaka: proces elasticsearch ne radi, port nije ispravan, ili je port blokiran vatrozidom na serveru gdje je instaliran elasticsearch.
Razmatramo kako poslati logove u Logstash s vatrozida Check Point.
S Check Point management serverom možete slati logove u Logstash preko sysloga, koristeći alat log_exporter, više detalja dostupno je ovdje. , ovdje ćemo ostaviti samo naredbu koja stvara tok:
cp_log_export add name check_point_syslog target-server <> target-port 5555 protocol tcp format generic read-mode semi-unified
<> — adresa servera na kojem radi Logstash, target-port 5555 — port na koji ćemo slati logove, slanje logova preko tcp može opteretiti server, stoga je u nekim slučajevima bolje koristiti udp.
Postavljamo INPUT u konfiguracijskoj datoteci Logstash.
Zadana konfiguracijska datoteka nalazi se u direktoriju /etc/logstash/conf.d/. Konfiguracijska datoteka sastoji se od tri smislene komponente: INPUT, FILTER, OUTPUT. U INPUT navodimo odakle sustav treba uzimati logove, u FILTER parsiramo log — postavljamo način dijeljenja poruke na polja i vrijednosti, u OUTPUT konfiguriramo izlazni tok — gdje će obrađeni logovi biti slani.
Prvo ćemo postaviti INPUT, razmotriti nekoliko tipova koji mogu biti — file, tcp i exe.
Tcp:
input {
tcp {
port => 5555
host => "10.10.1.205"
type => "checkpoint"
mode => "server"
}
}
mode => «server»
Ovo označava da Logstash prihvaća veze.
port => 5555
host => "10.10.1.205"
Prihvaćamo veze na IP adresi 10.10.1.205 (Logstash), port 5555 — port mora biti dopušten politikom vatrozida.
type => «checkpoint»
Označavamo dokument, što je vrlo korisno u slučaju kada imate više dolaznih veza. Kasnije za svaku vezu možete napisati svoj filtar koristeći logičku konstrukciju if.
File:
input {
file {
path => "/var/log/openvas_report/*"
type => "openvas"
start_position => "beginning"
}
}
Opis postavki:
path => "/var/log/openvas_report/*"
Odredite direktorij u kojem se nalaze datoteke koje treba pročitati.
type => "openvas"
Tip događaja.
start_position => "početak"
Pri promjeni datoteke, cijela se datoteka učitava, a ako se postavi "kraj", sustav čeka pojavu novih zapisa na kraju datoteke.
Exec:
input {
exec {
command => "ls -alh"
interval => 30
}
}
Ovaj ulaz pokreće (samo!) shell naredbu i njezin izlaz obavija u log poruku.
command => "ls -alh"
Naredba čiji nas izlaz zanima.
interval => 30
Interval poziva naredbe u sekundama.
Da bismo primili logove s vatrozida, postavljamo filtar tcp ili udp, ovisno o tome kako se logovi šalju u Logstash.
Postavljamo Output u konfiguracijskoj datoteci Logstash u debug modu kako bismo razumjeli kako izgleda log poruka.
Nakon što smo configurirali INPUT, trebamo razumjeti kako će izgledati log poruka i koje metode trebamo koristiti za konfiguraciju filtra (parsera) logova.
Za to ćemo koristiti filtar koji daje rezultat u stdout kako bismo pregledali izvornu poruku, cijela konfiguracijska datoteka u ovom trenutku izgledat će ovako:
input
{
tcp
{
port => 5555
type => "checkpoint"
mode => "server"
host => "10.10.1.205"
}
}
output
{
if [type] == "checkpoint"
{
stdout { codec=> json }
}
}
Pokrećemo naredbu za provjeru:
sudo /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/checkpoint.conf
Vidimo rezultat, slika je klikabilna:
Ako kopirate, izgledat će ovako:
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,
Gledajući ove poruke, shvaćamo da logovi imaju oblik: polje = vrijednost ili ključ = vrijednost, pa stoga odgovara filtru koji se zove kv. Da bismo ispravno odabrali filtar za svaku konkretnu situaciju, bilo bi korisno upoznati se s njima u tehničkoj dokumentaciji ili pitati prijatelja.
Postavljamo Filter
Na prethodnom koraku odabrali smo kv, dalje je prikazana konfiguracija ovog filtra:
filter {
if [type] == "checkpoint"{
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
Biramo simbol po kojem ćemo dijeliti polje i vrijednost — “=”. Ako imamo iste zapise u logu, čuvamo samo jedan primjerak u bazi, inače ćemo imati niz istih vrijednosti, to jest, ako imamo poruku “foo = some foo=some” zapisujemo samo foo = some.
Postavljamo ispravan Output u ElasticSearch
Nakon što je Filter postavljen, možemo izvoziti logove u bazu elasticsearch:
output
{
if [type] == "checkpoint"
{
elasticsearch
{
hosts => ["10.10.1.200:9200"]
index => "checkpoint-%{+YYYY.MM.dd}"
user => "tssolution"
password => "cool"
}
}
}
Ako je dokument potpisan tipom checkpoint, čuvamo događaj u bazi elasticsearch, koja prihvaća veze na 10.10.1.200 na portu 9200 po defaultu. Svaki dokument se čuva u određenom indeksu, u ovom slučaju čuvamo u indeks „checkpoint-” + trenutni datum. Svaki indeks može imati određeni skup polja, ili se automatski stvara pri pojavi novog polja u poruci, postavke polja i njihov tip možete pogledati u mappings.
Ako imate postavljenu autentifikaciju (o kojoj ćemo razgovarati kasnije), obavezno morate navesti akreditive za zapis u konkretni indeks, u ovom primjeru to je „tssolution” s lozinkom „cool”. Može se ograničiti prava korisnika za zapis logova samo u određeni indeks i nijedan drugi.
Pokrećemo Logstash.
Konfiguracijska datoteka 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"
}
}
}
Provjeravamo konfiguracijsku datoteku na ispravnost sastavljanja:
/usr/share/logstash/bin//logstash -f checkpoint.conf
Pokrećemo proces Logstash:
sudo systemctl start logstash
Provjeravamo je li proces pokrenut:
sudo systemctl status logstash
Provjeravamo je li otvoren socket:
netstat -nat |grep 5555
Provjeravamo logove u Kibani.
Nakon što je sve pokrenuto, idemo na Kibanu — Discover, uvjeravamo se da je sve pravilno postavljeno, slika je interaktivna!
Svi logovi su na svom mjestu i možemo vidjeti sva polja i njihove vrijednosti!
Zaključak
Razmotrili smo kako pisati konfiguracijski datoteku Logstash, kao rezultat smo dobili parser svih polja i vrijednosti. Sada možemo raditi s pretragom i izradom grafova za određena polja. U nastavku tečaja razmotrit ćemo vizualizaciju u Kibani, stvorit ćemo jednostavnu nadzornu ploču. Treba napomenuti da se konfiguracijska datoteka Logstash mora stalno dopisivati u određenim situacijama, na primjer, kada želimo zamijeniti vrijednost polja s broja na riječ. U budućim člancima to ćemo stalno raditi.
Stoga pratite novosti (, , , ), .
Izvor: habr.com
