Plugin Kafka pour Log Shipper

Ce document explique comment configurer le plugin Kafka v1.0.1 avec le module Log Shipper de la plateforme Netskope Cloud Exchange. Ce plugin ingère les logs Netskope Alerts, Events et WebTX [via Netskope LogStreaming] au format CEF et JSON depuis le locataire Netskope dans le sujet Kafka sur le serveur/cluster Kafka. Le plugin agira en tant que producteur pour publier le message dans le sujet Kafka.

Conditions préalables

Pour compléter cette configuration, vous avez besoin de :

Support des plugins Kafka

Ce plugin est utilisé pour transformer et ingérer les alertes, les événements et les journaux WebTX (via Netskope LogStreaming) dans le sujet Kafka sur le serveur/cluster Kafka. Le plugin agira en tant que producteur pour publier le message dans le sujet Kafka.

Type de données	Support
Alertes	Oui (informations compromises, politique, sites, logiciels malveillants, prévention des pertes de données, évaluation de la sécurité, liste de surveillance, quarantaine, remédiation, UBA, CTEP)
Événements	Oui (Page, Application, Audit, Infrastructure, Réseau, Incident)
WebTx Logs	Oui (via Netskope LogStreaming)
Journaux Syslog CE	Non pris en charge

Détails de l'API

Liste des API utilisées

Ce plugin utilise les bibliothèques kafka-python pour créer des producteurs et des consommateurs kafka.

La bibliothèque kafka-python est un client Python pur pour Apache Kafka qui permet aux développeurs d'interagir avec les clusters Kafka directement à partir d'applications Python. Il prend en charge les opérations Kafka essentielles telles que la production de messages vers des sujets, la consommation de messages à partir de sujets et l'exécution de tâches administratives telles que la création ou la suppression de sujets. La bibliothèque est facile à utiliser et s'intègre bien aux systèmes basés sur Python, ce qui la rend idéale pour créer des pipelines de données en temps réel ou des applications pilotées par les événements. Avec kafka-python, vous pouvez personnaliser les paramètres des groupes de consommateurs, gérer les commits de compensation et configurer les producteurs avec des garanties de livraison, le tout en utilisant du code Python familier.

Créer le consommateur

KafkaProducer(
     bootstrap_servers=kafka_broker_address,
     value_serializer=lambda x: x.encode("utf-8"),
     acks=ACKS,
     partitioner=DefaultPartitioner(),
     retries=RETRIES,
     linger_ms=LINGER_MS,
     client_id=self._get_user_agent(),
     batch_size=BATCH_SIZE,
     compression_type=COMPRESSION_TYPE,
     request_timeout_ms=TIMEOUT_MS,
)

Création du producteur

consumer = KafkaConsumer(
    bootstrap_servers=kafka_broker_address,
    security_protocol=kafka_security_protocol,
    ssl_cafile=tmp_ca_file,
    ssl_certfile=tmp_cert_file,
    ssl_keyfile=tmp_private_key_file,
    ssl_password=kafka_ssl_password,
    request_timeout_ms=TIMEOUT_MS,
)

Pour lister les sujets présents sur le cluster Kafka

consumer.topics()

Pour envoyer des données

producer.send(
    topic=kafka_topic_name,
    value=data
    if not isinstance(data, dict)
    else json.dumps(data),
)

Matrice de performance

Cette analyse des performances a été réalisée sur un CE à grande pile avec les spécifications de VM ci-dessous. Les relevés ci-dessous sont ajoutés en tenant compte du fait qu'il tire environ 10K alertes/événements en ~8 secondes en CE.

Description	Spécifications
Détails de la pile	Taille : Grande RAM : 32 GB CPU : 16 cœurs
Alertes/événements intégrés au SIEM	~200K EPM

Agent utilisateur

netskope-ce-5.1.1-cls-kafka-v1.0.1

Workflow

Cliquer sur « play » pour regarder une vidéo.

Obtenez vos paramètres de configuration Kafka

Les paramètres de configuration suivants sont nécessaires pour configurer le plugin Kafka pour Log Shipper. Contactez l'équipe de configuration du serveur Kafka afin d'obtenir tous les paramètres de configuration.

L'ingestion de tous les événements/alertes et WebTx se fait sur l'Offset Explorer (v3.0.2) ; voici les étapes pour configurer le cluster Kafka sur l'Offset Explorer.

1. Ajoutez une connexion New sur Offset Explorer, et fournissez toutes les informations pour le cluster.

   

2. Après une connexion réussie, allez dans le dossier Topics sous le cluster connecté.

   

3. Cliquez sur l'icône + pour ajouter un New sujet à votre cluster, et indiquez le nom du sujet, le nombre de partitions et le nombre de réplicas. Une fois terminé, cliquez sur Add.

   

4. Après une ingestion réussie depuis Cloud Exchange, vous pouvez trouver les données ingérées à Clusters > {cluster_name} > Topics > {topic_name} > Data.

   

Configurer le plugin Kafka

1. Dans Cloud Exchange, allez sur Settings > Plugin Store. Recherchez et sélectionnez le plugin Kafka v1.0.1 (CLS).

   

