2. Elastic stack : analyse des logs de sécurité. Logstash

Dans le passé, article Nous avons fait connaissance avec la stack ELK, des logiciels qui la composent. Et le premier défi auquel un ingénieur est confronté lorsqu'il travaille avec la stack ELK est d'envoyer des journaux pour stockage dans Elasticsearch pour une analyse ultérieure. Cependant, cela ne semble simple que sur le papier, Elasticsearch stocke les journaux sous forme de documents avec des champs et des valeurs définis, ce qui signifie que l'ingénieur doit utiliser divers outils pour analyser le message envoyé par les systèmes finaux. Cela peut être fait de plusieurs manières : en écrivant un programme qui ajoutera des documents à la base via l'API ou en utilisant des solutions déjà prêtes. Dans le cadre de ce cours, nous allons examiner la solution Logstash, qui fait partie de la stack ELK. Nous allons voir comment envoyer des journaux à partir des systèmes finaux vers Logstash, puis configurerons le fichier de configuration pour le parsing et le redirectionnement vers la base de données Elasticsearch. Pour cela, nous prenons comme système d'entrée les journaux du pare-feu Check Point.

Ce cours ne couvre pas l'installation de la stack ELK, car il existe une multitude d'articles sur ce sujet, nous allons nous pencher sur les aspects de configuration.

Établissons un plan d'action pour la configuration de Logstash :

