V prejšnjem smo smožemo ELK skupom, od kojih se sastoji. Prvi problem s kojim se inženjer suočava pri radu s ELK skupom je slanje logova za pohranu u Elasticsearch za daljnju analizu. Međutim, to je samo rečeno, Elasticsearch pohranjuje logove u obliku dokumenata s određenim poljima i vrijednostima, što znači da inženjer mora koristiti različite alate za parsiranje poruka koje šalju krajnji sustavi. To se može učiniti na nekoliko načina – napisati vlastiti program koji će pomoću API-a dodavati dokumente u bazu ili koristiti već gotova rješenja. U okviru ovog tečaja razmatrat ćemo rješenje Logstash, koje je dio ELK skupa. Pogledat ćemo kako možemo slati logove s krajnjih sustava u Logstash, a zatim ćemo konfigurirati konfiguracijski datoteku za parsiranje i preusmjeravanje u bazu podataka Elasticsearch. Za to ćemo kao ulazni sustav uzeti logove s vatrozida Check Point.
U okviru tečaja ne razmatra se instalacija ELK skupa, budući da postoji ogroman broj članaka na ovu temu, razmatrat ćemo konfiguracijske aspekte.
Sastavit ćemo plan djelovanja za konfiguraciju Logstasha:
- Provjera da li će Elasticsearch primati logove (provjera funkcionalnosti i otvorenosti porta).
- Razmatramo kako možemo slati događaje u Logstash, biramo način, implementiramo.
- Konfiguriramo Input u konfiguracijskoj datoteci Logstash.
- Konfiguriramo Output u konfiguracijskoj datoteci Logstash u načinu debagiranja, kako bismo vidjeli kako izgleda log poruka.
- Konfiguriramo Filter.
- Konfiguriramo ispravan Output u Elasticsearch.
- Pokrećemo Logstash.
- Provjeravamo logove u Kibani.
Pogledajmo detaljnije svaku točku:
Provjera da li će Elasticsearch primati logove
Za to možete provjeriti naredbom curl dostupnost u Elasticsearch iz sustava na kojem je pokrenut Logstash. Ako imate postavljenu autentifikaciju, također putem curla prosledite korisnika/lozinku, navodite port 9200 ako ga niste mijenjali. Ako dobijete odgovor poput onog navedenog u nastavku, sve je u redu.
[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 ~]$
Če odgovor ni prišel, lahko obstajajo različne napake: postopek elasticsearch ne deluje, port je napačen ali je port blokiran s požarnim zidom na strežniku, kjer je elasticsearch.
Ogledujemo si, kako lahko pošljemo dnevniške datoteke v Logstash prek požarnega zidu Check Point.
Z Check Point upravljalskim strežnikom lahko pošljemo dnevnike v Logstash preko sysloga, z uporabo orodja log_exporter, podrobnosti si lahko ogledate v tem. , tukaj pustimo samo ukaz, ki ustvari tok:
cp_log_export add name check_point_syslog target-server <> target-port 5555 protocol tcp format generic read-mode semi-unified
<> — naslov strežnika, na katerem teče Logstash, target-port 5555 — port, kamor bomo pošiljali dnevnike, pošiljanje dnevnikov prek tcp lahko obremeni strežnik, zato je v nekaterih primerih bolj primerno uporabiti udp.
Nastavljamo INPUT v konfiguracijski datoteki Logstash.
Privzeta konfiguracijska datoteka se nahaja v imeniku /etc/logstash/conf.d/. Konfiguracijska datoteka je sestavljena iz treh smiselnih delov: INPUT, FILTER, OUTPUT. V INPUT navedemo, od kod bo sistem vzel dnevnike, v FILTER parsamo dnevnik — nastavljamo, kako razdeliti sporočilo na polja in vrednosti, v OUTPUT nastavljamo odhodni tok — kam bodo poslani razparcani dnevniki.
Najprej nastavimo INPUT, preučimo nekatere tipe, ki so lahko — file, tcp in exe.
Tcp:
input {
tcp {
port => 5555
host => "10.10.1.205"
type => "checkpoint"
mode => "server"
}
}
mode => "server"
Pove, da Logstash sprejema povezave.
port => 5555
host => "10.10.1.205"
Sprejemamo povezave na IP naslovu 10.10.1.205 (Logstash), port 5555 — port mora biti dovoljen s politiko požarnega zida.
type => "checkpoint"
Označimo dokument, kar je zelo uporabno v primeru, da imate več vhodnih povezav. Kasneje za vsako povezavo lahko napišete svoj filter s pomočjo logične konstrukcije if.
File:
input {
file {
path => "/var/log/openvas_report/*"
type => "openvas"
start_position => "beginning"
}
}
Opis nastavitev:
path => "/var/log/openvas_report/*"
Določite imenik, ki ga je treba prebrati.
type => "openvas"
Tip dogodka.
start_position => "beginning"
Ob spremembi datoteke se prebere celotna datoteka; če nastavite "end", sistem čaka na pojav novih zapisov na koncu datoteke.
Exec:
input {
exec {
command => "ls -alh"
interval => 30
}
}
Ta vhod zažene (samo!) shell ukaz in njegov izhod ovije v log sporočilo.
command => "ls -alh"
Ukaz, katerega izhod nas zanima.
interval => 30
Interval klicanja ukaza v sekundah.
Za sprejem logov iz požarnega zidu vpišite filter tcp ili udp, odvisno od tega, kako se logi pošiljajo v Logstash.
Nastavite Output v konfiguracijski datoteki Logstash v načinu razhroščevanja, da razumete, kako izgleda log sporočilo.
Potem ko smo konfigurirali INPUT, moramo razumeti, kako bo izgledalo log sporočilo in katere metode je treba uporabiti za nastavitev filtra (parserja) logov.
Za to bomo uporabili filter, ki daje rezultat v stdout, da si ogledamo izvorno sporočilo; celotna trenutna konfiguracijska datoteka bo izgledala takole:
input
{
tcp
{
port => 5555
type => "checkpoint"
mode => "server"
host => "10.10.1.205"
}
}
output
{
if [type] == "checkpoint"
{
stdout { codec => json }
}
}
Zaženite ukaz za preverjanje:
sudo /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/checkpoint.conf
Vidimo rezultat, slika je klikabilna:
Če to kopiramo, bo izgledalo tako:
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,
Глядя на данные сообщения, понимаем, что логи имеют формат: поле = значение или ключ = значение, и значит, подходит фильтр, который называется kv. Чтобы правильно выбрать фильтр для каждого конкретного случая, было бы неплохо ознакомиться с ними в технической документации или спросить у коллег.
Настраиваем фильтр
На предыдущем этапе выбрали kv, далее представлена конфигурация этого фильтра:
filter {
if [type] == "checkpoint" {
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
Выбираем символ, по которому будем разделять поле и значение — “=”. Если в логах есть дубликаты записей, сохраняем в базе только один экземпляр, иначе получится массив одинаковых значений. То есть, если у нас есть сообщение “foo = some foo=some”, мы записываем только foo = some.
Настраиваем корректный вывод в ElasticSearch
После того как настроен фильтр, можно выгружать логи в базу elasticsearch:
output
{
if [type] == "checkpoint"
{
elasticsearch
{
hosts => ["10.10.1.200:9200"]
index => "checkpoint-%{+YYYY.MM.dd}"
user => "tssolution"
password => "cool"
}
}
}
Если документ помечен как checkpoint, сохраняем событие в базу elasticsearch, которая по умолчанию принимает соединения на 10.10.1.200 на порту 9200. Каждый документ сохраняется в определённый индекс, в данном случае — в индекс «checkpoint-» + текущая временная дата. Каждый индекс может иметь особый набор полей, или создаётся автоматически при появлении нового поля в сообщении; настройки полей и их тип можно посмотреть в mappings.
Если у вас настроена аутентификация (рассмотрим позже), обязательно должны быть указаны креды для записи в конкретный индекс, в данном примере это «tssolution» с паролем «cool». Можно разграничить права пользователей для записи логов только в определённый индекс.
Запускаем Logstash.
Конфигурационный файл 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"
}
}
}
Проверяем конфигурационный файл на правильность составления:
/usr/share/logstash/bin//logstash -f checkpoint.conf
Запускаем процесс Logstash:
sudo systemctl start logstash
Проверяем, что процесс запустился:
sudo systemctl status logstash
Проверяем, поднялся ли сокет:
netstat -nat | grep 5555
Provjeravamo logove u Kibani.
Ko končanju vseh procesov, odprete Kibano – Discover in se prepričajte, da je vse pravilno nastavljeno, slika je klikljiva!
Vsi logi so na voljo in lahko vidimo vsa polja ter njihove vrednosti!
Zaključek
Pregledali smo, kako napisati konfiguracijsko datoteko Logstash, in tako dobili parser vseh polj in vrednosti. Zdaj lahko delamo z iskanjem in oblikovanjem grafikonov po določenih poljih. V nadaljevanju tečaja bomo obravnavali vizualizacijo v Kibani in ustvarili preprost nadzorni pladenj. Omemba, da je konfiguracijsko datoteko Logstash treba nenehno dopolnjevati v določenih situacijah, recimo, ko želimo zamenjati vrednost polja iz števila v besedo. V naslednjih člankih bomo to pogosto počeli.
Zato spremljajte novosti (, , , ), .
Izvor: habr.com
