
A múltban megismerkedtünk a ELK stackkel, amely programtermékekből áll. Az első feladat, amellyel az mérnök szembesül az ELK stack használata során, az a naplók elküldése az Elasticsearch részére tárolásra, majd későbbi elemzés céljából. Ez azonban csak szavakban egyszerű, az Elasticsearch dokumentumok formájában tárolja a naplókat, meghatározott mezőkkel és értékekkel, tehát a mérnöknek különféle eszközöket kell használnia az üzenetek elemzésére, amelyeket a végső rendszerek küldenek. Ezt többféleképpen megvalósíthatja: írhat egy programot, amely API-n keresztül hozzáadja a dokumentumokat az adatbázishoz, vagy használhat már létező megoldásokat. A kurzus keretein belül a Logstashmegoldást fogjuk megvizsgálni, amely az ELK stack része. Megnézzük, hogyan küldhetjük el a naplókat a végső rendszerekről a Logstash-ba, majd konfigurációs fájlokat állítunk be a naplók elemzésére és az Elasticsearch adatbázisba való átirányítására. Ehhez a bemeneti rendszerként a Check Point tűzfal naplóit fogjuk használni.
A kurzus keretein belül nem foglalkozunk az ELK stack telepítésével, mivel erről rengeteg cikk található, inkább a konfigurációs aspektusokat fogjuk megvizsgálni.
Összeállítjuk a Logstash konfigurálásának cselekvési tervét:
- Ellenőrizzük, hogy az Elasticsearch képes-e naplókat fogadni (a port működőképessége és nyitottsága).
- Megvizsgáljuk, milyen módon küldhetünk eseményeket a Logstash-ba, kiválasztjuk a módszert, és megvalósítjuk.
- Beállítjuk az Input-ot a Logstash konfigurációs fájlban.
- Beállítjuk az Output-ot a Logstash konfigurációs fájljában debug módban, hogy lássuk, hogyan néz ki a naplóüzenet.
- Beállítjuk a Filtert.
- Beállítjuk a megfelelő Output-ot az Elasticsearch-ben.
- Elindítjuk a Logstash-t.
- Ellenőrizzük a naplókat a Kibana-ban.
Részletesebben megvizsgáljuk mindegyik pontot:
Ellenőrizzük, hogy az Elasticsearch képes-e naplókat fogadni
Ehhez a curl paranccsal ellenőrizhetjük az Elasticsearch elérhetőségét arról a rendszerről, amelyen a Logstash fut. Ha be van állítva az autentikáció, akkor a curl-en keresztül megadhatjuk a felhasználónevet/jelszót, és a 9200-as portot kell használnunk, ha azt nem módosítottuk. Ha a válasz hasonlít a lentiekhez, akkor minden rendben van.
[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 ~]$
Ha a válasz nem érkezett meg, akkor többféle hiba lehetséges: az elasticsearch folyamat nem működik, nem a megfelelő port van megadva, vagy a portot tűzfal blokkolja a szerveren, ahol az elasticsearch fut.
Megvizsgáljuk, hogyan küldhetjük el a naplókat a Logstash-be a Check Point tűzfalon keresztül.
A Check Point kezelő szerverről syslog-on keresztül küldhetjük a naplókat a Logstash-be a log_exporter segédprogram használatával, amelyről részletesebben itt olvashat. , itt csak azt a parancsot hagyjuk meg, amely létrehozza a folyamatot:
cp_log_export add name check_point_syslog target-server <> target-port 5555 protocol tcp format generic read-mode semi-unified
<> — a Logstash-t futtató szerver címe, target-port 5555 — a port, ahová a naplókat küldeni fogjuk. A naplók TCP-n keresztüli küldése megterhelheti a szervert, ezért bizonyos esetekben helyesebb az UDP használata.
Beállítjuk az INPUT-t a Logstash konfigurációs fájljában.

