クライアント認証を有効にするための 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. ログ・ソースが構成されている各イベント・プロセッサー上の /opt/qradar/conf/trusted_certificates/kafak/ に、kafka.client.keystore.jks と kafka.client.truststore.jks をコピーします。
      2. 各ブローカーに対して生成されたサーバー証明書 <filename>.der を、/opt/qradar/conf/trusted_certificates/ にコピーします。
  2. Kafka ブローカーでクライアント認証を構成します。
    1. ソケット・サーバーの設定セクションを探します。
    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 ブローカーを再始動します。