2. Saisissez un nom de configuration de plugin et sélectionnez un mappage. Désactivez l'option « Transformer les journaux bruts » si vous souhaitez ingérer vos alertes et événements au format JSON brut. 
   

   

3. Cliquez sur Next et entrez les paramètres de configuration :
   • Kafka Broker Address: DNS/adresse IP/FQDN du courtier Kafka auquel les données seront envoyées. Notez que le plugin n’a besoin que d’un seul courtier qui répondra aux requêtes de l’API de métadonnées.
   • Kafka Port: Adresse du port Kafka sur lequel le broker est configuré.
   • Kafka Security Protocol: Select le protocole de sécurité utilisant lequel l’authentification sera effectuée et les données seront envoyées au cluster Kafka.
   • Kafka CA Certificate: Certificat d'autorité de certification Kafka au format PEM. Ce paramètre de configuration n'est applicable que lorsque SSL est sélectionné comme protocole de sécurité Kafka.
   • Kafka Client Certificate: Certificat client Kafka au format PEM. Note : Ce paramètre de configuration n’est applicable que lorsque SSL est sélectionné comme protocole de sécurité Kafka.
   • Kafka Client Private Key: Clé privée client Kafka au format PEM. Ce paramètre de configuration n’est applicable que lorsque SSL est sélectionné comme protocole de sécurité Kafka.
   • Kafka SSL Private Key Password: Le mot de passe utilisé lors du chargement du certificat. Ce paramètre de configuration ne s’applique que lorsque SSL est sélectionné comme protocole de sécurité Kafka. Il n’est nécessaire que lorsque le fichier PEM est généré sans phrase secrète.
   • Kafka Topic Name: Nom du sujet Kafka auquel les journaux doivent être envoyés. Le nom du sujet Kafka ne doit contenir aucun espace.
   • Log Source Identifier: Ceci sera ajouté comme préfixe à tous les journaux. (L’identifiant source de journal ne doit pas contenir l’espace blanc).
     Tous les paramètres mentionnés ci-dessus sont nécessaires lorsque le protocole de sécurité est SSL. Si le protocole de sécurité est en texte clair, ajoutez Kafka Broker Address, Kafka Port et Topic Name.
     
     
4. Cliquez sur Save. La configuration du plugin sera disponible sur la page Cloud Log Shipper > Plugins.

   

Configurer une règle de gestion de l'expéditeur de logs pour Kafka

1. Va sur Business Rules. Par défaut, une règle métier filtre toutes les alertes et tous les événements. Si vous devez filtrer un type d'alerte ou d'événement spécifique, cliquez sur « Créer New règle » pour configurer une New règle métier en ajoutant le nom de la règle et le filtre.
   

   

2. Cliquer sur Enregistrer.

Configurer Log Shipper Log Delivery pour Kafka

1. Allez à Log Shipper > Log Delivery et cliquez sur Add Log Delivery Configuration.
2. Pour les alertes et événements, sélectionnez le plugin Source (Netskope CLS plugin), le plugin Destination (Kafka plugin), une règle métier, puis cliquez sur Save.
3. Pour l'ingestion de WebTx, sélectionnez le plugin Source (plugin AWS Netskope Log Streaming ou plugin Azure Netskope Log Streaming), le plugin Destination (plugin Kafka), puis cliquez sur Save.

Valider le plugin Kafka

Valider dans Cloud Exchange

Afin de valider le plugin workflow:

1. Cliquez sur Logging.
2. Recherchez les alertes ingérées à l'aide du filtre "message contains ingested".
3. Les journaux ingérés seront filtrés.
   

   

   

   

Valider dans Kafka

Le plugin Kafka a été conçu pour envoyer les données formatées CEF en les encodant en UTF-8, et les événements JSON en effectuant une sérialisation JSON à l'aide de json.dumps(), et l'encoder en UTF-8.

Il existe de nombreuses façons de valider l'envoi des données au serveur Kafka, mais ici, c'est l'Offset Explorer qui est utilisé pour le faire.

Dépannage du plugin Kafka

Le sujet est créé automatiquement dans Kafka lors de l'ingestion des données, même s'il a été supprimé.

What to do:

1. Ajoutez un sujet New dans Kafka.
2. Mettez à jour le plugin Kafka avec un nom de sujet New et sauvegardez-le.
3. Supprimez l'ancien sujet sur Kafka.

En suivant ces étapes, le sujet supprimé ne sera pas recréé dans Kafka, et les données seront ingérées dans le sujet nouvellement ajouté.

Réception du message "Kafka Broker is unreachable or Kafka cluster may be down" (Le courtier Kafka est inaccessible ou le cluster Kafka est peut-être en panne).

Vérifiez l'adresse du broker Kafka et le port Kafka indiqués dans les paramètres de configuration.

Ce problème peut être dû à l'une des raisons suivantes :

• Le serveur Kafka est actuellement en panne.
• L'espace disque du serveur est plein.

What to do:

1. Contactez votre service informatique et confirmez la raison de l'erreur parmi celles mentionnées ci-dessus.
2. Si Kafka est en panne, le redémarrer fonctionnera.
3. Si l'espace disque est plein, effacez les fichiers indésirables et videz l'espace, ou obtenez plus d'espace disque et redémarrez le serveur.