In het verleden we zijn bekend geraakt met de ELK stack, uit welke softwareproducten deze bestaat. De eerste taak waarmee een engineer bij het werken met de ELK stack wordt geconfronteerd, is het verzenden van logs voor opslag in Elasticsearch voor daaropvolgende analyse. Dit klinkt eenvoudig, maar Elasticsearch slaat logs op als documenten met bepaalde velden en waarden, wat betekent dat de engineer verschillende tools moet gebruiken om het bericht te parseren dat vanaf de eindsystemen wordt verzonden. Dit kan op verschillende manieren worden gedaan: een programma schrijven dat via de API documenten aan de database toevoegt of gebruikmaken van bestaande oplossingen. In deze cursus zullen we een oplossing bekijken Logstash, die deel uitmaakt van de ELK stack. We zullen zien hoe we logs van eindsystemen naar Logstash kunnen verzenden en vervolgens de configuratie van het configuratiebestand kunnen instellen voor parsing en doorverwijzing naar de Elasticsearch-database. Hiervoor nemen we logs van de firewall Check Point als inkomend systeem.
In de cursus zullen we de installatie van de ELK stack niet behandelen, aangezien er talloze artikelen over dit onderwerp bestaan; we zullen de configuratiecomponent behandelen.
We stellen een actieplan op voor de configuratie van Logstash:
- Controle dat Elasticsearch logs zal ontvangen (controle van de werking en openheid van de poort).
- We bekijken op welke manier we gebeurtenissen naar Logstash kunnen verzenden, kiezen een methode en implementeren deze.
- We configureren de Input in het configuratiebestand van Logstash.
- We configureren de Output in het configuratiebestand van Logstash in debugmodus, zodat we kunnen begrijpen hoe het logbericht eruitziet.
- We configureren de Filter.
- We configureren de correcte Output in Elasticsearch.
- We starten Logstash.
- We controleren de logs in Kibana.
Laten we elk punt in meer detail bekijken:
Controle dat Elasticsearch logs zal ontvangen.
Hiervoor kunt u met de curl-opdracht de toegang tot Elasticsearch vanuit het systeem dat Logstash draait controleren. Als u authenticatie heeft ingesteld, geven we ook de gebruiker/wachtwoord door via curl en geven we poort 9200 op als u deze niet heeft gewijzigd. Als er een antwoord komt zoals hieronder aangegeven, dan is alles in orde.
[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" : "Je weet wel, voor zoeken"
}
[elastic@elasticsearch ~]$
Als er geen antwoord komt, kunnen er verschillende soorten fouten zijn: het elasticsearch-proces werkt niet, de opgegeven poort is verkeerd, of de poort is geblokkeerd door de firewall op de server waar elasticsearch draait.
We bekijken hoe we logs naar Logstash kunnen verzenden vanaf de Check Point firewall.
Vanuit de Check Point management server kunnen logs via syslog naar Logstash worden verzonden met behulp van de utility log_exporter. Meer informatie hierover is te vinden in deze , hier laten we alleen de opdracht zien die de stream creƫert:
cp_log_export add name check_point_syslog target-server <> target-port 5555 protocol tcp format generic read-mode semi-unified
<> ā het adres van de server waarop Logstash draait, target-port 5555 ā de poort waarnaar we logs zullen verzenden. Het verzenden van logs via tcp kan de server belasten, daarom is het in sommige gevallen verstandiger om udp te gebruiken.
We configureren de INPUT in het configuratiebestand van Logstash.
Standaard bevindt het configuratiebestand zich in de map /etc/logstash/conf.d/. Het configuratiebestand bestaat uit 3 betekenisvolle delen: INPUT, FILTER, OUTPUT. In INPUT geef je op waar het systeem logs vandaan haalt, in FILTER parseren we de log ā we configureren hoe het bericht in velden en waarden moet worden opgedeeld, in OUTPUT configureren we de uitvoerstream ā waar de geparsed logs naartoe worden gestuurd.
Laten we eerst de INPUT instellen en bekijken welke types er kunnen zijn ā file, tcp en exe.
Tcp:
input {
tcp {
port => 5555
host => "10.10.1.205"
type => "checkpoint"
mode => "server"
}
}
mode => "server"
Geeft aan dat Logstash verbindingen accepteert.
port => 5555
host => "10.10.1.205"
Verbindingen accepteren via IP-adres 10.10.1.205 (Logstash), poort 5555 ā de poort moet zijn toegestaan volgens het firewallbeleid.
type => "checkpoint"
Markeert het document, wat zeer handig is als je meerdere inkomende verbindingen hebt. Vervolgens kan voor elke verbinding een eigen filter worden geschreven met behulp van een logische if-constructie.
File:
input {
file {
path => "/var/log/openvas_report/*"
type => "openvas"
start_position => "beginning"
}
}
Omschrijving van de instellingen:
path => "/var/log/openvas_report/*"
Geef de directory aan waarvan de bestanden moeten worden ingelezen.
type => "openvas"
Type gebeurtenis.
start_position => "beginning"
Bij wijziging van het bestand wordt het bestand volledig ingelezen. Als "end" is ingesteld, wacht het systeem op nieuwe vermeldingen aan het einde van het bestand.
Exec:
input {
exec {
command => "ls -alh"
interval => 30
}
}
Met deze input wordt (alleen!) een shell-commando uitgevoerd en de uitvoer wordt verpakt in een logbericht.
command => "ls -alh"
Het commando waarvan de uitvoer interessant is.
interval => 30
De interval om het commando in seconden aan te roepen.
Om logs van de firewall te ontvangen, schrijven we een filter. tcp of udp, afhankelijk van hoe de logs naar Logstash worden verzonden.
Stel de output in het configuratiebestand van Logstash in debug-modus in, om te begrijpen hoe het logbericht eruitziet.
Nadat we de INPUT hebben geconfigureerd, moeten we begrijpen hoe het logbericht eruit zal zien en welke methoden we moeten gebruiken om de filter (parser) instellingen te configureren.
Hiervoor gebruiken we een filter dat het resultaat in stdout geeft, zodat we het oorspronkelijke bericht kunnen bekijken. Het volledige configuratiebestand ziet er op dit moment zo uit:
input
{
tcp
{
port => 5555
type => "checkpoint"
mode => "server"
host => "10.10.1.205"
}
}
output
{
if [type] == "checkpoint"
{
stdout { codec=> json }
}
}
Voer de opdracht uit ter controle:
sudo /usr/share/logstash/bin//logstash -f /etc/logstash/conf.d/checkpoint.conf
We zien het resultaat, de afbeelding is klikbaar:
Als we dit kopiƫren, ziet het er zo uit:
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,
Kijkend naar de gegeven berichten begrijpen we dat de logs de vorm hebben: veld = waarde of key = value, en dus is de filter die hiervoor geschikt is kv. Om de juiste filter voor elk specifiek geval te kiezen, zou het handig zijn om ze in de technische documentatie door te nemen of een vriend te vragen.
Filter instellen
In de vorige stap hebben we kv gekozen, hieronder is de configuratie van deze filter:
filter {
if [type] == "checkpoint"{
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
We kiezen het teken waarmee we veld en waarde zullen splitsen ā '='. Als we dubbele vermeldingen in de log hebben, bewaren we alleen Ć©Ć©n exemplaar in de database, anders vormt je een array met dezelfde waarden, dus als we het bericht "foo = some foo=some" hebben, schrijven we alleen foo = some.
De juiste Output instellen in ElasticSearch
Zodra de Filter is ingesteld, kunnen we de logs naar de database exporteren. elasticsearch:
output
{
if [type] == "checkpoint"
{
elasticsearch
{
hosts => ["10.10.1.200:9200"]
index => "checkpoint-%{+YYYY.MM.dd}"
user => "tssolution"
password => "cool"
}
}
}
Als het document is getekend met het type checkpoint, slaan we het evenement op in de elasticsearch-database, die standaard verbindingen accepteert op 10.10.1.200 op poort 9200. Elk document wordt opgeslagen in een specifieke index, in dit geval slaan we het op in de index "checkpoint-" + de huidige tijdsdatum. Elke index kan een specifieke set velden hebben, of wordt automatisch aangemaakt wanneer er een nieuw veld in het bericht verschijnt; de instellingen van velden en hun typen kunnen in mappings worden bekeken.
Als je authenticatie hebt ingesteld (waar we later naar zullen kijken), moeten de inloggegevens voor het schrijven naar een specifieke index absoluut worden opgegeven; in dit voorbeeld is dat "tssolution" met het wachtwoord "cool". Je kunt gebruikersrechten beperken voor het schrijven van logs alleen naar een bepaalde index en niet naar andere indices.
We starten Logstash.
Configuratiebestand van 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"
}
}
}
We controleren het configuratiebestand op de correctheid van de samenstelling:
/usr/share/logstash/bin//logstash -f checkpoint.conf
We starten het Logstash-proces:
sudo systemctl start logstash
We controleren of het proces is gestart:
sudo systemctl status logstash
We controleren of de socket is opgestart:
netstat -nat |grep 5555
We controleren de logs in Kibana.
Nadat alles is gestart, gaan we naar Kibana ā Discover, en zorgen we ervoor dat alles correct is ingesteld en de afbeelding klikbaar is!
Alle logs zijn op hun plaats en we kunnen alle velden en hun waarden zien!
Conclusie
We hebben besproken hoe we een configuratiebestand voor Logstash moeten schrijven, en daardoor hebben we een parser voor alle velden en waarden gekregen. Nu kunnen we werken met zoeken en het maken van grafieken op basis van specifieke velden. In de volgende cursus bekijken we de visualisatie in Kibana en maken we een eenvoudige dashboard. Het is vermeldenswaard dat het configuratiebestand van Logstash in bepaalde situaties voortdurend moet worden aangevuld, bijvoorbeeld wanneer we de waarde van een veld van een cijfer op een woord willen veranderen. In de volgende artikelen zullen we dit voortdurend doen.
Dus houd de updates in de gaten (, , , ), .
Bron: habr.com
