6.3. 認証

手順

1. ユーザー証明書の署名に使用される CA (認証局) の公開鍵を含む JKS (Java Keystore) トラストストアを準備します。

2. すべてのクラスターノードの Kafka 設定プロパティーファイルを次のように編集します。

   • ssl.truststore.location プロパティーを使用して JKS トラストストアへのパスを指定します。
   • トラストストアがパスワードで保護される場合は、ssl.truststore.password プロパティーを使用してパスワードを設定します。

   • ssl.client.auth プロパティーを required に設定します。

     TLS クライアント認証の設定

     ssl.truststore.location=/path/to/truststore.jks
     ssl.truststore.password=123456
     ssl.client.auth=required

3. Kafka ブローカーを (再) 起動します。

手順

1. /opt/kafka/config/jaas.conf JAAS 設定ファイルを編集または作成して、PlainLoginModule を有効にし、許可されるユーザー名とパスワードを指定します。

   このファイルがすべての Kafka ブローカーで同じであることを確認します。

   JAAS 設定

   KafkaServer {
       org.apache.kafka.common.security.plain.PlainLoginModule required
       user_admin="123456"
       user_user1="123456"
       user_user2="123456";
   };

2. すべてのクラスターノードの Kafka 設定プロパティーファイルを次のように編集します。

   • listener.security.protocol.map プロパティーを使用して、特定のリスナーで SASL PLAIN 認証を有効にします。SASL_PLAINTEXT または SASL_SSL を指定します。

   • sasl.enabled.mechanisms プロパティーを PLAIN に設定します。

     SASL plain 設定

     listeners=INSECURE://:9092,AUTHENTICATED://:9093,REPLICATION://:9094
     listener.security.protocol.map=INSECURE:PLAINTEXT,AUTHENTICATED:SASL_PLAINTEXT,REPLICATION:PLAINTEXT
     sasl.enabled.mechanisms=PLAIN

3. KAFKA_OPTS 環境変数を使用して Kafka ブローカーを (再) 起動し、JAAS 設定を Kafka ブローカーに渡します。

   su - kafka
   export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/jaas.conf"; /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

手順

1. /opt/kafka/config/jaas.conf JAAS 設定ファイルを編集または作成して、ScramLoginModule を有効にします。

   このファイルがすべての Kafka ブローカーで同じであることを確認します。

   JAAS 設定

   KafkaServer {
       org.apache.kafka.common.security.scram.ScramLoginModule required;
   };

2. すべてのクラスターノードの Kafka 設定プロパティーファイルを次のように編集します。

   • listener.security.protocol.map プロパティーを使用して、特定のリスナーで SASL SCRAM 認証を有効にします。SASL_PLAINTEXT または SASL_SSL を指定します。

   • sasl.enabled.mechanisms オプションを SCRAM-SHA-256 または SCRAM-SHA-512 に設定します。

     以下に例を示します。

     listeners=INSECURE://:9092,AUTHENTICATED://:9093,REPLICATION://:9094
     listener.security.protocol.map=INSECURE:PLAINTEXT,AUTHENTICATED:SASL_PLAINTEXT,REPLICATION:PLAINTEXT
     sasl.enabled.mechanisms=SCRAM-SHA-512

3. KAFKA_OPTS 環境変数を使用して Kafka ブローカーを (再) 起動し、JAAS 設定を Kafka ブローカーに渡します。

   su - kafka
   export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/jaas.conf"; /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

手順

1. sasl.mechanism.inter.broker.protocol プロパティーを使用して、Kafka 設定でブローカー間 SASL メカニズムを指定します。

   ブローカー間 SASL メカニズム

   sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512

2. (オプション) SCRAM メカニズムを使用している場合は、SCRAM ユーザーを追加 して、Kafka クラスターに SCRAM 認証情報を登録します。

   これにより、Kafka クラスターの実行前に、認証用の認証情報がブートストラップ用に更新されるようになります。

3. username フィールドと password フィールドを使用して、KafkaServer JAAS コンテキストでのブローカー間通信用のユーザー名とパスワードを指定します。

   ブローカー間 JAAS コンテキスト

   KafkaServer {
       org.apache.kafka.common.security.plain.ScramLoginModule required
       username="admin"
       password="123456"
       # ...
   };

1. サービスプリンシパル
2. Kafka ブローカー (Kerberos ログインを使用するため)
3. ZooKeeper (Kerberos ログインを使用するため)
4. プロデューサーおよびコンシューマークライアント (Kerberos 認証を使用して Kafka にアクセスするため)

