
Երեկ եղել է հայտարարված դյուրիական զրույցի սարք, իսկ այսում մենք այն կկառուցենք։ Մենք ծանոթացանք ELK ստեկի, որոնցից կազմված ծրագրային արտադրանքներ կան։ Եվ առաջին խնդիրը, որի հետ գործ инженер ս сталкивается, աշխատելով ELK ստեկի հետ, դա լոգերի ուղարկումն է՝ պահպանելու համար elasticsearch-ում՝ հետագա վերապատրաստման համար։ Սակայն, սա պարզապես խոսքում է, elasticsearch պահում է լոգերը որպես փաստաթղթեր որոշակի դաշտերով և արժեքներով, ինչը նշանակում է, որ инженерն անհրաժեշտ է օգտագործել տարբեր գործիքներ, որպեսզի ձևաբանի հաղորդումը, որը ուղարկվում է վերջնական համակարգերից։ Սա կարելի է անել մի քանի ձևերով՝ ինքնուրույն գրելով ծրագիր, որն API-ով կավելացնի փաստաթղթեր տվյալների բազա, կամ օգտագործելով արդեն պատրաստ լուծումներ։ Այս կուրսի շրջանակներում մենք կքննարկենք լուծում Logstash, որը պատկանում է ELK ստեկին։ Մենք կտեսնենք, թե ինչպես կարող ենք ուղարկել լոգեր վերջնական համակարգերից Logstash-ին, և ապա կկարգավորենք կոնֆիգուրացիոն ֆայլը՝ ձևաբանում և ուղղարկելու համար Elasticsearch տվյալների բազա։ Այս համար որպես մուտքային համակարգ վերցնում ենք Check Point զոդիչի լոգեր։
Այս կուրսի շրջանակներում ELK ստեկի տեղադրումը չքննարկվում, քանի որ այս թեմայի շուրջ շատ հոդվածներ կան, մենք կքննարկենք կոնֆիգուրացիոն բաղադրիչը։
Կազմենք Logstash-ի կոնֆիգուրացման գործողությունների պլան․
- Նախ, ստուգում ենք, թե արդյոք elasticsearch-ի կողմից լոգերը կդունին (ստուգում ենք աշխատանքունակությունը և պորտի բաց լինելը):
- Կքննարկենք, թե ինչպես կարող ենք ուղարկել իրադարձություններ Logstash-ին, ընտրենք կերպը, իրականացնել:
- Կարգավորում ենք Logstash-ի կոնֆիգուրացիոն ֆայլի Input-ը:
- Կարգավորում ենք Logstash-ի կոնֆիգուրացիոն ֆայլի Output-ը դեպագի ռեժիմում, որպեսզի հասկանալ, թե ինչպես է տեսվում լոգի հաղորդման տեղեկությունը:
- Կարգավորում ենք Filter:
- Կարգավորում ենք ճիշտ Output-ը Elasticsearch-ում:
- Սկսեցնում ենք Logstash-ը:
- Ստուգում ենք լոգերը Kibana-ում:
Բարձրապես ուշադրություն ենք դարձնում յուրաքանչյուր կետին․
Ստուգում ենք, թե արդյոք elasticsearch-ը կհետևի լոգերին:
Այս համար կարող եք ստուգել CURL հրամանով, որպեսզի մուտք ունենաք Elasticsearch-ից այն համակարգից, որտեղ Logstash-ը տեղադրված է: Եթե դուք կարգավորել եք աութենտիկացիան, ապա նաև CURL-ով փոխանցեք օգտվող/գործիք, նշեք 9200 պորտը, եթե այն չեք փոխել: Եթե վերևում նշվածին նման պատասխան է գալիս, ուրեմն ամեն ինչ լավ է:
[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" : "Դուք գիտեք, որոնման համար"
}
[elastic@elasticsearch ~]$
Եթե պատասխան չի եկել, ապա կարող է լինել մի քանի տեսակի սխալներ. elasticsearch-ը չի աշխատում, նշված է սխալ պորտ, կամ պորտը արգելափակված է ֆեղ Firewall-ի շնորհիվ այն սերվերում, որտեղ տեղադրված է elasticsearch-ը։
Կքննարկենք, թե ինչպես ուղարկել լոգերը Logstash-ին Check Point զոդիչից:
Check Point-ի կառավարման սերվերը կարող է ուղարկել լոգերը Logstash-ի միջոցով syslog, օգտագործելով log_exporter գործիքը, ավելի մանրամասն կարող եք ծանոթանալ դրան այս , այստեղ կթողնենք միայն այն հրամանն, որը ստեղծում է հոսքը:
cp_log_export add name check_point_syslog target-server <> target-port 5555 protocol tcp format generic read-mode semi-unified
<> — այն սերվերի հասցեն, որտեղ աշխատում է Logstash, target-port 5555 — այն портը, որի վրա ներկրելու ենք լոգերը, TCP-ով լոգերի ուղարկումը կարող է բեռնել սերվերը, այդ պատճառով որոշ դեպքերում ավելի ճիշտ է UDP օգտագործելը:
Կարգավորում ենք INPUT Logstash-ի config ֆայլում

