このトピックでは、E-MapReduce（EMR）Kafka で認証用の LDAP ユーザーを設定する方法について説明し、例を示します。
前提条件
Dataflow クラスタが作成され、クラスタに対して Kafka サービスと OpenLDAP サービスが選択されています。クラスタの作成方法の詳細については、「Dataflow Kafka クラスタの作成」をご参照ください。
制限事項
このトピックは、EMR V3.44.0 以降または EMR V5.10.0 以降のマイナーバージョンの Kafka クラスタに適用されます。
EMR OpenLDAP サービスまたは外部 LDAP サービスがクラスタにデプロイされていることを確認する必要があります。
このトピックでは、EMR OpenLDAP サービスを使用します。
注意事項
認証用のユーザーグループを設定する場合、LDAP サービスに対して memberOf オーバーレイ機能が有効になっているか、memberOf プロパティを LDAP ユーザーに対して設定できることを確認する必要があります。
認証用の LDAP ユーザーの設定
スーパーユーザーを作成します。
説明スーパーユーザーを作成した場合は、この手順をスキップします。
Kafka スーパーユーザーは、すべての Kafka リソースにアクセスできます。この例では、スーパーユーザーを使用して、Kafka のブローカーノードとコンポーネントにアクセスします。この例では、Kafka スーパーユーザーが EMR OpenLDAP サービスに追加されます。
SSH モードでクラスタの master-1-1 ノードに接続します。詳細については、「クラスタへのログオン」をご参照ください。
kafka.ldif という名前のファイルを作成し、次の情報をファイルに追加します。
この例では、
ldapコマンドを実行して、UID が kafka、パスワードが kafka-secret の LDAP ユーザーを追加します。dn: uid=kafka,ou=people,o=emr cn: kafka sn: kafka objectClass: inetOrgPerson userPassword: kafka-secret uid: kafka次のコマンドを実行して、LDAP ユーザーを追加します。
ldapadd -H ldap://master-1-1:10389 -f kafka.ldif -D ${uid} -w ${rootDnPW}説明${uid}: ${uid} を [admin_dn] パラメーターの値に置き換えます。 admin_dn パラメーターの値は、EMR コンソールの OpenLDAP サービスページの [設定] タブで取得できます。${rootDnPW}: ${rootDnPW} を [admin_pwd] パラメーターの値に置き換えます。 admin_pwd パラメーターの値は、EMR コンソールの OpenLDAP サービスページの [設定] タブで取得できます。10389: OpenLDAP サービスのリスニングポート。
LDAP ユーザーを追加した後、次のコマンドを実行してユーザーに関する情報を表示できます。
ldapsearch -w ${rootDnPW} -D ${uid} -H ldap://master-1-1:10389 -b uid=kafka,ou=people,o=emr
Kafka サービスページの [設定] タブに移動します。
上部のナビゲーションバーで、クラスターが存在するリージョンを選択し、ビジネス要件に基づいてリソースグループを選択します。
EMR on ECS ページで、目的のクラスタを見つけ、[アクション] 列の [サービス] をクリックします。
[サービス] タブで、Kafka サービスを見つけ、[設定] をクリックします。
設定項目を変更し、設定を保存します。
Kafka サービスページの [設定] タブで、[server.properties] タブをクリックします。
[kafka.ssl.config.type] 設定項目の値を [CUSTOM] に変更します。
[authorizer.class.name] 設定項目の値を [kafka.security.ldap.authorizer.simpleldapauthorizer] に変更します。
設定を保存します。
Kafka サービスページの [設定] タブで、[保存] をクリックします。
表示されるダイアログボックスで、[実行理由] パラメーターを設定し、[設定を自動的に更新] をオンにします。
設定項目を追加し、設定を保存します。
[server.properties] タブで、[設定項目の追加] をクリックします。
[設定項目の追加] ダイアログボックスで、次の表に記載されている設定項目を追加します。
設定項目
値
説明
super.users
User:kafka
kafkaは、作成したスーパーユーザーの名前です。ビジネス要件に基づいてユーザー名を置き換えます。ユーザーの作成方法の詳細については、「手順 1」をご参照ください。
listener.name.${listener}.sasl.enabled.mechanisms
PLAIN
${listener} をリスナーの名前に置き換えます。例: sasl_ssl。
listener.name.${listener}.plain.sasl.jaas.config
org.apache.kafka.common.security.plain.PlainLoginModule required username="kafka" password="kafka-secret" emr.kafka.security.ldap.user.name.attribute="uid" emr.kafka.security.ldap.user.base.dn="ou=people,o=emr" emr.kafka.security.ldap.group.name.attribute="cn" emr.kafka.security.ldap.group.base.dn="ou=groups,o=emr" emr.kafka.security.ldap.admin.name.attribute="uid" emr.kafka.security.ldap.admin.base.dn="o=emr" emr.kafka.security.ldap.url="ldaps://master-1-1:10636" emr.kafka.security.ldap.bind.user="admin" emr.kafka.security.ldap.bind.user.password="WMMuhh3P**********" emr.kafka.security.ldap.user.member.of.attribute="memberOf" emr.kafka.security.ldap.group.authorization.support="true" ;${listener} をリスナーの名前に置き換えます。例: sasl_ssl。
この設定項目の次のオプションの値を、LDAP ユーザーの実際の構成に基づいて置き換えます。
username: 作成したスーパーユーザーの名前。 Kafka サービスは、スーパーユーザーを使用してノードにアクセスします。mr.kafka.security.ldap.user.name.attribute: LDAP ユーザーの name プロパティ。このプロパティを使用してユーザー名を取得できます。emr.kafka.security.ldap.user.base.dn: LDAP ユーザーのベース識別名（DN）。emr.kafka.security.ldap.group.name.attribute: LDAP ユーザーグループの name プロパティ。このプロパティを使用してユーザーグループの名前を取得できます。
emr.kafka.security.ldap.group.base.dn: LDAP ユーザーグループのベース DN。emr.kafka.security.ldap.admin.name.attribute: LDAP 管理ユーザーの name プロパティ。emr.kafka.security.ldap.admin.base.dn: LDAP 管理ユーザーのベース DN。emr.kafka.security.ldap.url: OpenLDAP サービスの URL。emr.kafka.security.ldap.bind.use: ユーザーグループ認証に使用できる LDAP 管理者のユーザー名。emr.kafka.security.ldap.bind.user.password: LDAP 管理者のパスワード。emr.kafka.security.ldap.user.member.of.attribute: ユーザーが属するグループのプロパティ。このプロパティを使用して、ユーザーが属するグループを取得できます。emr.kafka.security.ldap.group.authorization.support: グループ認証をサポートするかどうかを指定します。このパラメーターを true に設定し、ユーザーが属するグループに権限がある場合、ユーザーはグループの権限を継承します。
listener.name.${listener}.plain.sasl.server.callback.handler.class
kafka.security.ldap.authenticator.LdapAuthenticateCallbackHandler
${listener} をリスナーの名前に置き換える必要があります。例: sasl_ssl。
sasl.mechanism.inter.broker.protocol
PLAIN
なし。
sasl.enabled.mechanisms
PLAIN
なし。
他の設定項目を変更します。
ビジネス要件に基づいて、次の設定項目を変更します。
説明kafka.client.jaas.content パラメーターの値のユーザー名とパスワードは、ビジネス要件に基づいて置き換える必要があります。スーパーユーザーを使用することをお勧めします。
設定ファイル
パラメーター
値
説明
[kafka_client_jaas.conf]
kafka.client.jaas.content
KafkaClient { org.apache.kafka.common.security.plain.PlainLoginModule required username="kafka" password="kafka-secret"; }; // ユーザー名とパスワードをクライアントに合わせて置き換えます。このパラメーターの値はセミコロン（;）で終わる必要があります。
[schema-registry.properties]
schema_registry_opts
-Djava.security.auth.login.config=/etc/taihao-apps/kafka-conf/kafka-conf/kafka_client_jaas.confこのパラメーターに既に値がある場合は、必要な情報を既存の値の末尾に追加します。
[kafka-rest.properties]
kafkarest_opts
-Djava.security.auth.login.config=/etc/taihao-apps/kafka-conf/kafka-conf/kafka_client_jaas.confこのパラメーターに既に値がある場合は、必要な情報を既存の値の末尾に追加します。
Kafka サービスを再起動します。
Kafka サービスページの [設定] タブで、 を選択します。
表示されるダイアログボックスで、[実行理由] パラメーターを設定し、[OK] をクリックします。
[確認] メッセージで、[OK] をクリックします。
例
この例の操作は、オープンソース Kafka 2.4.1 に適用されます。他の Kafka バージョンで認証用の LDAP ユーザーを設定する方法の詳細については、「Apache Kafka」を参照してください。
テストユーザーとテストユーザーグループを作成します。
OpenLDAP サービスで memberOf オーバーレイ機能が無効になっている場合は、次の操作を実行して機能を有効にします。
説明memberOf オーバーレイ機能を有効にするために使用される方法は、使用する LDAP サービスによって異なります。次の操作は参考用です。
OpenLDAP サービスがデプロイされているすべてのノードで memberOf オーバーレイ機能を有効にする必要があります。
必要な設定ファイルの次のパラメーターを、ビジネス要件に基づいて変更する必要があります。例：
olcModulepath: 32 ビットオペレーティングシステムの場合は、このパラメーターを /usr/lib/openldap に設定します。dn: cn=module{0},cn=config: cn=module{0}.ldif ファイルが既に /etc/openldap/slapd.d/cn=config ディレクトリに存在する場合は、cn=module{0}.ldif の 0 を 1 に変更する必要があります。 cn=module{1}.ldif ファイルが既に存在する場合は、cn=module{1}.ldif の 1 を 2 に変更する必要があります。ファイル名のすべての数字は同じルールに従います。dn: olcOverlay={0}memberof,olcDatabase={2}hdb,cn=config: /etc/openldap/slapd.d/cn=config ディレクトリの olcDatabase パラメーターの値に基づいて olcDatabase パラメーターを設定できます。たとえば、olcDatabaseパラメーターの値が {2}hdb.ldif の場合は、olcDatabase パラメーターを hdb に設定する必要があります。
vim memberof_config.ldifコマンドを実行して、memberof_config.ldif ファイルを開き、次の情報をファイルに追加します。dn: cn=module{0},cn=config cn: module{0} objectClass: olcModuleList objectclass: top olcModuleload: memberof.la olcModulePath: /usr/lib64/openldap dn: olcOverlay={0}memberof,olcDatabase={2}hdb,cn=config objectClass: olcConfig objectClass: olcMemberOf objectClass: olcOverlayConfig objectClass: top olcOverlay: memberof olcMemberOfDangling: ignore olcMemberOfRefInt: TRUE olcMemberOfGroupOC: groupOfNames olcMemberOfMemberAD: member olcMemberOfMemberOfAD: memberOf // 必要なパラメーターを変更します。vim refint1.ldifコマンドを実行して、refint1.ldif ファイルを開き、次の情報をファイルに追加します。dn: cn=module{0},cn=config add: olcmoduleload olcmoduleload: refintvim refint2.ldifコマンドを実行して、refint2.ldif ファイルを開き、次の情報をファイルに追加します。dn: olcOverlay=refint,olcDatabase={2}hdb,cn=config objectClass: olcConfig objectClass: olcOverlayConfig objectClass: olcRefintConfig objectClass: top olcOverlay: refint olcRefintAttribute: memberof member manager owner // 必要なパラメーターを変更します。次のコマンドを実行して、設定ファイルをロードします。
ldapadd -Q -Y EXTERNAL -H ldapi:/// -f memberof_config.ldif ldapmodify -Q -Y EXTERNAL -H ldapi:/// -f refint1.ldif ldapadd -Q -Y EXTERNAL -H ldapi:/// -f refint2.ldif
テストユーザーを作成します。
ldapadd コマンドを実行して、テストユーザーを作成します。ファイル名として kafka-users.ldif を指定し、次の情報をファイルに追加します。
dn: uid=kafka-user1,ou=people,o=emr cn: kafka-user1 sn: kafka-user1 uid: kafka-user1 objectClass: inetOrgPerson userPassword: kafka-secret dn: uid=kafka-user2,ou=people,o=emr cn: kafka-user2 sn: kafka-user2 uid: kafka-user2 objectClass: inetOrgPerson userPassword: kafka-secretテストユーザーグループを作成します。
ldapadd コマンドを実行して、テストユーザーグループを作成します。ファイル名として kafka-groups.ldif を指定し、次の情報をファイルに追加します。
dn: cn=kafka-group1,ou=groups,o=emr cn: kafka-group1 objectClass: groupOfNames member: uid=kafka-user1,ou=people,o=emr member: uid=kafka-user2,ou=people,o=emr dn: cn=kafka-group2,ou=groups,o=emr cn: kafka-group2 objectClass: groupOfNames member: uid=kafka-user1,ou=people,o=emrOpenLDAP サーバーで次のコマンドを実行して、テストユーザーの memberOf プロパティを表示します。
ldapsearch -Q -Y EXTERNAL -H ldapi:/// -b ou=people,o=emr memberOf返された情報では、kafka-user1 の memberOf プロパティには kafka-group1 と kafka-group2 が含まれ、kafka-user2 の memberOf プロパティには kafka-group2 が含まれます。
クライアント設定ファイルを作成します。
client.properties という名前の設定ファイルを作成し、次の情報をファイルに追加します。
security.protocol=SASL_PLAINTEXT sasl.mechanism=PLAIN // クライアントごとにユーザー名とパスワードを置き換えます。 sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="kafka" password="kafka-secret";kafka-user1.properties という名前の設定ファイルを作成し、次の情報をファイルに追加します。
security.protocol=SASL_PLAINTEXT sasl.mechanism=PLAIN // クライアントごとにユーザー名とパスワードを置き換えます。 sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="kafka-user1" password="kafka-secret";kafka-user2.properties という名前の設定ファイルを作成し、次の情報をファイルに追加します。
security.protocol=SASL_PLAINTEXT sasl.mechanism=PLAIN // クライアントごとにユーザー名とパスワードを置き換えます。 sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="kafka-user2" password="kafka-secret";
test という名前のテストトピックを作成します。
kafka-topics.sh --bootstrap-server core-1-1:9092 --command-config client.properties --create --topic test --replication-factor 3 --partitions 2必要な権限をテストユーザーとテストユーザーグループに付与します。
権限の付与に使用されるコマンドの詳細については、「認証と ACL」を参照してください。
// kafka-group2 ユーザーグループに権限を付与します。 kafka-acls.sh --authorizer-properties zookeeper.connect=$KAFKA_ZOOKEEPER --add --allow-principal User:kafka-group2 --allow-host "*" --operation All --cluster kafka-cluster kafka-acls.sh --authorizer-properties zookeeper.connect=$KAFKA_ZOOKEEPER --add --allow-principal User:kafka-group2 --allow-host "*" --operation All --topic test // kafka-group1 ユーザーグループに権限を付与します。 kafka-acls.sh --authorizer-properties zookeeper.connect=$KAFKA_ZOOKEEPER --add --deny-principal User:kafka-group1 --deny-host "*" --operation All --topic test // kafka-user1 ユーザーに権限を付与します。 kafka-acls.sh --authorizer-properties zookeeper.connect=$KAFKA_ZOOKEEPER --add --deny-principal User:kafka-user1 --deny-host "*" --operation Read --topic test // kafka-user2 ユーザーに権限を付与します。 kafka-acls.sh --authorizer-properties zookeeper.connect=$KAFKA_ZOOKEEPER --add --allow-principal User:kafka-user2 --allow-host "*" --operation Read --topic test次のコマンドを実行して、テストユーザーとテストユーザーグループの権限を表示できます。
kafka-acls.sh --authorizer-properties zookeeper.connect=$KAFKA_ZOOKEEPER --list --topic test次の情報が返されます。
Current ACLs for resource `Topic:LITERAL:test`: User:kafka-group2 has Allow permission for operations: All from hosts: * User:kafka-user2 has Allow permission for operations: Read from hosts: * User:kafka-user1 has Deny permission for operations: Read from hosts: * User:kafka-group1 has Deny permission for operations: All from hosts: *権限を確認します。
kafka-user1 ユーザーは、kafka-group2 ユーザーグループに属しているため、test トピックにデータを書き込むことができます。
kafka-console-producer.sh --broker-list core-1-1:9092 --producer.config ./kafka-user1.properties --topic testコードを書いて test トピックにアクセスします。
// コードを記述します。 >a >b >c >dkafka-user1 ユーザーは、test トピックからデータを読み取ることができません。
// コンシューマーグループに必要な権限を付与します。 kafka-acls.sh --authorizer-properties zookeeper.connect=$KAFKA_ZOOKEEPER --add --allow-principal User:kafka-user1 --allow-host "*" --operation All --group kka-user1-consumer kafka-console-consumer.sh --bootstrap-server core-1-1:9092 --consumer.config ./kafka-user1.properties --topic test --group kafka-user1-consumer結果は、認証が失敗し、ユーザーが test トピックにアクセスできないことを示しています。
kafka-user2 ユーザーは、test トピックにデータを書き込むことができません。
kafka-console-producer.sh --broker-list core-1-1:9092 --producer.config ./kafka-user2.properties --topic test // コードを記述します。 >a // 認証が失敗します。結果は、認証が失敗したことを示しています。
kafka-user2 ユーザーは、test トピックからデータを読み取ることができます。
// コンシューマーグループに必要な権限を付与します。 kafka-acls.sh --authorizer-properties zookeeper.connect=$KAFKA_ZOOKEEPER --add --allow-principal User:kafka-user2 --allow-host "*" --operation All --group kafka-user2-consumer kafka-console-consumer.sh --bootstrap-server core-1-1:9092 --consumer.config ./kafka-user2.properties --topic test --group kafka-user2-consumer --from-beginning // データは想定どおりに消費されます。