ログシッパー用Kafkaプラグイン

このドキュメントでは、 Netskope Cloud Exchangeの Log Shipper モジュールを使用して Kafka v1.0.1 プラグインを構成する方法について説明します。 このプラグインは、NetskopeテナントからNetskopeアラート、イベント、およびWebTX（Netskope LogStreaming経由）ログをCEFおよびJSON形式で取り込み、Kafkaサーバー/クラスタ上のKafkaトピックに格納します。このプラグインはプロデューサーとして機能し、メッセージをKafkaトピックに公開します。

前提条件

この設定を完了するには、以下が必要です。

Kafkaプラグインのサポート

このプラグインは アラート、イベント、および WebTX ( Netskope LogStreaming 経由) ログを変換して、Kafka サーバー/クラスタ上の Kafka トピックに取り込みます。 このプラグインはプロデューサーとして機能し、メッセージをKafkaトピックに公開します。

データ型	サポート
アラート	はい (侵害されたクレデンシャル、ポリシー、マルサイト、マルウェア、DLP、セキュリティ評価、ウォッチリスト、検疫、修復、UBA、CTEP)
イベント	はい（ページ、アプリケーション、監査、インフラストラクチャ、ネットワーク、インシデント）
WebTx Logs	はい（Netskope LogStreaming経由）
Syslog CEログ	サポートされていません

APIの詳細

使うAPI一覧

このプラグインは、kafka-python ライブラリを使用して、kafka プロデューサーとコンシューマーを作成します。

kafka-pythonライブラリは、Apache Kafka用の純粋なPythonクライアントであり、開発者がPythonアプリケーションからKafkaクラスタと直接やり取りすることを可能にします。トピックへのメッセージ送信、トピックからのメッセージ受信、トピックの作成や削除といった管理タスクの実行など、Kafkaの基本的な操作をサポートします。このライブラリは使いやすく Pythonベースのシステムとの統合性も高く、リアルタイムデータパイプラインやイベント駆動型アプリケーションの構築に最適です。 kafka-python を使用すると、コンシューマー グループの設定をカスタマイズしたり、オフセット コミットを管理したり、配信保証付きのプロデューサーを設定したりできます。 使い慣れた Python コードで。

消費者の創造

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

Kafkaクラスタに存在するトピックを一覧表示するには

consumer.topics()

データを送信する

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

パフォーマンスマトリックス

このパフォーマンス測定は、下記のVM仕様を持つ大規模スタックCE上で実施されました。以下の測定値は、CE環境で約8秒で約1万件のアラート/イベントを取得することを考慮に入れて追加されています。

デスクリプション	仕様
スタックの詳細	サイズ：大 RAM：32 GB CPU：16コア
SIEMに取り込まれたアラート/イベント	約20万EPM

ユーザーエージェント

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

ワークフロー

再生ボタンをクリックして動画をご覧ください。

Kafkaの設定パラメータを取得する

Log Shipper 用の Kafka プラグインを設定するには、以下の設定パラメータが必要です。すべての設定パラメータを入手するには、Kafkaサーバーの設定チームにお問い合わせください。

すべてのイベント/アラートとWebTxの取り込みはOffset Explorer（v3.0.2）で行われます。以下はOffset ExplorerでKafkaクラスタを構成する手順です。

1. Offset Explorerで新しい接続を追加し、クラスターに関するすべての情報を入力してください。

   

2. 接続が成功したら、接続先のクラスターの下にあるTopicsフォルダーに移動してください。

   

3. +アイコンをクリックしてクラスターに新しいトピックを追加し、トピック名、パーティション数、レプリカ数を入力してください。完了したら、 Addをクリックしてください。

   

4. Cloud Exchangeで取り込みに成功した後、取り込んだデータは Clusters > {cluster_name} > Topics > {topic_name} > Dataで確認できます。

   

Kafkaプラグインの設定

1. Cloud Exchange で、 Settings > Plugin Storeに移動します。Kafka v1.0.1 (CLS)プラグインを検索して選択してください。

   

2. プラグイン構成名を入力し、マッピングを選択してください。アラートとイベントをJSON形式で生データとして取り込む場合は、 「生ログを変換する」のトグルを無効にしてください。
   

   

