Plugin Kafka pour Log Shipper

Ce document explique comment configurer Kafka v1.1.0 plugin avec le module Log Shipper de la plateforme Netskope Cloud Exchange. Ce plugin est utilisé pour envoyer des alertes Netskope (DLP (Prévention des pertes de données), Malware, Policy, Compromised Credential, Malsite, Quarantine, Remediation, Security Assessment, Watchlist, UBA, CTEP, périphérique, Content), des événements (Page, Application, Audit, Infrastructure, Network, Incident, Endpoint, Client Status) et des données WebTx vers un sujet Kafka. Ce plugin prend en charge l'ingestion aux formats JSON et CEF.

Conditions préalables

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

Support des plugins Kafka

Ce plugin est utilisé pour envoyer les alertes Netskope (DLP (Prévention des pertes de données), Malware, Policy, Compromised Credential, Malsite, Quarantine, Remediation, Security Assessment, Watchlist, UBA, CTEP, périphérique, Content), les événements (Page, Application, Audit, Infrastructure, Network, Incident, Endpoint, Client Status) et les données WebTx vers un sujet Kafka. Ce plugin prend en charge l’ingestion au format JSON et CEF.

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,
)

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,
)

consumer.topics()

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

Matrice de performance

Cette analyse des performances est réalisée sur un CE à grande pile avec ces spécifications de VM. Ces relevés sont ajoutés en tenant compte du fait qu'il récupérera environ 10 000 alertes/événements en ~8 secondes dans le CE.

Agent utilisateur

netskope-ce-6.1.0-cls-kafka-v1.1.0

Workflow

Regardez une vidéo

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, accédez au 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 dans Settings > Plugin Store. Recherchez et sélectionnez le plugin Kafka v1.1.0 (CLS).

   

2. Ajoutez un nom de configuration de plugin et sélectionnez le mappage souhaité. Si vous souhaitez ingérer vos alertes/événements/webtx au format JSON brut, remplacez le format par JSON.

   

3. Cliquez sur Next et entrez les paramètres de configuration :

   • Kafka Broker Address: DNS/adresse IP/FQDN du courtier Kafka vers lequel les données seront envoyées. Remarque : le plugin n'a besoin que d'un seul courtier capable de répondre à la requête de l'API de métadonnées.
   • Kafka Port: Port sur lequel le courtier Kafka configuré est à l’écoute. La valeur doit être un nombre entier compris entre 1 et 65 535.
   • Kafka Security Protocol: Select le protocole de sécurité avec lequel l'authentification sera effectuée et les données seront envoyées au cluster Kafka. La sélection de « SSL » nécessitera le remplissage des champs Kafka CA Certificate, Kafka Client Certificate, Kafka Client Private Key et Kafka SSL Private Key Password pour établir une connexion TLS mutuelle.
   • Kafka CA Certificate: Certificat d'autorité de certification (CA) Kafka au format PEM, utilisé pour vérifier l'identité du courtier Kafka. Requis uniquement lorsque « SSL » est sélectionné comme protocole de sécurité Kafka.
   • Kafka Client Certificate: Certificat client Kafka au format PEM, utilisé par le plugin pour s'authentifier auprès du broker Kafka. Requis uniquement lorsque « SSL » est sélectionné comme protocole de sécurité Kafka.
   • Kafka Client Private Key: Clé privée du client Kafka au format PEM, associée au certificat du client Kafka. Requis uniquement lorsque « SSL » est sélectionné comme protocole de sécurité Kafka.
   • Kafka SSL Private Key Password: Mot de passe pour la clé privée du client Kafka. Requis uniquement si le fichier PEM de la clé privée a été généré avec une phrase secrète. Requis uniquement lorsque « SSL » est sélectionné comme protocole de sécurité Kafka.
   • Kafka Topic Name: Nom du sujet Kafka vers lequel les journaux doivent être envoyés. Remarque : le nom du sujet Kafka ne doit contenir aucun espace et doit déjà exister sur le cluster Kafka configuré. Ce plugin ne prend pas en charge la création de sujets sur Kafka.
   • Log Source Identifier: Cette valeur sera ajoutée en tant que préfixe à tous les journaux ingérés.
   • Exclude Timestamp Field: Select « Yes » pour ingérer les données sans le préfixe d'horodatage. Cette option ne s'applique qu'aux données au format JSON.
   • Exclude Log Source Identifier Field: Select Oui" pour ingérer les données sans le champ Log Source Identifier. Cette option ne s'applique qu'aux données au format JSON.
   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 Plaintext, alors Kafka Broker Address, Kafka Port, Topic Name, Exclude Timestamp Field and Exclude Log Source Identifier Field sont requis.

   

   

   

   Tous les paramètres ci-dessus sont nécessaires lorsque le protocole de sécurité est SSL. Si le protocole de sécurité est Plaintext, alors Kafka Broker Address, Kafka Port, Topic Name, Exclude Timestamp Field and Exclude Log Source Identifier Field sont requis.

4. Cliquer sur Save. La configuration de votre plugin sera disponible sur la page Log Shipper > Plugins.

   

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

1. Accédez aux règles de gestion. Par défaut, une règle de gestion filtre toutes les alertes et tous les événements. 
2. Si vous souhaitez filtrer un type d'alerte ou d'événement spécifique, cliquer sur Create New Rule et configurer une New règle de gestion en ajoutant le nom de la règle et le filtre.
3. Cliquez sur Save.

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. Accédez à la fonction d'enregistrement.
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.

Exemple d'alertes ingérées au format CEF :

Exemples d'événements ingérés au format CEF :

Exemple de Webtx ingéré au format CEF :

Exemple d’alerte ingérée au format JSON :

Exemple d'événement ingéré au format JSON :

Exemple de Webtx ingéré au format JSON :

Exemple de données JSON ingérées avec mappage personnalisé :

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. Supprimer l'ancien sujet dans 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 courtier Kafka et le port Kafka fournis 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.