2. Elastic stack: turvaloogide analüüs. Logstash

Eelmisel artiklis me kohtusime ELK steki, millest see koosneb. Esimene ülesanne, millega insener ELK steki kasutamisel silmitsi seisab, on logide saatmine nende salvestamiseks elasticsearch'i hilisemaks analüüsiks. Kuid see on vaid sõnades lihtne, elasticsearch salvestab logid dokumentide kujul, millel on kindlad väljad ja väärtused, seega peab insener kasutama erinevaid tööriistu, et analüüsida sõnumit, mis saadetakse lõpp süsteemidest. Seda saab teha mitmel viisil — kirjutama oma programmi, mis API kaudu lisab dokumendid andmebaasi, või kasutama juba olemasolevaid lahendusi. Käesolevas kursuses vaatleme lahendust Logstash, mis on osa ELK stekist. Vaadake, kuidas logisid lõpp süsteemidest Logstash'i saata, ja seejärel seadistame konfigureerimisfaili parsimiseks ja andmebaasi Elasticsearch edastamiseks. Selleks võtame sisend süsteemina Check Point'i tulemüüri logid.

Kursuse raames ei käsitleta ELK steki paigaldamist, kuna sellel teemal on palju artikleid, keskendume konfiguratsiooni aspektile.

Koostame tegevusplaani Logstash'i konfigureerimiseks:

1. Kontrollime, et elasticsearch võtaks logisid vastu (kontrollime porti töökindlust ja avatud olekut).
2. Arutame, kuidas saame sündmusi Logstash'i edastada, valime viisi ja rakendame selle.
3. Seadistame Logstash'i konfigureerimisfailis sisendi.
4. Seadistame Logstash'i konfigureerimisfailis väljundi debug-režiimis, et mõista, milline välja näeb logisõnum.
5. Seadistame filtri.
6. Seadistame korrektse väljundi ElasticSearch'is.
7. Käivitame Logstash'i.
8. Kontrollime logisid Kibanas.

Vaatleme iga punkti lähemalt:

Kontrollime, et elasticsearch võtaks logisid vastu

Selle jaoks saab kontrollida curl'i käsuga juurdepääsu Elasticsearch'ile süsteemist, kuhu Logstash on paigaldatud. Kui teil on seadistatud autentimine, edastame ka curl'i kaudu kasutajanime/parooli, märkides pordi 9200, kui te ei ole seda muutnud. Kui vastus on sarnane alltoodud näitele, siis on kõik korras.

[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 ~]$

Kui vastust ei tulnud, võivad põhjused olla mitmed: elasticsearchi protsess ei pruugi töötada, port on vale või port on tulemüüriga blokeeritud serveris, kus elasticsearch on paigaldatud.

Vaatame, kuidas logisid Logstashi saata Check Point tulemüürist.

Check Pointi haldusserverist saate logisid Logstashi saata syslogi kaudu, kasutades log_exporter utiliiti, mille kohta saate rohkem teavet järgmistest. artiklis, siin jätame ainult käsu, mis loob voolu:

cp_log_export add name check_point_syslog target-server <> target-port 5555 protocol tcp format generic read-mode semi-unified

<> — serveri aadress, kus Logstash töötab, target-port 5555 — port, kuhu logisid saadame, logide saatmine tcp kaudu võib serverit koormata, seetõttu on mõnes olukorras õigem kasutada udp-d.

Seame Logstashi konfiguratsioonifailis INPUT-i üles.

Vaikimisi asub konfiguratsioonifail kataloogis /etc/logstash/conf.d/. Konfiguratsioonifail koosneb kolmest mõtestatud osast: INPUT, FILTER, OUTPUT. INPUT määra, kust süsteem logisid saab, FILTER parsi logi — seadista, kuidas jagada sõnumid väljadeks ja väärtusteks, OUTPUT seadista väljundivoog — kuhu parsilogid saadetakse.

Esmalt seadistame INPUT-i, vaatame mõningaid tüüpe, mis võivad olla — file, tcp ja exe.

Tcp:

input {
tcp {
    port => 5555
    host => "10.10.1.205"
    type => "checkpoint"
    mode => "server"
}
}

mode => «server»
Tähendab, et Logstash aktsepteerib ühendusi.

port => 5555
host => "10.10.1.205"
Võtame ühendusi IP-aadressilt 10.10.1.205 (Logstash), port 5555 — port peab olema tulemüüri poliitika poolt lubatud.

type => «checkpoint»
Märgistame dokumendi, mis on väga mugav, kui teil on mitu sissetulevat ühendust. Edasi võib iga ühenduse jaoks kirjutada oma filtri loogilise konstruktsiooniga if.

File:

input {
  file {
    path => "/var/log/openvas_report/*"
    type => "openvas"
    start_position => "beginning"
    }
}

