🥇2. Elastic stack: analýza security logov. Logstash

2. Elastic stack: analýza bezpečnostných logov. Logstash

V minulosti článku stretli sme sa s ELK stackom, z ktorých softvérových produktov sa skladá. Prvá úloha, s ktorou sa inžinier stretáva pri práci s ELK stackom, je zasielanie logov na uloženie do Elasticsearch na následnú analýzu. Avšak, to je len teoreticky jednoduché, Elasticsearch uchováva logy vo forme dokumentov s určitými poliami a hodnotami, čo znamená, že inžinier musí pomocou rôznych nástrojov spracovať správu, ktorá sa posiela z koncových systémov. Môže to spraviť niekoľkými spôsobmi – buď napísať vlastný program, ktorý cez API pridáva dokumenty do databázy, alebo využiť už existujúce riešenia. V rámci tohto kurzu budeme skúmať riešenie Logstash, ktoré je súčasťou ELK stacku. Pozrieme sa, ako môžeme poslať logy z koncových systémov do Logstash, a potom budeme nastavovať konfiguračný súbor na spracovanie a presmerovanie do databázy Elasticsearch. Na tento účel vezmeme logy z firewallu Check Point ako vstupný systém.

V rámci kurzu sa nenastavuje ELK stack, pretože existuje obrovské množstvo článkov na túto tému, budeme sa sústrediť na konfiguračné aspekty.

Sformulujeme plán akcií na konfiguráciu Logstash:

Kontrola, že Elasticsearch bude prijímať logy (overenie funkčnosti a otvorenosti portu).
Zvážime, akým spôsobom môžeme zasielať udalosti do Logstash, vyberieme spôsob a zrealizujeme ho.
Nastavíme Input v konfiguračnom súbore Logstash.
Nastavíme Output v konfiguračnom súbore Logstash v režime debugu, aby sme pochopili, ako vyzerá logová správa.
Nastavíme Filter.
Nastavíme správny Output v ElasticSearch.
Spustíme Logstash.
Skontrolujeme logy v Kibana.

Podrobnejšie preskúmame každý bod:

Kontrola, že Elasticsearch bude prijímať logy

Na to môžete skontrolovať prístup do Elasticsearch z systému, na ktorom je nasadený Logstash, príkazom curl. Ak máte nastavenú autentifikáciu, cez curl tiež posielame používateľské meno/heslo a uvádzame port 9200, ak ste ho nezmenili. Ak dostanete odpoveď typu, aká je uvedená nižšie, všetko je v poriadku.

[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 ~]$

Ak nedošlo k odpovedi, môže byť niekoľko typov chýb: proces elasticsearch nefunguje, bola zadaná nesprávna portová adresa alebo je port blokovaný firewallom na serveri, kde sa nachádza elasticsearch.

Zvažujeme, ako poslať logy do Logstash cez firewall check point.

S Check Point management serverom je možné poslať logy do Logstash cez syslog, pričom sa používa nástroj log_exporter, s ktorým sa môžete podrobnejšie oboznámiť tu. článku, tu necháme len príkaz, ktorý vytvára 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 ktorom beží Logstash, target-port 5555 — port, na ktorý budeme posielať logy, posielanie logov po TCP môže zaťažiť server, preto je v niektorých prípadoch lepšie používať UDP.

Nastavujeme INPUT v konfiguračnom súbore Logstash.

2. Elastic stack: analýza bezpečnostných logov. Logstash

Predvolený konfiguračný súbor sa nachádza v adresári /etc/logstash/conf.d/. Konfiguračný súbor pozostáva z 3 významných častí: INPUT, FILTER, OUTPUT. INPUT uvádzame, odkiaľ systém vezme logy, v FILTER parsing log — nastavujeme, ako rozdeľovať správy na polia a hodnoty, v OUTPUT nastavujeme výstupný tok — kam budú odosielané rozparsované logy.

Najprv nastavíme INPUT, zvažujme niektoré typy, ktoré môžu byť — file, tcp a exe.

Tcp:

input {
tcp {
    port => 5555
    host => “10.10.1.205”
    type => "checkpoint"
    mode => "server"
}
}

mode => «server»
Hovorí o tom, že Logstash prijíma pripojenia.

port => 5555
host => “10.10.1.205”
Prijímame pripojenia na IP adrese 10.10.1.205 (Logstash), port 5555 — port by mal byť povolený politikou firewallu.

type => «checkpoint»
Označujeme dokument, čo je veľmi pohodlné v prípade, ak máte niekoľko prichádzajúcich pripojení. Pre každé pripojenie je možné napísať svoj filter pomocou logickej konštrukcie if.

File:

input {
  file {
    path => "\/var\/log\/openvas_report\/*"
    type => "openvas"
    start_position => "beginning"
    }
}

