W poprzedniej zapoznaliśmy się z stosem ELK, z jakich produktów oprogramowania się składa. Pierwszym zadaniem, z którym musi zmierzyć się inżynier pracujący ze stosem ELK, jest wysyłanie logów do przechowywania w Elasticsearch w celu dalszej analizy. Jednak to tylko w teorii, Elasticsearch przechowuje logi w postaci dokumentów z określonymi polami i wartościami, co oznacza, że inżynier musi użyć różnych narzędzi, aby sparsować wiadomości wysyłane z systemów końcowych. Można to zrobić na kilka sposobów — sami napisać program, który za pomocą API doda dokumenty do bazy lub skorzystać z gotowych rozwiązań. W ramach tego kursu będziemy omawiać rozwiązanie Logstash, które jest częścią stosu ELK. Zobaczymy, jak można wysyłać logi z systemów końcowych do Logstash, a następnie skonfigurujemy plik konfiguracyjny do parsowania i przekierowywania do bazy danych Elasticsearch. W tym celu weźmiemy logi z zapory ogniowej Check Point jako system wejściowy.
W ramach kursu nie rozważamy instalacji stosu ELK, ponieważ istnieje ogromna ilość artykułów na ten temat, skupimy się na konfiguracji.
Sporządzimy plan działania dotyczący konfiguracji Logstash:
- Sprawdzenie, czy Elasticsearch będzie akceptować logi (sprawdzenie funkcjonalności i otwartości portu).
- Rozważamy, w jaki sposób możemy wysyłać zdarzenia do Logstash, wybieramy metodę, implementujemy.
- Konfigurujemy Input w pliku konfiguracyjnym Logstash.
- Konfigurujemy Output w pliku konfiguracyjnym Logstash w trybie debug, aby zrozumieć, jak wygląda wiadomość logu.
- Konfigurujemy Filter.
- Konfigurujemy prawidłowy Output w Elasticsearch.
- Uruchamiamy Logstash.
- Sprawdzamy logi w Kibana.
Szczegółowo omówimy każdy punkt:
Sprawdzenie, czy Elasticsearch będzie akceptował logi
W tym celu można sprawdzić za pomocą polecenia curl dostęp do Elasticsearch z systemu, na którym uruchomiono Logstash. Jeśli masz skonfigurowane uwierzytelnianie, to również za pomocą curl przekażemy użytkownika/hasło, określając port 9200, jeśli go nie zmieniłeś. Jeśli otrzymasz odpowiedź, jak poniżej, wszystko jest w porzą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 ~]$
Jeśli odpowiedź nie nadeszła, może być kilka rodzajów błędów: proces elasticsearch nie działa, port jest błędny lub port jest zablokowany przez zaporę sieciową na serwerze, na którym znajduje się elasticsearch.
Rozważamy, jak można wysłać logi do Logstash z firewalla Check Point.
Z serwera zarządzania Check Point można wysłać logi do Logstash przez syslog, używając narzędzia log_exporter, więcej szczegółów można znaleźć w tej , tutaj pozostawimy tylko polecenie, które tworzy strumień:
cp_log_export add name check_point_syslog target-server <> target-port 5555 protocol tcp format generic read-mode semi-unified
<> — adres serwera, na którym uruchomiony jest Logstash, target-port 5555 — port, na który będziemy wysyłać logi, wysyłanie logów przez TCP może obciążyć serwer, dlatego w niektórych przypadkach lepiej ustawić UDP.
Konfigurujemy INPUT w pliku konfiguracyjnym Logstash.
Domyślnie plik konfiguracyjny znajduje się w katalogu /etc/logstash/conf.d/. Plik konfiguracyjny składa się z 3 sensownych części: INPUT, FILTER, OUTPUT. W INPUT określamy, skąd system będzie pobierał logi, w FILTER parsujemy log — konfigurujemy, jak podzielić komunikat na pola i wartości, w OUTPUT konfigurujemy strumień wyjściowy — dokąd będą wysyłane sparsowane logi.
Najpierw skonfigurujemy INPUT, zapoznajmy się z niektórymi typami, które mogą być — file, tcp i exe.
Tcp:
input {
tcp {
port => 5555
host => "10.10.1.205"
type => "checkpoint"
mode => "server"
}
}
mode => "server"
Mówi, że Logstash akceptuje połączenia.
port => 5555
host => "10.10.1.205"
Akceptujemy połączenia pod adresem IP 10.10.1.205 (Logstash), port 5555 — port musi być dozwolony przez politykę zapory.
type => "checkpoint"
Oznaczamy dokument, co jest bardzo wygodne w przypadku, gdy masz kilka przychodzących połączeń. W następnych krokach dla każdego połączenia można napisać własny filtr za pomocą konstrukcji logicznej if.
File:
input {
file {
path => "/var/log/openvas_report/*"
type => "openvas"
start_position => "beginning"
}
}
Opis ustawień:
path => "/var/log/openvas_report/*"
Wskazujemy katalog, z którego należy odczytać pliki.
typ => «openvas»
Typ zdarzenia.
start_position => «początek»
Przy zmianie pliku, odczytuje cały plik, jeśli ustawione jest «end», to system czeka na pojawienie się nowych wpisów na końcu pliku.
Exec:
input {
exec {
command => "ls -alh"
interval => 30
}
}
Na podstawie tego inputu uruchamiana jest (tylko!) komenda shell i jej wynik jest opakowywany w wiadomości logu.
command => «ls -alh»
Komenda, której wynik nas interesuje.
interval => 30
Interwał wywołania komendy w sekundach.
Aby odbierać logi z zapory sieciowej, definiujemy filtr tcp lub udp, w zależności od tego, jak logi są wysyłane do Logstash.
Konfigurujemy Output w pliku konfiguracyjnym Logstash w trybie debugowania, aby zrozumieć, jak wygląda wiadomość logu.
Po skonfigurowaniu INPUT, musimy zrozumieć, jak będzie wyglądać wiadomość logu, jakie metody należy użyć do ustawienia filtra (parsera) logów.
W tym celu użyjemy filtra, który zwraca wynik do stdout, aby móc zobaczyć oryginalną wiadomość, a cały plik konfiguracyjny na obecny moment będzie wyglądał tak:
input
{
tcp
{
port => 5555
typ => "checkpoint"
tryb => "serwer"
host => “10.10.1.205”
}
}
output
{
if [typ] == "checkpoint"
{
stdout { codec=> json }
}
}
Uruchamiamy komendę do sprawdzenia:
sudo /usr/share/logstash/bin//logstash -f /etc/logstash/conf.d/checkpoint.conf
Widoczny wynik, obrazek jest klikalny:
Jeśli skopiować to, będzie wyglądać tak:
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,
Patrząc na te komunikaty, rozumiemy, że logi mają postać: pole = wartość lub key = value, co oznacza, że odpowiedni filtr nazywa się kv. Aby właściwie dobrać filtr do każdej konkretnej sytuacji, warto zapoznać się z nimi w dokumentacji technicznej lub zapytać znajomego.
Konfigurujemy filtr
Na poprzednim etapie wybraliśmy kv, poniżej przedstawiono konfigurację tego filtra:
filter {
if [type] == "checkpoint"{
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
Wybieramy znak, po którym będziemy dzielić pole i wartość — “=”. Jeśli w logu są identyczne wpisy, zapisujemy w bazie tylko jeden egzemplarz, w przeciwnym razie otrzymamy tablicę z identycznymi wartościami, to znaczy, jeśli mamy komunikat “foo = some foo=some”, zapisujemy tylko foo = some.
Konfigurujemy poprawny output w ElasticSearch
Po skonfigurowaniu filtra można eksportować logi do bazy elasticsearch:
output
{
if [type] == "checkpoint"
{
elasticsearch
{
hosts => ["10.10.1.200:9200"]
index => "checkpoint-%{+YYYY.MM.dd}"
user => "tssolution"
password => "cool"
}
}
}
Jeśli dokument jest podpisany typem checkpoint, zapisujemy zdarzenie w bazie elasticsearch, która domyślnie akceptuje połączenia na 10.10.1.200 na porcie 9200. Każdy dokument jest zapisywany w określonym indeksie, w tym przypadku zapisujemy w indeksie „checkpoint-” + aktualna data. Każdy indeks może mieć określony zestaw pól lub być tworzony automatycznie przy pojawieniu się nowego pola w komunikacie, ustawienia pól i ich typ można obejrzeć w mappings.
Jeśli masz skonfigurowaną autoryzację (omówimy to później), musisz podać dane logowania do zapisu w konkretnym indeksie, w tym przykładzie jest to „tssolution” z hasłem „cool”. Można rozdzielić prawa użytkowników do zapisywania logów tylko w określonym indeksie i żadnym innym.
Uruchamiamy Logstash.
Plik konfiguracyjny 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"
}
}
}
Sprawdzamy plik konfiguracyjny pod kątem poprawności:
/usr/share/logstash/bin//logstash -f checkpoint.conf
Uruchamiamy proces Logstash:
sudo systemctl start logstash
Sprawdzamy, czy proces został uruchomiony:
sudo systemctl status logstash
Sprawdzamy, czy gniazdo działa:
netstat -nat |grep 5555
Sprawdzamy logi w Kibana.
Po uruchomieniu wszystkiego, przechodzimy do Kibany - Discover, upewniając się, że wszystko jest poprawnie skonfigurowane, a obrazek jest klikalny!
Wszystkie logi są dostępne i możemy zobaczyć wszystkie pola i ich wartości!
Podsumowanie
Omówiliśmy, jak napisać plik konfiguracyjny Logstash, w efekcie uzyskaliśmy parser wszystkich pól i wartości. Teraz możemy pracować z wyszukiwaniem i tworzeniem wykresów na podstawie określonych pól. W dalszej części kursu omówimy wizualizację w Kibanie, stworzymy prosty dashboard. Należy wspomnieć, że plik konfiguracyjny Logstash wymaga ciągłych aktualizacji w niektórych sytuacjach, na przykład, gdy chcemy zamienić wartość pola z liczby na słowo. W kolejnych artykułach będziemy to regularnie robić.
Więc śledźcie aktualizacje (, , , ), .
Źródło: habr.com
