Viimases me kohtusime ELK virn, millistest tarkvaratoodetest see koosneb. Ja esimene ülesanne, millega insener ELK-virnaga töötades silmitsi seisab, on logide saatmine elastsearchi salvestamiseks hilisemaks analüüsiks. See on aga lihtsalt huul, elasticsearch salvestab logid teatud väljade ja väärtustega dokumentide kujul, mis tähendab, et insener peab lõppsüsteemidest saadetava sõnumi sõelumiseks kasutama erinevaid tööriistu. Seda saab teha mitmel viisil – kirjutada ise programm, mis API abil dokumendid andmebaasi lisab, või kasutada valmislahendusi. Sellel kursusel kaalume lahendust Logstash, mis on osa ELK pinust. Vaatame, kuidas saame lõpp-punktisüsteemidest logisid Logstashi saata, ja seejärel seadistame konfiguratsioonifaili, et sõeluda ja suunata Elasticsearchi andmebaasi. Selleks võtame sissetuleva süsteemina Check Pointi tulemüüri logid.
Kursus ei hõlma ELK virna paigaldamist, kuna sellel teemal on tohutult palju artikleid; käsitleme konfiguratsioonikomponenti.
Koostame Logstashi konfigureerimise tegevuskava:
- Kontrollige, kas elasticsearch aktsepteerib logisid (pordi funktsionaalsuse ja avatuse kontrollimine).
- Mõtleme, kuidas saaksime sündmusi Logstashi saata, valida meetodi ja seda rakendada.
- Konfigureerime sisendi Logstashi konfiguratsioonifailis.
- Konfigureerime väljundi Logstashi konfiguratsioonifailis silumisrežiimis, et mõista, kuidas logiteade välja näeb.
- Filtri seadistamine.
- Õige väljundi seadistamine ElasticSearchis.
- Logstash käivitub.
- Palkide kontrollimine Kibanas.
Vaatame iga punkti üksikasjalikumalt:
Kontrollige, kas elasticsearch aktsepteerib logisid
Selleks saate kasutada käsku curl, et kontrollida juurdepääsu Elasticsearchile süsteemist, kuhu Logstash on juurutatud. Kui teil on autentimine seadistatud, edastame kasutaja/parooli ka curli kaudu, kui te pole seda muutnud, täpsustades pordi 9200. Kui saate allolevale sarnase vastuse, on kõik korras.
[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 ~]$
Kui vastust ei saada, võib esineda mitut tüüpi tõrkeid: elasticsearch protsess ei tööta, on määratud vale port või pordi blokeerib tulemüür serveris, kuhu elasticsearch on installitud.
Vaatame, kuidas saate kontrollpunkti tulemüürist Logstashi logisid saata
Check Pointi haldusserverist saad logisid saata Logstashi läbi syslogi kasutades utiliiti log_exporter, selle kohta saad lähemalt lugeda siit , jätame siia ainult käsu, mis loob voo:
cp_log_export lisamise nimi check_point_syslog sihtserver < > sihtport 5555 protokolli tcp vormingu üldine lugemisrežiim poolühendatud
< > - serveri aadress, millel Logstash töötab, sihtport 5555 - port, kuhu logid saadame, logide saatmine tcp kaudu võib serverit laadida, nii et mõnel juhul on õigem kasutada udp-d.
SISENDI seadistamine Logstashi konfiguratsioonifailis
Vaikimisi asub konfiguratsioonifail kataloogis /etc/logstash/conf.d/. Konfiguratsioonifail koosneb 3 tähendusrikkast osast: INPUT, FILTER, OUTPUT. IN INPUT näitame, kust süsteem logid võtab FILTER parsi logi – määrake, kuidas sõnum väljadeks ja väärtusteks jagada OUTPUT konfigureerime väljundvoo - kuhu parsitud logid saadetakse.
Kõigepealt konfigureerime SISEND, kaalume mõningaid tüüpe, mis võivad olla - fail, tcp ja exe.
TCP:
input {
tcp {
port => 5555
host => “10.10.1.205”
type => "checkpoint"
mode => "server"
}
}
režiim => "server"
Näitab, et Logstash aktsepteerib ühendusi.
port => 5555
host => “10.10.1.205”
Aktsepteerime ühendusi IP-aadressi 10.10.1.205 (Logstash), pordi 5555 kaudu – port peab tulemüüripoliitikaga lubatud olema.
tüüp => "kontrollpunkt"
Märgistame dokumendi, väga mugav, kui teil on mitu sissetulevat ühendust. Seejärel saate iga ühenduse jaoks kirjutada oma filtri, kasutades loogilist if-konstruktsiooni.
File:
input {
file {
path => "/var/log/openvas_report/*"
type => "openvas"
start_position => "beginning"
}
}
Seadete kirjeldus:
path => "/var/log/openvas_report/*"
Märgime kataloogi, kus faile tuleb lugeda.
tüüp => "openvas"
Sündmuse tüüp.
algus_positsioon => "algus"
Faili muutmisel loeb see kogu faili; kui määrate "lõpp", ootab süsteem uute kirjete ilmumist faili lõppu.
Teostaja:
input {
exec {
command => "ls -alh"
interval => 30
}
}
Seda sisendit kasutades käivitatakse (ainult!) shellikäsk ja selle väljund muudetakse logiteateks.
käsk => "ls -alh"
Käsk, mille väljund meid huvitab.
intervall => 30
Käskude kutsumise intervall sekundites.
Tulemüürist logide vastuvõtmiseks registreerime filtri tcp või Udp, olenevalt sellest, kuidas logid Logstashi saadetakse.
Konfigureerime väljundi Logstashi konfiguratsioonifailis silumisrežiimis, et mõista, kuidas logiteade välja näeb
Pärast INPUT-i konfigureerimist peame mõistma, kuidas logiteade välja näeb ja milliseid meetodeid tuleb logifiltri (parser) konfigureerimiseks kasutada.
Selleks kasutame algse sõnumi vaatamiseks filtrit, mis väljastab tulemuse stdout-i, hetkel näeb täielik konfiguratsioonifail välja selline:
input
{
tcp
{
port => 5555
type => "checkpoint"
mode => "server"
host => “10.10.1.205”
}
}
output
{
if [type] == "checkpoint"
{
stdout { codec=> json }
}
}
Kontrollimiseks käivitage käsk:
sudo /usr/share/logstash/bin//logstash -f /etc/logstash/conf.d/checkpoint.conf
Näeme tulemust, pilt on klikitav:
Kui kopeerite selle, näeb see välja järgmine:
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,
Neid sõnumeid vaadates saame aru, et logid näevad välja kujul: väli = väärtus või võti = väärtus, mis tähendab, et filter nimega kv sobib. Iga konkreetse juhtumi jaoks õige filtri valimiseks oleks hea nendega tutvuda tehnilises dokumentatsioonis või küsida sõbralt.
Filtri seadistamine
Viimases etapis valisime kv, selle filtri konfiguratsioon on esitatud allpool:
filter {
if [type] == "checkpoint"{
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
Valime sümboli, millega jagame välja ja väärtuse - “=”. Kui meil on logis identsed kirjed, salvestame andmebaasi ainult ühe eksemplari, vastasel juhul saate tulemuseks identsete väärtuste massiivi, st kui meil on teade "foo = mõned foo=some", kirjutame ainult foo = mõned.
Õige väljundi seadistamine ElasticSearchis
Kui filter on konfigureeritud, saate logid andmebaasi üles laadida elastneotsing:
output
{
if [type] == "checkpoint"
{
elasticsearch
{
hosts => ["10.10.1.200:9200"]
index => "checkpoint-%{+YYYY.MM.dd}"
user => "tssolution"
password => "cool"
}
}
}
Kui dokument on allkirjastatud kontrollpunkti tüübiga, salvestame sündmuse elasticsearch andmebaasi, mis vaikimisi aktsepteerib ühendusi 10.10.1.200 pordis 9200. Iga dokument salvestatakse kindlasse registrisse, sel juhul salvestame indeksisse “kontrollpunkt-” + praegune kellaaeg. Igal registril võib olla kindel väljade komplekt või see luuakse automaatselt, kui sõnumisse ilmub uus väli; välja seadeid ja nende tüüpi saab vaadata vastendustes.
Kui teil on autentimine seadistatud (vaatame seda hiljem), tuleb täpsustada konkreetsesse registrisse kirjutamise mandaadid, antud näites on see “tssolution” parooliga “cool”. Saate eristada kasutajaõigusi logide kirjutamiseks ainult konkreetsesse registrisse ja mitte enam.
Käivitage Logstash.
Logstashi 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äivitage Logstashi protsess:
sudo systemctl start logstash
Kontrollime, kas protsess on alanud:
sudo systemctl olek logstash
Kontrollime, kas pistikupesa on üleval:
netstat -nat |grep 5555
Palkide kontrollimine Kibanas.
Kui kõik töötab, minge Kibana - Discover, veenduge, et kõik on õigesti seadistatud, pilt on klõpsatav!
Kõik logid on paigas ja näeme kõiki välju ja nende väärtusi!
Järeldus
Vaatasime, kuidas kirjutada Logstashi konfiguratsioonifaili ja selle tulemusena saime kõikidest väljadest ja väärtustest parseri. Nüüd saame töötada konkreetsete väljade otsimise ja joonistamisega. Järgmisena vaatleme Kibana visualiseerimist ja loome lihtsa armatuurlaua. Tasub mainida, et Logstashi konfiguratsioonifaili tuleb teatud olukordades pidevalt uuendada, näiteks kui soovime välja väärtuse asendada numbrist sõnaga. Järgmistes artiklites teeme seda pidevalt.
Nii et jääge lainel (, , , ), .
Allikas: www.habr.com