クライアント認証を有効にするための Apache Kafka の構成

このタスクでは、Apache Kafka でクライアント認証を有効にする方法について説明します。

始める前に

  1. Kafka サーバーが使用するポートがファイアウォールによってブロックされていないことを確認します。
  2. Kafka コンシューマー (QRadar) と Kafka ブローカーの間のクライアント認証を有効にするには、クラスター内のブローカーとクライアントごとに鍵と証明書を生成する必要があります。 また、証明書は認証局 (CA) によって署名されている必要があります。

このタスクについて

以下のステップでは、CA を生成し、それを使用してクライアントおよびブローカーの証明書に署名を行い、その証明書をクライアントおよびブローカーのトラストストアに追加します。 また、Java keytool と OpenSSL を使用して、鍵と証明書を生成します。 別の方法として、外部 CA を複数の CA (1 つはブローカー証明書に署名するため、もう 1 つはクライアント証明書に署名するため) と組み合わせて使用することもできます。

手順

  1. トラストストア、鍵ストア、秘密鍵、および CA 証明書を生成します。
    注: 以下のコマンドの PASSWORD、VALIDITY、SERVER_ALIAS、および CLIENT_ALIAS は、適切な値に置き換えてください。
    1. サーバーの鍵ストアを作成します。
      注:

      ブローカー証明書の共通名 (CN) は、サーバー/ホストの完全修飾ドメイン名 (FQDN) と一致している必要があります。 QRadar によって使用される Kafka コンシューマー・クライアントは、CN を DNS ドメイン・ネームと比較して、悪意のあるブローカーではなく正しいブローカーに接続していることを確認します。 サーバーの鍵ストアの生成時には、CN/姓名の値に必ず FQDN を入力してください。

      keytool -keystore kafka.server.keystore.jks -alias SERVER_ALIAS
 -validity VALIDITY -genkey
      例
      keytool -keystore kafka.server.keystore.jks -alias server.hostname
 -validity 365 -genkey
    2. CA 証明書を生成します。
      注:

      この CA 証明書は、すべてのブローカー証明書およびクライアント証明書に署名するために使用できます。

      openssl req -new -x509 -keyout ca-key -out ca-cert -days VALIDITY
      例
      openssl req -new -x509 -keyout ca-key -out ca-cert -days 365
    3. サーバーのトラストストアを作成し、CA 証明書をインポートします。
      keytool -keystore kafka.server.truststore.jks -alias CARoot
 -import -file ca-cert
    4. クライアントのトラストストアを作成し、CA 証明書をインポートします。
      keytool -keystore kafka.client.truststore.jks -alias CARoot
 -import -file ca-cert
    5. サーバー証明書を生成し、CA を使用して署名します。
      keytool -keystore kafka.server.keystore.jks -alias SERVER_ALIAS
 -certreq -file cert-file
      openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-file -out
 cert-signed -days VALIDITY -CAcreateserial
      例
      keytool -keystore kafka.server.keystore.jks -alias server.hostname 
-certreq -file cert-file 
      openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-file -out
 cert-signed -days 365 -CAcreateserial
    6. サーバーの鍵ストアに CA 証明書をインポートします。
      keytool -keystore kafka.server.keystore.jks -alias CARoot
 -import -file ca-cert
    7. 署名済みのサーバー証明書をサーバーの鍵ストアにインポートします。
      keytool -keystore kafka.server.keystore.jks -alias SERVER_ALIAS -import
 -file cert-signed
      例
      keytool -keystore kafka.server.keystore.jks -alias server.hostname
 -import -file cert-signed
    8. サーバー証明書をバイナリー DER ファイルにエクスポートします。
      注: keytool -exportcert コマンドはデフォルトで DER 形式を使用します。 証明書は、Kafka と通信する任意の EP の trusted_certificates/ ディレクトリーに配置します。 構成で使用するすべてのブートストラップ・サーバーに対してサーバー証明書が必要です。 それ以外の場合、 QRadar はサーバーとの TLS ハンドシェークを拒否します。
      keytool -exportcert -keystore kafka.server.keystore.jks -alias
 SERVER_ALIAS -file SEVER_ALIAS.der
      例
      keytool -exportcert -keystore kafka.server.keystore.jks -alias
 server.hostname -file server.hostname.der
    9. クライアントの鍵ストアを生成します。
      keytool -keystore kafka.client.keystore.jks -alias CLIENT_ALIAS
 -validity VALIDITY -genkey
      例
      keytool -keystore kafka.client.keystore.jks -alias client.hostname
 -validity 365 -genkey
    10. クライアント証明書を生成し、CA を使用して署名します。
      keytool -keystore kafka.client.keystore.jks -alias CLIENT_ALIAS
 -certreq -file client-cert-file
      openssl x509 -req -CA ca-cert -CAkey ca-key -in client-cert-file -out
 client-cert-signed -days VALIDITY -CAcreateserial
      例
      keytool -keystore kafka.client.keystore.jks -alias client.hostname
 -certreq -file client-cert-file 
      openssl x509 -req -CA ca-cert -CAkey ca-key -in client-cert-file
 -out client-cert-signed -days 365 -CAcreateserial
    11. CA 証明書をクライアントの鍵ストアにインポートします。
      keytool -keystore kafka.client.keystore.jks -alias CARoot
 -import -file ca-cert
    12. 署名済みのクライアント証明書をクライアントの鍵ストアにインポートします。
      keytool -keystore kafka.client.keystore.jks -alias CLIENT_ALIAS
 -import -file client-cert-signed
      例
      keytool -keystore kafka.client.keystore.jks -alias client.hostname
 -import -file client-cert-signed
    13. クライアントの鍵ストアとトラストストアを QRadarにコピーします。
      1. kafka.client.keystore.jksおよびkafka.client.truststore.jksを、ログ・ソースが構成されている各イベント・プロセッサー上の/opt/qradar/conf/trusted_certificates/kafka/にコピーします。
      2. ブローカーごとに生成されたサーバー証明書 <filename>.der を /opt/qradar/conf/trusted_certificates/にコピーします。
  2. Kafka ブローカーでクライアント認証を構成します。
    1. 「ソケット・サーバー設定 (Socket Server Settings)」 セクションを見つけます。
    2. 以下のオプションの 1 つを実行します。
      • SASL 認証を使用していない場合は、listeners=PLAINTEXT://:<port>をlisteners=SSL://:<PORT>に変更し、security.inter.broker.protocol=SSLを追加します。
      • SASL 認証を使用している場合は、listeners=PLAINTEXT://:<port>をlisteners=SASL_SSL://:<PORT>に変更し、security.inter.broker.protocol=SASL_SSLを追加します。
    3. ブローカー間、およびブローカーとクライアント間で暗号化された通信を行うために、以下のプロパティーを追加します。 必要に応じて、パス、ファイル名、およびパスワードを調整してください。 以下のプロパティーは、サーバーのトラストストアと鍵ストアです。

      security.inter.broker.protocol=SSL

      ssl.client.auth=required

      ssl.keystore.location=/somefolder/kafka.server.keystore.jks

      ssl.keystore.password=test1234

      ssl.key.password=test1234

      ssl.truststore.location=/somefolder/kafka.server.truststore.jks

      ssl.truststore.password=test1234
      重要: パスワードは server.propertiesにプレーン・テキストで保管されるため、ファイルへのアクセスはファイル・システムの許可によって制限されることをお勧めします。
    4. server.properties が変更された Kafka ブローカーを再始動します。