Popis nastavení:
path => «\/var\/log\/openvas_report\/*»
Zadáme adresár, v ktorom je potrebné čítať súbory.

type => "openvas"
Typ udalosti.

start_position => "beginning"
Pri zmene súboru sa súbor prečíta celý, ak nastavíte „end“, systém čaká na nové záznamy na konci súboru.

Exec:

input {
  exec {
    command => "ls -alh"
    interval => 30
  }
}

Na tejto vstupnej definícii sa spustí (iba!) shell príkaz a jeho výstup sa zabalí do logovej správy.

command => "ls -alh"
Príkaz, ktorého výstup nás zaujíma.

interval => 30
Interval volania príkazu v sekundách.

Aby sme mohli prijímať logy z firewallu, definujeme filter tcp alebo udp, v závislosti od toho, ako sú logy odosielané do Logstashi.

Nastavujeme výstup v konfiguračnom súbore Logstash v debug móde, aby sme pochopili, ako vyzerá logová správa.

Potom, čo sme nastavili INPUT, musíme pochopiť, ako bude vyzerať logová správa, aké metódy je potrebné použiť na nastavenie filtra (parsera) logov.

Na to použijeme filter, ktorý poskytuje výsledok do stdout, aby sme si mohli pozrieť pôvodnú správu; momentálne bude celý konfiguračný súbor vyzerať takto:

input 
{
         tcp 
         {
                port => 5555
  	  	type => "checkpoint"
    		mode => "server"
                host => "10.10.1.205"
   	 }
}

output 
{
	if [type] == "checkpoint" 
       {
		stdout { codec => json }
	}
}

Spúšťame príkaz na overenie:
sudo /usr/share/logstash/bin//logstash -f /etc/logstash/conf.d/checkpoint.conf
Vidíme výsledok, obrázok je klikateľný:

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="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,

Na základe týchto správ rozumieme, že logy majú formát: pole = hodnota alebo kľúč = hodnota, a preto sa hodí filter, ktorý sa nazýva kv. Pre správny výber filtra pre každý konkrétny prípad by bolo dobré zoznámiť sa s nimi v technickej dokumentácii alebo sa opýtať priateľa.

Nastavujeme Filter

Na predchádzajúcom kroku sme vybrali kv, ďalej je uvedená konfigurácia tohto filtra:

filter {
if [type] == "checkpoint"{
	kv {
		value_split => "="
		allow_duplicate_values => false
	}
}
}

Vyberáme znak, podľa ktorého budeme deliť pole a hodnotu — “=”. Ak máme v logu rovnaké záznamy, uchovávame v databáze len jeden prípad, inak dostaneme pole rovnakých hodnôt, takže ak máme správu “foo = some foo=some”, zapisujeme len foo = some.

Nastavujeme správny Output v ElasticSearch

Akonáhle je nastavený Filter, môžeme exportovať logy 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 checkpoint, ukladáme udalosť do databázy Elasticsearch, ktorá prijíma pripojenia na 10.10.1.200 na porte 9200 štandardne. Každý dokument sa uloží do konkrétneho indexu, v tomto prípade do indexu „checkpoint-” + aktuálny časový dátum. Každý index môže mať určitú sadu polí alebo sa vytvára automaticky pri objavení nového poľa v správe, nastavenia polí a ich typ si môžete pozrieť v mappings.

Ak máte nastavenú autentifikáciu (o ktorej budeme diskutovať neskôr), musia byť určite uvedené kredity na zápis do konkrétneho indexu, v tomto príklade „tssolution“ s heslom „cool“. Môžete obmedziť práva používateľov na zapisovanie logov iba do určitého indexu a žiadneho iného.

Spúšťame 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 konfiguračný súbor na správnosť zloženia:
/usr/share/logstash/bin//logstash -f checkpoint.conf
2. Elastic stack: analýza bezpečnostných logov. Logstash

Spúšťame proces Logstash:
sudo systemctl start logstash

Kontrolujeme, či sa proces spustil:
sudo systemctl status logstash

2. Elastic stack: analýza bezpečnostných logov. Logstash

Kontrolujeme, či sa socket zdvihol:
netstat -nat |grep 5555

Skontrolujeme logy v Kibana.

Po spustení všetkého prejdeme do Kibany — Objav, a uistíme sa, že je všetko správne nastavené, obrázok je klikateľný!

Všetky logy sú na svojom mieste a môžeme vidieť všetky polia a ich hodnoty!

Záver

Prešli sme si, ako napísať konfiguračný súbor Logstash, a výsledkom je parser všetkých polí a hodnôt. Teraz môžeme pracovať s vyhľadávaním a tvorbou grafov podľa určitých polí. V ďalšej časti kurzu sa pozrieme na vizualizáciu v Kibane a vytvoríme jednoduchý dashboard. Je potrebné spomenúť, že konfiguračný súbor Logstash je potrebné neustále dopisovať v určitých situáciách, napríklad keď chceme nahradiť hodnotu poľa z čísla na slovo. V nasledujúcich článkoch to budeme neustále robiť.

Tak sa držte aktualizácií (Telegram, Facebook, VK, Blog TS Solution), Yandex.Zen.

Zdroj: habr.com