
Në të kaluarën u takuam me stakun ELK, nga cilat produkte software përbëhet ai. Detyra e parë me të cilën përballet inxhinieri gjatë punës me stakun ELK është dërgimi i logëve për ruajtje në elasticsearch për analizin e mëvonshëm. Megjithatë, kjo është e lehtë vetëm në fjalë; elasticsearch ruan logët si dokumente me fusha dhe vlera të caktuara, dhe kjo do të thotë se inxhinieri duhet, duke përdorur mjete të ndryshme, të analizojë mesazhin që dërgohet nga sistemet përfundimtare. Kjo mund të bëhet në disa mënyra - ose të shkruash vetë një program që nëpërmjet API-së do të shtojë dokumente në bazë, ose të përdorësh zgjidhje të gatshme. Në këtë kurs ne do të shqyrtojmë zgjidhjen Logstash, e cila është pjesë e stakut ELK. Ne do të shohim si mund të dërgojmë logët nga sistemet përfundimtare në Logstash, dhe më pas do të konfiguroni skedarin e konfigurimit për analizimin dhe ridrejtuar në bazën e të dhënave Elasticsearch. Për këtë, si sistem hyrës do të marrim logët nga firewall-i Check Point.
Në kursin e këtij kursi nuk trajtohet instalimi i stakut ELK, pasi ekziston një numër të madh artikujsh në këtë temë; ne do të shqyrtojmë aspektin e konfigurimit.
Do të përgatitim një plan veprimi për konfigurimin e Logstash:
- Kontrolloni që elasticsearch do të pranojë logët (verifikimi i funksionalitetit dhe hapjes së portës).
- Shqyrtojmë se si mund të dërgojmë ngjarjet në Logstash, zgjedhim mënyrën dhe e realizojmë.
- Konfigurojmë Input në skedarin e konfigurimit të Logstash.
- Konfigurojmë Output në skedarin e konfigurimit të Logstash në mënyrë debug, për t'u kuptuar si duket mesazhi i logut.
- Konfigurojmë Filter.
- Konfigurojmë Output-in e saktë në ElasticSearch.
- Nisim Logstash.
- Kontrollojmë logët në Kibana.
Do të shqyrtojmë më në detaje secilën pikë:
Kontrolloni që elasticsearch do të pranojë logët
Për këtë mund të kontrolloni me komandën curl qasje në Elasticsearch nga sistemi ku është vendosur Logstash. Nëse keni vendosur autentifikimin, atëherë gjithashtu përmes curl dërgojmë përdoruesin/faqen, duke specifikuar portin 9200 nëse nuk e keni ndryshuar. Nëse përgjigja vjen e ngjashme me atë të dhënë më poshtë, atëherë gjithçka është në rregull.
[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 ~]$
Nëse përgjigja nuk ka arritur, mund të ketë disa lloje gabimesh: procesi elasticsearch nuk funksionon, porti është i gabuar, ose porti është bllokuar nga firewall-i në serverin ku ndodhet elasticsearch.
Të shohim si të dërgojmë log-et në Logstash nga firewall-i Check Point
Me serverin e menaxhimit Check Point, mund të dërgojmë log-et në Logstash përmes syslog, duke përdorur utilitarin log_exporter, më shumë detaje mund të gjeni në lidhjen e mëposhtme. , këtu do të lëmë vetëm komandën që krijon rrjedhën:
cp_log_export add name check_point_syslog target-server <> target-port 5555 protocol tcp format generic read-mode semi-unified
<> — adresa e serverit ku funksionon Logstash, target-port 5555 — porta në të cilën do të dërgojmë log-et, dërgimi i log-eve përmes tcp mund të ngarkojë serverin, prandaj në disa raste është më e saktë të përdorim udp.
Konfigurojmë INPUT në skedarin konfigurues të Logstash

