Pile ELK pour la surveillance

Introduction

IBM Security Verify génère un riche flux d'événements ; centraliser ces logs est essentiel pour l'observabilité et la mise en conformité. Ce guide montre comment envoyer les événements Verify vers une pile ELK (Elasticsearch + Logstash + Kibana) à l'aide d'un outil Node.js issu du dépôt IBM Security Expert Labs.

Prérequis

Architecture

Le process Node.js interroge l'API Verify, 1000 événements par appel. Les événements sont formatés en bulk upload et envoyés à Elasticsearch. Une tombstone (fichier JSON) stocke le dernier timestamp traité pour reprendre après redémarrage.

Lors du premier démarrage, l'outil lit la configuration qui fournit :

• Informations de connexion pour votre tenant IBM Security Verify
• Informations de connexion pour votre pile ELK
• Chemin du fichier qui contient la valeur tombstone pour continuer après redémarrage

Outil d'intégration

L'extrait de code suivant montre l'initialisation des connexions, la configuration du filtre, et le code pour effectuer un arrêt gracieux à la sortie du processus :

// Initialize the request must wait for it!
await req.init(config);
await elkReq.init(config);

// Setting up the filter to process 1k events, getting the tombstone's
// initial value and then appending it to the filter
let filter = "?all_events=yes&sort_order=asc&size=1000";
let ts = JSON.parse(req.getTombstone(config.tombstone.fileName));
let filter2 = filter + '&from='+ ts.last; 

let run = true;
// Enable grace full shutdown
process.on('SIGINT', function() {
  console.log("Caught interrupt signal - shutting down gracefully");
  run = false;
});

En utilisant les données de configuration, l'outil initialise les connexions à votre tenant IBM Security Verify et à votre pile ELK.

Il génère ensuite le filtre qui sera utilisé lors de la demande d'événements d'IBM Security Verify. Ce filtre commence par des entrées pour demander 1000 événements triés par ordre croissant. Il ajoute ensuite une entrée pour spécifier un point de départ basé sur la valeur lue dans le fichier tombstone.

Le code suivant montre la boucle d'exécution de l'outil. Il s'agit d'une boucle infinie qui tire les événements d'IBM Security Verify (1000 événements à la fois comme défini par le filtre).

while (run) {
    let list = await req.get(uri+filter2, { "Content-Type": "applications/json", "Accept": "application/json"});
    list = list.response.events;
    // check if the list is empty and sleep if needed
    if (list.events.length == 0) {
      log.debug("Pausing - ", stats.found);
      await req.sleep(30000);
    }

Si aucun événement n'est retourné, l'outil dort pendant 30 secondes avant de réessayer.

Si des événements sont retournés, le code suivant s'exécute pour traiter les événements et les envoyer à votre pile ELK en tant qu'upload en masse :

else {
        stats.found += list.events.length;
        log.trace("Count: ", stats.found);
        let payload = '';
        list.events.forEach( function (event, i) {
            payload += '{"create":{"_index": "' + elk.index + '", "_id": "' + event.id + '"}}\r\n';  
            payload += JSON.stringify(event) + '\r\n';
      });

      elkReq.postEvents(payload); 
      // print the event:
      let backupTimeStamp = list.search_after.time;
      // update the search filter 
      filter2 = filter + '&from='+list.search_after.time; 
      ts.last = list.search_after.time; 
      req.putTombstone(config.tombstone.fileName, ts); 

    }
  } // end of while

Lors de la réception d'un upload en masse, la pile ELK s'attend à ce que chaque enregistrement soit composé de deux lignes. La première ligne est formatée JSON metadata. La deuxième ligne est une chaîne brute qui contient l'événement lui-même.

Les métadonnées indiquent à la pile ELK où créer l'événement (_index) et lui donnent un id (_id) pour s'assurer que l'événement est unique. L'outil utilise l'id d'événement d'IBM Security Verify comme id pour l'événement dans la pile ELK. Cela garantit que si le même événement est envoyé plusieurs fois, il ne créera pas plusieurs enregistrements dans la pile ELK.

À la fin de chaque upload réussi, le fichier tombstone est mis à jour afin que l'outil continue après ces événements s'il est redémarré.

Installation de la pile ELK

La pile ELK utilisée lors de la création de ce guide était v7.13.2 déployée sur une boîte RedHat en tant qu'utilisateur root en suivant ces étapes.

