V poslednom sme sa stretli zásobník ELK, z akých softvérových produktov sa skladá. A prvou úlohou, ktorej inžinier čelí pri práci so zásobníkom ELK, je odosielanie protokolov na uloženie v elasticsearch na následnú analýzu. Toto je však len slovná služba, elasticsearch ukladá protokoly vo forme dokumentov s určitými poľami a hodnotami, čo znamená, že inžinier musí použiť rôzne nástroje na analýzu správy odoslanej z koncových systémov. Dá sa to urobiť niekoľkými spôsobmi – napíšte si sami program, ktorý bude pridávať dokumenty do databázy pomocou API, alebo použite hotové riešenia. V tomto kurze zvážime riešenie Logstash, ktorý je súčasťou zásobníka ELK. Pozrieme sa na to, ako môžeme posielať protokoly z koncových systémov do Logstash, a potom nastavíme konfiguračný súbor na analýzu a presmerovanie do databázy Elasticsearch. Aby sme to dosiahli, berieme protokoly z firewallu Check Point ako prichádzajúci systém.
Kurz nezahŕňa inštaláciu zásobníka ELK, pretože na túto tému existuje veľké množstvo článkov; zvážime konfiguračný komponent.
Zostavme akčný plán pre konfiguráciu Logstash:
- Kontrola, či elasticsearch bude akceptovať protokoly (kontrola funkčnosti a otvorenosti portu).
- Zvažujeme, ako môžeme odosielať udalosti do Logstashe, zvoliť metódu a implementovať ju.
- Vstup nakonfigurujeme v konfiguračnom súbore Logstash.
- Výstup konfigurujeme v konfiguračnom súbore Logstash v režime ladenia, aby sme pochopili, ako vyzerá správa protokolu.
- Nastavenie filtra.
- Nastavenie správneho výstupu v ElasticSearch.
- Spustí sa Logstash.
- Kontrola denníkov v Kibane.
Pozrime sa na každý bod podrobnejšie:
Skontrolujte, či elasticsearch bude akceptovať protokoly
Ak to chcete urobiť, môžete použiť príkaz curl na kontrolu prístupu k Elasticsearch zo systému, na ktorom je Logstash nasadený. Ak máte nakonfigurovanú autentifikáciu, prenesieme používateľa/heslo aj cez curl, špecifikujúc port 9200, ak ste ho nezmenili. Ak dostanete odpoveď podobnú tej nižšie, potom je všetko v poriadku.
[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 ~]$
Ak odpoveď nedostanete, môže ísť o niekoľko typov chýb: proces elasticsearch nebeží, je zadaný nesprávny port alebo je port zablokovaný bránou firewall na serveri, kde je elasticsearch nainštalovaný.
Pozrime sa, ako môžete odosielať protokoly do Logstash z brány firewall kontrolného bodu
Zo servera správy Check Point môžete posielať protokoly do Logstash cez syslog pomocou utility log_exporter, viac si o tom môžete prečítať tu , tu ponecháme iba príkaz, ktorý vytvorí stream:
cp_log_export pridať názov check_point_syslog cieľový-server < > cieľový port 5555 protokol tcp formát generický režim čítania polozjednotený
< > - adresa servera, na ktorom beží Logstash, cieľový port 5555 - port, na ktorý budeme posielať protokoly, odosielanie protokolov cez tcp môže zaťažiť server, takže v niektorých prípadoch je správnejšie použiť udp.
Nastavenie INPUT v konfiguračnom súbore Logstash
Štandardne sa konfiguračný súbor nachádza v adresári /etc/logstash/conf.d/. Konfiguračný súbor pozostáva z 3 zmysluplných častí: INPUT, FILTER, OUTPUT. IN INPUT uvádzame, odkiaľ bude systém brať logy, v FILTER analyzovať protokol - nastaviť spôsob rozdelenia správy na polia a hodnoty, v VÝKON konfigurujeme výstupný tok - kam sa budú odosielať analyzované protokoly.
Najprv nakonfigurujme INPUT, zvážte niektoré typy, ktoré môžu byť - súbor, tcp a exe.
TCP:
input {
tcp {
port => 5555
host => “10.10.1.205”
type => "checkpoint"
mode => "server"
}
}
režim => "server"
Označuje, že Logstash prijíma pripojenia.
port => 5555
hostiteľ => "10.10.1.205"
Akceptujeme pripojenia cez IP adresu 10.10.1.205 (Logstash), port 5555 - port musí byť povolený politikou brány firewall.
typ => "kontrolný bod"
Označíme dokument, veľmi výhodné, ak máte niekoľko prichádzajúcich spojení. Následne si pre každé pripojenie môžete napísať vlastný filter pomocou konštrukcie logického if.
Súbor:
input {
file {
path => "/var/log/openvas_report/*"
type => "openvas"
start_position => "beginning"
}
}
Popis nastavení:
cesta => "/var/log/openvas_report/*"
Označíme adresár, v ktorom je potrebné čítať súbory.
typ => "openvas"
Typ udalosti.
start_position => "začiatok"
Pri zmene súboru načíta celý súbor, ak nastavíte „koniec“, systém čaká, kým sa na konci súboru objavia nové záznamy.
Exec:
input {
exec {
command => "ls -alh"
interval => 30
}
}
Pomocou tohto vstupu sa spustí (len!) príkaz shellu a jeho výstup sa zmení na protokolovú správu.
príkaz => "ls -alh"
Príkaz, ktorého výstup nás zaujíma.
interval => 30
Interval vyvolania príkazu v sekundách.
Aby sme mohli prijímať protokoly z brány firewall, registrujeme filter tcp alebo udpv závislosti od toho, ako sa protokoly odosielajú do Logstash.
Výstup konfigurujeme v konfiguračnom súbore Logstash v režime ladenia, aby sme pochopili, ako vyzerá správa protokolu
Potom, čo sme nakonfigurovali INPUT, musíme pochopiť, ako bude správa protokolu vyzerať a aké metódy je potrebné použiť na konfiguráciu filtra protokolu (analyzátora).
Na tento účel použijeme filter, ktorý odošle výsledok do stdout, aby sa zobrazila pôvodná správa; kompletný konfiguračný súbor bude momentálne vyzerať takto:
input
{
tcp
{
port => 5555
type => "checkpoint"
mode => "server"
host => “10.10.1.205”
}
}
output
{
if [type] == "checkpoint"
{
stdout { codec=> json }
}
}
Spustite príkaz a skontrolujte:
sudo /usr/share/logstash/bin//logstash -f /etc/logstash/conf.d/checkpoint.conf
Vidíme výsledok, na obrázok sa dá kliknúť:
Ak to skopírujete, bude to vyzerať takto:
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,
Pri pohľade na tieto správy chápeme, že protokoly vyzerajú takto: pole = hodnota alebo kľúč = hodnota, čo znamená, že je vhodný filter s názvom kv. Aby ste si vybrali ten správny filter pre každý konkrétny prípad, bolo by dobré sa s nimi oboznámiť v technickej dokumentácii, prípadne sa opýtať známeho.
Nastavenie filtra
V poslednej fáze sme vybrali kv, konfigurácia tohto filtra je uvedená nižšie:
filter {
if [type] == "checkpoint"{
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
Vyberieme symbol, ktorým rozdelíme pole a hodnotu - „=“. Ak máme v logu identické záznamy, uložíme do databázy iba jednu inštanciu, inak skončíme s poľom identických hodnôt, to znamená, že ak máme hlásenie “foo = some foo=some” napíšeme len foo = nejaký.
Nastavenie správneho výstupu v ElasticSearch
Po nakonfigurovaní filtra môžete nahrať protokoly do databázy ElasticSearch:
output
{
if [type] == "checkpoint"
{
elasticsearch
{
hosts => ["10.10.1.200:9200"]
index => "checkpoint-%{+YYYY.MM.dd}"
user => "tssolution"
password => "cool"
}
}
}
Ak je dokument podpísaný typom kontrolného bodu, udalosť uložíme do databázy elasticsearch, ktorá štandardne prijíma spojenia 10.10.1.200 na porte 9200. Každý dokument sa ukladá do konkrétneho indexu, v tomto prípade ukladáme do indexu „kontrolný bod-“ + aktuálny čas dátum. Každý index môže mať špecifickú množinu polí alebo sa vytvorí automaticky, keď sa v správe objaví nové pole; nastavenia polí a ich typ je možné zobraziť v mapovaniach.
Ak máte nakonfigurovanú autentifikáciu (pozrieme sa na to neskôr), musíte zadať poverenia pre zápis do konkrétneho indexu, v tomto príklade je to „tssolution“ s heslom „cool“. Môžete rozlišovať práva používateľov na zapisovanie protokolov iba do konkrétneho indexu a nie viac.
Spustite Logstash.
Konfiguračný súbor 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"
}
}
}
Kontrolujeme správnosť konfiguračného súboru:
/usr/share/logstash/bin//logstash -f checkpoint.conf
Spustite proces Logstash:
sudo systemctl spustiť logstash
Skontrolujeme, či sa proces začal:
sudo systemctl status logstash
Skontrolujeme, či je zásuvka zapnutá:
netstat -nat |grep 5555
Kontrola denníkov v Kibane.
Keď všetko beží, prejdite na Kibana - Objavte, uistite sa, že je všetko správne nakonfigurované, na obrázok sa dá kliknúť!
Všetky protokoly sú na svojom mieste a vidíme všetky polia a ich hodnoty!
Záver
Pozreli sme sa na to, ako napísať konfiguračný súbor Logstash, a ako výsledok sme dostali analyzátor všetkých polí a hodnôt. Teraz môžeme pracovať s vyhľadávaním a vykresľovaním konkrétnych polí. Ďalej sa v kurze pozrieme na vizualizáciu v Kibane a vytvoríme si jednoduchý dashboard. Za zmienku stojí, že konfiguračný súbor Logstash je potrebné v určitých situáciách neustále aktualizovať, napríklad keď chceme nahradiť hodnotu poľa z čísla na slovo. V ďalších článkoch to budeme robiť neustále.
Takže zostaňte naladení (, , , ), .
Zdroj: hab.com