Pile ELK pour la surveillance

Introduction

IBM Verify fournit un riche ensemble d'événements collectés dans le cadre de ses opérations et de ses transactions. L'utilisation de ces événements pour obtenir des informations, ce qui implique de les ingérer et de les stocker dans un endroit central, est essentielle à toute solution.

Ce guide montrera comment intégrer IBM Verify avec un déploiement de pile ELK ( Elasticsearch, Logstash, Kibana) en lisant des événements à partir de IBM Verify et en les envoyant à la pile ELK. Pour ce faire, nous utiliserons un outil fonctionnant sous nodejS.

Prérequis

Vous aurez besoin d'un identifiant et d'un secret de client de votre locataire IBM Verify qui dispose de l'autorisation Read reports requise pour lire les données d'événement. Consultez ce guide pour créer un client API.

Vous devrez vous familiariser avec la pile d'outils nodeJS-based d' IBM Security Expert Labs.
Vous pouvez le télécharger ou le cloner ici.

Vous devez être familiarisé avec la pile ELK.

Architecture

L'outil d'intégration décrit dans ce guide fonctionne comme un processus nodeJS qui extrait les événements de IBM Verify, prépare les événements et les introduit dans la pile ELK. Chaque fois qu'il réussit à tirer ou à pousser, il écrit une pierre tombale afin de pouvoir reprendre là où il s'est arrêté après un redémarrage.

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

• Informations sur la connexion de votre locataire IBM Verify
• Informations de connexion pour votre pile ELK
• Chemin d'accès au fichier contenant la valeur de la pierre tombale pour la poursuite de l'activité après le redémarrage

Outil d'intégration

L'extrait de code suivant montre l'initialisation des connexions, la configuration du filtre et le code permettant d'effectuer un arrêt en douceur lors de 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;
});

À l'aide des données de configuration, l'outil initialise les connexions à votre locataire IBM Verify et à votre pile ELK.

Il génère ensuite le filtre qui sera utilisé pour demander des événements à IBM Verify. Ce filtre commence par les entrées de la requête 1000 événements triées 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 de base.

Le code suivant montre la boucle d'exécution de l'outil. Il s'agit d'une boucle sans fin qui extrait des événements de IBM Verify (1000 événements à la fois, tel que 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 renvoyé, l'outil se met en veille pendant 30 secondes avant de réessayer.

Si des événements sont renvoyés, le code suivant s'exécute pour traiter les événements et les envoyer à votre pile ELK sous la forme d'un téléchargement 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 téléchargement en masse, la pile ELK s'attend à ce que chaque enregistrement soit composé de deux lignes. La première ligne est constituée de métadonnées au format JSON. 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 identifiant (_id) pour s'assurer que l'événement est unique. L'outil utilise l'identifiant de l'événement provenant de IBM Verify comme identifiant de 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 téléchargement réussi, le fichier de base est mis à jour afin que l'outil puisse continuer à fonctionner 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 la v7.13.2 déployée sur une machine RedHat en tant qu'utilisateur root en suivant les étapes suivantes.

1. Téléchargez la pile ELK en suivant les instructions ci-dessous :
   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 devez lier la pile à une adresse IP externe. Ce point est abordé dans les étapes suivantes.

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

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

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

5. Démarrer 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 :
   ht tp://yo ur-hostname.example.com:9200
   La réponse est un objet JSON contenant les détails du cluster :

7. Testez la connectivité à l'interface web de Kibana. Par défaut, l'interface utilisateur Kibana fonctionne sur le port 5601. Pour accéder à l'interface utilisateur, saisissez l'adresse IP ou le nom d'hôte où vous hébergez la pile ELK et ajoutez :5601 à la fin. Par exemple :
   ht tp://yo ur-hostname.example.com:5601

Configuration de l'outil

Dans le fichier env.js de l'outil d'intégration, configurez l'identifiant du client de l'API de votre locataire IBM Verify et le secret de l'API.

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

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

Dans la section tombstone de env.js, vous pouvez passer un fichier JSON contenant l'heure de la pierre tombale que vous souhaitez vérifier. Pour fileName, indiquez le chemin d'accès au fichier de base.

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

Le fichier de pierre tombale a le format suivant :

{
  "last": "16149608629898"
}

La valeur de last est une époque en millisecondes et vous pouvez utiliser 0 pour recommencer depuis le début si vous le souhaitez. Vous verrez que la valeur de last sera mise à jour régulièrement au fur et à mesure de l'exécution de l'outil.

Exécuter l'outil d'intégration"

Pour lancer 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 essayez d'accéder à l'onglet "Découverte", vous ne verrez pas l'index qui vous permettra d'obtenir les informations dont vous avez besoin. Pour accéder à ces informations, cliquez sur le bouton hamburger en haut à gauche, puis sur Gestion des piles

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

Saisissez 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 allons pouvoir le visualiser. Cliquez sur le bouton hamburger et sous la section "Analytics", cliquez sur "Discover".

Un menu déroulant apparaît dans la partie gauche de l'écran. Sélectionnez l'index que vous souhaitez consulter dans le menu déroulant.

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

Félicitations ! Vous pouvez désormais analyser les événements de votre locataire IBM Verify dans Kibana.

💎

Martin Schmidt, IBM Security Expert Labs

Martin Alvarez, IBM Security Expert Labs