V minulosti seznámili jsme se s ELK stackem, z jakých softwarových produktů se skládá. A první úkol, se kterým se inženýr při práci s ELK stackem setkává, je odeslání logů pro uložení do Elasticsearch pro následnou analýzu. Avšak, to je jen na slovech jednoduché, Elasticsearch uchovává logy jako dokumenty s určitými poli a hodnotami, což znamená, že inženýr musí pomocí různých nástrojů zpracovat zprávu, která je odesílána z koncových systémů. To lze provést několika způsoby – napsat program, který pomocí API přidává dokumenty do databáze, nebo použít již hotová řešení. V rámci tohoto kurzu budeme zvažovat řešení Logstash, které je součástí ELK stacku. Podíváme se, jak lze odeslat logy z koncových systémů do Logstash a poté nastavíme konfigurační soubor pro zpracování a přesměrování do databáze Elasticsearch. Jako vstupní systém bereme logy z firewallu Check Point.
V rámci kurzu nebude probírána instalace ELK stacku, protože existuje obrovské množství článků na toto téma, budeme se zaměřovat na konfiguraci.
Sestavíme plán akcí pro konfiguraci Logstash:
- Ověření, že Elasticsearch bude přijímat logy (kontrola funkčnosti a dostupnosti portu).
- Zvažujeme, jakým způsobem můžeme odeslat události do Logstash, vybereme metodu a realizujeme ji.
- Nastavujeme Input v konfiguračním souboru Logstash.
- Nastavujeme Output v konfiguračním souboru Logstash v režimu debug, abychom pochopili, jak vypadá log zpráva.
- Nastavujeme Filter.
- Nastavujeme správný Output v Elasticsearch.
- Spouštíme Logstash.
- Kontrolujeme logy v Kibana.
Podíváme se podrobněji na každý bod:
Ověření, že Elasticsearch bude přijímat logy
K tomu můžeme zkontrolovat příkazem curl přístup do Elasticsearch ze systému, na kterém je nasazen Logstash. Pokud máte nastavenu autentifikaci, pak také prostřednictvím curl přidáme uživatelské jméno/a heslo, uvádíme port 9200, pokud jste ho neměnili. Pokud přijde odpověď podobná níže uvedené, znamená to, že je vše v pořádku.
[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" : "You Know, for Search"
}
[elastic@elasticsearch ~]$
Pokud odpověď nedorazí, může být několik typů chyb: elastický proces nefunguje, je uveden nesprávný port nebo je port zablokován firewallem na serveru, kde běží Elasticsearch.
Podíváme se, jak můžeme posílat logy do Logstash přes firewall Check Point.
Z Management Serveru Check Point lze posílat logy do Logstash přes syslog, pomocí nástroje log_exporter, podrobněji si o něm můžete přečíst zde. , zde uvedeme jen příkaz, který vytváří tok:
cp_log_export add name check_point_syslog target-server <> target-port 5555 protocol tcp format generic read-mode semi-unified
<> — adresa serveru, na kterém běží Logstash, target-port 5555 — port, na který budeme posílat logy, odesílání logů přes tcp může zatížit server, proto v některých případech je lepší nastavit tcp.
Nastavujeme INPUT v konfiguračním souboru Logstash.
Konfigurační soubor se standardně nachází v adresáři /etc/logstash/conf.d/. Konfigurační soubor se skládá ze 3 logických částí: INPUT, FILTER, OUTPUT. V INPUT uvádíme, odkud si systém vezme logy, v FILTER parsujeme log — nastavujeme jak rozdělit zprávu na pole a hodnoty, v OUTPUT nastavujeme výstupní tok — kam budou odesílány zpracované logy.
Nejprve nastavíme INPUT, podíváme se na některé typy, které mohou být — file, tcp a exe.
TCP:
input {
tcp {
port => 5555
host => "10.10.1.205"
type => "checkpoint"
mode => "server"
}
}
mode => „server“
Označuje, že Logstash přijímá spojení.
port => 5555
host => "10.10.1.205"
Přijímáme spojení na IP adrese 10.10.1.205 (Logstash), port 5555 — port by měl být povolen politikou firewallu.
type => „checkpoint“
Označujeme dokument, což je velmi užitečné, pokud máte více příchozích spojení. V budoucnu pro každé spojení lze napsat vlastní filtr pomocí logické konstrukce if.
File:
input {
file {
path => "/var/log/openvas_report/*"
type => "openvas"
start_position => "beginning"
}
}
Popis nastavení:
path => "/var/log/openvas_report/*"
Uveďte adresář, ze kterého je třeba načíst soubory.
type => "openvas"
Typ události.
start_position => "beginning"
Při změně souboru se načte celý soubor. Pokud nastavíte "end", systém čeká na nové záznamy na konci souboru.
Exec:
input {
exec {
command => "ls -alh"
interval => 30
}
}
Tento vstup spouští (pouze!) shell příkaz a jeho výstup je obalen do log zprávy.
command => "ls -alh"
Příkaz, jehož výstup nás zajímá.
interval => 30
Interval spuštění příkazu v sekundách.
Abychom mohli přijímat logy z firewallu, nastavíme filtr tcp nebo udp, v závislosti na tom, jak jsou logy odesílány do Logstash.
Nastavujeme Output v konfiguračním souboru Logstash v režimu ladění, abychom pochopili, jak vypadá log zpráva.
Poté, co jsme nakonfigurovali INPUT, musíme pochopit, jak bude log zpráva vypadat, jaké metody je třeba použít pro nastavení filtru (parzera) logů.
K tomu použijeme filtr, který vydává výstup do stdout, abychom si prohlédli původní zprávu. Celý 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 }
}
}
Spouštíme příkaz pro kontrolu:
sudo /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/checkpoint.conf
Vidíme výsledek, obrázek je klikací:
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="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,"
Při pohledu na tato oznámení chápeme, že logy mají formát: pole = hodnota nebo klíč = hodnota, což znamená, že se hodí filtr, který se nazývá kv. Abychom správně vybrali filtr pro každý konkrétní případ, bylo by dobré se s nimi seznámit v technické dokumentaci nebo sezeptat přítele.
Nastavujeme filtr
V předchozím kroku jsme zvolili kv, níže je uvedena konfigurace tohoto filtru:
filter {
if [type] == "checkpoint" {
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
Vyberáme symbol, podle kterého budeme oddělovat pole a hodnotu — "=". Pokud máme v logu stejné záznamy, uchováváme v databázi pouze jeden exemplář, jinak byste měli pole se stejnými hodnotami, což znamená, že pokud máme zprávu „foo = some foo=some“, zapíšeme pouze foo = some.
Nastavujeme správný výstup do ElasticSearch
Jakmile je filtr nastaven, můžete vyvážet logy 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 checkpoint, ukládáme událost do databáze elasticsearch, která standardně přijímá připojení na 10.10.1.200 na portu 9200. Každý dokument je ukládán do určitého indexu, v tomto případě ukládáme do indexu „checkpoint-” + aktuální časové datum. Každý index může mít určitou sadu polí, nebo je vytvořen automaticky při výskytu nového pole ve zprávě, nastavení polí a jejich typ lze zkontrolovat v mappings.
Pokud máte nastavenou autentizaci (probereme později), musíte uvést údaje pro zápis do konkrétního indexu, v tomto příkladu je to „tssolution“ s heslem „cool“. Můžete omezit práva uživatelů pro zápis logů pouze do určitého indexu a žádného dalšího.
Spouštíme 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 konfigurační soubor na správnost sestavení:
/usr/share/logstash/bin//logstash -f checkpoint.conf
Spouštíme proces Logstash:
sudo systemctl start logstash
Kontrolujeme, že proces byl spuštěn:
sudo systemctl status logstash
Kontrolujeme, zda se socket zvedl:
netstat -nat |grep 5555
Kontrolujeme logy v Kibana.
Poté, co je vše spuštěno, přejdeme do Kibana — Discover a ujistíme se, že je vše správně nastaveno, obrázek je klikací!
Všechny logy jsou na místě a můžeme vidět všechna pole a jejich hodnoty!
Závěr
Prozkoumali jsme, jak napsat konfigurační soubor Logstash, a výsledek byl parser všech polí a hodnot. Nyní můžeme pracovat s vyhledáváním a sestavováním grafů podle určitých polí. V dalším kurzu se podíváme na vizualizaci v Kibana a vytvoříme jednoduchý dashboard. Je třeba zmínit, že konfigurační soubor Logstash je potřeba průběžně doplňovat v určitých situacích, například když chceme změnit hodnotu pole z čísla na slovo. V následujících článcích to budeme dělat pravidelně.
Tak sledujte aktualizace (, , , ), .
Zdroj: habr.com
