1. Home
2. Platform
3. セキュリティ
4. 認証

実行中クラスターへのセキュリティの追加

このトピックでは、セキュリティ（SSL または SASL）を実行中のクラスターに追加する方法について説明します。

SSL または SASL 認証を実行するブローカーおよびクライアントへのセキュリティの追加

サポートされているプロトコルの 1 つまたは複数を使用して、実行中のクラスターをセキュアにできます。これは段階的に実行します。

1. 追加の（複数）セキュアポートを開くため、クラスターノードを段階的に再起動します。
2. PLAINTEXT ポートではなく、セキュアポートを使用してクライアントを再起動します（クライアントとブローカー間の接続をセキュアにしていると想定しています）。
3. ブローカー間のセキュリティを有効にするために、（必要な場合）再びクラスターを段階的に再起動します。
4. 最後に PLAINTEXT ポートを閉じるために段階的に再起動します。

セキュリティプロトコルを構成する具体的な手順は、それぞれ SSL および SASL のセクションに記載されています。これらの手順に従って、対象の（複数）プロトコルに対してセキュリティを有効にします。

このセキュリティ実装では、ブローカーとクライアント間、ブローカーとブローカー間のやりとりに異なるプロトコルを構成できます。これらは再起動ごとに有効にする必要があります。ブローカーまたはクライアントあるいはその両方が引き続きやりとりできるよう、PLAINTEXT ポートは開いたままにする必要があります。

段階的に再起動するとき、エンドユーザーのダウンタイムを回避するための ローリング再起動 の推奨事項を考慮します。

たとえば、ブローカーとクライアント間およびブローカーとブローカー間のやりとりを SSL で暗号化する場合、次のとおり実行します。

1. 1 回目の段階的再起動で、SSL ポートを各ノードで開きます。

   listeners=PLAINTEXT://broker1:9091,SSL://broker1:9092
   

   Copy

   注釈

   Kafka 1.1 以降では、リスナーを 動的に 追加または削除することで、一部のブローカー構成をブローカーを再起動することなくアップデートできます。新しいリスナーを追加するとき、リスナーのプレフィックス listener.name.{listenerName}. を使用して、リスナーのセキュリティ構成を指定します。新しいリスナーで SASL を使用する場合、リスナーとメカニズムのプレフィックスを付けた JAAS 構成プロパティ sasl.jaas.config を指定します。詳細については「JAAS」を参照してください。

2. 次にクライアントを再起動し、新しく開かれたセキュアポートを接続先に構成を変更します。

   bootstrap.servers=[broker1:9092,...]
   security.protocol=SSL
   ...etc
   

   Copy

   詳細については、「SSL による暗号化」を参照してください。

3. サーバーの 2 回目の段階的再起動で、ブローカー間のプロトコルとして SSL を使用するように Apache Kafka® に指示します（同じ SSL ポートを使用します）。

   listeners=PLAINTEXT://broker1:9091,SSL://broker1:9092
   security.inter.broker.protocol=SSL
   

   Copy

4. 最後の再起動では、次のとおり PLAINTEXT ポートを閉じて、クラスターをセキュアにします。

   listeners=SSL://broker1:9092
   security.inter.broker.protocol=SSL
   

   Copy

他の方法として、ブローカーとブローカー間およびブローカーとクライアント間のやりとりに異なるプロトコルを使用できるように、複数のポートを開くことができます。SSL 暗号化を全体（つまりブローカーとブローカー間およびブローカーとクライアント間のやりとり）に使用し、ブローカーとクライアント間の接続には SASL 認証を追加する場合、次の手順を実行します。

1. 1 回目の再起動中に、次のとおり 2 つの追加ポートを開きます。

   listeners=PLAINTEXT://broker1:9091,SSL://broker1:9092,SASL_SSL://broker1:9093
   

   Copy

2. もう一度クライアントを再起動して、新しく開いた SASL と SSL セキュアポートを接続先としてそれぞれの構成を変更します。

   bootstrap.servers=[broker1:9093,...]
   security.protocol=SASL_SSL
   ...etc
   

   Copy

   詳細については、「JAAS を使用した SASL による認証」を参照してください。

3. サーバーの 2 回目の再起動により、ポート 9092 で開いている SSL ポートを使用して、暗号化されたブローカー間のやり取りを実行するように、クラスターを切り替えます。

   listeners=PLAINTEXT://broker1:9091,SSL://broker1:9092,SASL_SSL://broker1:9093
   security.inter.broker.protocol=SSL
   

   Copy

