2. Elastic stack: analys av säkerhetsloggar. Logstash

Tidigare artikeln Vi har blivit bekanta med ELK-stacken, vilken programvara den består av. Den första uppgiften en ingenjör ställs inför när man arbetar med ELK-stacken är att skicka loggar för lagring i Elasticsearch för efterföljande analys. Men det är bara teori; Elasticsearch lagrar loggar som dokument med specifika fält och värden, vilket innebär att ingenjören måste använda olika verktyg för att bearbeta meddelandet som skickas från de slutliga systemen. Detta kan göras på flera sätt – antingen skriva ett program själv som via API lägger till dokument i databasen eller använda redan befintliga lösningar. Inom ramen för denna kurs kommer vi att titta på en lösning Logstash, som är en del av ELK-stacken. Vi kommer att se hur vi kan skicka loggar från de slutliga systemen till Logstash och sedan konfigurera konfigurationsfilen för att analysera och omdirigera till Elasticsearch-databasen. För detta tar vi loggar från Check Point-brandväggen som vår ingångssystem.

I kursen kommer vi inte att titta på installationen av ELK-stacken, eftersom det finns ett stort antal artiklar om detta ämne; vi kommer att fokusera på konfigurationsaspekten.

Vi ska upprätta en handlingsplan för Logstash-konfigurationen:

1. Kontrollera att Elasticsearch kommer att ta emot loggar (testa funktionaliteten och öppenheten av porten).
2. Vi undersöker hur vi kan skicka händelser till Logstash, väljer en metod och implementerar den.
3. Vi konfigurerar Input i Logstash-konfigurationsfilen.
4. Vi konfigurerar Output i Logstash-konfigurationsfilen i debug-läge för att förstå hur loggmeddelandet ser ut.
5. Vi konfigurerar Filter.
6. Vi konfigurerar korrekt Output i Elasticsearch.
7. Vi startar Logstash.
8. Vi kontrollerar loggarna i Kibana.

Vi kommer att gå igenom varje punkt mer detaljerat:

Kontrollera att Elasticsearch kommer att ta emot loggar

För detta kan du kontrollera tillgången till Elasticsearch från systemet där Logstash är installerat med kommandot curl. Om du har autentisering konfigurerad, överför även användarnamn/lösenord via curl och ange port 9200 om du inte har ändrat den. Om svaret ser ut som nedan, så är allt i sin ordning.

[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" : "Du vet, för sökning"
}
[elastic@elasticsearch ~]$

Om inget svar kom, kan det finnas flera typer av fel: elasticsearch-processen fungerar inte, fel port anges, eller porten blockeras av brandväggen på servern där elasticsearch är installerad.

Vi undersöker hur man kan skicka loggar till Logstash genom brandväggen Check Point.

Med Check Point management server kan loggar skickas till Logstash via syslog, med hjälp av verktyget log_exporter, mer information om detta kan hittas i den här artikeln, här lämnar vi bara kommandot som skapar en ström:

cp_log_export add name check_point_syslog target-server <> target-port 5555 protocol tcp format generic read-mode semi-unified

<> — adressen till servern där Logstash körs, target-port 5555 — porten som vi kommer att skicka loggar till, att skicka loggar via tcp kan belasta servern, så i vissa fall är det mer korrekt att använda udp.

Vi konfigurerar INPUT i Logstash-konfigurationsfilen.

Som standard ligger konfigurationsfilen i katalogen /etc/logstash/conf.d/. Konfigurationsfilen består av 3 meningsfulla delar: INPUT, FILTER, OUTPUT. I INPUT angiver vi från vilket källsystem loggarna ska hämtas, i FILTER vi parser loggen — ställer in hur meddelandet ska delas upp i fält och värden, i OUTPUT konfigurerar vi utströmmen — vart de parsade loggarna ska skickas.

Först konfigurerar vi INPUT, vi tittar på några typer som kan finnas — file, tcp och exe.

Tcp:

input {
tcp {
    port => 5555
    host => "10.10.1.205"
    type => "checkpoint"
    mode => "server"
}
}

mode => "server"
Visar att Logstash accepterar anslutningar.

port => 5555
host => "10.10.1.205"
Accepterar anslutningar till IP-adressen 10.10.1.205 (Logstash), port 5555 — porten måste vara tillåten av brandväggens policy.

type => "checkpoint"
Markerar dokumentet, vilket är mycket praktiskt om du har flera inkommande anslutningar. Därefter kan en egen filterkonfiguration skrivas för varje anslutning med hjälp av en logisk if-konstruktion.

File:

input {
  file {
    path => "/var/log/openvas_report/*"
    type => "openvas"
    start_position => "beginning"
    }
}

Beskrivning av inställningarna:
path => "/var/log/openvas_report/*"
Ange katalogen vars filer behöver läsas.

type => "openvas"
Händelsetyp.

start_position => "beginning"
Vid filändring läses hela filen, om "end" anges kommer systemet att vänta på att nya poster dyker upp i slutet av filen.