Alapértelmezés szerint a konfigurációs fájl az /etc/logstash/conf.d/ könyvtárban található. A konfigurációs fájl 3 érdemi részből áll: INPUT, FILTER, OUTPUT. INPUT be állítjuk, hogy honnan szerzi be a rendszer a naplókat, a FILTER részben analizáljuk a naplót – beállítjuk, hogy hogyan osszuk fel az üzenetet mezőkre és értékekre, az OUTPUT részben beállítjuk a kimeneti folyamatot – hová küldjük a feldolgozott naplókat.
Először beállítjuk az INPUT-ot, megvizsgálunk néhány lehetséges típust, mint a file, tcp és exe.
TCP:
input {
tcp {
port => 5555
host => "10.10.1.205"
type => "checkpoint"
mode => "server"
}
}
mode => "server"
Megmondja, hogy a Logstash kapcsolatokat fogad.
port => 5555
host => "10.10.1.205"
Kapcsolatokat fogadunk a 10.10.1.205 IP címen (Logstash), 5555 port — a portot a tűzfal szabályai engedélyezniük kell.
type => "checkpoint"
Megjelöljük a dokumentumot, ami nagyon hasznos, ha több bejövő kapcsolatunk van. A későbbiekben minden egyes kapcsolatra saját szűrőt írhatunk logikai if feltétellel.
File:
input {
file {
path => "/var/log/openvas_report/*"
type => "openvas"
start_position => "beginning"
}
}
A beállítások leírása:
path => "/var/log/openvas_report/*"
Megadjuk a könyvtárat, amelynek fájljait olvasni kell.
type => "openvas"
Eseménytípus.
start_position => "beginning"
A fájl módosításakor az egész fájlt beolvassa; ha az "end"-t állítjuk be, a rendszer a fájl végén új bejegyzések megjelenésére vár.
Exec:
input {
exec {
command => "ls -alh"
interval => 30
}
}
Ez az input csak egy shell parancsot indít el, és annak kimenetét egy naplóüzenetbe csomagolja.
command => "ls -alh"
A parancs, amelynek a kimenete érdekel.
interval => 30
A parancs hívásának intervalluma másodpercben.
A tűzfal logjaival való fogadáshoz szűrőt adunk meg tcp vagy udp, attól függően, hogy hogyan küldik a logokat a Logstash-be.
Beállítjuk az Output-ot a Logstash konfigurációs fájlban hibakeresés módban, hogy megértsük, hogyan néz ki a naplóüzenet.
Miután konfiguráltuk az INPUT-ot, meg kell értenünk, hogyan fog kinézni a naplóüzenet, és milyen módszereket kell használni a logok szűrésének (elemző) beállításához.
Ebből egy stdout-ra kimeneti szűrőt fogunk használni, hogy lássuk az eredeti üzenetet, a teljes konfigurációs fájl ebben a pillanatban így fog kinézni:
input
{
tcp
{
port => 5555
type => "checkpoint"
mode => "server"
host => "10.10.1.205"
}
}
output
{
if [type] == "checkpoint"
{
stdout { codec => json }
}
}
Futtassuk a parancsot a teszteléshez:
sudo /usr/share/logstash/bin//logstash -f /etc/logstash/conf.d/checkpoint.conf
Látjuk az eredményt, a kép kattintható:
Ha ezt lemásoljuk, így fog kinézni:
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,
A megadott üzenetek alapján látható, hogy a naplók formátuma: mező = érték vagy kulcs = érték, ami azt jelenti, hogy a kv nevű szűrő alkalmazható. A megfelelő szűrő kiválasztásához célszerű lenne megismerkedni velük a műszaki dokumentációban, vagy megkérdezni egy barátot.
Szűrő beállítása
Az előző lépésben a kv-t választottuk, itt a szűrő konfigurációja látható:
filter {
if [type] == "checkpoint"{
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
Válasszuk ki a szimbólumot, amivel elválasztjuk a mezőt és az értéket — "=", ha a naplóban azonos bejegyzések vannak, akkor csak egy példányt tárolunk az adatbázisban, különben olyan tömböt kapunk, amely az azonos értékeket tartalmazza, tehát ha a üzenet "foo = some foo=some", csak a foo = some-t írjuk be.
Helyes output beállítása az ElasticSearch-ben
A szűrő beállítása után lehetőség van a naplók betöltésére az adatbázisba elasticsearch:
output
{
if [type] == "checkpoint"
{
elasticsearch
{
hosts => ["10.10.1.200:9200"]
index => "checkpoint-%{+YYYY.MM.dd}"
user => "tssolution"
password => "cool"
}
}
}
Ha a dokumentum típusa checkpoint, az eseményt elmentjük az elasticsearch adatbázisba, amely alapértelmezetten a 10.10.1.200 IP-címen 9200-as porton fogadja a kapcsolatokat. Minden dokumentum egy meghatározott indexbe kerül, ebben az esetben a "checkpoint-" + aktuális időpontot használjuk. Minden index meghatározott mezőkkel rendelkezhet, vagy automatikusan létrejön, ha új mező jelenik meg az üzenetben, a mezők és azok típusa a mappings-ban tekinthető meg.
Ha be van állítva a hitelesítés (erről később), feltétlenül meg kell adni a hitelesítő adatokat egy adott indexbe történő íráshoz, jelen példában ez a "tssolution" a "cool" jelszóval. Korlátozhatjuk a felhasználói jogokat, hogy csak egy adott indexbe írhassanak naplókat és másba ne.
Logstash elindítása.
Logstash konfigurációs fájl:
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"
}
}
}
A konfigurációs fájl helyességének ellenőrzése:
/usr/share/logstash/bin//logstash -f checkpoint.conf

Logstash folyamat elindítása:
sudo systemctl start logstash
Ellenőrizzük, hogy a folyamat elindult:
sudo systemctl status logstash

Ellenőrizzük, hogy a socket működik-e:
netstat -nat |grep 5555
![]()
Ellenőrizzük a naplókat a Kibana-ban.
Miután minden elindult, lépjünk be a Kibana — Discover-be, és győződjünk meg róla, hogy minden helyesen van beállítva, a kép pedig kattintható!
Minden napló a helyén van, és láthatjuk az összes mezőt és azok értékeit!
Zárás
Megvizsgáltuk, hogyan írjunk Logstash konfigurációs fájlt, amelynek eredményeként megkaptuk az összes mező és érték parserét. Most már tudunk dolgozni a kereséssel és a grafikonok létrehozásával bizonyos mezők alapján. A tanfolyam további részeiben a Kibana vizualizációját fogjuk megnézni, és létrehozunk egy egyszerű irányítópultot. Érdemes megemlíteni, hogy a Logstash konfigurációs fájlt bizonyos helyzetekben folyamatosan frissíteni kell, például amikor egy mező értékét számról szóra szeretnénk váltani. A következő cikkekben ezt folyamatosan fogjuk csinálni.
Úgyhogy kövessétek az frissítéseket (, , , ), .
Forrás: habr.com
