
Në të kaluarën më takuam me stack-un ELK, nga cilat produkte softuerike përbëhet. Detyra e parë me të cilën përballet inxhinieri kur punon me stack-un ELK është dërgimi i logeve për ruajtje në Elasticsearch për analizë të mëvonshme. Megjithatë, kjo është thjesht në fjalë; Elasticsearch ruan loget si dokumente me fusha dhe vlera të caktuara, dhe kështu inxhinieri, duke përdorur mjetet e ndryshme, duhet të parserojë mesazhin që dërgohet nga sistemet përfundimtare. Kjo mund të bëhet në disa mënyra: mund të shkruani vetë një program që do të shtojë dokumente në databazë përmes API-së, ose të përdorni zgjidhje të gatshme. Gjatë këtij kursi, ne do të shqyrtojmë zgjidhjen Logstash, e cila është pjesë e stack-ut ELK. Do të shohim si mund të dërgojmë loge nga sistemet përfundimtare në Logstash dhe pastaj do të konfigurojmë skedarin konfiguruese për parserimin dhe redirektimin në databazën Elasticsearch. Për këtë, do të marrim loget nga firewall-i Check Point si sistemin hyrës.
Në kuadër të kursit nuk do të shqyrtojmë instalimin e ELK stack, pasi ekzistojnë shumë artikuj mbi këtë temë; do të përqendrohemi në komponentin e konfigurimit.
Do të përgatisim një plan veprimi për konfigurimin e Logstash:
- Kontrolli që elasticsearch do të pranojë logjet (kontrollimi i funksionimit dhe hapjes së portit).
- Shikojmë se si mund të dërgojmë ngjarje në Logstash, zgjedhim një metodë, dhe e realizojmë.
- Konfigurojmë Input në skedarin e konfigurimit të Logstash.
- Konfigurojmë Output në skedarin e konfigurimit të Logstash në mënyrën debug, për të kuptuar si duket mesazhi i logut.
- Konfigurojmë Filter.
- Konfigurojmë Output-in e saktë në ElasticSearch.
- Nisim Logstash.
- Kontrollojmë logjet në Kibana.
Të shqyrtojmë më në detaje çdo pikë:
Kontrolli që elasticsearch do të pranojë logjet
Për këtë mund të verifikoni me komandën curl qasjen në Elasticsearch nga sistemi ku është vendosur Logstash. Nëse keni konfigurur autentikimin, atëherë gjithashtu nëpërmjet curl jepni përdoruesin/fjalëkalimin, duke treguar portin 9200 nëse nuk e keni ndryshuar. Nëse merrni një përgjigje të ngjashme me atë që është treguar 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ërgjigjja nuk ka ardhur, mund të ketë disa lloje gabimesh: procesi elasticsearch nuk funksionon, porti i caktuar nuk është i saktë, ose porti është bllokuar nga firewalla në serverin ku ndodhet elasticsearch.
Të shqyrtojmë si mund të dërgojmë loget në Logstash nga firewalle check point.
Me serverin e menaxhimit Check Point, loget mund të dërgohen në Logstash përmes syslog, duke përdorur utilitarin log_exporter, më shumë detaje mund të gjenden në këtë , 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 po punon Logstash, target-port 5555 â porti nĂ« tĂ« cilin do tĂ« dĂ«rgojmĂ« logjet, dĂ«rgimi i logjeve me tcp mund ta ngarkojĂ« serverin, prandaj nĂ« disa raste Ă«shtĂ« mĂ« korrekt tĂ« pĂ«rdorim udp.
Konfigurojmë INPUT në skedarin e konfigurimit të Logstash.

