Kafka Plugin for Log Shipper

This document explains how to configure the Kafka v1.0.1 plugin with the Log Shipper module of the Netskope Cloud Exchange platform. This plugin ingests Netskope Alerts, Events, and WebTX[via Netskope LogStreaming] logs in CEF and JSON format from the Netskope Tenant into the Kafka topic on the Kafka server/cluster. The plugin will act as a producer to publish the message to the Kafka topic.

Prerequisites

To complete this configuration, you need:

Kafka Plugin Support

This plugin is used to transform and ingest the alerts, events, and WebTX (via Netskope LogStreaming) logs to the Kafka topic on the Kafka server/cluster. The plugin will act as a producer to publish the message to the Kafka topic.

Data Type	Support
Alerts	Yes (Compromised Credential, Policy, Malsite, Malware, DLP, Security Assessment, Watchlist, Quarantine, Remediation, UBA, CTEP)
Events	Yes (Page, Application, Audit, Infrastructure, Network, Incident)
WebTx Logs	Yes (via Netskope LogStreaming)
Syslog CE Logs	Not Supported

API Details

List of APIs Used

This plugin uses kafka-python libraries to create kafka producer and consumer.

The kafka-python library is a pure Python client for Apache Kafka that enables developers to interact with Kafka clusters directly from Python applications. It supports essential Kafka operations such as producing messages to topics, consuming messages from topics, and performing administrative tasks like creating or deleting topics. The library is easy to use and integrates well with Python-based systems, making it ideal for building real-time data pipelines or event-driven applications. With kafka-python, you can customize consumer group settings, manage offset commits, and configure producers with delivery guarantees, all using familiar Python code.

Creating the Consumer

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

Creating the Producer

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

To List the Topics Present on the Kafka cluster

consumer.topics()

To Send Data

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

Performance Matrix

This performance reading is conducted on a Large Stack CE with the below-mentioned VM specifications. The below readings are added with the consideration that it will pull around 10K alerts/events in ~8 seconds in CE.

Description	Specification
Stack details	Size: Large RAM: 32 GB CPU: 16 Cores
Alerts/Events ingested to SIEM	~200K EPM

User Agent

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

Workflow

Click play to watch a video.

Get your Kafka Configuration Parameters

The following configuration parameters are needed to configure the Kafka plugin for Log Shipper. Reach out to the Kafka server configuration team in order to get all the configuration parameters.

The ingestion of all the Events/Alerts and WebTx is done on the Offset Explorer (v3.0.2); here are the steps to configure the Kafka cluster on Offset Explorer.

1. Add a new Connection on Offset Explorer, and provide all the information for the cluster.

   

2. After a successful connection, go to the Topics folder under the connected cluster.

   

3. Click the + icon to add a new Topic in your cluster, and provide the Name of the Topic, Partition Count, and Replica Count. When finished, click Add.

   

4. After successful ingestion from Cloud Exchange, you can find the ingested data at Clusters > {cluster_name} > Topics > {topic_name} > Data.

   

Configure the Kafka Plugin

1. In Cloud Exchange, go to Settings > Plugin Store. Search for and select the Kafka v1.0.1 (CLS) plugin.

   

2. Enter a plugin configuration name and select a Mapping. Disable the Transform the raw logs toggle if you want to ingest your alerts and events in raw JSON format. 
   

   

3. Click Next and enter the configuration parameters:
   • Kafka Broker Address: DNS/IP Address/FQDN of Kafka broker to which data will be sent. Note that the plugin just needs one broker that will respond to Metadata API requests.
   • Kafka Port: Kafka Port address to which broker is configured.
   • Kafka Security Protocol: Select the security protocol using which authentication will be performed and data will be sent to the Kafka cluster.
   • Kafka CA Certificate: Kafka CA Certificate in PEM format. This configuration parameter is only applicable when SSL is selected as Kafka Security Protocol.
   • Kafka Client Certificate: Kafka Client Certificate in PEM format. Note: This configuration parameter is only applicable when SSL is selected as Kafka Security Protocol.
   • Kafka Client Private Key: Kafka Client Private Key in PEM format. This configuration parameter is only applicable when SSL is selected as Kafka Security Protocol.
   • Kafka SSL Private Key Password: The password that is used while loading the certificate. This configuration parameter only applies when SSL is selected as Kafka Security Protocol. It is only needed when the PEM file is generated without a passphrase.
   • Kafka Topic Name: Kafka Topic Name to which the logs should be sent. The Kafka Topic Name should not have any spaces in it.
   • Log Source Identifier: This will be added as prefix to all the logs. (The log source identifier should not contain the whitespace).
     All above mentioned parameters are needed when Security Protocol is SSL. If the security Protocol is Plaintext, add Kafka Broker Address, Kafka Port and Topic Name.
     
     
4. Click Save. The plugin configuration will be available on the Cloud Log Shipper > Plugins page.

   

Configure a Log Shipper Business Rule for Kafka

1. Go to Business Rules. By default, there is a business rule that filters all alerts and events. If you need to filter out any specific type of alert or event, click Create New Rule to configure a new business rule by adding the rule name and filter.
   

   

2. Click Save.

Configure Log Shipper Log Delivery for Kafka

1. Go to Log Shipper > Log Delivery and click Add Log Delivery Configuration.
2. For alerts and events, select the Source plugin (Netskope CLS plugin), the Destination plugin (Kafka plugin), and a business rule, and then click Save.
3. For ingestion of WebTx, select the Source plugin (AWS Netskope Log Streaming or Azure Netskope Log Streaming plugin), the Destination plugin (Kafka plugin), and then click Save.

Validate the Kafka Plugin

Validate in Cloud Exchange

In order to validate the plugin workflow:

1. Click Logging.
2. Search for ingested alerts with the filter message contains ingested.
3. The ingested logs will be filtered.
   

   

   

   

Validate in Kafka

The Kafka plugin was designed to send the CEF formatted data by encoding it to UTF-8, and JSON events by performing JSON serialization using json.dumps(), and then encoding it to UTF-8.

There are many ways to validate that the data is sent to the Kafka server, but here the Offset Explorer is used to validate it.

Troubleshooting the Kafka Plugin

Topic is created automatically in Kafka while ingesting data, even when it was deleted

What to do:

1. Add a new topic in Kafka.
2. Update the Kafka plugin with a new topic name and save it.
3. Delete the old topic on Kafka.

Following these steps will not create the deleted topic again on Kafka, and data will be ingested to the newly added topic.

Receiving “Kafka Broker is unreachable or Kafka cluster might be down

Verify Kafka Broker Address and Kafka Port provided in configuration parameters.

This issue might be due to one of these reasons:

• The Kafka server is actually down.
• The disk space for the server is full.

What to do:

1. Reach out to your IT and confirm which of the above reasons causes the error.
2. If Kafka is down, restarting it will work.
3. If disk space is full, either clear the unwanted files and empty the space, or get more disk space and restart the server.