🥇2. Elastic stack: анализ на security логовете. Logstash

2. Elastic stack: анализ на security логов. Logstash

В предходната статията с кого се запознахме с 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 могат да се изпращат логове в 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.

2. Elastic stack: анализ на security логов. Logstash

По подразбиране конфигурационният файл се намира в директорията /etc/logstash/conf.d/. Конфигурационният файл се състои от три логически части: 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 => "начало"
При промяна на файла, се чете целия файл; ако е зададено "край", системата очаква нови записи в края на файла.

Exec:

input {
  exec {
    command => "ls -alh"
    interval => 30
  }
}

По този вход се стартира (само!) shell команда и нейният изход се обгръща в лог съобщение.

command => "ls -alh"
Команда, чийто изход ни интересува.

interval => 30
Интервал за извикване на командата в секунди.

За да приемем логовете от защитната стена, определяме филтър tcp или udp, в зависимост от начина, по който се изпращат логовете в Logstash.

Конфигурираме Output в конфигурационния файл на Logstash в режим дебъг, за да разберем как изглежда лог съобщението.

След като конфигурираме INPUT, трябва да разберем как ще изглежда лог съобщението и какви методи трябва да използваме за настройка на филтъра (парсера) на логовете.

За целта ще използваме филтър, който дава резултати в 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
Виждаме резултата, картинката еClickable:

Ако копираме това, ще изглежда така:

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. За да изберем правилния филтър за всеки конкретен случай, би било добре да се запознаем с тях в техническата документация или да попитаме приятел.

Настройваме Filter

На предишния етап избрахме 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
2. Elastic stack: анализ на security логов. Logstash

Стартираме процеса Logstash:
sudo systemctl start logstash

Проверяваме дали процесът е стартирал:
sudo systemctl status logstash

2. Elastic stack: анализ на security логов. Logstash

Проверяваме дали сокетът е активен:
netstat -nat |grep 5555

Проверяваме логовете в Kibana.

След като всичко е пуснато, влизаме в Kibana — Discover и се уверяваме, че всичко е настроено правилно, изображението е кликаемo!

Всички логове са на място и можем да видим всички полета и техните стойности!

Заключение

Разгледахме как да пишем конфигурационен файл за Logstash и в резултат получихме парсер за всички полета и стойности. Сега можем да работим с търсене и визуализиране на графики на определени полета. В следващите части на курса ще разгледаме визуализацията в Kibana и ще създадем прост дашборд. Трябва да споменем, че конфигурационният файл на Logstash трябва да се дописва постоянно в определени ситуации, например, когато искаме да заменим стойност на поле от число на дума. В следващите статии ще го правим постоянно.

Така че следете за обновленията (Telegram, Facebook, VK, TS Solution Blog), Яндекс.Дзен.

Източник: habr.com