1. Kerberos KDC を使用してサービスプリンシパルと、プリンシパルキーを保存するキータブを作成します。

   Kerberos プリンシパルのドメイン名が大文字であることを確認してください。

   以下に例を示します。

   • zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM
   • kafka/node1.example.redhat.com@EXAMPLE.REDHAT.COM
   • producer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM

   • consumer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM

     ZooKeeper サービスプリンシパルは、Kafka config/server.properties ファイルの zookeeper.connect 設定と同じホスト名にする必要があります。

     zookeeper.connect=node1.example.redhat.com:2181

     ホスト名が異なる場合は localhost が使用されるため、認証に失敗します。

2. ホストにディレクトリーを作成し、キータブファイルを追加します。

   以下に例を示します。

   /opt/kafka/krb5/zookeeper-node1.keytab
   /opt/kafka/krb5/kafka-node1.keytab
   /opt/kafka/krb5/kafka-producer1.keytab
   /opt/kafka/krb5/kafka-consumer1.keytab

3. kafka ユーザーがディレクトリーにアクセスできることを確認します。

   chown kafka:kafka -R /opt/kafka/krb5

1. opt/kafka/config/jaas.conf ファイルを作成または変更して、ZooKeeper クライアントおよびサーバー操作をサポートします。

   Client {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true 1
       storeKey=true 2
       useTicketCache=false 3
       keyTab="/opt/kafka/krb5/zookeeper-node1.keytab" 4
       principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM"; 5
   };
   
   Server {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true
       storeKey=true
       useTicketCache=false
       keyTab="/opt/kafka/krb5/zookeeper-node1.keytab"
       principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };
   
   QuorumServer {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true
       storeKey=true
       keyTab="/opt/kafka/krb5/zookeeper-node1.keytab"
       principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };
   
   QuorumLearner {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true
       storeKey=true
       keyTab="/opt/kafka/krb5/zookeeper-node1.keytab"
       principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };

   1

   true に設定し、キータブからプリンシパルキーを取得します。

   2

   true に設定し、プリンシパルキーを保存します。

   3

   true に設定し、チケットキャッシュから Ticket Granting Ticket (TGT) を取得します。

   4

   keyTab プロパティーは、Kerberos KDC からコピーされた keytab ファイルの場所を示します。その場所とファイルは、kafka ユーザーが読み取り可能になものにする必要があります。

   5

   principal プロパティーは、KDC ホストで作成された完全修飾プリンシパル名と一致するように設定され、その形式は SERVICE-NAME/FULLY-QUALIFIED-HOST-NAME@DOMAIN-NAME に従います。

2. opt/kafka/config/zookeeper.properties を編集して、更新された JAAS 設定を使用します。

   # ...
   
   requireClientAuthScheme=sasl
   jaasLoginRenew=3600000 1
   kerberos.removeHostFromPrincipal=false 2
   kerberos.removeRealmFromPrincipal=false 3
   quorum.auth.enableSasl=true 4
   quorum.auth.learnerRequireSasl=true 5
   quorum.auth.serverRequireSasl=true
   quorum.auth.learner.loginContext=QuorumLearner 6
   quorum.auth.server.loginContext=QuorumServer
   quorum.auth.kerberos.servicePrincipal=zookeeper/_HOST 7
   quorum.cnxn.threads.size=20

   1

   ログイン更新の頻度をミリ秒単位で制御します。これは、チケットの更新間隔に合わせて調整できます。デフォルトは 1 時間です。

   2

   ホスト名がログインプリンシパル名の一部として使用されるかどうかを指定します。クラスターのすべてのノードで単一の keytab を使用する場合、これは true に設定されます。ただし、トラブルシューティングのために、各ブローカーホストに個別のキータブと完全修飾プリンシパルを生成することが推奨されます。

   3

   Kerberos ネゴシエーションのプリンシパル名からレルム名を削除するかどうかを制御します。この設定は、false にすることを推奨します。

   4

   ZooKeeper サーバーおよびクライアントの SASL 認証メカニズムを有効にします。

   5

   RequireSasl プロパティーは、マスター選出などのクォーラムイベントに SASL 認証を必要とするかどうかを制御します。

   6

   loginContext プロパティーは、指定されたコンポーネントの認証設定に使用される JAAS 設定のログインコンテキストの名前を識別します。loginContext 名は、opt/kafka/config/jaas.conf ファイルの関連セクションの名前に対応します。

   7

   識別に使用されるプリンシパル名を形成するために使用される命名規則を制御します。プレースホルダー _HOST は、実行時に server.1 プロパティーによって定義されたホスト名に自動的に解決されます。

3. JVM パラメーターで ZooKeeper を起動し、Kerberos ログイン設定を指定します。

   su - kafka
   export EXTRA_ARGS="-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/opt/kafka/config/jaas.conf"; /opt/kafka/bin/zookeeper-server-start.sh -daemon /opt/kafka/config/zookeeper.properties

   デフォルトのサービス名 (zookeeper) を使用していない場合は、-Dzookeeper.sasl.client.username=NAME パラメーターを使用して名前を追加します。

   注記

   /etc/krb5.conf を場所として使用している場合は、ZooKeeper、Kafka、Kafka プロデューサーおよびコンシューマーの起動時に -Djava.security.krb5.conf=/etc/krb5.conf を指定する必要はありません。