Առաջին հերթին, config ֆայլին գտնվում է /etc/logstash/conf.d/ հատվածում։ Config ֆայլը բաղկացած է 3 իմաստային մասից՝ INPUT, FILTER, OUTPUT։ Ա INPUT մենք նշում ենք, թե որտեղից համակարգը հանդես է գալիս լոգերը, FILTER անընդհատ ենք լոգը — կարգավորում ենք, թե ինչպես բաժանել հաղորդագրությունը դաշտերի և արժեքների, OUTPUT կարգավորում ենք ելքային հոսքը — ուր կմեկնվեն բաժանված լոգերը:
Առաջին հերթին կարգավորում ենք INPUT-ը, անդրադարձնենք որոշ տեսակների, որոնք կարող են լինել — file, tcp և exec։
Tcp:
input {
tcp {
port => 5555
host => “10.10.1.205”
type => "checkpoint"
mode => "server"
}
}
mode => «server»
ընդսախոսում է, որ Logstash-ը ընդունում է միացումները։
port => 5555
host => “10.10.1.205”
Անդունում ենք միացումները IP հասցեից 10.10.1.205 (Logstash), порт 5555 — պորտը պետք է թույլ տրվի ֆայերուոլի քաղաքականությամբ:
type => «checkpoint»
Նշում ենք փաստաթուղթը, շատ հարմար է, եթե ունեք մի քանի մուտքային միացումներ։ Այլևս, յուրաքանչյուր միացման համար կարելի է գրել իր ֆիլտրը տրամաբանական կառուցվածքի միջոցով if:
File:
input {
file {
path => "\/var\/log\/openvas_report\/*"
type => "openvas"
start_position => "beginning"
}
}
Կարգավորումների նկարագրություն:
path => «\/var\/log\/openvas_report\/*»
Նշում ենք այդ каталогը, այն ֆայլերը, որոնք անհրաժեշտ է ընթերցել։
type => «openvas»
Համազօրի տեսակը։
start_position => «beginning»
Ֆայլի փոփոխման դեպքում ընթերցում է ֆայլը ամբողջությամբ, իսկ եթե «end» դնում եք, ապա համակարգը սպասում է ֆայլի վերջում նոր գրանցումների հայտնմանը։
Exec:
input {
exec {
command => "ls -alh"
interval => 30
}
}
Այս ինպուտի միջոցով կատարվում է (միայն!) shell հրամանի գործարկում և դրա արդյունքը փաթեթավորվում է լոգի հաղորդագրությունում։
command => «ls -alh»
Հրաման, որի արդյունքը մեզ հետաքրքրում է։
interval => 30
Հրամանի կատարելու ինտերվալը վայրկյաններով։
Միջնորդային ֆլիտրի միջոցով, այսինքն, ըստ ուղարկելու ձևի TCP կամ UDP, անհրաժեշտ է կարգավորել ֆիլտրը: tcp կամ udp, կախված նրանից, թե ինչպես են լոգերը ուղարկվում Logstash-ի:
Կարգավորում ենք Output Logstash-ի config ֆայլում debug ռեժիմում, որպեսզի հասկանալ, թե ինչպես է նայում լոգի հաղորդագրությունը:
Երբ INPUT-ը արդեն կարգավորված է, անհրաժեշտ է հասկանալ, թե ինչպես է նայում լոգի հաղորդագրությունը, ինչ մեթոդներ անհրաժեշտ են ֆիլտրը (parse) կարգավորելու համար:
Այս նպատակով օգտագործելու ենք ֆիլտր, որը արդյունք է տալիս stdout, որպեսզի դիտենք սկզբնական հաղորդագրությունը, ամբողջ config ֆայլը ներկայումս կունենա այս տեսքը:
թողարկում
{
tcp
{
port => 5555
type => "checkpoint"
mode => "server"
host => "10.10.1.205"
}
}
տպագրություն
{
if [type] == "checkpoint"
{
stdout { codec => json }
}
}
Դուրս ենք բերում մուտքային հրամանը՝ ստուգելու համար:
sudo /usr/share/logstash/bin//logstash -f /etc/logstash/conf.d/checkpoint.conf
Մեր արդյունքը տեսնում ենք, պատկերը սեղմելու قابل է:
Եթե սա պատճենեք, սա կտեսնվի:
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,
Տվյալ հաղորդումների վրա նայելով, հասկանում ենք, որ լոգերը դիտարկում են՝ դաշտ = արժեք կամ key = value, այսինքն՝ հարմար է kv ֆիլտրը։ Կեչանկ պետք բացահայտել Ֆիլտրի հենց այն կոնկրետ դեպքերին, լավ կլինի ծանոթանալ տեխնիկական փաստաթղթին կամ հարցնել ընկերոջ տարրերի մասին։
Ձևավորում ենք ֆիլտր:
Ներկոն փուլում ընտրել ենք kv, հաջորդում է այս ֆիլտրի կոնֆիգուրացիան:
ֆիլտր {
if [type] == "checkpoint" {
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
Ընտրում ենք սիմվոլը, որով բաժանելու ենք դաշտը և արժեքը՝ «=»։ Եթե մեր լոգում նույնական գրառումներ կան, պահում ենք տվյալներում միայն մեկ օրինակ, հակառակ դեպքում կունենաք նույնական արժեքների զանգված, այսինքն՝ եթե ունենք հաղորդագրություն «foo = some foo=some», գրանցում ենք միայն foo = some։
Ձևավորում ենք ճիշտ Output-ը ElasticSearch-ում:
ֆիլտրի կարգավորումից հետո հնարավոր է լոգերը տեղափոխել տվյալների բազա: elasticsearch:
տպագրություն
{
if [type] == "checkpoint"
{
elasticsearch
{
hosts => ["10.10.1.200:9200"]
index => "checkpoint-%{+YYYY.MM.dd}"
user => "tssolution"
password => "cool"
}
}
}
Եթե տեքստը ստորագրված է checkpoint-ի տեսակով, պահում ենք դեպքը ElasticSearch տվյալների բազայում, որը ընդունում է միացում 10.10.1.200 հասցեից 9200 նավահանգստի վրա։ Յուրաքանչյուր փաստաթուղթ պահվում է հատուկ ինդեքսում, այս դեպքում պահում ենք «checkpoint-» + ներկա ժամանակաքաների ամսաթիվը։ Յուրաքանչյուր ինդեքս կարող է անհատական դաշտերի հավաքածու ունենալ, կամ ավտոմատորեն ստեղծվել, երբ հաղորդման մեջ նոր դաշտ է հասանելի՝ դաշտերի կարգավորումների և դրանց տեսակների վերաբերյալ տեղեկություն կարելի է ստանալ mappings-ում։
Եթե ունեք հաստատված նույնականացում (մենք հետագայում կքննարկենք), պետք է նշել գիտակցված տվյալների մուտքագրելու համար, այս օրինակով «tssolution» տալով «cool» գաղտնաբառը։ Կարող եք սահմանել օգտատերերի իրավունքները միայն օգնելու համար տվյալների մուտքագրելու համար կոնկրետ ինդեքսում և ոչ մի այլում:
Մուտք գործեք Logstash:
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"
}
}
}
Մենք ստուգում ենք կոնֆիգուրացիոն ֆայլի կազմման ճիշտությունը:
/usr/share/logstash/bin//logstash -f checkpoint.conf

Մենք մեկնարկում ենք Logstash գործընթացը:
sudo systemctl start logstash
Մենք տեսնում ենք, որ գործընթացը մեկնարկվել է:
sudo systemctl status logstash

Մենք ստուգում ենք, թե արդյոք սոեքտը բարձրացել է:
netstat -nat |grep 5555
![]()
Ստուգում ենք լոգերը Kibana-ում:
Ամբողջացումից հետո մտնում ենք Kibana — Discover, համոզվելու համար, որ ամեն ինչ ճիշտ է կարգավորված, պատկերն ամբողջական է:
Բոլոր լոգերը տեղակայված են և մենք կարող ենք տեսնել բոլոր դաշտերն ու դրանց արժեքները:
Ավարտ
Մենք դիտարկել ենք, թե ինչպես գրել Logstash կոնֆիգուրացիոն ֆայլ, արդյունքում ստացել ենք բոլոր դաշտերի և արժեքների թեքող: Այժմ մենք կարող ենք աշխատել որոնման և որոշակի դաշտերի հիման վրա գրաֆիկներ կազմելու հետ: Course-ի ընթացքում մենք դիտարկելու ենք Kibana-ի վիզուալիզացիան, Created простой дашбоर्ड: Worth mentioning, որ Logstash կոնֆիգուրացիոն ֆայլը պահանջվում է միշտ լրացնել որոշակի իրավիճակներում, օրինակ, երբ ցանկանում ենք փոխել դաշտի արժեքը թվից բառի: Հաջորդ հոդվածներում մենք մշտապես կանենք դա:
Հետևեք թարմացումներին (, , , ), .
Ընտանիք: habr.com
