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/kraft/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/kraft/server.properties

手順

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

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

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

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

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

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

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

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

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

   以下に例を示します。

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

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

   以下に例を示します。

   /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 ファイルを以下の要素で修正します。

   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/kraft/server.properties

4. Kafka プロデューサーおよびコンシューマークライアントの設定して 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