1. opt/kafka/config/jaas.conf ファイルを以下の要素で修正します。

   KafkaServer {
       com.sun.security.auth.module.Krb5LoginModule required
       useKeyTab=true
       storeKey=true
       keyTab="/opt/kafka/krb5/kafka-node1.keytab"
       principal="kafka/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };
   KafkaClient {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true
       storeKey=true
       useTicketCache=false
       keyTab="/opt/kafka/krb5/kafka-node1.keytab"
       principal="kafka/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };

2. Kafka クラスターの各ブローカーを設定するには、config/server.properties ファイルのリスナー設定を変更して、リスナーが SASL/GSSAPI ログインを使用するようにします。

   リスナーのセキュリティープロトコルのマップに SASL プロトコルを追加し、不要なプロトコルを削除します。

   以下に例を示します。

   # ...
   broker.id=0
   # ...
   listeners=SECURE://:9092,REPLICATION://:9094 1
   inter.broker.listener.name=REPLICATION
   # ...
   listener.security.protocol.map=SECURE:SASL_PLAINTEXT,REPLICATION:SASL_PLAINTEXT 2
   # ..
   sasl.enabled.mechanisms=GSSAPI 3
   sasl.mechanism.inter.broker.protocol=GSSAPI 4
   sasl.kerberos.service.name=kafka 5
   ...

   1

   クライアントとの汎用通信用のセキュアなリスナー (通信用 TLS をサポート)、およびブローカー間通信用のレプリケーションリスナーの 2 つが設定されています。

   2

   TLS に対応しているリスナーのプロトコル名は SASL_PLAINTEXT になります。コネクターが TLS に対応していない場合、プロトコル名は SASL_PLAINTEXT になります。SSL が必要ない場合は、ssl.* プロパティーを削除できます。

   3

   Kerberos 認証における SASL メカニズムは GSSAPI になります。

   4

   ブローカー間通信の Kerberos 認証。

   5

   認証要求に使用するサービスの名前は、同じ Kerberos 設定を使用している他のサービスと区別するために指定されます。

3. Kafka ブローカーを起動し、JVM パラメーターを使用して Kerberos ログイン設定を指定します。

   su - kafka
   export KAFKA_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/opt/kafka/config/jaas.conf"; /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

   ブローカーおよび ZooKeeper クラスターが設定されていて、Kerberos ベース以外の認証システムで動作している場合は、ZooKeeper およびブローカークラスターを起動し、ログで設定エラーを確認できます。

   ブローカーおよび Zookeeper インスタンスを起動すると、Kerberos 認証用にクラスターが設定されました。

1. プロデューサーまたはコンシューマー設定ファイルに Kerberos 設定を追加します。

   以下に例を示します。

   /opt/kafka/config/producer.properties

   # ...
   sasl.mechanism=GSSAPI 1
   security.protocol=SASL_PLAINTEXT 2
   sasl.kerberos.service.name=kafka 3
   sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \ 4
       useKeyTab=true  \
       useTicketCache=false \
       storeKey=true  \
       keyTab="/opt/kafka/krb5/producer1.keytab" \
       principal="producer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   # ...

   1

   Kerberos (GSSAPI) 認証の設定。

   2

   Kerberos は SASL プレーンテキスト (ユーザー名/パスワード) セキュリティープロトコルを使用します。

   3

   Kerberos KDC で設定された Kafka のサービスプリンシパル (ユーザー)。

   4

   jaas.conf で定義されたものと同じプロパティーを使用した JAAS の設定。

   /opt/kafka/config/consumer.properties

   # ...
   sasl.mechanism=GSSAPI
   security.protocol=SASL_PLAINTEXT
   sasl.kerberos.service.name=kafka
   sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
       useKeyTab=true  \
       useTicketCache=false \
       storeKey=true  \
       keyTab="/opt/kafka/krb5/consumer1.keytab" \
       principal="consumer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   # ...

2. クライアントを実行して、Kafka ブローカーからメッセージを送受信できることを確認します。

   プロデューサークライアント:

   export KAFKA_HEAP_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Dsun.security.krb5.debug=true"; /opt/kafka/bin/kafka-console-producer.sh --producer.config /opt/kafka/config/producer.properties  --topic topic1 --bootstrap-server node1.example.redhat.com:9094

   コンシューマークライアント:

   export KAFKA_HEAP_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Dsun.security.krb5.debug=true"; /opt/kafka/bin/kafka-console-consumer.sh --consumer.config /opt/kafka/config/consumer.properties  --topic topic1 --bootstrap-server node1.example.redhat.com:9094