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

本書では、Kafka v1.1.0の構成方法について説明します。Netskope Cloud ExchangeプラットフォームのLog Shipperモジュールを使用したプラグイン。このプラグインは、Netskopeのアラート（DLP、マルウェア、ポリシー、侵害されたクレデンシャル、Malsite、検疫、修復、セキュリティ評価、ウォッチリスト、UBA、CTEP、デバイス、コンテンツ）、イベント（ページ、アプリケーション、監査、インフラストラクチャ、ネットワーク、インシデント、エンドポイント、クライアントステータス）、およびWebTxデータをKafkaトピックに送信するために使用されます。このプラグインは、JSONおよびCEF形式での取り込みをサポートしています。

前提条件

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

Kafkaプラグインのサポート

このプラグインは、Netskopeのアラート（DLP、マルウェア、ポリシー、侵害されたクレデンシャル、Malsite、検疫、修復、セキュリティ評価、ウォッチリスト、UBA、CTEP、デバイス、コンテンツ）、イベント（ページ、アプリケーション、監査、インフラストラクチャ、ネットワーク、インシデント、エンドポイント、クライアントステータス）、およびWebTxデータをKafkaトピックに送信するために使用されます。このプラグインは、JSONおよび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),
)

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

このパフォーマンス測定は、これらのVM仕様を備えたLarge Stack CE上で実施されています。これらの測定値は、CEにおいて約8秒間で約10,000件のアラート/イベントをプルすることを考慮して追加されています。

ユーザーエージェント

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

ワークフロー

動画を見る

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

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.1.0 (CLS)プラグインを検索して選択します。

   

2. プラグイン構成名を追加し、目的のマッピングを選択してください。アラート/イベント/WebTxを未加工のJSON形式で取り込む場合は、形式をJSONに変更してください。

   

3. Nextをクリックして、設定パラメータを入力してください。

   • Kafka Broker Address: データを送信するKafkaブローカのDNS/IPアドレス/FQDN。注：プラグインには、Metadata APIリクエストに応答するブローカが1つだけ必要です。
   • Kafka Port: 構成されたKafkaブローカがリッスンしているポート。値は1～65535の範囲の整数である必要があります。
   • Kafka Security Protocol: 認証を実行し、Kafkaクラスターにデータを送信するために使用するセキュリティプロトコルを選択します。「SSL」を選択すると、相互TLS接続を確立するために、Kafka CA証明書、Kafkaクライアント証明書、Kafkaクライアント秘密鍵、およびKafka SSL秘密鍵パスワードの各フィールドが必要になります。
   • Kafka CA Certificate: Kafkaブローカの身元を確認するために使用される、PEM形式のKafka CA証明書。Kafkaセキュリティプロトコルとして「SSL」が選択されている場合にのみ必要です。
   • Kafka Client Certificate: プラグインがKafkaブローカに対して自身を認証するために使用する、PEM形式のKafkaクライアント証明書。Kafkaセキュリティプロトコルとして「SSL」が選択されている場合にのみ必要です。
   • Kafka Client Private Key: Kafkaクライアント証明書とペアになる、PEM形式のKafkaクライアント秘密鍵。Kafkaセキュリティプロトコルとして「SSL」が選択されている場合にのみ必要です。
   • Kafka SSL Private Key Password: Kafkaクライアント秘密鍵のパスワード。秘密鍵のPEMファイルがパスフレーズ付きで生成された場合にのみ必要です。Kafkaセキュリティプロトコルとして「SSL」が選択されている場合にのみ必要です。
   • Kafka Topic Name: ログの送信先となるKafkaトピックの名前。注：Kafkaトピック名に空白を含めることはできません。また、設定済みのKafkaクラスター上に既に存在している必要があります。このプラグインは、Kafka上でのトピック作成をサポートしていません。
   • Log Source Identifier: この値は、取り込まれるすべてのログのプレフィックスとして追加されます。
   • Exclude Timestamp Field: Select ‘Yes’ to ingest the データ without the timestamp prefix.このオプションは、JSON形式のデータにのみ適用されます。
   • Exclude Log Source Identifier Field: ログソース識別子フィールドなしでデータを取り込む場合は、「はい」 Select 。 このオプションはJSON形式のデータにのみ適用されます。
   セキュリティプロトコルがSSLの場合、上記のすべてのパラメーターが必要です。セキュリティプロトコルがPlaintextの場合、Kafka Broker Address, Kafka Port, Topic Name, Exclude Timestamp Field and Exclude Log Source Identifier Fieldが必要です.

   

   

   

   セキュリティプロトコルがSSLの場合、上記のすべてのパラメータが必要です。セキュリティプロトコルがPlaintextの場合、Kafka Broker Address, Kafka Port, Topic Name, Exclude Timestamp Field and Exclude Log Source Identifier Fieldが必要です.

4. Saveをクリックしてください。プラグイン設定は、Log Shipper > Pluginsページで利用可能です。

   

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

1. ビジネスルールに移動します。デフォルトでは、すべてのアラートとイベントをフィルタリングするビジネスルールが存在します。 
2. 特定の種類のアラートやイベントを除外する必要がある場合は、Create New Ruleをクリックし、ルール名とフィルタを追加して新しいビジネスルールを構成してください。
3. Saveをクリックしてください。

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サーバーに送信されたことを検証する方法はたくさんありますが、ここではオフセットエクスプローラーが 検証に使用されます。

CEF形式で取り込まれたサンプルアラート：

CEF形式の取り込み済みイベントサンプル：

CEF形式の取り込み済みWebtxのサンプル：

JSON形式での取り込みアラートのサンプル：

JSON形式の取り込み済みイベントのサンプル：

JSON形式の取り込み済みWebtxサンプル：

カスタムマッピングを使用した取り込み済みJSONデータのサンプル：

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. ディスク容量がいっぱいになった場合は、不要なファイルを削除して空き容量を確保するか、ディスク容量を増設してサーバーを再起動してください。