W przeszłości spotkaliśmy się Stos ELKz jakich produktów programowych się składa. Pierwszym zadaniem, jakie staje przed inżynierem podczas pracy ze stosem ELK, jest wysyłanie logów do przechowywania w Elasticsearch w celu późniejszej analizy. To jednak tylko frazesy, Elasticsearch przechowuje logi w formie dokumentów z określonymi polami i wartościami, co oznacza, że inżynier musi używać różnych narzędzi do analizowania wiadomości wysyłanej z systemów końcowych. Można to zrobić na kilka sposobów – napisz samodzielnie program, który będzie dodawać dokumenty do bazy za pomocą API, lub skorzystaj z gotowych rozwiązań. W tym kursie rozważymy rozwiązanie Logstash, który jest częścią stosu ELK. Przyjrzymy się, jak możemy wysyłać logi z systemów końcowych do Logstash, a następnie skonfigurujemy plik konfiguracyjny do analizy i przekierowania do bazy danych Elasticsearch. W tym celu pobieramy logi z zapory sieciowej Check Point jako systemu przychodzącego.
Kurs nie obejmuje instalacji stosu ELK, ponieważ istnieje ogromna liczba artykułów na ten temat, rozważymy komponent konfiguracyjny.
Przygotujmy plan działania dla konfiguracji Logstash:
- Sprawdzenie czy Elasticsearch przyjmie logi (sprawdzenie funkcjonalności i otwartości portu).
- Zastanawiamy się, w jaki sposób możemy wysyłać zdarzenia do Logstash, wybieramy metodę i wdrażamy ją.
- Konfigurujemy dane wejściowe w pliku konfiguracyjnym Logstash.
- Konfigurujemy dane wyjściowe w pliku konfiguracyjnym Logstash w trybie debugowania, aby zrozumieć, jak wygląda komunikat dziennika.
- Konfigurowanie filtra.
- Konfigurowanie prawidłowego wyjścia w ElasticSearch.
- Uruchamia się Logstash.
- Sprawdzanie logów w Kibanie.
Przyjrzyjmy się każdemu punktowi bardziej szczegółowo:
Sprawdzanie, czy Elasticsearch zaakceptuje logi
Aby to zrobić, możesz użyć polecenia curl, aby sprawdzić dostęp do Elasticsearch z systemu, w którym wdrożono Logstash. Jeśli masz skonfigurowane uwierzytelnianie, przesyłamy również użytkownika/hasło poprzez curl, podając port 9200, jeśli go nie zmieniłeś. Jeśli otrzymasz odpowiedź podobną do poniższej, to wszystko jest w porzą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 ~]$
Jeśli odpowiedź nie zostanie odebrana, może wystąpić kilka rodzajów błędów: proces Elasticsearch nie jest uruchomiony, określono niewłaściwy port lub port jest blokowany przez zaporę ogniową na serwerze, na którym zainstalowano Elasticsearch.
Przyjrzyjmy się, jak możesz wysyłać dzienniki do Logstash z zapory ogniowej punktu kontrolnego
Z serwera zarządzającego Check Point możesz wysyłać logi do Logstash poprzez syslog za pomocą narzędzia log_exporter, więcej na ten temat możesz przeczytać tutaj , tutaj pozostawimy jedynie polecenie tworzące strumień:
cp_log_export dodaj nazwę check_point_syslog serwer docelowy < > port docelowy, protokół 5555, format TCP, ogólny tryb odczytu, częściowo ujednolicony
< > - adres serwera na którym działa Logstash, docelowy-port 5555 - port na który będziemy wysyłać logi, wysyłanie logów poprzez tcp może załadować serwer, dlatego w niektórych przypadkach bardziej poprawne jest użycie udp.
Konfigurowanie wejścia w pliku konfiguracyjnym Logstash
Domyślnie plik konfiguracyjny znajduje się w katalogu /etc/logstash/conf.d/. Plik konfiguracyjny składa się z 3 znaczących części: WEJŚCIA, FILTRA, WYJŚCIA. W WEJŚCIE wskazujemy skąd system będzie pobierał logi, w FILTER analizuj dziennik — skonfiguruj sposób dzielenia wiadomości na pola i wartości, w WYDAJNOŚĆ konfigurujemy strumień wyjściowy – dokąd będą przesyłane przeanalizowane logi.
Najpierw skonfigurujmy WEJŚCIE, rozważmy niektóre typy, które mogą być - plik, tcp i exe.
TCP:
input {
tcp {
port => 5555
host => “10.10.1.205”
type => "checkpoint"
mode => "server"
}
}
tryb => „serwer”
Wskazuje, że Logstash akceptuje połączenia.
port => 5555
gospodarz => „10.10.1.205”
Akceptujemy połączenia poprzez adres IP 10.10.1.205 (Logstash), port 5555 - port musi być dozwolony przez politykę firewalla.
wpisz => „punkt kontrolny”
Zaznaczamy dokument, co jest bardzo wygodne, jeśli masz kilka połączeń przychodzących. Następnie dla każdego połączenia możesz napisać własny filtr, korzystając z konstrukcji logicznej if.
Plik:
input {
file {
path => "/var/log/openvas_report/*"
type => "openvas"
start_position => "beginning"
}
}
Opis ustawień:
ścieżka => "/var/log/openvas_report/*"
Wskazujemy katalog, w którym pliki mają zostać odczytane.
wpisz => „openvas”
Typ wydarzenia.
start_position => „początek”
Podczas zmiany pliku czyta cały plik; jeśli ustawisz „koniec”, system będzie czekał, aż na końcu pliku pojawią się nowe rekordy.
Wykonanie:
input {
exec {
command => "ls -alh"
interval => 30
}
}
Używając tego wejścia, uruchamiane jest (tylko!) polecenie powłoki, a jego dane wyjściowe zamieniane są na komunikat dziennika.
polecenie => "ls -alh"
Polecenie, którego wyjście nas interesuje.
interwał => 30
Interwał wywołania polecenia w sekundach.
Aby otrzymywać logi z firewalla rejestrujemy filtr tcp lub udp, w zależności od sposobu przesyłania dzienników do Logstash.
Konfigurujemy dane wyjściowe w pliku konfiguracyjnym Logstash w trybie debugowania, aby zrozumieć, jak wygląda komunikat dziennika
Po skonfigurowaniu INPUT musimy zrozumieć, jak będzie wyglądał komunikat dziennika i jakich metod należy użyć, aby skonfigurować filtr dziennika (parser).
Aby to zrobić, użyjemy filtra, który wyśle wynik na standardowe wyjście w celu wyświetlenia oryginalnej wiadomości; kompletny plik konfiguracyjny w tej chwili będzie wyglądał następująco:
input
{
tcp
{
port => 5555
type => "checkpoint"
mode => "server"
host => “10.10.1.205”
}
}
output
{
if [type] == "checkpoint"
{
stdout { codec=> json }
}
}
Uruchom polecenie, aby sprawdzić:
sudo /usr/share/logstash/bin//logstash -f /etc/logstash/conf.d/checkpoint.conf
Widzimy wynik, zdjęcie można kliknąć:
Jeśli go skopiujesz, 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="[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,
Patrząc na te komunikaty rozumiemy, że logi wyglądają następująco: pole = wartość lub klucz = wartość, co oznacza, że odpowiedni jest filtr o nazwie kv. Aby dobrać odpowiedni filtr do konkretnego przypadku, warto zapoznać się z nimi w dokumentacji technicznej lub zapytać znajomego.
Konfigurowanie filtra
Na ostatnim etapie wybraliśmy kv, konfigurację tego filtra przedstawiono poniżej:
filter {
if [type] == "checkpoint"{
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
Wybieramy symbol, według którego podzielimy pole i wartość - „=”. Jeżeli w logu mamy identyczne wpisy to zapisujemy tylko jedną instancję w bazie, w innym przypadku otrzymamy tablicę identycznych wartości, czyli jeśli mamy komunikat „foo = Some foo=some” to piszemy tylko foo = niektórzy.
Konfigurowanie prawidłowego wyjścia w ElasticSearch
Po skonfigurowaniu filtra możesz przesyłać dzienniki do bazy danych wyszukiwanie elastyczne:
output
{
if [type] == "checkpoint"
{
elasticsearch
{
hosts => ["10.10.1.200:9200"]
index => "checkpoint-%{+YYYY.MM.dd}"
user => "tssolution"
password => "cool"
}
}
}
Jeżeli dokument jest podpisany typem checkpoint to zapisujemy zdarzenie do bazy Elasticsearch, która domyślnie przyjmuje połączenia w dniu 10.10.1.200 na porcie 9200. Każdy dokument zapisywany jest do określonego indeksu, w tym przypadku zapisujemy do indeksu „checkpoint-” + bieżąca data czasu. Każdy indeks może mieć określony zestaw pól lub jest tworzony automatycznie, gdy w wiadomości pojawi się nowe pole; ustawienia pól i ich typ można przeglądać w mapowaniach.
Jeśli masz skonfigurowane uwierzytelnianie (przyjrzymy się temu później), należy określić dane uwierzytelniające do zapisu do określonego indeksu, w tym przykładzie jest to „tssolution” z hasłem „cool”. Możesz rozróżnić uprawnienia użytkowników, aby zapisywać logi tylko do określonego indeksu i nie więcej.
Uruchom Logstasha.
Plik konfiguracyjny Logstasha:
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 poprawność pliku konfiguracyjnego:
/usr/share/logstash/bin//logstash -f checkpoint.conf
Rozpocznij proces Logstash:
sudo systemctl uruchom logstash
Sprawdzamy, czy proces się rozpoczął:
sudo systemctl status logstash
Sprawdźmy, czy gniazdo jest włączone:
netstat -nat |grep 5555
Sprawdzanie logów w Kibanie.
Po uruchomieniu wszystkiego przejdź do Kibana - Discover, upewnij się, że wszystko jest poprawnie skonfigurowane, obraz można kliknąć!
Wszystkie logi są na swoim miejscu i możemy zobaczyć wszystkie pola i ich wartości!
wniosek
Przyjrzeliśmy się, jak napisać plik konfiguracyjny Logstash, w wyniku czego otrzymaliśmy parser wszystkich pól i wartości. Teraz możemy pracować z wyszukiwaniem i kreśleniem określonych pól. W dalszej części kursu przyjrzymy się wizualizacji w Kibanie i stworzymy prosty dashboard. Warto wspomnieć, że plik konfiguracyjny Logstash wymaga ciągłej aktualizacji w niektórych sytuacjach, np. gdy chcemy zamienić wartość pola z liczby na słowo. W kolejnych artykułach będziemy to robić stale.
Bądźcie na bieżąco (, , , ), .
Źródło: www.habr.com