4. 最後の再起動では、次のとおり PLAINTEXT ポートを閉じて、クラスターをセキュアにします。

   listeners=SSL://broker1:9092,SASL_SSL://broker1:9093
   security.inter.broker.protocol=SSL
   

   Copy

実行中の ZooKeeper クラスターへのセキュリティの追加

このセクションでは、 SASL （TLS あり、なし）または mTLS セキュリティを実行中の ZooKeeper クラスターに追加する方法について説明します。

SASL セキュリティの ZooKeeper への追加

セキュリティが無効な Kafka を実行していてクラスターをセキュアにする場合、次の手順を実行して、運用の停止が最小限になるように ZooKeeper の認証を有効にする必要があります。

注釈

Kafka クラスターが動作していない（ZooKeeper にコントローラーノードなし）とき ZooKeeper のセキュリティの移行を実施すると、コントローラーノードが作成されますが、null 値が入ります。このシナリオでは、リーダーの選出は、Kafka クラスターを後から起動した時に正しく機能しません。

1. zookeeper.properties で、次のとおり ZooKeeper セキュリティを有効にするために認証プロバイダーを追加します。

   authProvider.sasl=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
   

   Copy

2. 次のとおり ZooKeeper の JAAS ファイルをエクスポートしてから、ZooKeeper を再起動します。

   KAFKA_OPTS=-Djava.security.auth.login.config=<path>/zookeeper_server_jaas.conf
   

   Copy

   zookeeper_server_jaas.conf の内容は以下のようになります。

   Server {
      org.apache.zookeeper.server.auth.DigestLoginModule required
      user_super="adminsecret"
      user_bob="bobsecret";
   };
   

   Copy

3. 前に示した JAAS ログインファイルで ZooKeeper のローリング再起動を実行します。これによりブローカーが認証できます。

4. ZooKeeper で認証するために、ブローカーに対して JAAS ファイルを作成します。クライアント情報をブローカーの JAAS 構成に追加し、JAAS ファイルをエクスポートします。

   export KAFKA_OPTS=-Djava.security.auth.login.config=<path>/kafka_server_jaas.conf
   

   Copy

   ブローカーの JAAS ファイル（kafka_server_jaas.conf）の内容は次のようになります。

   Client {
      org.apache.zookeeper.server.auth.DigestLoginModule required
      username="bob"
      password="bobsecret";
   };
   

   Copy

5. ブローカーのローリング再起動を実行します。このとき構成パラメーター zookeeper.set.acl を true に設定します。これにより、znode を作成する際にセキュアな ACL の使用が有効になります。

6. ZkSecurityMigrator ツールを実行します。使用するスクリプト bin/zookeeper-security-migration で zookeeper.acl を secure に設定します。このツールは対応するサブツリー内をすべて検索し、znode の ACL を変更します。

セキュリティがブローカーと ZooKeeper 間で有効になっていることを検証する場合、次のとおり実行します。

1. ブローカーの JAAS 構成をエクスポートします。

   export KAFKA_OPTS=-Djava.security.auth.login.config=<path>/kafka_server_jaas.conf
   

   Copy

2. kafka-topic コマンドを使用して、test という名前の新しいトピックを作成します。

   bin/kafka-topics --bootstrap-server localhost:9092 --create --topic test --partitions 2 --replication-factor 2
   

   Copy

3. zookeeper-shell にログインし、新しく作成された znode の ACL で ACL が有効であることをチェックします。

   bin/zookeeper-shell <zk_host>:<zk_port>
   
   [zk: localhost:12181(CONNECTED) 9] getAcl /config/topics/test
   'world,'anyone
   : r
   'sasl,'bob
   : cdrwa
   

   Copy

セキュアなクラスターで認証をオフにする場合は、次の手順を実行します。

1. JAAS ログインファイルを設定して、ブローカーのローリング再起動を実行します。これによりブローカーで認証が有効になりますが、zookeeper.set.acl が false に設定されます。ローリング再起動の最後に、ブローカーはセキュアな ACL での znode の作成を停止しますが、まだすべての znode を認証し操作できます。
2. ZkSecurityMigrator ツールを実行します。使用するスクリプト bin/zookeeper-security-migration で zookeeper.acl を unsecure に設定します。このツールは対応するサブツリー内をすべて検索し、znode の ACL を変更します。
3. ブローカーの 2 回目のローリング再起動を実行します。このとき、JAAS ログインファイルを設定するシステムプロパティを除外します。

