V minulosti jsme se potkali ELK zásobník, z jakých softwarových produktů se skládá. A prvním úkolem, kterému inženýr čelí při práci se zásobníkem ELK, je odesílání protokolů pro uložení v elasticsearch pro následnou analýzu. To je však jen řeč, elasticsearch ukládá protokoly ve formě dokumentů s určitými poli a hodnotami, což znamená, že technik musí použít různé nástroje k analýze zprávy odesílané z koncových systémů. To lze provést několika způsoby – sami si napište program, který bude dokumenty do databáze přidávat pomocí API, nebo použijte hotová řešení. V tomto kurzu se podíváme na řešení Logstash, který je součástí zásobníku ELK. Podíváme se na to, jak můžeme odesílat protokoly z koncových systémů do Logstashe, a poté nastavíme konfigurační soubor, který bude analyzovat a přesměrovat do databáze Elasticsearch. Abychom toho dosáhli, bereme jako příchozí systém protokoly z firewallu Check Point.
Kurz se nezabývá instalací ELK stacku, protože na toto téma existuje velké množství článků, budeme zvažovat konfigurační komponentu.
Pojďme sestavit akční plán pro konfiguraci Logstash:
- Kontrola, že elasticsearch bude přijímat protokoly (kontrola funkčnosti a otevřenosti portu).
- Zvažujeme, jak můžeme odesílat události do Logstashe, zvolit metodu a implementovat ji.
- Vstup nakonfigurujeme v konfiguračním souboru Logstash.
- Výstup nakonfigurujeme v konfiguračním souboru Logstash v režimu ladění, abychom pochopili, jak vypadá zpráva protokolu.
- Nastavení filtru.
- Nastavení správného výstupu v ElasticSearch.
- Logstash se spustí.
- Kontrola protokolů v Kibaně.
Podívejme se na každý bod podrobněji:
Kontrola, že elasticsearch bude přijímat protokoly
Chcete-li to provést, můžete použít příkaz curl ke kontrole přístupu k Elasticsearch ze systému, na kterém je Logstash nasazen. Pokud máte nakonfigurovanou autentizaci, pak také přeneseme uživatele/heslo přes curl s uvedením portu 9200, pokud jste jej nezměnili. Pokud obdržíte odpověď podobnou té níže, pak je vše v pořádku.
[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 ~]$
Pokud odpověď neobdržíte, může se jednat o několik typů chyb: proces elasticsearch neběží, je zadán nesprávný port nebo je port blokován bránou firewall na serveru, kde je elasticsearch nainstalován.
Podívejme se, jak můžete odeslat protokoly do Logstashe z firewallu kontrolního bodu
Ze serveru pro správu Check Point můžete posílat protokoly na Logstash přes syslog pomocí nástroje log_exporter, více si o tom můžete přečíst zde , zde ponecháme pouze příkaz, který vytvoří stream:
cp_log_export přidat název check_point_syslog cílový-server < > cílový port 5555 protokol tcp formát generický režim čtení polosjednocený
< > - adresa serveru, na kterém běží Logstash, cílový port 5555 - port, na který budeme posílat logy, odesílání logů přes tcp může zatížit server, takže v některých případech je správnější použít udp.
Nastavení INPUT v konfiguračním souboru Logstash
Ve výchozím nastavení je konfigurační soubor umístěn v adresáři /etc/logstash/conf.d/. Konfigurační soubor se skládá ze 3 smysluplných částí: INPUT, FILTER, OUTPUT. V VSTUP uvedeme, odkud bude systém brát logy, v FILTER parse the log - nastavení způsobu rozdělení zprávy na pole a hodnoty, in VÝSTUP konfigurujeme výstupní proud - kam se budou odesílat analyzované protokoly.
Nejprve nakonfigurujeme INPUT, zvažte některé typy, které mohou být - soubor, tcp a exe.
TCP:
input {
tcp {
port => 5555
host => “10.10.1.205”
type => "checkpoint"
mode => "server"
}
}
režim => "server"
Označuje, že Logstash přijímá připojení.
port => 5555
hostitel => "10.10.1.205"
Přijímáme připojení přes IP adresu 10.10.1.205 (Logstash), port 5555 - port musí být povolen zásadami brány firewall.
typ => "kontrolní bod"
Označíme dokument, velmi výhodné, pokud máte několik příchozích spojení. Následně si pro každé připojení můžete napsat svůj vlastní filtr pomocí logické konstrukce if.
Soubor:
input {
file {
path => "/var/log/openvas_report/*"
type => "openvas"
start_position => "beginning"
}
}
Popis nastavení:
cesta => "/var/log/openvas_report/*"
Označíme adresář, ve kterém je třeba soubory načíst.
typ => "openvas"
Typ události.
start_position => "začátek"
Při změně souboru načte celý soubor, pokud nastavíte „end“, systém čeká, až se na konci souboru objeví nové záznamy.
Exec:
input {
exec {
command => "ls -alh"
interval => 30
}
}
Pomocí tohoto vstupu se spustí (pouze!) příkaz shellu a jeho výstup se změní na zprávu protokolu.
příkaz => "ls -alh"
Příkaz, jehož výstup nás zajímá.
interval => 30
Interval vyvolání příkazu v sekundách.
Abychom mohli přijímat protokoly z firewallu, registrujeme filtr tcp nebo udp, v závislosti na tom, jak jsou protokoly odesílány do Logstash.
Výstup nakonfigurujeme v konfiguračním souboru Logstash v režimu ladění, abychom pochopili, jak vypadá zpráva protokolu
Poté, co jsme nakonfigurovali INPUT, musíme pochopit, jak bude zpráva protokolu vypadat a jaké metody je třeba použít ke konfiguraci filtru protokolu (analyzátoru).
K tomu použijeme filtr, který vypíše výsledek na stdout, aby se zobrazila původní zpráva; kompletní konfigurační soubor bude v tuto chvíli vypadat takto:
input
{
tcp
{
port => 5555
type => "checkpoint"
mode => "server"
host => “10.10.1.205”
}
}
output
{
if [type] == "checkpoint"
{
stdout { codec=> json }
}
}
Spusťte příkaz pro kontrolu:
sudo /usr/share/logstash/bin//logstash -f /etc/logstash/conf.d/checkpoint.conf
Vidíme výsledek, na obrázek se dá kliknout:
Pokud to zkopírujete, bude to vypadat 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,
Při pohledu na tyto zprávy chápeme, že protokoly vypadají takto: pole = hodnota nebo klíč = hodnota, což znamená, že je vhodný filtr nazvaný kv. Abyste si vybrali ten správný filtr pro každý konkrétní případ, bylo by dobré se s nimi seznámit v technické dokumentaci, případně se zeptat známého.
Nastavení filtru
V poslední fázi jsme vybrali kv, konfigurace tohoto filtru je uvedena níže:
filter {
if [type] == "checkpoint"{
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
Vybereme symbol, kterým rozdělíme pole a hodnotu - „=“. Pokud máme v logu identické záznamy, uložíme do databáze pouze jednu instanci, jinak skončíte s polem identických hodnot, to znamená, že pokud máme hlášku “foo = some foo=some” zapíšeme pouze foo = nějaké.
Nastavení správného výstupu v ElasticSearch
Jakmile je filtr nakonfigurován, můžete nahrát protokoly do databáze elasticsearch:
output
{
if [type] == "checkpoint"
{
elasticsearch
{
hosts => ["10.10.1.200:9200"]
index => "checkpoint-%{+YYYY.MM.dd}"
user => "tssolution"
password => "cool"
}
}
}
Pokud je dokument podepsán typem kontrolního bodu, uložíme událost do databáze elasticsearch, která standardně přijímá připojení 10.10.1.200 na portu 9200. Každý dokument se ukládá do určitého indexu, v tomto případě ukládáme do indexu „checkpoint-“ + aktuální čas datum. Každý index může mít specifickou sadu polí nebo je vytvořen automaticky, když se ve zprávě objeví nové pole; nastavení polí a jejich typ lze zobrazit v mapováních.
Pokud máte nastavenou autentizaci (podíváme se na to později), musí být zadány přihlašovací údaje pro zápis do konkrétního indexu, v tomto příkladu je to „tssolution“ s heslem „cool“. Můžete rozlišovat uživatelská práva k zápisu protokolů pouze do určitého indexu a nic víc.
Spusťte Logstash.
Konfigurační soubor 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ávnost konfiguračního souboru:
/usr/share/logstash/bin//logstash -f checkpoint.conf
Spusťte proces Logstash:
sudo systemctl start logstash
Zkontrolujeme, zda proces začal:
sudo systemctl status logstash
Zkontrolujeme, zda je zásuvka aktivní:
netstat -nat |grep 5555
Kontrola protokolů v Kibaně.
Poté, co vše běží, přejděte na Kibana - Objevte, ujistěte se, že je vše správně nakonfigurováno, obrázek je klikatelný!
Všechny protokoly jsou na svém místě a vidíme všechna pole a jejich hodnoty!
Závěr
Podívali jsme se na to, jak napsat konfigurační soubor Logstash, a jako výsledek jsme dostali analyzátor všech polí a hodnot. Nyní můžeme pracovat s vyhledáváním a vykreslováním pro konkrétní pole. Dále se v kurzu podíváme na vizualizaci v Kibaně a vytvoříme si jednoduchý dashboard. Za zmínku stojí, že konfigurační soubor Logstash je potřeba v určitých situacích neustále aktualizovat, například když chceme nahradit hodnotu pole z čísla na slovo. V dalších článcích to budeme dělat neustále.
Takže zůstaňte naladěni (, , , ), .
Zdroj: www.habr.com