NĂ« mĂ«nyrĂ« default, skedari i konfigurimit ndodhet nĂ« drejtori /etc/logstash/conf.d/. Skedari i konfigurimit pĂ«rbĂ«het nga 3 pjesĂ« kuptimplote: INPUT, FILTER, OUTPUT. INPUT ne specifikojmĂ« nga ku sistemi do tĂ« marrĂ« logjet, nĂ« FILTER ne analizojmĂ« logun â konfigurojmĂ« se si ta ndajmĂ« mesazhin nĂ« fusha dhe vlera, nĂ« OUTPUT konfigurojmĂ« rrjedhĂ«n dalĂ«se â ku do tĂ« dĂ«rgohen logjet e pĂ«rpunuara.
SĂ« pari, tĂ« konfigurojmĂ« INPUT, le tĂ« shqyrtojmĂ« disa lloje 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 lidhje.
port => 5555
host => â10.10.1.205â
PranojmĂ« lidhje me IP adresĂ«n 10.10.1.205 (Logstash), porti 5555 â porti duhet tĂ« jetĂ« i lejuar nga politika e firewall-it.
type => «checkpoint»
Kemi dokumentin, shumë e dobishme në rast se keni disa lidhje hyrëse. Më vonë, për secilën lidhje mund të shkruani filtrin tuaj me ndihmën e konstruktit logjik if.
File:
input {
file {
path => "/var/log/openvas_report/*"
type => "openvas"
start_position => "beginning"
}
}
Përshkrimi i renditjeve:
path => "/var/log/openvas_report/*"
Specifikoni direktorinë, skedarët në të cilët duhet të lexohen.
type => "openvas"
Lloji i ngjarjes.
start_position => "beginning"
Nëse ndodhen ndryshime në skedar, lexon skedarin në tërësi, nëse vendoset "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!) komanda shell dhe dalja e saj paketizohet në mesazh 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 firewalle, shkruajmë filterin tcp ose udp, në varësi të mënyrës se si dërgohen logët në Logstash.
Konfigurojmë Output në skedarin e konfigurimit të Logstash në mënyrën debug, që të kuptojmë si duket mesazhi i logut.
Pas pas, pasi ne kemi konfiguruar INPUT, është e nevojshme të kuptojmë si do të duket mesazhi i log-ut, cilat metoda duhet të përdorim për të konfiguruar filtrin (parserin) e log-eve.
Për këtë do të përdorim një filtër që jep rezultat në stdout për të parë mesazhin origjinal, skedari i plotë i konfigurimit 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 }
}
}
Nisjmë komandën për verifikim:
sudo /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/checkpoint.conf
Shohim rezultatin, imazhi është klikues:
Nëse e kopjojmë, 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
Duke të dhënat e mesazhit, kuptojmë se loget kanë format: fushë = vlerë ose çelës = vlerë, prandaj filtrimi i quajtur kv është i përshtatshëm. Për të zgjedhur siç duhet filtrin për çdo rast të veçantë, do të ishte mirë t'i studiojmë ato në dokumentacionin teknik, ose të pyesim një mik.
Konfigurojmë Filtrin
Në hapin e kaluar, zgjodhëm kv, më poshtë jepet konfigurimi i këtij filtri:
filter {
if [type] == "checkpoint" {
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
Zgjidhni simbolin me tĂ« cilin do tĂ« ndajmĂ« fushĂ«n dhe vlerĂ«n â â=â. NĂ«se kemi regjistrime tĂ« barabarta nĂ« log, ruajmĂ« vetĂ«m njĂ« ekzemplar nĂ« bazĂ«, pĂ«rndryshe do tĂ« krijojmĂ« njĂ« varg me vlera tĂ« ngjashme. Pra, nĂ«se kemi mesazh âfoo = some foo=someâ shkruajmĂ« vetĂ«m foo = some.
Konfiguroni Output-in e saktë në ElasticSearch
Pasi të jetë konfigurimi i Filter-it, mund të eksportoni log-ët 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Ă« nĂ«nshkruar me tipin checkpoint, ruajmĂ« ngjarjen nĂ« bazĂ«n elasticsearch, e cila pranon lidhje nĂ« 10.10.1.200 nĂ« portin 9200 nga standart. Ădo dokument ruhet nĂ« njĂ« indeks tĂ« caktuar, nĂ« kĂ«tĂ« rast ruajmĂ« nĂ« indeksin «checkpoint-» + data e tanishme. Ădo indeks mund tĂ« ketĂ« njĂ« grup tĂ« caktuar fushash, ose krijohet automatikisht me shfaqjen e njĂ« fushe tĂ« re nĂ« mesazh, konfigurimet e fushave dhe tipin e tyre mund tĂ« studiohen nĂ« mappings.
Nëse keni konfiguruar autentifikimin (do ta shqyrtojmë më vonë), do të duhet patjetër të jepen kredencialet për të shkruar në një indeks të caktuar, në këtë rast "tssolution" me fjalëkalimin "cool". Mund të kufizoni të drejtat e përdoruesve për të shkruar logjet vetëm në një indeks të caktuar dhe asnjë tjetër.
Nisni Logstash.
Skedari i konfigurimit të 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"
}
}
}
Kontrolloni skedarin e konfigurimit për saktësinë e përgatitjes:
/usr/share/logstash/bin//logstash -f checkpoint.conf

Nisni procesin Logstash:
sudo systemctl start logstash
Kontrolloni nëse procesi u nis:
sudo systemctl status logstash

Kontrolloni nëse soketi u ngrit:
netstat -nat |grep 5555
![]()
Kontrollojmë logjet në Kibana.
Pasi gjithçka Ă«shtĂ« nisur, hyni nĂ« Kibana â Discover, sigurohuni qĂ« gjithçka Ă«shtĂ« konfiguruar siç duhet, imazhi Ă«shtĂ« klikues!
Të gjitha logjet janë në vend dhe ne mund të shohim të gjitha fushat dhe vlerat e tyre!
Përfundimi
Ne shqyrtuam se si të shkruajmë një skedë konfiguruese për Logstash, duke rezultuar në 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ë trajtojmë vizualizimin në Kibana dhe do të krijojmë një panel të thjeshtë. Duhet përmendur se skeda konfiguruese e Logstash duhet të përmirësohet vazhdimisht në situata të caktuara, për shembull, kur duam të zëvendësojmë vlerën e një fushe nga një numër në një fjalë. Në artikujt e ardhshëm, do ta bëjmë këtë vazhdimisht.
Pra, ndiqni përditësimet (, , , ), .
Burimi: habr.com
