使用LDAP用戶鑒權(quán)
本文為您介紹EMR Kafka如何啟用LDAP用戶進(jìn)行鑒權(quán),以及詳細(xì)使用的示例。
前提條件
已創(chuàng)建DataFlow集群,且選擇了Kafka和OpenLDAP服務(wù)。創(chuàng)建集群詳情,請參見創(chuàng)建DataFlow Kafka集群。
使用限制
適用于EMR-3.44.0之后版本、EMR-5.10.0之后版本的EMR Kafka集群。
集群中已有EMR OpenLDAP服務(wù)或者外部LDAP服務(wù)。
本文以EMR OpenLDAP服務(wù)為例。
注意事項
如果需要使用用戶組鑒權(quán),則需要確保LDAP服務(wù)啟用memberOf overlay特性,或者LDAP用戶支持memberOf屬性設(shè)置。
啟用LDAP用戶鑒權(quán)
創(chuàng)建super users。
說明如果您已經(jīng)創(chuàng)建好super user,則可以跳過此步驟。
Kafka super user可以訪問Kafka的所有資源,本文super user用戶用于Broker節(jié)點以及組件之間的相互訪問。本文以EMR OpenLDAP服務(wù)為例,添加Kafka super user用戶。
通過SSH方式連接集群master-1-1節(jié)點,詳情請參見登錄集群。
創(chuàng)建kafka.ldif文件,文件內(nèi)容如下。
本文是通過
ldap
命令,添加uid為kafka,密碼為kafka-secret的LDAP用戶。dn: uid=kafka,ou=people,o=emr cn: kafka sn: kafka objectClass: inetOrgPerson userPassword: kafka-secret uid: kafka
執(zhí)行以下命令,添加LDAP用戶。
ldapadd -H ldap://master-1-1:10389 -f kafka.ldif -D ${uid} -w ${rootDnPW}
說明${uid}
:為EMR控制臺OpenLDAP服務(wù)配置頁面admin_dn的參數(shù)值。${rootDnPW}
:為EMR控制臺OpenLDAP服務(wù)配置頁面admin_pwd的參數(shù)值。10389
:為OpenLDAP服務(wù)的監(jiān)聽端口。
添加成功后,您可以執(zhí)行以下命令,查看該LDAP用戶信息。
ldapsearch -w ${rootDnPW} -D ${uid} -H ldap://master-1-1:10389 -b uid=kafka,ou=people,o=emr
進(jìn)入Kafka服務(wù)的配置頁面。
在頂部菜單欄處,根據(jù)實際情況選擇地域和資源組。
單擊目標(biāo)集群操作列的集群服務(wù)。
在集群服務(wù)頁面,單擊Kafka服務(wù)區(qū)域的配置。
修改配置項并保存。
在配置頁面,單擊server.properties頁簽。
修改配置項kafka.ssl.config.type的值為CUSTOM。
修改配置項authorizer.class.name的值為kafka.security.ldap.authorizer.SimpleLdapAuthorizer。
保存配置項。
單擊保存。
在彈出的對話框中,輸入執(zhí)行原因。
新增自定義配置項并保存。
在server.properties頁簽,單擊新增配置項。
在新增配置項對話框中,添加以下配置。
參數(shù)
參數(shù)值
說明
super.users
User:kafka
kafka
為已創(chuàng)建好的用戶名,請根據(jù)實際情況替換。創(chuàng)建用戶詳情,請參見步驟1。
listener.name.${listener}.sasl.enabled.mechanisms
PLAIN
${listener}需要替換成具體的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}需要替換成具體的listener的名稱,例如sasl_ssl。
參數(shù)值也應(yīng)根據(jù)實際情況替換LDAP的相關(guān)配置:
username
:需要使用super user,Kafka服務(wù)使用該用戶進(jìn)行節(jié)點間訪問。mr.kafka.security.ldap.user.name.attribute
:LDAP用戶名稱屬性,用于獲取用戶名。emr.kafka.security.ldap.user.base.dn
:LDAP用戶base dn。emr.kafka.security.ldap.group.name.attribute:LDAP用戶組名稱屬性,用于獲取用戶組名稱。
emr.kafka.security.ldap.group.base.dn
:LDAP用戶組base dn。emr.kafka.security.ldap.admin.name.attribute
:LDAP admin用戶屬性。emr.kafka.security.ldap.admin.base.dn
:LDAP admin用戶base dn。emr.kafka.security.ldap.url
:LDAP URL。emr.kafka.security.ldap.bind.use
:LDAP管理員用戶名稱,用于用戶組鑒權(quán)使用。emr.kafka.security.ldap.bind.user.password
:LDAP管理員用戶密碼。emr.kafka.security.ldap.user.member.of.attribute
:用戶所屬組的屬性,用于獲取用戶所屬組。emr.kafka.security.ldap.group.authorization.support
:是否支持組鑒權(quán)。如果支持,則當(dāng)用戶所屬組有權(quán)限時,該用戶繼承組的權(quán)限。
listener.name.${listener}.plain.sasl.server.callback.handler.class
kafka.security.ldap.authenticator.LdapAuthenticateCallbackHandler
${listener}需要替換成具體的listener的名稱,例如sasl_ssl。
sasl.mechanism.inter.broker.protocol
PLAIN
無
sasl.enabled.mechanisms
PLAIN
無
修改其他配置項。
您需要根據(jù)業(yè)務(wù)需求,修改以下配置。
說明需要根據(jù)實際情況替換kafka.client.jaas.content參數(shù)值中的用戶名和密碼,建議使用super user用戶。
配置文件
參數(shù)
參數(shù)值
說明
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
如果該配置已經(jīng)有參數(shù)值,則將新增的內(nèi)容追加到已有的參數(shù)值之后。
kafka-rest.properties
kafkarest_opts
-Djava.security.auth.login.config=/etc/taihao-apps/kafka-conf/kafka-conf/kafka_client_jaas.conf
如果該配置已經(jīng)有參數(shù)值,則將新增的內(nèi)容追加到已有的參數(shù)值之后。
重啟Kafka服務(wù)。
在Kafka服務(wù)的配置頁面,選擇 。
在彈出的對話框中,輸入執(zhí)行原因,單擊確定。
在確認(rèn)對話框中,單擊確定。
使用示例
本示例操作適用于開源Kafka 2.4.1,其它版本的Kafka操作請參見Apache Kafka。
創(chuàng)建測試用戶以及用戶組。
如果OpenLdap沒有開啟memberOf overlay特性,則需要開啟此功能。
說明以下配置方式只作為參考,不同的LDAP服務(wù),開啟memberOf overlay的方式不同。
需要在所有OpenLDAP節(jié)點下配置生效。
以下所有配置文件的參數(shù)請根據(jù)實際情況修改,例如:
olcModulepath
:32位OS系統(tǒng)為/usr/lib/openldap。dn: cn=module{0},cn=config
:如果/etc/openldap/slapd.d/cn=config目錄下已經(jīng)存在cn=module{0}.ldif文件,則需要修改module后面的數(shù)字為1;如果存在cn=module{1}.ldif,則需要修改module后面的數(shù)字為2,以此類推。dn: olcOverlay={0}memberof,olcDatabase={2}hdb,cn=config
:根據(jù)/etc/openldap/slapd.d/cn=config目錄olcDatabase的值來配置相應(yīng)的值。例如,如果存在olcDatabase={2}hdb.ldif
,則此處的值應(yīng)為hdb。
執(zhí)行命令
vim memberof_config.ldif
,新增配置文件memberof_config.ldif,文件內(nèi)容如下。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
執(zhí)行命令
vim refint1.ldif
,新增配置文件refint1.ldif,文件內(nèi)容如下。dn: cn=module{0},cn=config add: olcmoduleload olcmoduleload: refint
執(zhí)行命令
vim refint2.ldif
,新增配置文件refint2.ldif,文件內(nèi)容如下。dn: olcOverlay=refint,olcDatabase={2}hdb,cn=config objectClass: olcConfig objectClass: olcOverlayConfig objectClass: olcRefintConfig objectClass: top olcOverlay: refint olcRefintAttribute: memberof member manager owner
執(zhí)行以下命令,加載配置文件。
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
創(chuàng)建測試用戶。
通過ldapadd命令的方式添加測試用戶。ldif文件命名為kafka-users.ldif,文件內(nèi)容如下。
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
創(chuàng)建測試用戶組。
通過ldapadd命令的方式添加測試用戶組。ldif文件命名為kafka-groups.ldif,文件內(nèi)容如下。
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=emr
在LDAP所在服務(wù)器執(zhí)行以下命令,查看測試用戶memberOf屬性。
ldapsearch -Q -Y EXTERNAL -H ldapi:/// -b ou=people,o=emr memberOf
根據(jù)返回信息,可以看見kafka-user1的memberOf屬性包含kafka-group1和kafka-group2,kafka-user2的memberOf屬性包含kafka-group2。
創(chuàng)建客戶端配置文件。
創(chuàng)建client.properties客戶端配置文件,內(nèi)容如下所示。
security.protocol=SASL_PLAINTEXT sasl.mechanism=PLAIN #根據(jù)不同的客戶端身份,替換username以及password。 sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="kafka" password="kafka-secret";
創(chuàng)建kafka-user1.properties客戶端配置文件,內(nèi)容如下所示。
security.protocol=SASL_PLAINTEXT sasl.mechanism=PLAIN #根據(jù)不同的客戶端身份,替換username以及password sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="kafka-user1" password="kafka-secret";
創(chuàng)建kafka-user2.properties客戶端配置文件,內(nèi)容如下所示。
security.protocol=SASL_PLAINTEXT sasl.mechanism=PLAIN #根據(jù)不同的客戶端身份,替換username以及password sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="kafka-user2" password="kafka-secret";
創(chuàng)建名稱為test的測試Topic。
kafka-topics.sh --bootstrap-server core-1-1:9092 --command-config client.properties --create --topic test --replication-factor 3 --partitions 2
授權(quán)操作。
詳細(xì)的授權(quán)命令操作,請參見Authorization and ACLs。
#kafka-group2權(quán)限 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權(quán)限 kafka-acls.sh --authorizer-properties zookeeper.connect=$KAFKA_ZOOKEEPER --add --deny-principal User:kafka-group1 --deny-host "*" --operation All --topic test #kafka-user1權(quán)限 kafka-acls.sh --authorizer-properties zookeeper.connect=$KAFKA_ZOOKEEPER --add --deny-principal User:kafka-user1 --deny-host "*" --operation Read --topic test #kafka-user2權(quán)限 kafka-acls.sh --authorizer-properties zookeeper.connect=$KAFKA_ZOOKEEPER --add --allow-principal User:kafka-user2 --allow-host "*" --operation Read --topic test
您可以執(zhí)行以下命令,查看授權(quán)結(jié)果。
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: *
權(quán)限驗證。
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 >d
kafka-user1不能讀test。
#授權(quán)consumer group 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
輸出結(jié)果提示鑒權(quán)不通過無法訪問test。
kafka-user2不能寫test。
kafka-console-producer.sh --broker-list core-1-1:9092 --producer.config ./kafka-user2.properties --topic test #提示符后輸入測試消息 >a #提示鑒權(quán)不通過
輸出結(jié)果提示鑒權(quán)不通過。
kafka-user2可以讀test。
#授權(quán)consumer group 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 #正常消費數(shù)據(jù)