В миналото запознахме се с ELK стека, от какви софтуерни продукти се състои. Първата задача, с която се сблъсква инженерът при работа с ELK стека, е изпращането на логовете за съхранение в Elasticsearch за последващ анализ. Въпреки че на пръв поглед изглежда просто, Elasticsearch съхранява логовете под формата на документи с определени полета и стойности, което означава, че инженерът трябва да използва различни инструменти, за да парсне съобщението, което се изпраща от крайни системи. Това може да се направи по няколко начина — да напише програма, която чрез API добавя документи в базата, или да използва вече готови решения. В рамките на този курс ще разгледаме решение Logstash, което е част от ELK стека. Ще видим как можем да изпратим логовете от крайни системи в Logstash и след това ще настроим конфигурационния файл за парсване и пренасочване в базата данни Elasticsearch. За целта ще вземем логовете от защитната стена Check Point като входна система.
В рамките на курса не разглеждаме инсталирането на ELK стека, тъй като съществува огромно количество статии по тази тема; ще разгледаме конфигурационната част.
Създаваме план за действие по конфигурация на Logstash:
- Проверка дали Elasticsearch ще приема логовете (проверка за работоспособност и отвореност на порта).
- Разглеждаме по какъв начин можем да изпратим събития в Logstash, избираме метод и го реализираме.
- Настройваме Input в конфигурационния файл на Logstash.
- Настройваме Output в конфигурационния файл на Logstash в режим на дебъг, за да разбера как изглежда лог съобщението.
- Настройваме 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" : "You Know, for Search"
}
[elastic@elasticsearch ~]$
Ако отговорът не дойде, може да има няколко вида грешки: elasticsearch процесът не работи, неправилен порт е указан или портът е блокиран от фаервола на сървъра, на който е инсталиран elasticsearch.
Нека разгледаме как можем да изпратим логовете в Logstash с помощта на Check Point.
С Check Point management server можем да изпратим логовете в 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.
По подразбиране конфигурационният файл се намира в директорията /etc/logstash/conf.d/. Конфигурационният файл се състои от 3 смислени части: INPUT, FILTER, OUTPUT. В INPUT указваме откъде системата ще взема логовете, в FILTER парсим логовете — настройваме как да разделим съобщението на полета и стойности; в OUTPUT настройваме изходящия поток — къде ще бъдат изпратени парсираните логове.
Първо ще настроим INPUT и ще разгледаме някои типове, които могат да бъдат — file, tcp и exe.
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, в зависимост от начина на изпращане на логовете в Logstash.
Настройваме изхода в конфигурационния файл на Logstash в режим на отстраняване на грешки, за да разберем как изглежда лог съобщението.
След като конфигурираме входа, трябва да разберем как ще изглежда лог съобщението и какви методи е необходимо да използваме за настройване на филтъра (парсера) за логовете.
За целта ще използваме филтър, който извежда резултата в stdout, за да прегледаме оригиналното съобщение; актуалният конфигурационен файл ще изглежда така:
input
{
tcp
{
port => 5555
type => "checkpoint"
mode => "server"
host => “10.10.1.205"
}
}
output
{
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,
Гледайки на данните съобщения, разбираем, че логовете имат вид: поле = стойност или ключ = стойност, следователно подходящият филтър се нарича kv. За да изберем правилния филтър за всеки конкретен случай, би било полезно да се запознаем с тях в техническата документация или да попитаме приятел.
Настройваме Филтри.
На предишния етап избрахме kv, по-долу е представена конфигурацията на този филтър:
filter {
if [type] == "checkpoint"{
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
Избираме символа, по който ще разделяме полето и стойността — “=”. Ако в логовете имаме идентични записи, запазваме в базата само един екземпляр, в противен случай ще получите масив от идентични стойности, тоест, ако имаме съобщение “foo = some foo=some”, записваме само foo = some.
Настройваме коректен Output в ElasticSearch.
След като филтърът е настроен, можем да изнесем логовете в базата. elasticsearch:
output
{
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 и в резултат получихме парсер за всички полета и стойности. Сега можем да работим с търсене и съставяне на графики по определени полета. В следващите материали ще разгледаме визуализацията в Kibana и ще създадем прост дашборд. Трябва да спомена, че конфигурационният файл на Logstash трябва постоянно да се дописва в определени ситуации, например, когато искаме да заменим стойността на поле от число на дума. В последващи статии ще го правим постоянно.
Така че следете обновленията (, , , ), .
Източник: habr.com
