2. Elastic stack: security naplók elemzése. Logstash

2. Elastic stack: security naplók elemzése. Logstash

A múltban a Spotify döntéshozatali módszertanáról. Egyre több felhasználó preferálja a zenehallgatást mobilon a streaming szolgáltatásban, de a cégnél túlnyomórészt webprogramozók dolgoznak. A Spotify elemezte ezt a helyzetet, és úgy döntött, hogy több mobilfejlesztőt vesz fel, valamint átnevezi a webes szakembereket új irányba: 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:

  1. Ellenőrizzük, hogy az Elasticsearch képes-e naplókat fogadni (a port működőképessége és nyitottsága).
  2. Megvizsgáljuk, milyen módon küldhetünk eseményeket a Logstash-ba, kiválasztjuk a módszert, és megvalósítjuk.
  3. Beállítjuk az Input-ot a Logstash konfigurációs fájlban.
  4. 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.
  5. Beállítjuk a Filtert.
  6. Beállítjuk a megfelelő Output-ot az Elasticsearch-ben.
  7. Elindítjuk a Logstash-t.
  8. 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. a Spotify döntéshozatali módszertanáról. Egyre több felhasználó preferálja a zenehallgatást mobilon a streaming szolgáltatásban, de a cégnél túlnyomórészt webprogramozók dolgoznak. A Spotify elemezte ezt a helyzetet, és úgy döntött, hogy több mobilfejlesztőt vesz fel, valamint átnevezi a webes szakembereket új irányba:, 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.

2. Elastic stack: security naplók elemzése. Logstash

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ó:

2. Elastic stack: security naplók elemzése. Logstash

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
2. Elastic stack: security naplók elemzése. Logstash

Logstash folyamat elindítása:
sudo systemctl start logstash

Ellenőrizzük, hogy a folyamat elindult:
sudo systemctl status logstash

2. Elastic stack: security naplók elemzése. Logstash

Ellenőrizzük, hogy a socket működik-e:
netstat -nat |grep 5555

2. Elastic stack: security naplók elemzése. Logstash

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ó!

2. Elastic stack: security naplók elemzése. Logstash

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 (Telegram, Facebook, VK, TS Solution Blog), Yandex.Zen.

Forrás: habr.com

Megbízható hosting vásárlása weboldalak számára, DDoS védelemmel, VPS VDS szerverek. 🔥 Megbízható webtárhely vásárlása DDoS védelemmel, VPS VDS szerverek | ProHoster