
Letzte Woche Wir haben uns mit dem ELK-Stack, aus welchen Softwareprodukten er besteht. Die erste Aufgabe, mit der ein Ingenieur bei der Arbeit mit dem ELK-Stack konfrontiert ist, besteht darin, Protokolle zur Speicherung in Elasticsearch für eine anschließende Analyse zu senden. Das klingt zunächst einfach; Elasticsearch speichert Protokolle in Form von Dokumenten mit bestimmten Feldern und Werten, was bedeutet, dass der Ingenieur verschiedene Werkzeuge verwenden muss, um die Nachricht zu analysieren, die von den Endsystemen gesendet wird. Dies kann auf verschiedene Weisen erfolgen – indem man selbst ein Programm schreibt, das über die API Dokumente zur Datenbank hinzufügt, oder indem man bereits fertige Lösungen verwendet. Im Rahmen dieses Kurses werden wir eine Lösung betrachten Logstash, die Teil des ELK-Stacks ist. Wir werden sehen, wie man Protokolle von Endsystemen nach Logstash senden kann, und anschließend die Konfigurationsdatei für das Parsen und die Umleitung in die Elasticsearch-Datenbank konfigurieren. Dazu verwenden wir als eingehendes System die Protokolle von der Firewall Check Point.
Im Rahmen des Kurses wird die Installation des ELK-Stacks nicht behandelt, da es hierzu eine Vielzahl von Artikeln gibt; wir werden uns auf die Konfigurationsaspekte konzentrieren.
Wir erstellen einen Aktionsplan zur Konfiguration von Logstash:
- Überprüfung, dass Elasticsearch Protokolle akzeptiert (Überprüfung der Funktionsfähigkeit und der Offenheit des Ports).
- Wir betrachten, wie wir Ereignisse an Logstash senden können, wählen eine Methode aus und setzen sie um.
- Wir konfigurieren den Input in der Konfigurationsdatei von Logstash.
- Wir konfigurieren den Output in der Konfigurationsdatei von Logstash im Debug-Modus, um zu verstehen, wie eine Protokollnachricht aussieht.
- Wir konfigurieren den Filter.
- Wir konfigurieren den korrekten Output in Elasticsearch.
- Wir starten Logstash.
- Wir überprüfen die Protokolle in Kibana.
Wir betrachten jeden Punkt im Detail:
Überprüfung, dass Elasticsearch Protokolle akzeptiert
Dazu kann man mit dem curl-Befehl den Zugriff auf Elasticsearch aus dem System überprüfen, auf dem Logstash ausgeführt wird. Wenn eine Authentifizierung eingerichtet ist, übergeben wir auch Benutzername/Passwort über curl und geben den Port 9200 an, falls Sie ihn nicht geändert haben. Wenn eine Antwort in etwa wie die unten angegebene kommt, ist alles in Ordnung.
[elastic@elasticsearch ~]$ curl -u <> : <> -sS -XGET "<>:9200"
{
"name" : "elastic-1",
"cluster_name" : "projekt",
"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" : "Sie wissen schon, für die Suche"
}
[elastic@elasticsearch ~]$
Wenn die Antwort nicht kommt, kann es mehrere Arten von Fehlern geben: der elasticsearch-Prozess funktioniert nicht, der falsche Port wurde angegeben, oder der Port wird von der Firewall auf dem Server, auf dem elasticsearch läuft, blockiert.
Wir betrachten, wie man Protokolle von der Firewall Check Point an Logstash senden kann.
Von der Check Point Management-Server können Protokolle über Syslog an Logstash gesendet werden, unter Verwendung des Tools log_exporter, das näher beschrieben wird in diesem , hier lassen wir nur den Befehl, der den Stream erstellt:
cp_log_export add name check_point_syslog target-server <> target-port 5555 protocol tcp format generic read-mode semi-unified
<> — die Adresse des Servers, auf dem Logstash läuft, target-port 5555 — der Port, an den wir Protokolle senden werden, das Senden von Protokollen über tcp kann den Server belasten, daher ist es in einigen Fällen sinnvoller, udp zu verwenden.
Wir konfigurieren INPUT in der Konfigurationsdatei von Logstash.