移行ツールの実行方法の一例を次に示します。

bin/zookeeper-security-migration --zookeeper.acl=secure --zookeeper.connect=localhost:2181

Copy

このコマンドを実行して、パラメーターの完全なリストを表示します。

bin/zookeeper-security-migration --help

Copy

SASL セキュリティでの TLS 暗号化の構成

SASL セキュリティで TLS 暗号化を構成するには、以下の手順を実行します。

1. zookeeper.properties で、次のとおり ZooKeeper セキュリティを有効にするために認証プロバイダーを追加します。

   authProvider.sasl=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
   

   Copy

   オプションとして、TLS 暗号化を有効にするため、次の構成を指定します（keyStore と trustStore で必ずキャメルケースを使用します）。この時点で、clientPort と secureClientPort があります。

   authProvider.x509=org.apache.zookeeper.server.auth.X509AuthenticationProvider
   secureClientPort=2182
   serverCnxnFactory=org.apache.zookeeper.server.NettyServerCnxnFactory
   ssl.keyStore.location=<path-to-zookeeper-keystore>
   ssl.keyStore.password=<zookeeper-keystore-password>
   ssl.trustStore.location=<path-to-zookeeper-truststore>
   ssl.trustStore.password=<zookeeper-truststore-password>
   ssl.clientAuth=none
   

   Copy

   重要

   ZooKeeper では、ZooKeeper サーバーのキーストアにあるキーパスワードをキーストアのパスワード自体と異なる値に設定することはできません。キーパスワードは、必ずキーストアのパスワードと同じになるように設定してください。

2. 次のとおり ZooKeeper の JAAS ファイルをエクスポートしてから、ZooKeeper を再起動します。

   KAFKA_OPTS=-Djava.security.auth.login.config=<path>/zookeeper_server_jaas.conf
   

   Copy

   zookeeper_server_jaas.conf の内容は以下のようになります。

   Server {
      org.apache.zookeeper.server.auth.DigestLoginModule required
      user_super="adminsecret"
      user_bob="bobsecret";
   };
   

   Copy

3. 前に示した JAAS ログインファイルで ZooKeeper のローリング再起動を実行します。これによりブローカーが認証できます。

4. ZooKeeper で認証するために、ブローカーに対して JAAS ファイルを作成します。クライアント情報をブローカーの JAAS 構成に追加し、JAAS ファイルをエクスポートします。

   export KAFKA_OPTS=-Djava.security.auth.login.config=<path-to-kafka_server_jaas.conf>
   

   Copy

   ブローカーの JAAS ファイル（kafka_server_jaas.conf）の内容は次のようになります。

   Client {
      org.apache.zookeeper.server.auth.DigestLoginModule required
      username="bob"
      password="bobsecret";
   };
   

   Copy

オプションで、ZooKeeper に対して TLS 暗号化を有効にする場合、これらのブローカー構成を追加します（truststore でキャメルケースを使用しません）。

# Connect to the ZooKeeper port configured for TLS
zookeeper.connect=zk1:2182,zk2:2182,zk3:2182
# Required to use TLS to ZooKeeper (default is false)
zookeeper.ssl.client.enable=true
# Required to use TLS to ZooKeeper
zookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty
# Define trust store to use TLS to ZooKeeper; ignored unless zookeeper.ssl.client.enable=true
zookeeper.ssl.truststore.location=<path-to-kafka-truststore>
zookeeper.ssl.truststore.password=<kafka-truststore-password>

Copy

1. ブローカーのローリング再起動を実行します。このとき zookeeper.set.acl=true を設定します。これにより znode を作成するとき、セキュアな ACL の使用が有効になります。
2. ZkSecurityMigrator ツールを実行します。使用するスクリプト bin/zookeeper-security-migration で zookeeper.acl を secure に設定します。このツールは対応するサブツリー内をすべて検索し、znode の ACL を変更します。

オプションで、ZooKeeper の TLS 暗号化を有効にする場合、ZooKeeper の TLS 暗号化ポート（2182 など）に接続し、--zk-tls-config-file <path-to-tls-configs> （2 個のダッシュに注意）にファイルのこの内容を指定します（ truststore はキャメルケースではありません）。

