Iepriekšējā Mēs iepazināmies ar ELK steku, no kādiem programmatūras produktiem tas sastāv. Un pirmais uzdevums, ar kuru saskaras inženieris, strādājot ar ELK steku, ir nosūtīt žurnālus uz glabāšanu elasticsearch, lai tos vēlāk analizētu. Tomēr tas ir tikai vārdu līmenī, elasticsearch glabā žurnālus noteiktu lauku un vērtību veidā, tāpēc inženierim ir jāizmanto dažādi rīki, lai analizētu ziņu, kas tiek nosūtīta no gala sistēmām. To var izdarīt vairācos veidos — pašam uzrakstīt programmu, kas caur API pievieno dokumentus datubāzei, vai izmantot jau gatavus risinājumus. Šajā kursā mēs apskatīsim risinājumu Logstash, kas ir daļa no ELK steka. Mēs aplūkosim, kā nosūtīt žurnālus no gala sistēmām Logstash, un pēc tam konfigurēsim konfigurācijas failu, lai analizētu un novirzītu datus uz Elasticsearch datubāzi. Tam mēs izmantosim žurnālus no Check Point ugunsmūra kā ieejas sistēmu.
Kursā netiks aplūkota ELK steka uzstādīšana, jo par šo tēmu ir milzum daudz rakstu, mēs piedāvāsim pievērsties konfigurācijas aspektiem.
Sastādīsim rīcības plānu Logstash konfigurācijai:
- Pārbaudām, vai elasticsearch pieņems žurnālus (pārbaudām, vai ports strādā un ir atvērts).
- Apspriežam, kā mēs varam nosūtīt notikumus uz Logstash, izvēlamies metodi, realizējam.
- Konfigurējam Input Logstash konfigurācijas failā.
- Konfigurējam Output Logstash konfigurācijas failā debug režīmā, lai saprastu, kā izskatās žurnāla ziņojums.
- Konfigurējam Filtru.
- Konfigurējam pareizu Output Elasticsearch.
- Palaidam Logstash.
- Pārbaudām žurnālus Kibana.
Apskatīsim katru punktu sīkāk:
Pārbaude, vai elasticsearch pieņems žurnālus
Lai to izdarītu, varat pārbaudīt ar curl komandu piekļuvi Elasticsearch no sistēmas, kurā ir izvietots Logstash. Ja jums ir iestatīta autentifikācija, caur curl arī nosūtām lietotājvārdu/paroli, norādot portu 9200, ja to neesat mainījis. Ja atbilde izskatās kā norādīts zemāk, viss ir kārtībā.
[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 ~]$
Jaudājuma trūkums var liecināt par vairākām kļūdām: elasticsearch process nedarbojas, ir norādīts nepareizs ports vai ports ir bloķēts ugunsdzēsības sienā serverī, kurā izvietots elasticsearch.
Apsveram, kā nosūtīt žurnālus uz Logstash caur starpniekserveri Check Point.
No Check Point pārvaldības servera var nosūtīt žurnālus uz Logstash, izmantojot syslog, izmantojot rīku log_exporter, vairāk sīkāk varat iepazīties ar to šeit. , šeit atstāsim tikai komandu, kas izveido plūsmu:
cp_log_export add name check_point_syslog target-server <> target-port 5555 protocol tcp format generic read-mode semi-unified
<> — tas ir servera adrese, uz kura darbojas Logstash, target-port 5555 — ports, uz kuru tiks nosūtīti žurnāli, žurnālu nosūtīšana pa tcp var noslogot serveri, tāpēc dažos gadījumos labāk izmantot udp.
Iestatām INPUT Logstash konfigurācijas failā.
Pēc noklusējuma konfigurācijas fails atrodas direktorijā /etc/logstash/conf.d/. Konfigurācijas fails sastāv no trim nozīmīgām daļām: INPUT, FILTER, OUTPUT. INPUT mēs norādām, no kurienes sistēma ņems žurnālus, FILTER mēs analizējam žurnālu — iestatām, kā dalīt ziņas laukos un vērtībās, OUTPUT mēs iestatām izejošo plūsmu — kur nosūtīt analizētos žurnālus.
Vispirms iestatām INPUT, aplūkosim dažādas veidu iespējas — file, tcp un exe.
Tcp:
input {
tcp {
port => 5555
host => “10.10.1.205”
type => "checkpoint"
mode => "server"
}
}
mode => «server»
Norāda, ka Logstash pieņem savienojumus.
port => 5555
host => “10.10.1.205”
Pieņemam savienojumus pa IP adresi 10.10.1.205 (Logstash), ports 5555 — šim portam jābūt atļautam ugunsdzēsības sienas politikā.
type => «checkpoint»
Marķējam dokumentu, kas ir ļoti ērti, ja jums ir vairāki ienākošie savienojumi. Nākotnē katram savienojumam var uzrakstīt savu filtru, izmantojot loģisko struktūru if.
File:
input {
file {
path => "\/var\/log\/openvas_report\/"
type => "openvas"
start_position => "beginning"
}
}
Iestatījumu apraksts:
path => «\/var\/log\/openvas_report\/*»
Norādiet direktoriju, failus, kuri jāizlasa.
type => «openvas»
Notikuma tips.
start_position => «beginning»
Izmaiņu gadījumā fails tiek nolasīts pilnībā, ja uzstādīts «end», sistēma gaida jaunu ierakstu parādīšanos faila beigās.
Exec:
input {
exec {
command => "ls -alh"
interval => 30
}
}
Balstoties uz šo ievadi, tiks palaista (tikai!) shell komanda, un tās izvade tiks iekļauta žurnāla ziņojumā.
command => «ls -alh»
Komanda, kuras izvadei ir mūsu interese.
interval => 30
Komandas izsaukšanas intervāls sekundēs.
Lai saņemtu žurnālus no ugunsmūra, jānorāda filtrs tcp vai udp, atkarībā no tā, kā žurnāli tiek sūtīti uz Logstash.
Mēs konfigurējam Output Logstash konfigurācijas failā debuge režīmā, lai saprastu, kā izskatās žurnāla ziņojums.
Pēc INPUT konfigurēšanas nepieciešams saprast, kā izskatīsies žurnāla ziņojums, kādas metodes ir jāizmanto filtrēšanas (parsing) konfigurācijai.
Tāpēc mēs izmantosim filtrus, kas izvada rezultātu stdout, lai apskatītu sākotnējo ziņojumu, pilns konfigurācijas fails šobrīd izskatīsies šādi:
input
{
tcp
{
port => 5555
type => "checkpoint"
mode => "server"
host => “10.10.1.205”
}
}
output
{
if [type] == "checkpoint"
{
stdout { codec=> json }
}
}
Palaidiet komandu, lai pārbaudītu:
sudo /usr/share/logstash/bin//logstash -f /etc/logstash/conf.d/checkpoint.conf
Mēs redzam rezultātu, attēls ir klikšķināms:
Ja to kopē, tas izskatīsies šādi:
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,
Apskatot šos ziņojumus, saprotam, ka žurnāli ir veidoti šādā formātā: lauks = vērtība vai atslēga = vērtība, tāpēc ir piemērots filtrs, kas saucas kv. Lai pareizi izvēlētos filtrus katram konkrētajam gadījumam, būtu labi iepazīties ar tehnisko dokumentāciju vai jautāt draugam.
Iestatām Filtru
Iepriekšējā posmā izvēlējāmies kv, tālāk ir sniegta šī filtra konfigurācija:
filter {
if [type] == "checkpoint"{
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
Izvēlamies simbolu, pēc kura dalīsim lauku un vērtību — “=”. Ja mūsu žurnālā ir vienādi ieraksti, saglabājam datu bāzē tikai vienu eksemplāru, pretējā gadījumā jums izveidosies masīvs ar identiskām vērtībām, tātad, ja saņemam ziņojumu “foo = some foo=some”, ierakstām tikai foo = some.
Iestatām pareizu Izvadi ElasticSearch
Pēc tam, kad ir iestatīts Filtrs, varam izvietot žurnālus datu bāzē elasticsearch:
output
{
if [type] == "checkpoint"
{
elasticsearch
{
hosts => ["10.10.1.200:9200"]
index => "checkpoint-%{+YYYY.MM.dd}"
user => "tssolution"
password => "cool"
}
}
}
Ja dokuments ir parakstīts ar tipu checkpoint, saglabājam notikumu datu bāzē elasticsearch, kas pieņem savienojumus uz 10.10.1.200 uz 9200 portu pēc noklusējuma. Katra dokuments tiek saglabāts noteiktā indeksā, šajā gadījumā saglabājam indeksā "checkpoint-" + pašreizējā laika datums. Katram indeksam var būt noteikts lauku kopums vai tas tiek automātiski izveidots, kad parādās jauns lauks ziņojumā, lauku iestatījumus un to tipus var apskatīt mappings.
Ja jums ir iestatīta autentifikācija (par to runāsim vēlāk), obligāti jānorāda akreditīvi, lai ierakstītu konkrētā indeksā, šajā piemērā tas ir "tssolution" ar paroli "cool". Varat ierobežot lietotāju tiesības, lai ierakstītu žurnālus tikai konkrētā indeksā, un nekādus citus.
Palaižam Logstash.
Logstash konfigurācijas fails:
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"
}
}
}
Pārbaudām konfigurācijas failu uz pareizību:
/usr/share/logstash/bin//logstash -f checkpoint.conf
Palaižam Logstash procesu:
sudo systemctl start logstash
Pārbaudām, vai process ir uzsākts:
sudo systemctl status logstash
Pārbaudām, vai pieslēgums ir uzstādīts:
netstat -nat |grep 5555
Pārbaudām žurnālus Kibana.
Kad viss ir uzsākts, mēs ieejam Kibana - Discover un pārliecināmies, ka viss ir pareizi konfigurēts, attēls ir noklikšķināms!
Visi žurnāli ir vietā, un mēs varam redzēt visus laukus un to vērtības!
Secinājums
Mēs apskatījām, kā rakstīt Logstash konfigurācijas failu, rezultātā saņēmām visus lauku un vērtību parseru. Tagad mēs varam strādāt ar meklēšanu un veidot grafikus pēc noteiktiem laukiem. Tālāk kursā mēs apskatīsim vizualizāciju Kibana, izveidosim vieglu paneli. Jāatzīmē, ka Logstash konfigurācijas fails ir jāpievieno nepārtraukti noteiktās situācijās, piemēram, kad vēlamies mainīt lauka vērtību no skaitļa uz vārdu. Nākamajos rakstos mēs to pastāvīgi darīsim.
Tātad sekojiet līdzi jaunumiem (, , , ), .
Avots: habr.com