Standardmäßig befindet sich die Konfigurationsdatei im Verzeichnis /etc/logstash/conf.d/. Die Konfigurationsdatei besteht aus 3 sinnvollen Teilen: INPUT, FILTER, OUTPUT. In INPUT geben wir an, woher das System die Protokolle beziehen wird, in FILTER parsen wir das Protokoll – wir konfigurieren, wie die Nachricht in Felder und Werte aufgeteilt wird, in OUTPUT konfigurieren wir den Ausgangsstrom – wohin die geparsten Protokolle gesendet werden.
Zuerst konfigurieren wir INPUT, betrachten einige Typen, die möglich sind – file, tcp und exe.
Tcp:
input {
tcp {
port => 5555
host => "10.10.1.205"
type => "checkpoint"
mode => "server"
}
}
mode => "server"
zeigt an, dass Logstash Verbindungen annimmt.
port => 5555
host => "10.10.1.205"
Wir nehmen Verbindungen über die IP-Adresse 10.10.1.205 (Logstash) entgegen, der Port 5555 – der Port muss durch die Firewall-Policy erlaubt werden.
type => "checkpoint"
Wir kennzeichnen das Dokument, was sehr praktisch ist, wenn Sie mehrere eingehende Verbindungen haben. Später kann für jede Verbindung ein eigener Filter mit einer logischen Konstruktion if geschrieben werden.
File:
input {
file {
path => "/var/log/openvas_report/*"
type => "openvas"
start_position => "beginning"
}
}
Beschreibung der Einstellungen:
path => "/var/log/openvas_report/*"
Wir geben das Verzeichnis an, aus dem die Dateien gelesen werden müssen.
type => „openvas“
Ereignistyp.
start_position => „beginning“
Beim Ändern der Datei wird die Datei vollständig gelesen; wenn „end“ festgelegt ist, wartet das System auf neue Einträge am Ende der Datei.
Exec:
input {
exec {
command => "ls -alh"
interval => 30
}
}
Bei diesem Input wird (nur!) ein Shell-Befehl ausgeführt und dessen Ausgabe in eine Log-Nachricht eingebettet.
command => „ls -alh“
Der Befehl, dessen Ausgabe uns interessiert.
interval => 30
Intervall für den Aufruf des Befehls in Sekunden.
Um Logs von der Firewall zu empfangen, definieren wir einen Filter tcp oder udp, je nachdem, wie die Logs an Logstash gesendet werden.
Wir konfigurieren Output in der Logstash-Konfigurationsdatei im Debug-Modus, um zu verstehen, wie die Log-Nachricht aussieht.
Nachdem wir INPUT konfiguriert haben, müssen wir verstehen, wie die Log-Nachricht aussehen wird und welche Methoden für die Filtereinstellung (Parser) der Logs benötigt werden.
Dafür verwenden wir einen Filter, der das Ergebnis an stdout ausgibt, um die ursprüngliche Nachricht anzuzeigen. Die komplette Konfigurationsdatei sieht zum aktuellen Zeitpunkt so aus:
input
{
tcp
{
port => 5555
type => "checkpoint"
mode => "server"
host => “10.10.1.205”
}
}
output
{
if [type] == "checkpoint"
{
stdout { codec=> json }
}
}
Wir führen den Befehl zur Überprüfung aus:
sudo /usr/share/logstash/bin//logstash -f /etc/logstash/conf.d/checkpoint.conf
Wir sehen das Ergebnis, das Bild ist klickbar:
Wenn man es kopiert, sieht es so aus:
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,
Wenn wir uns die vorliegenden Nachrichten ansehen, verstehen wir, dass die Protokolle die Form haben: Feld = Wert oder key = value, daher passt der Filter, der als kv bezeichnet wird. Um den richtigen Filter für jeden spezifischen Fall auszuwählen, wäre es sinnvoll, sich in der technischen Dokumentation darüber zu informieren oder einen Freund zu fragen.
Filter einstellen
Im vorherigen Schritt haben wir kv ausgewählt, hier ist die Konfiguration dieses Filters:
filter {
if [type] == "checkpoint"{
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
Wir wählen das Zeichen, nach dem wir Feld und Wert trennen — "= ". Wenn wir im Protokoll identische Einträge haben, speichern wir nur ein Exemplar in der Datenbank, andernfalls erhalten Sie ein Array von identischen Werten, das heißt, wenn wir die Nachricht „foo = some foo=some“ haben, speichern wir nur foo = some.
Korrekte Ausgabe in ElasticSearch einstellen
Nachdem der Filter eingerichtet ist, können Sie die Protokolle in die Datenbank exportieren. elasticsearch:
output
{
if [type] == "checkpoint"
{
elasticsearch
{
hosts => ["10.10.1.200:9200"]
index => "checkpoint-%{+YYYY.MM.dd}"
user => "tssolution"
password => "cool"
}
}
}
Wenn das Dokument mit dem Typ checkpoint signiert ist, speichern wir das Ereignis in die ElasticSearch-Datenbank, die standardmäßig Verbindungen auf 10.10.1.200 am Port 9200 akzeptiert. Jedes Dokument wird in einem bestimmten Index gespeichert; in diesem Fall speichern wir im Index „checkpoint-" + aktuelles Datum. Jeder Index kann eine bestimmte Menge an Feldern haben oder wird automatisch erstellt, wenn ein neues Feld in der Nachricht erscheint; die Einstellungen der Felder und deren Typen können in den Mappings eingesehen werden.
Wenn Sie die Authentifizierung eingerichtet haben (darauf werden wir später eingehen), müssen unbedingt die Anmeldedaten für das Schreiben in einen bestimmten Index angegeben werden; in diesem Beispiel handelt es sich um „tssolution“ mit dem Passwort „cool“. Es kann festgelegt werden, dass Benutzer nur in einem bestimmten Index Protokolle schreiben dürfen und nicht in andere.
Logstash starten.
Konfigurationsdatei von 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"
}
}
}
Wir überprüfen die Konfigurationsdatei auf Korrektheit:
/usr/share/logstash/bin//logstash -f checkpoint.conf