1. Téléchargez la pile ELK en utilisant les instructions suivantes :
   https://www.elastic.co/guide/en/elastic-stack/current/installing-elastic-stack.html

Par défaut, tout est configuré pour localhost (127.0.0.1). Pour permettre l'accès depuis l'extérieur de l'hôte, vous devrez lier la pile à une adresse IP externe. Ceci est couvert dans les étapes suivantes.

2. En tant qu'utilisateur root, éditez le fichier /etc/elasticsearch/elasticsearch.yml et éditez les lignes spécifiant network.host et discover.seed_hosts (voir exemple ci-dessous) :

3. En tant qu'utilisateur root, éditez le fichier /etc/logstash/logstash.yml et éditez les lignes spécifiant http.host dans les paramètres de l'API HTTP (voir exemple ci-dessous) :

4. En tant qu'utilisateur root, éditez le fichier /etc/kibana/kibana.yml et éditez les lignes spécifiant server.host et elasticserch.hosts (voir exemples ci-dessous) :

5. Démarrez les composants de la pile ELK. Pour ce faire, exécutez les commandes suivantes dans l'ordre :

sudo systemctl start elasticsearch.service
sudo systemctl start kibana.service
sudo systemctl start logstash.service

6. Testez la connectivité au service Elasticsearch. Ouvrez un navigateur et naviguez vers :
   http://votre-hostname.example.com:9200
   La réponse est un objet JSON avec les détails du cluster :

7. Testez la connectivité à l'interface web Kibana. Par défaut, l'interface utilisateur Kibana fonctionne sur le port 5601, donc pour accéder à l'interface utilisateur, tapez l'adresse ip ou le nom d'hôte où vous hébergez la pile ELK et ajoutez :5601 à la fin. Par exemple :
   http://votre-hostname.example.com:5601

Configuration de l'outil

Dans le fichier env.js pour l'outil d'intégration, configurez votre ID client API de tenant IBM Security Verify et le secret api.

Dans la section elk de env.js, pour ui spécifiez l'adresse IP ou le nom de domaine où vous hébergez la pile ELK et ajoutez :9200 à la fin. pour index, spécifiez l'index dans lequel vous voulez mettre les données. Si un index avec le nom fourni n'existe pas, alors un index avec ce nom sera créé lors de la première exécution de l'outil.

elk: {
      ui: 'http://votre-hostname.example.com:9200',
      index: 'votreIndexIci'
     },

Dans la section tombstone de env.js vous pouvez passer un fichier JSON contenant le temps tombstone à partir duquel vous souhaitez vérifier. Pour fileName, spécifiez le chemin vers le fichier tombstone.

tombstone: {
      fileName: './ts.json'
    },

Le fichier tombstone a le format suivant :

{
  "last": "16149608629898"
}

La valeur pour last est une époque en millisecondes et vous pouvez utiliser 0 pour commencer depuis le début si vous le souhaitez. Vous verrez que la valeur pour last sera mise à jour régulièrement pendant que l'outil fonctionne.

Exécuter l'outil d'intégration

Pour exécuter l'outil, utilisez la commande suivante :

node src/events/send.js

Création du tableau de bord ELK

Pour continuer, vous devez avoir envoyé des données à Logstash.

Si vous tentez d'aller dans l'onglet découvrir maintenant, vous ne verrez pas l'index où vous pouvez obtenir les informations dont vous avez besoin. Pour pouvoir accéder à ces informations, cliquez sur le bouton hamburger en haut à gauche puis cliquez sur stack management

Sous la section Kibana, cliquez sur "Index Patterns" et "Create Index Pattern"

Tapez le nom de l'index et cliquez sur suivant

Dans "Time field", sélectionnez "time" et cliquez sur "create index pattern"

Maintenant que le modèle d'index est créé, nous pourrons le voir. Cliquez sur le bouton hamburger et sous la section "Analytics", cliquez sur "Discover".

Sur la partie gauche de l'écran, il y aura un menu déroulant. Sélectionnez l'index que vous souhaitez voir dans ce menu déroulant.

Vous pouvez cliquer sur le bouton calendrier en haut à droite pour changer l'intervalle de temps des événements que vous regardez

Félicitations ! Vous pouvez maintenant analyser les événements de votre tenant IBM Security Verify dans Kibana.

💎

Martin Schmidt, IBM Security Expert Labs

Martin Alvarez, IBM Security Expert Labs