zookeeper.ssl.client.enable=true
zookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty
zookeeper.ssl.truststore.location=<path-to-kafka-truststore>
zookeeper.ssl.truststore.password=<kafka-truststore-password>

Copy

セキュリティがブローカーと ZooKeeper 間で有効になっていることを検証するには、次のとおり実行します。

1. ブローカーの JAAS 構成をエクスポートします。

   export KAFKA_OPTS=-Djava.security.auth.login.config=<path-to-kafka_server_jaas.conf>
   

   Copy

2. test という名前の新しいトピックを作成します。

   bin/kafka-topics --bootstrap-server <kafka-host>:<kafka-port> --create --topic test --partitions 2 --replication-factor 2
   

   Copy

3. zookeeper-shell にログインし、新しく作成された znode の ACL で ACL が有効であることをチェックします。

   bin/zookeeper-shell <zk_host>:<zk_port>
   
   [zk: localhost:12181(CONNECTED) 9] getAcl /config/topics/test
   'world,'anyone
   : r
   'sasl,'bob
   : cdrwa
   

   Copy

オプションで、ZooKeeper の TLS 暗号化を有効にする場合、TLS ポートに zookeeper-shell で接続します。このとき、コマンドラインフラグ -zk-tls-config-file <path-to-tls-configs> （シングルダッシュを使用）にファイルのこの内容を含めます（truststore でキャメルケースを使用しません）。

zookeeper.ssl.client.enable=true
zookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty
zookeeper.ssl.truststore.location=<path-to-kafka-truststore>
zookeeper.ssl.truststore.password=<kafka-truststore-password>

Copy

TLS 暗号化を有効にした場合、ここで ZooKeeper の clientPort 構成を削除できます。

mTLS セキュリティの ZooKeeper への追加

運用の中断を最小限に抑えて、mTLS セキュリティを実行中のクラスターに追加するには、次の手順を実行します。

1. ZooKeeper で SASL または mTLS あるいは両方の認証を有効にします。mTLS を有効にする場合、ここに示すとおり非 TLS ポートと TLS ポートを設定します。

   clientPort=2181
   secureClientPort=2182
   serverCnxnFactory=org.apache.zookeeper.server.NettyServerCnxnFactory
   authProvider.x509=org.apache.zookeeper.server.auth.X509AuthenticationProvider
   ssl.keyStore.location=<path-to-zookeeper-keystore>
   ssl.keyStore.password=<zookeeper-keystore-password>
   ssl.trustStore.location=<path-to-zookeeper-truststore>
   ssl.trustStore.password=<zookeeper-truststore-password>
   

   Copy

   重要

   ZooKeeper では、ZooKeeper サーバーのキーストアにあるキーパスワードをキーストアのパスワード自体と異なる値に設定することはできません。キーパスワードは、必ずキーストアのパスワードと同じになるように設定してください。

2. ブローカーのローリング再起動を実行します。

   必要に応じて、JAAS ログインファイルを指定、または ZooKeeper の mTLS 構成を定義する（TLS が有効な ZooKeeper ポートへの接続を含む）、あるいは両方を実行します。これによりブローカーを ZooKeeper で認証できます。

   ローリング再起動の最後に、ブローカーは znode を厳格な ACL で操作できますが、それらの ACL で znode を作成しません。ZooKeeper とは異なり、Kafka では TLS 関連の構成にキャメルケース名を使用しないことに注意してください（たとえば、Kafka では zookeeper.ssl.keystore.location を使用しますが、ZooKeeper では ssl.keyStore.location を使用します）。

   # Connect to the ZooKeeper port configured for TLS
   zookeeper.connect=zk1:2182,zk2:2182,zk3:2182
   # Required to use TLS to ZooKeeper (default is false)
   zookeeper.ssl.client.enable=true
   # Required to use TLS to ZooKeeper
   zookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty
   # Define key/trust stores to use TLS to ZooKeeper; ignored unless zookeeper.ssl.client.enable=true
   zookeeper.ssl.keystore.location=<path-to-kafka-keystore>
   zookeeper.ssl.keystore.password=<kafka-keystore-password>
   zookeeper.ssl.truststore.location=<path-to-kafka-truststore>
   zookeeper.ssl.truststore.password=<kafka-truststore-password>
   

   Copy