Exec:

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

Med denna input körs (endast!) shell-kommandot och dess utdata omvandlas till loggmeddelande.

command => "ls -alh"
Kommandot vars utdata intresserar oss.

interval => 30
Intervallet för att anropa kommandot i sekunder.

För att ta emot loggar från brandväggen anger vi en filter tcp eller udp, beroende på hur loggarna skickas till Logstash.

Vi konfigurerar Output i Logstash-konfigurationsfilen i debugläge för att förstå hur loggmeddelandet ser ut.

När vi har konfigurerat INPUT måste vi förstå hur loggmeddelandet kommer att se ut och vilka metoder som behöver användas för att konfigurera filtret (parsern) för loggar.

För detta kommer vi att använda ett filter som ger resultat i stdout för att se det ursprungliga meddelandet, hela konfigurationsfilen ser för tillfället ut så här:

input 
{
         tcp 
         {
                port => 5555
  	  	type => "checkpoint"
    		mode => "server"
                host => "10.10.1.205"
   	 }
}

output 
{
	if [type] == "checkpoint" 
       {
		stdout { codec => json }
	}
}

Vi kör kommandot för att kontrollera:
sudo /usr/share/logstash/bin//logstash -f /etc/logstash/conf.d/checkpoint.conf
Vi ser resultatet, bilden är klickbar:

Om vi kopierar detta skulle det se ut så här:

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,

Genom att titta på dessa meddelanden förstår vi att loggarna har formen: fält = värde eller key = value, vilket innebär att filtret som kallas kv passar. För att korrekt välja ett filter för varje specifikt fall skulle det vara bra att bekanta sig med dem i den tekniska dokumentationen, eller att fråga en vän.

Inställning av Filter

I föregående steg valde vi kv, här presenteras konfigurationen för detta filter:

filter {
if [type] == "checkpoint"{
	kv {
		value_split => "="
		allow_duplicate_values => false
	}
}
}

Vi väljer symbolen som vi ska dela fältet och värdet med — “=”. Om vi har identiska poster i loggen, sparar vi endast ett exemplar i databasen, annars får du en matris av identiska värden, det vill säga om vi har meddelandet “foo = some foo=some” sparar vi endast foo = some.

Ställer in korrekt Output i ElasticSearch

När filtret är inställt kan loggarna exporteras till databasen , där serveradressen och mönstret för indexnamnet anges för att skicka det formaterade dokumentet::

output 
{
if [type] == "checkpoint"
{
 	elasticsearch 
        {
		hosts => ["10.10.1.200:9200"]
		index => "checkpoint-%{+YYYY.MM.dd}"
    		user => "tssolution"
    		password => "cool"
  	}
}
}

Om dokumentet är signerat med typen checkpoint, sparar vi händelsen i elasticsearch-databasen, som tar emot anslutningar på 10.10.1.200 på port 9200 som standard. Varje dokument sparas i ett specifikt index, i detta fall sparar vi i indexet "checkpoint-" + nuvarande tidsdatum. Varje index kan ha en viss uppsättning fält, eller skapas automatiskt när ett nytt fält dyker upp i meddelandet; inställningarna för fälten och deras typ kan ses i mappings.

Om du har aktiverat autentisering (vi kommer att diskutera detta senare), måste du se till att ange kredentialer för att skriva in ett specifikt index, i detta exempel är det "tssolution" med lösenordet "cool". Du kan avgränsa användarrättigheter för att skriva loggar endast till ett visst index och inte till något annat.

Startar Logstash.

Logstash-konfigurationsfil:

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"
  	}
}
}

Kontrollerar konfigurationsfilen för korrekt sammansättning:
/usr/share/logstash/bin//logstash -f checkpoint.conf


Startar Logstash-processen:
sudo systemctl start logstash

Kontrollerar att processen har startat:
sudo systemctl status logstash

Kontrollerar att socketen är uppe:
netstat -nat |grep 5555

Vi kontrollerar loggarna i Kibana.

När allt har startat går vi in i Kibana – Discover, och kontrollerar att allt är korrekt inställt, bilden är klickbar!

Alla loggar är på plats och vi kan se alla fält och deras värden!

Slutsats

Vi har gått igenom hur man skriver en Logstash-konfigurationsfil, vilket resulterade i en parser för alla fält och värden. Nu kan vi arbeta med sökningar och skapa diagram baserat på specifika fält. I den här kursen kommer vi att titta på visualisering i Kibana och skapa en enkel instrumentpanel. Det är värt att nämna att Logstash-konfigurationsfilen behöver ständigt uppdateras i vissa situationer, till exempel när vi vill byta ett fälts värde från siffra till ord. I kommande artiklar kommer vi ständigt att göra detta.

Så håll utkik efter uppdateringar (Telegram, Facebook, VK, TS Solution Blog), Yandex.Dzen.

Källa: habr.com