Wir starten den Logstash-Prozess:
sudo systemctl start logstash
Wir überprüfen, ob der Prozess gestartet ist:
sudo systemctl status logstash

Wir überprüfen, ob der Socket hochgefahren ist:
netstat -nat |grep 5555
![]()
Wir überprüfen die Protokolle in Kibana.
Nachdem alles gestartet ist, gehen wir zu Kibana — Discover und stellen sicher, dass alles richtig eingerichtet ist, das Bild klickbar ist!
Alle Logs sind vorhanden und wir können alle Felder und deren Werte sehen!
Fazit
Wir haben uns angesehen, wie man eine Konfigurationsdatei für Logstash schreibt, und dabei einen Parser für alle Felder und Werte erstellt. Jetzt können wir mit der Suche und der Erstellung von Grafiken für bestimmte Felder arbeiten. Im weiteren Verlauf des Kurses werden wir die Visualisierung in Kibana betrachten und ein einfaches Dashboard erstellen. Es sei erwähnt, dass die Konfigurationsdatei von Logstash in bestimmten Situationen ständig aktualisiert werden muss, zum Beispiel, wenn wir den Wert eines Feldes von einer Zahl in ein Wort ändern wollen. In den folgenden Artikeln werden wir dies ständig tun.
Also haltet euch über die Updates auf dem Laufenden (, , , ), .
Quelle: habr.com