Seadete kirjeldus:
path => «/var/log/openvas_report/*»
Määrake kataloog, mille failid tuleb lugeda.

type => "openvas"
Sündmuse tüüp.

start_position => "beginning"
Faili muutmisel loetakse fail täielikult, kui seadistada "end", siis ootab süsteem, et uued kirjed ilmuksid faili lõppu.

Exec:

input {
  exec {
    command => "ls -alh"
    interval => 30
  }
}

Selle sisendi tõttu käivitatakse (ainult!) shell käsk ja selle väljund ümbritsetakse logi sõnumiga.

command => "ls -alh"
Käsk, mille väljund meid huvitab.

interval => 30
Käsu väljakutsumise intervall sekundites.

Logid tulemüürist vastuvõtmiseks määrame filtri tcp või udp, sõltuvalt sellest, kuidas logi edastatakse Logstashi.

Seame Logstashi konfigureerimisfailis väljundi debugeerimise režiimis, et mõista, kuidas logi sõnum välja näeb.

Kui oleme INPUTi konfigureerinud, peame aru saama, kuidas logi sõnum välja näeb, milliseid meetodeid on vaja logide filtri (parsi) seadistamiseks kasutada.

Selleks kasutame filtrit, mis annab tulemuse stdout-is, et näha algsõnumit, kogu konfigureerimisfail näeb hetkel välja järgmine:

input 
{
         tcp 
         {
                port => 5555
  	  	type => "checkpoint"
    		mode => "server"
                host => "10.10.1.205"
   	 }
}

output 
{
	if [type] == "checkpoint" 
       {
		stdout { codec => json }
	}
}

Käivitame kontrollimise käsu:
sudo /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/checkpoint.conf
Näeme tulemust, pilt on klikkitav:

Kui kopeerite, näeb see välja nii:

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,

Vaadates neid sõnumeid, mõistame, et logid on kujul: väli = väärtus või key = value, seega sobib filter, mida nimetatakse kv. Selleks, et õigesti valida filter iga konkreetse juhtumi jaoks, oleks hea nendega tutvuda tehnilises dokumentatsioonis või küsida sõbralt.

Seame Filter'i

Eelmisel sammul valisime kv, nüüd on esitatud selle filtri konfiguratsioon:

filter {
if [type] == "checkpoint"{
	kv {
		value_split => "="
		allow_duplicate_values => false
	}
}
}

Valime sümboli, mille põhjal jagame väli ja väärtus — “=”. Kui meil on logis sama sisuga kirjeid, salvestame andmebaasi vaid ühe eksemplari, vastasel juhul tekib teil massiiv sama väärtuse küljes, seega kui meil on sõnum “foo = some foo=some”, salvestame ainult foo = some.

Seame korrektse väljundi ElasticSearch'i

Pärast seda, kui filter on seadistatud, saab logisid andmebaasi laadida elasticsearch:

output 
{
if [type] == "checkpoint"
{
 	elasticsearch 
        {
		hosts => ["10.10.1.200:9200"]
		index => "checkpoint-%{+YYYY.MM.dd}"
    		user => "tssolution"
    		password => "cool"
  	}
}
}

Kui dokument on märgitud tüübi checkpoint, salvestame sündmuse andmebaasi elasticsearch, mis aktsepteerib ühendusi 10.10.1.200 aadressil port 9200 vaikesätetena. Iga dokument salvestatakse kindlasse indeksisse, antud juhul salvestame indeksi «checkpoint-» + praegune kuupäev. Iga indeks võib omada kindlat atribuutide kogumit või luuakse automaatselt, kui sõnumis ilmneb uus väli; välja seadistusi ja nende tüüpe saab vaadata mappings.

Kui teil on seadistatud autentimine (arutame hiljem), peavad kindlasti olema määratud kredentsid kirjutamiseks konkreetse indeksisse, antud näites on see «tssolution» koos parooliga «cool». Võib piirata kasutajate õigusi logide kirjutamiseks ainult kindlasse indeksisse ja mitte kuhugi mujale.

Käivitame Logstash'i.

Logstash'i konfiguratsioonifail:

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"
  	}
}
}

Kontrollime konfiguratsioonifaili õigsust:
/usr/share/logstash/bin//logstash -f checkpoint.conf


Käivitame Logstash'i protsessi:
sudo systemctl start logstash

Kontrollime, kas protsess on käivitatud:
sudo systemctl status logstash

Kontrollime, kas sokkel on üles tõusnud:
netstat -nat |grep 5555

Kontrollime logisid Kibanas.

Kui kõik on käivitatud, siseneme Kibana - Discover, veendudes, et kõik on õigesti seadistatud, pilt on klikitav!

Kõik logid on kohal ja me näeme kõiki välju ja nende väärtusi!

Kokkuvõte

Me arutasime, kuidas kirjutada Logstashi konfiguratsioonifaili, mille tulemusel saime parseri kõigi väljade ja väärtuste jaoks. Nüüd saame töötada otsingu ja graafikute koostamisega teatud väljade põhjal. Kursuse järgmistes osades käsitleme visualiseerimist Kibanas ja loome lihtsa armatuurlaual. Tuleb mainida, et Logstashi konfiguratsioonifaili tuleb teatud olukordades pidevalt uuendada, näiteks kui soovime asendada väljaväärtuse numbrist sõnaga. Järgmistes artiklites teeme seda pidevalt.

Seega jälgige uuendusi (Telegram, Facebook, VK, TS Solution Blog), Yandex.Zen.

Allikas: habr.com