3. mTLS が有効になったら、ZooKeeper で非 TLS ポートを無効にします。

   secureClientPort=2182
   serverCnxnFactory=org.apache.zookeeper.server.NettyServerCnxnFactory
   authProvider.x509=org.apache.zookeeper.server.auth.X509AuthenticationProvider
   ssl.keyStore.location=<path-to-zookeeper-keystore>
   ssl.keyStore.password=<zookeeper-keystore-password>
   ssl.trustStore.location=<path-to-zookeeper-truststore>
   ssl.trustStore.password=<zookeeper-truststore-password>
   

   Copy

4. 2 回目のブローカーのローリング再起動を実行します。このとき zookeeper.set.acl=true を設定します。これにより znode を作成するとき、セキュアな ACL の使用が有効になります。

   # Connect to the ZooKeeper port configured for TLS
   zookeeper.connect=zk1:2182,zk2:2182,zk3:2182
   # Required to use TLS to ZooKeeper (default is false)
   zookeeper.ssl.client.enable=true
   # Required to use TLS to ZooKeeper
   zookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty
   # Define key/trust stores to use TLS to ZooKeeper; ignored unless zookeeper.ssl.client.enable=true
   zookeeper.ssl.keystore.location=<path-to-kafka-keystore>
   zookeeper.ssl.keystore.password=<kafka-keystore-password>
   zookeeper.ssl.truststore.location=<path-to-kafka-truststore>
   zookeeper.ssl.truststore.password=<kafka-truststore-password>
   # Tell broker to create ACLs on znodes
   zookeeper.set.acl=true
   

   Copy

5. ZkSecurityMigrator ツールを実行します。使用するスクリプト bin/zookeeper-security-migration で zookeeper.acl を secure に設定します。このツールは対応するサブツリー内をすべて検索し、znode の ACL を変更します。mTLS を有効にしているので、--zk-tls-config-file <file> オプションを指定します。

   ./bin/zookeeper-security-migration.sh --zookeeper.acl=secure --zookeeper.connect=localhost:2182 --zk-tls-config-file <path-to-tls-config-file.properties>
   

   Copy

   TLS の構成は以下のようになります。

   zookeeper.ssl.client.enable=true
   zookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty
   zookeeper.ssl.truststore.location=<path-to-kafka-truststore>
   zookeeper.ssl.truststore.password=<kafka-truststore-password>
   

   Copy

このセキュアなクラスターで mTLS 認証をオフにするには、次の手順を実行します。

1. JAAS ログインファイルを設定する、または ZooKeeper の mTLS 構成を定義する、あるいは両方を実行して、ブローカーのローリング再起動を実行します。これによりブローカーで認証できますが、zookeeper.set.acl を false に設定します。ローリング再起動の最後に、ブローカーはセキュアな ACL での znode の作成を停止しますが、まだすべての znode を認証し操作できます。

2. ZkSecurityMigrator ツールを実行します。使用するスクリプト bin/zookeeper-security-migration で zookeeper.acl を unsecure に設定します。このツールは対応するサブツリー内をすべて検索し、znode の ACL を変更します。TLS 構成を設定する必要がある場合、--zk-tls-config-file <file> を指定します。

   --zookeeper.connect=localhost:2182
   

   Copy

3. mTLS を無効にしている場合、ZooKeeper で非 TLS ポートを有効にします。

   clientPort=2181
   secureClientPort=2182
   serverCnxnFactory=org.apache.zookeeper.server.NettyServerCnxnFactory
   authProvider.x509=org.apache.zookeeper.server.auth.X509AuthenticationProvider
   ssl.keyStore.location=<path-to-zookeeper-keystore>
   ssl.keyStore.password=<zookeeper-keystore-password>
   ssl.trustStore.location=<path-to-zookeeper-truststore>
   ssl.trustStore.password=<zookeeper-truststore-password>
   

   Copy

4. ブローカーの 2 回目のローリング再起動を実行します。このとき必要に応じて JAAS ログインファイルの設定、または ZooKeeper の mTLS 構成（非 TLS が有効な ZooKeeper ポートへの接続を含む）の削除、あるいは両方を実行したシステムプロパティが除外されます。

5. mTLS を無効にしている場合、Kafka で TLS ポートを無効にします。

   clientPort=2181
   

   Copy

ZooKeeper アンサンブルの移行

ZooKeeper アンサンブルで認証も有効にする必要があります。これには、アンサンブルのローリング再起動の実行と、複数のプロパティの設定が必要です。詳細については、次の ZooKeeper のドキュメントを参照してください。

• Apache ZooKeeper ドキュメント
• Apache ZooKeeper wiki