Nga të dhënat e zakonshme, skedari konfigurues ndodhet në direktorinë /etc/logstash/conf.d/. Skeda përbëhet nga 3 pjesë të kuptueshme: INPUT, FILTER, OUTPUT. INPUT ne tregojmë nga ku sistemi do të marrë log-et, në FILTER parsimin e log-ut — konfiguroni si të ndahen mesazhet në fusha dhe vlera, në OUTPUT konfiguroni rrjedhën dalëse — ku do të dërgohen log-et e ndara.
Së pari do të konfigurojmë INPUT, duke shqyrtuar disa tipe që mund të jenë — file, tcp dhe exe.
Tcp:
input {
tcp {
port => 5555
host => "10.10.1.205"
type => "checkpoint"
mode => "server"
}
}
mode => «server»
Tregon se Logstash pranon lidhjet.
port => 5555
host => "10.10.1.205"
Pranojmë lidhjet në adresën IP 10.10.1.205 (Logstash), porta 5555 — porta duhet të jetë e lejuar nga politika e firewall-it.
type => «checkpoint»
Markojmë dokumentin, shumë e dobishme nëse keni disa lidhje të ardhshme. Më vonë, për secilën lidhje mund të shkruani filtrin tuaj duke përdorur konstrukcionin logjik if.
File:
input {
file {
path => "/var/log/openvas_report/*"
type => "openvas"
start_position => "beginning"
}
}
Përshkrimi i parametrave:
path => «/var/log/openvas_report/*»
Specifikoni direktorinë, skedarët që duhet të lexohen.
type => "openvas"
Lloji i ngjarjes.
start_position => "beginning"
Kur ndërroni skedarin, lexon skedarin në tërësi; nëse vendosni "end", sistemi pret shfaqjen e regjistrimeve të reja në fund të skedarit.
Exec:
input {
exec {
command => "ls -alh"
interval => 30
}
}
Me këtë input, ekzekutohet (vetëm!) një komandë shell dhe dalja e saj mbyllet brenda një mesazhi log.
command => "ls -alh"
Komanda, dalja e së cilës na intereson.
interval => 30
Intervali i thirrjes së komandës në sekonda.
Për të pranuar log-ët nga firewalla, ruajmë filtrin tcp или udp, në varësi të mënyrës se si dërgohen log-ët në Logstash.
Konfigurojmë Output në skedarin konfiguruese të Logstash në mënyrën debug, për të kuptuar si duket mesazhi log.
Pasi të kemi konfiguruar INPUT, është e nevojshme të kuptojmë si do duket mesazhi log, cilat metoda duhet të përdoren për të konfiguruar filtrin (parser) e log-ëve.
Për këtë, do të përdorim një filtër që jep rezultat në stdout për të parë mesazhin origjinal; skedari komplekse konfiguruese deri tani do të duket kështu:
input
{
tcp
{
port => 5555
type => "checkpoint"
mode => "server"
host => "10.10.1.205"
}
}
output
{
if [type] == "checkpoint"
{
stdout { codec=> json }
}
}
Ekzekutojmë komandën për kontroll:
sudo /usr/share/logstash/bin//logstash -f /etc/logstash/conf.d/checkpoint.conf
Shikojmë rezultatin, imazhi është klikues:
Nëse e kopjoni, do të duket kështu:
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,
Duke parë këto mesazhe, kuptojmë se log-et kanë formatin: fushë = vlerë ose çelës = vlerë, pra përshtatet filtri, i quajtur kv. Për të zgjedhur saktë filtrin për çdo rast specifik, do ishte mirë të njiheshim me ta në dokumentacionin teknik, ose të pyesnim një mik.
Konfigurojmë Filtrin
Në fazën e kaluar zgjodhëm kv, më poshtë paraqitet konfigurimi i këtij filtri:
filter {
if [type] == "checkpoint"{
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
Zgjidhim simbolin me të cilin do ndajmë fushën dhe vlerën — “=”. Nëse kemi në log ekzemplarë të njëjtë, ruajmë në bazë vetëm një kopje, përndryshe do të kemi një masiv të vlerave të njëjta, pra nëse kemi mesazhin “foo = some foo=some” shkruajmë vetëm foo = some.
Konfigurojmë Output-in e saktë në ElasticSearch
Pasi të jetë konfiguruar Filtri, mund të eksportohet log-et në bazë elasticsearch:
output
{
if [type] == "checkpoint"
{
elasticsearch
{
hosts => ["10.10.1.200:9200"]
index => "checkpoint-%{+YYYY.MM.dd}"
user => "tssolution"
password => "cool"
}
}
}
Nëse dokumenti është i nënshkruar me tipin checkpoint, ruajmë ngjarjen në bazën elasticsearch, e cila pranon lidhje në 10.10.1.200 në portin 9200 si defolt. Çdo dokument ruhet në një indeks të caktuar, në këtë rast ruajmë në indeksin "checkpoint-" + datën aktuale. Çdo indeks mund të ketë një grup të caktuar fushash, ose krijohet automatikisht kur shfaqet një fushë e re në mesazh, konfigurimet e fushave dhe tipin e tyre mund të shihen në mappings.
Nëse keni konfiguruar autorsinë (do ta shqyrtojmë më vonë), patjetër duhet të saktësohen kredencialet për të shkruar në një indeks të caktuar, në këtë shembull "tssolution" me fjalëkalim "cool". Mund të ndahen të drejtat e përdoruesve për të regjistruar log-et vetëm në një indeks të caktuar dhe askund tjetër.
Nisim Logstash.
Skeda konfiguruese e Logstash:
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"
}
}
}
Kontrollojmë skedën konfiguruese për saktësinë e përgatitjes:
/usr/share/logstash/bin//logstash -f checkpoint.conf

Nisim procesin Logstash:
sudo systemctl start logstash
Kontrollojmë nëse procesi është nisur:
sudo systemctl status logstash

Kontrollojmë nëse socket-i është ngritur:
netstat -nat |grep 5555
![]()
Kontrollojmë logët në Kibana.
Pas pas që gjithçka është nisur, hyjmë në Kibana — Discover dhe sigurohemi që gjithçka është konfiguruar siç duhet, figura është e klikuar!
Të gjitha log-et janë në vend dhe ne mund të shohim të gjitha fushat dhe vlerat e tyre!
Përfundim
Kemi shqyrtuar se si të shkruajmë skedarin e konfigurimit Logstash, dhe në rezultat kemi marrë një parser për të gjitha fushat dhe vlerat. Tani mund të punojmë me kërkimin dhe krijimin e grafikëve për fusha të caktuara. Më pas në kurs do të shqyrtojmë vizualizimin në Kibana dhe do të krijojmë një dashboard të thjeshtë. Duhet të përmendim se skedari i konfigurimit Logstash kërkon që të shkruhet vazhdimisht në disa situata, për shembull, kur dëshirojmë të zëvendësojmë vlerën e një fushe nga numër në fjalë. Në artikujt e ardhshëm do ta bëjmë këtë vazhdimisht.
Prandaj, qëndroni të informuar për azhurnimet (, , , ), .
Burimi: habr.com
