U posljednjih smo se susreli ELK stog, od kojih se softverskih proizvoda sastoji. A prvi zadatak s kojim se inženjer suočava kada radi s ELK stogom je slanje zapisa za pohranu u elasticsearch radi naknadne analize. Međutim, ovo su samo riječi, elasticsearch pohranjuje zapisnike u obliku dokumenata s određenim poljima i vrijednostima, što znači da inženjer mora koristiti različite alate za raščlanjivanje poruke koja se šalje s krajnjih sustava. To možete učiniti na nekoliko načina - sami napišite program koji će pomoću API-ja dodavati dokumente u bazu ili koristite gotova rješenja. U ovom tečaju razmotrit ćemo rješenje logstash, koji je dio ELK skupa. Pogledat ćemo kako možemo slati zapise iz sustava krajnjih točaka u Logstash, a zatim ćemo postaviti konfiguracijsku datoteku za analizu i preusmjeravanje na Elasticsearch bazu podataka. Da bismo to učinili, uzimamo zapise iz vatrozida Check Point kao dolazni sustav.
Tečaj ne pokriva instalaciju ELK stacka, budući da postoji ogroman broj članaka o ovoj temi; razmotrit ćemo konfiguracijsku komponentu.
Napravimo akcijski plan za Logstash konfiguraciju:
- Provjera hoće li elasticsearch prihvatiti zapise (provjera funkcionalnosti i otvorenosti porta).
- Razmatramo kako možemo poslati događaje u Logstash, odabrati metodu i implementirati je.
- Konfiguriramo unos u Logstash konfiguracijskoj datoteci.
- Konfiguriramo izlaz u konfiguracijskoj datoteci Logstash u načinu rada za otklanjanje pogrešaka kako bismo razumjeli kako izgleda poruka dnevnika.
- Postavljanje filtra.
- Postavljanje ispravnog izlaza u ElasticSearch.
- Logstash se pokreće.
- Provjera dnevnika u Kibani.
Pogledajmo svaku točku detaljnije:
Provjera prihvaća li elasticsearch zapise
Da biste to učinili, možete koristiti naredbu curl za provjeru pristupa Elasticsearchu sa sustava na kojem je Logstash postavljen. Ako ste konfigurirali autentifikaciju, također prenosimo korisnika/lozinku putem curla, navodeći port 9200 ako ga niste promijenili. Ako dobijete odgovor sličan donjem, onda je sve u redu.
[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 ~]$
Ako se odgovor ne primi, može postojati nekoliko vrsta grešaka: proces elasticsearch se ne izvodi, naveden je pogrešan port ili je port blokiran vatrozidom na poslužitelju na kojem je instaliran elasticsearch.
Pogledajmo kako možete poslati zapise u Logstash s vatrozida kontrolne točke
S poslužitelja za upravljanje Check Point možete slati zapise u Logstash putem syslog-a pomoću uslužnog programa log_exporter, više o tome možete pročitati ovdje , ovdje ćemo ostaviti samo naredbu koja kreira stream:
cp_log_export dodaj naziv check_point_syslog target-server < > target-port 5555 protokol tcp format generički način čitanja poluunificiran
< > - adresa servera na kojem radi Logstash, target-port 5555 - port na koji ćemo slati logove, slanje logova preko tcp-a može opteretiti server pa je u nekim slučajevima ispravnije koristiti udp.
Postavljanje INPUT-a u Logstash konfiguracijskoj datoteci
Prema zadanim postavkama, konfiguracijska datoteka se nalazi u direktoriju /etc/logstash/conf.d/. Konfiguracijska datoteka se sastoji od 3 smislena dijela: INPUT, FILTER, OUTPUT. U INPUT označavamo odakle će sustav uzimati zapise, u FILTER parse the log - postavite kako podijeliti poruku u polja i vrijednosti, u PROIZVODNJA konfiguriramo izlazni tok - gdje će se poslati raščlanjeni dnevnici.
Prvo, konfigurirajmo INPUT, razmotrimo neke vrste koje mogu biti - datoteka, tcp i exe.
TCP:
input {
tcp {
port => 5555
host => “10.10.1.205”
type => "checkpoint"
mode => "server"
}
}
način => "poslužitelj"
Označava da Logstash prihvaća veze.
port => 5555
domaćin => “10.10.1.205”
Prihvaćamo veze putem IP adrese 10.10.1.205 (Logstash), port 5555 - port mora biti dopušten prema pravilima vatrozida.
tip => "kontrolna točka"
Označavamo dokument, što je vrlo zgodno ako imate nekoliko dolaznih veza. Nakon toga, za svaku vezu možete napisati vlastiti filtar pomoću logičke if konstrukcije.
File:
input {
file {
path => "/var/log/openvas_report/*"
type => "openvas"
start_position => "beginning"
}
}
Opis postavki:
put => "/var/log/openvas_report/*"
Označavamo direktorij u kojem se datoteke trebaju čitati.
tip => "openvas"
Vrsta događaja.
start_position => "početak"
Prilikom mijenjanja datoteke, čita cijelu datoteku, ako postavite “end”, sustav čeka da se novi zapisi pojave na kraju datoteke.
Izvršitelj:
input {
exec {
command => "ls -alh"
interval => 30
}
}
Koristeći ovaj unos, pokreće se (samo!) naredba ljuske i njen se izlaz pretvara u poruku dnevnika.
naredba => "ls -alh"
Naredba čiji nas izlaz zanima.
interval => 30
Interval pozivanja naredbe u sekundama.
Kako bismo primali zapise s vatrozida, registriramo filtar tcp ili pud, ovisno o tome kako se dnevnici šalju u Logstash.
Konfiguriramo izlaz u Logstash konfiguracijskoj datoteci u načinu rada za otklanjanje pogrešaka kako bismo razumjeli kako izgleda poruka dnevnika
Nakon što smo konfigurirali INPUT, moramo razumjeti kako će poruka dnevnika izgledati i koje metode je potrebno koristiti za konfiguraciju filtera dnevnika (parser).
Da bismo to učinili, upotrijebit ćemo filtar koji šalje rezultat u stdout kako bismo vidjeli izvornu poruku; kompletna 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 }
}
}
Pokrenite naredbu za provjeru:
sudo /usr/share/logstash/bin//logstash -f /etc/logstash/conf.d/checkpoint.conf
Vidimo rezultat, slika se može kliknuti:
Ako ga 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="[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,
Gledajući ove poruke, shvaćamo da dnevnici izgledaju ovako: polje = vrijednost ili ključ = vrijednost, što znači da je filter nazvan kv prikladan. Kako biste odabrali pravi filter za svaki konkretan slučaj, bilo bi dobro da se s njima upoznate u tehničkoj dokumentaciji ili pitate prijatelja.
Postavljanje filtra
U posljednjoj fazi odabrali smo kv, konfiguracija ovog filtra prikazana je u nastavku:
filter {
if [type] == "checkpoint"{
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
Odaberemo simbol kojim ćemo podijeliti polje i vrijednost - “=”. Ako imamo identične unose u dnevniku, spremamo samo jednu instancu u bazu podataka, inače ćete završiti s nizom identičnih vrijednosti, odnosno, ako imamo poruku “foo = some foo=some” pišemo samo foo = neki.
Postavljanje ispravnog izlaza u ElasticSearch
Nakon što je filter konfiguriran, možete učitati zapise u bazu podataka elastično pretraživanje:
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 kontrolne točke, spremamo događaj u bazu podataka elasticsearch, koja prema zadanim postavkama prihvaća veze na 10.10.1.200 na portu 9200. Svaki dokument se sprema u određeni indeks, u ovom slučaju spremamo u indeks “kontrolna točka-” + trenutni datum datum. Svaki indeks može imati određeni skup polja ili se kreira automatski kada se novo polje pojavi u poruci; postavke polja i njihov tip mogu se vidjeti u preslikavanjima.
Ako imate konfiguriranu provjeru autentičnosti (pogledat ćemo je kasnije), vjerodajnice za pisanje u određeni indeks moraju biti navedene, u ovom primjeru to je “tssolution” s lozinkom “cool”. Možete razlikovati korisnička prava za pisanje dnevnika samo u određeni indeks i ništa više.
Pokrenite Logstash.
Logstash konfiguracijska datoteka:
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 ispravnost konfiguracijske datoteke:
/usr/share/logstash/bin//logstash -f checkpoint.conf
Pokrenite Logstash proces:
sudo systemctl pokreni logstash
Provjeravamo je li proces započeo:
sudo systemctl status logstash
Provjerimo je li utičnica uključena:
netstat -nat |grep 5555
Provjera dnevnika u Kibani.
Nakon što se sve pokrene, idite na Kibana - Discover, provjerite je li sve ispravno konfigurirano, slika se može kliknuti!
Svi zapisnici su na mjestu i možemo vidjeti sva polja i njihove vrijednosti!
Zaključak
Pogledali smo kako napisati Logstash konfiguracijsku datoteku, i kao rezultat dobili smo parser svih polja i vrijednosti. Sada možemo raditi s pretraživanjem i crtanjem za određena polja. Zatim ćemo u tečaju pogledati vizualizaciju u Kibani i izraditi jednostavnu nadzornu ploču. Vrijedno je spomenuti da Logstash konfiguracijsku datoteku treba stalno ažurirati u određenim situacijama, primjerice kada želimo zamijeniti vrijednost polja iz broja u riječ. U narednim člancima to ćemo stalno činiti.
Zato ostanite s nama, , , ), .
Izvor: www.habr.com