3. Nextをクリックして、設定パラメータを入力してください。
   • Kafka Broker Addressデータが送信されるKafka ブローカのDNS/IPアドレス/FQDN。 このプラグインはメタデータAPIリクエストに応答するブローカーが1つだけ必要だという点に注意してください。
   • Kafka Port: ブローカーが設定されているKafkaポートアドレス。
   • Kafka Security Protocol: Select 使うセキュリティプロトコル 使う 認証が行われ、データがKafkaクラスタに送信されます。
   • Kafka CA CertificateKafka CA証明書（PEM形式）この設定パラメータは、KafkaのセキュリティプロトコルとしてSSLを選択した場合にのみ適用されます。
   • Kafka Client CertificateKafkaクライアント証明書（PEM形式）注：この設定パラメータは、KafkaのセキュリティプロトコルとしてSSLを選択した場合にのみ適用されます。
   • Kafka Client Private KeyKafkaクライアントの秘密鍵（PEM形式）。この設定パラメータは、KafkaのセキュリティプロトコルとしてSSLを選択した場合にのみ適用されます。
   • Kafka SSL Private Key Password: 証明書を読み込む際に使用するパスワード 証明書を読み込む際に使用するパスワード。 この設定パラメータは、KafkaのセキュリティプロトコルとしてSSLが選択された場合にのみ適用されます。これは、PEMファイルがパスフレーズなしで生成される場合にのみ必要です。
   • Kafka Topic Nameログを送信するKafkaトピック名。Kafkaトピック名にはスペースを含めてはいけません。
   • Log Source Identifierこれはすべてのログのプレフィックスとして追加されます。（ログソース識別子には空白文字を含めないでください。）
     上記のパラメータはすべて、セキュリティプロトコルがSSLの場合に必要です。セキュリティプロトコルが平文の場合は、 Kafka Broker Address, Kafka Port とTopic Name.
     を追加します。
     
4. Saveをクリックしてください。プラグインの設定はCloud Log Shipper > Pluginsページで確認できます。

   

Kafka 用のログシッパービジネスルールを設定する

1. Business Rulesへ移動してください。デフォルトでは、すべてのアラートとイベントをフィルタリングするビジネスルールが設定されています。特定のタイプのアラートまたはイベントをフィルターで除外する必要がある場合は、 「新しいルールの作成」をクリックして、ルール名とフィルターを追加して新しいビジネス ルールを構成します。
   

   

2. 「保存」をクリックしてください。

Kafka の Log Shipper ログ配信の設定

1. Log Shipper > Log Deliveryに移動してAdd Log Delivery Configurationをクリックしてください。
2. アラートやイベントについては、Sourceプラグイン(Netskope CLSプラグイン)、Destinationプラグイン(Kafkaプラグイン)、ビジネスルールを選択し、 Saveをクリックします。
3. WebTx を取り込むには、ソース プラグイン (AWS Netskope Log Streaming または Azure Netskope Log Streaming プラグイン)、宛先プラグイン (Kafka プラグイン) を選択し、 Saveをクリックします。

Kafkaプラグインを検証する

Cloud Exchangeで検証する

プラグインのワークフローを検証するために：

1. 「ログ記録」をクリックします。
2. フィルターメッセージに「insted」が含まれるようにして、取り込まれたアラートを検索します。
3. 取り込まれたログはフィルタリングされます。
   

   

   

   

Kafkaで検証する

Kafkaプラグインは、CEF形式のデータをUTF-8にエンコードして送信し、JSONイベントをJSONシリアライゼーションを実行して送信するように設計されています。 json.dumps()、 そしてそれをUTF-8にエンコードする。

データがKafkaサーバーに送信されたことを検証する方法はたくさんありますが、ここではオフセットエクスプローラーが 検証に使用されます。

Kafkaプラグインのトラブルシューティング

トピックは、削除された場合でも、データの取り込み中に Kafka に自動的に作成されます。

What to do:

1. Kafkaに新しいトピックを追加します。
2. Kafkaプラグインを新しいトピック名で更新し、保存してください。
3. Kafka上の古いトピックを削除します。

これらの手順を実行しても、削除されたトピックがKafka上に再度作成されることはなく、データは新しく追加されたトピックに取り込まれます。

「Kafkaブローカに到達できないか、Kafkaクラスターがダウンしている可能性があります」というエラーメッセージが表示される

設定パラメータで指定された Kafka ブローカ アドレスと Kafka ポートを確認してください。

この問題は、以下のいずれかの理由による可能性があります。

• Kafkaサーバーが実際にダウンしています。
• サーバーのディスク容量がいっぱいです。

What to do:

1. IT部門に連絡して、上記のどの原因がエラーを引き起こしているのかを確認してください。
2. Kafkaがダウンしている場合は、再起動すれば解決します。
3. ディスク容量がいっぱいになった場合は、不要なファイルを削除して空き容量を確保するか、ディスク容量を増設してサーバーを再起動してください。