第14章 Kerberos(GSSAPI)認証の使用

Kerberos KDC でプリンシパルキーを保存するサービスプリンシパルおよびキータブを作成します。

以下に例を示します。

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

以下に例を示します。

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

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

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

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";
};

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

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 パラメーターを使用して名前を追加します。

以下の要素で 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";
};

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

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
...

JVM パラメーターを使用して、Kafka ブローカーを起動し、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 認証に対して設定されます。

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

以下に例を示します。

# ...
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";
# ...

# ...
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";
# ...

クライアントを実行して、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