1. Vérification que Elasticsearch acceptera les journaux (vérification de la fonctionnalité et de l'ouverture du port).
2. Nous examinons comment nous pouvons envoyer des événements à Logstash, choisissons une méthode, puis la mettons en œuvre.
3. Nous configurons l'Input dans le fichier de configuration Logstash.
4. Nous configurons l'Output dans le fichier de configuration Logstash en mode débug, afin de comprendre à quoi ressemble un message log.
5. Nous configurons le Filter.
6. Nous configurons un Output correct dans Elasticsearch.
7. Nous lançons Logstash.
8. Nous vérifions les journaux dans Kibana.

Nous examinerons chaque point plus en détail :

Vérifiez que Elasticsearch acceptera les journaux

Pour cela, vous pouvez vérifier avec la commande curl l'accès à Elasticsearch depuis le système sur lequel Logstash est déployé. Si vous avez configuré une authentification, passez également le nom d'utilisateur/mot de passe via curl, en précisant le port 9200 si vous ne l'avez pas changé. Si vous recevez une réponse similaire à celle indiquée ci-dessous, cela signifie que tout est en ordre.

[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" : "Vous savez, pour la recherche"
}
[elastic@elasticsearch ~]$

Si la réponse n'est pas arrivée, cela peut être dû à plusieurs types d'erreurs : le processus elasticsearch ne fonctionne pas, un mauvais port est spécifié, ou le port est bloqué par le pare-feu sur le serveur où elasticsearch est installé.

Voyons comment envoyer des logs à Logstash depuis un pare-feu Check Point.

Depuis le serveur de gestion Check Point, vous pouvez envoyer des logs à Logstash via syslog, en utilisant l'outil log_exporter ; plus de détails peuvent être consultés ici. article, ici nous laisserons seulement la commande qui crée le flux :

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

<> — l'adresse du serveur où Logstash fonctionne, target-port 5555 — le port vers lequel nous enverrons les logs ; l'envoi de logs par tcp peut surcharger le serveur, donc dans certains cas, il est plus approprié d'utiliser udp.

Configurons l'INPUT dans le fichier de configuration Logstash.

Par défaut, le fichier de configuration se trouve dans le répertoire /etc/logstash/conf.d/. Le fichier de configuration se compose de 3 parties significatives : INPUT, FILTER, OUTPUT. INPUT nous spécifions d'où le système prendra les logs, dans FILTER nous analysons le log — nous configurons comment diviser le message en champs et valeurs, dans OUTPUT nous configurons le flux sortant — où les logs analysés seront envoyés.

Commençons par configurer l'INPUT, voyons quelques types possibles — file, tcp et exe.

Tcp :

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

mode => «server»
Indique que Logstash accepte les connexions.

port => 5555
host => “10.10.1.205”
Nous acceptons les connexions à l'adresse IP 10.10.1.205 (Logstash), le port 5555 — ce port doit être autorisé par la politique de pare-feu.

type => «checkpoint»
Nous étiquetons le document, très pratique si vous avez plusieurs connexions entrantes. Par la suite, pour chaque connexion, vous pouvez écrire votre propre filtre à l'aide d'une structure logique if.

File :

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

Description des paramètres :
path => «/var/log/openvas_report/*»
Indiquez le répertoire contenant les fichiers à lire.

type => «openvas»
Type d'événement.

start_position => «beginning»
Lorsqu'un fichier est modifié, le fichier entier est lu. Si «end» est spécifié, le système attend l'arrivée de nouvelles entrées à la fin du fichier.

Exec:

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

Pour cette entrée, une commande shell est lancée (uniquement !) et sa sortie est encapsulée dans un message de log.

command => «ls -alh»
Commande dont nous voulons la sortie.

interval => 30
Intervalle d'appel de la commande en secondes.

Pour recevoir les logs du pare-feu, il faut spécifier le filtre tcp ou udp, selon la façon dont les logs sont envoyés à Logstash.

Configurez la sortie dans le fichier de configuration de Logstash en mode débogage pour comprendre à quoi ressemble le message de log.

Après avoir configuré l'INPUT, il est nécessaire de comprendre à quoi ressemblera le message de log et quels méthodes il faudra utiliser pour configurer le filtre (parseur) des logs.

Pour cela, nous utiliserons un filtre qui fournit le résultat en stdout pour voir le message d'origine, le fichier de configuration complet ressemblera donc à ceci :

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

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

Exécutez la commande pour vérification :
sudo /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/checkpoint.conf
Nous voyons le résultat, l'image est cliquable :

Si vous copiez cela, cela ressemblera à ceci :

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,

En regardant ces messages, nous comprenons que les journaux sont au format : champ = valeur ou clé = valeur, ce qui signifie que le filtre appelé kv est approprié. Pour bien choisir le filtre pour chaque cas particulier, il serait bon de les consulter dans la documentation technique ou de demander à un ami.

Configurer le filtre

À l'étape précédente, nous avons choisi kv, la configuration de ce filtre est présentée ci-dessous :

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

Choisissons le symbole sur lequel nous allons diviser le champ et la valeur — '='. Si nous avons des enregistrements identiques dans le journal, nous conservons dans la base seulement une seule instance, sinon vous obtiendrez un tableau de valeurs identiques, c'est-à-dire que si nous avons le message "foo = some foo=some", nous enregistrons seulement foo = some.

Configurer la sortie correcte dans ElasticSearch

Une fois le filtre configuré, il est possible d'exporter les journaux vers la base. elasticsearch:

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

Si le document est signé comme type checkpoint, nous sauvegardons l'événement dans la base Elasticsearch, qui accepte les connexions sur 10.10.1.200 sur le port 9200 par défaut. Chaque document est sauvegardé dans un index spécifique, dans ce cas, nous enregistrons dans l'index "checkpoint-" + la date actuelle. Chaque index peut avoir un ensemble de champs spécifiques ou être créé automatiquement lors de l'apparition d'un nouveau champ dans le message, les paramètres des champs et leur type peuvent être consultés dans les mappings.

Si vous avez configuré l'authentification (que nous examinerons plus tard), des identifiants doivent nécessairement être fournis pour l'écriture dans un index spécifique, dans cet exemple, il s'agit de « tssolution » avec le mot de passe « cool ». Il est possible de restreindre les droits des utilisateurs pour écrire des journaux uniquement dans un index défini et pas dans d'autres.

Lancer Logstash.

Fichier de configuration de 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"
  	}
}
}

Vérifions le fichier de configuration pour s'assurer qu'il est correctement rédigé :
/usr/share/logstash/bin//logstash -f checkpoint.conf


Démarrer le processus Logstash :
sudo systemctl start logstash

Vérifiez que le processus a démarré :
sudo systemctl status logstash

Vérifions si le socket est actif :
netstat -nat |grep 5555

Nous vérifions les journaux dans Kibana.

Après avoir tout lancé, accédez à Kibana — Discover, et assurez-vous que tout est correctement configuré, l'image est cliquable !

Tous les journaux sont en place et nous pouvons voir tous les champs et leurs valeurs !

Conclusion

Nous avons vu comment rédiger un fichier de configuration Logstash, ce qui nous a permis d'obtenir un parseur de tous les champs et valeurs. Maintenant, nous pouvons travailler sur la recherche et la création de graphiques basés sur des champs spécifiques. Dans la suite du cours, nous aborderons la visualisation dans Kibana et créerons un tableau de bord simple. Il est important de mentionner que le fichier de configuration Logstash doit être continuellement mis à jour dans certaines situations, par exemple, lorsque nous voulons remplacer la valeur d'un champ d'un chiffre par un mot. Dans les articles suivants, nous allons constamment le faire.

Alors restez à l'écoute pour les mises à jour (Telegram, Facebook, VK, Blog de solutions TS), Yandex.Zen.

Source : habr.com