E-MapReduce:Use LDAP user authentication

Step 1: Create a superuser

Skip this step if you already have a Kafka superuser.

A Kafka superuser has access to all Kafka resources. In this setup, the superuser is used to access broker nodes and Kafka components. This step adds a Kafka superuser to the EMR OpenLDAP service.

1. Connect to the master-1-1 node of the cluster over SSH. See Log on to a cluster.

2. Create a file named kafka.ldif with the following content. This adds an LDAP user with UID kafka and password kafka-secret.

   dn: uid=kafka,ou=people,o=emr
   cn: kafka
   sn: kafka
   objectClass: inetOrgPerson
   userPassword: kafka-secret
   uid: kafka

3. Run the following command to add the LDAP user:

   ldapadd -H ldap://master-1-1:10389 -f kafka.ldif -D ${uid} -w ${rootDnPW}

   Replace the placeholders as follows:

   Placeholder	Description	Where to find it
   ${uid}	Value of the admin_dn parameter	Configure tab of the OpenLDAP service page in the EMR console
   ${rootDnPW}	Value of the admin_pwd parameter	Configure tab of the OpenLDAP service page in the EMR console

   10389 is the default listening port of the OpenLDAP service.

   To verify that the user was added, run:

   ldapsearch -w ${rootDnPW} -D ${uid} -H ldap://master-1-1:10389 -b uid=kafka,ou=people,o=emr

Step 2: Go to the Kafka service configuration page

1. Log on to the EMR on ECS console.

2. In the top navigation bar, select the region where your cluster resides and select a resource group.

3. On the EMR on ECS page, find your cluster and click Services in the Actions column.

4. On the Services tab, find Kafka and click Configure.

Step 3: Update existing configuration items

1. On the Configure tab, click the server.properties tab.

2. Set kafka.ssl.config.type to CUSTOM.

3. Set authorizer.class.name to kafka.security.ldap.authorizer.SimpleLdapAuthorizer.

4. Click Save. In the dialog box, set the Execution Reason and turn on Automatically Update Configurations.

Step 4: Add new configuration items

1. On the server.properties tab, click Add Configuration Item.

2. In the Add Configuration Item dialog box, add the following items:

   Configuration item	Value	Description
   super.users	User:kafka	The name of the superuser you created in Step 1. Replace kafka with your superuser name.
   listener.name.${listener}.sasl.enabled.mechanisms	PLAIN	Replace ${listener} with your listener name, for example, sasl_ssl.
   listener.name.${listener}.plain.sasl.jaas.config	See below	Replace ${listener} with your listener name. Update the LDAP-related values based on your configuration.
   listener.name.${listener}.plain.sasl.server.callback.handler.class	kafka.security.ldap.authenticator.LdapAuthenticateCallbackHandler	Replace ${listener} with your listener name.
   sasl.mechanism.inter.broker.protocol	PLAIN
   sasl.enabled.mechanisms	PLAIN

   For the listener.name.${listener}.plain.sasl.jaas.config item, use the following value and replace each option based on your LDAP configuration:

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

   The following table describes the LDAP-related options:

   Option	Description
   username	Name of the superuser. Kafka uses this account to access broker nodes.
   emr.kafka.security.ldap.user.name.attribute	LDAP attribute used to identify a user's name (used to look up the username).
   emr.kafka.security.ldap.user.base.dn	Base distinguished name (DN) for LDAP users.
   emr.kafka.security.ldap.group.name.attribute	LDAP attribute used to identify a group's name.
   emr.kafka.security.ldap.group.base.dn	Base DN for LDAP user groups.
   emr.kafka.security.ldap.admin.name.attribute	LDAP attribute used to identify the admin user's name.
   emr.kafka.security.ldap.admin.base.dn	Base DN for the LDAP admin user.
   emr.kafka.security.ldap.url	URL of the OpenLDAP service.
   emr.kafka.security.ldap.bind.user	LDAP administrator username, used for user group authentication.
   emr.kafka.security.ldap.bind.user.password	Password for the LDAP administrator.
   emr.kafka.security.ldap.user.member.of.attribute	LDAP attribute that records which groups a user belongs to.
   emr.kafka.security.ldap.group.authorization.support	Set to true to enable group-based authorization. When enabled, users inherit permissions from the groups they belong to.

3. Update the following configuration files based on your requirements:

   Replace the username and password in kafka.client.jaas.content with your actual credentials. Use the superuser credentials for broker-level access.

   Configuration file	Parameter	Value	Note
   kafka_client_jaas.conf	kafka.client.jaas.content	KafkaClient { org.apache.kafka.common.security.plain.PlainLoginModule required username="kafka" password="kafka-secret"; };	The value must end with a semicolon (;).
   schema-registry.properties	schema_registry_opts	-Djava.security.auth.login.config=/etc/taihao-apps/kafka-conf/kafka-conf/kafka_client_jaas.conf	If a value already exists, append this to the end.
   kafka-rest.properties	kafkarest_opts	-Djava.security.auth.login.config=/etc/taihao-apps/kafka-conf/kafka-conf/kafka_client_jaas.conf	If a value already exists, append this to the end.

Step 5: Restart the Kafka service

1. On the Configure tab, choose More > Restart.

2. In the dialog box, set the Execution Reason and click OK.

3. In the Confirm dialog, click OK.

Step 1: Create test users and user groups

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

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

ldapsearch -Q -Y EXTERNAL -H ldapi:/// -b ou=people,o=emr memberOf

Step 2: Create client configuration files

Each client configuration file specifies the credentials for a particular user. All examples use SASL_PLAINTEXT as the security protocol.

In a production environment, use SASL_SSL instead of SASL_PLAINTEXT to prevent credentials from being transmitted in plaintext. See the Apache Kafka security documentation for details.

Create the following three configuration files. Replace the username and password in each file based on the client.

client.properties (superuser):

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

Step 3: Create a test topic

kafka-topics.sh --bootstrap-server core-1-1:9092 --command-config client.properties --create --topic test --replication-factor 3 --partitions 2

Step 4: Grant permissions

Kafka access control lists (ACLs) follow the format: "Principal P is [Allowed|Denied] Operation O from Host H on Resource R." For full ACL documentation, see Apache Kafka authorization and ACLs.

If a resource has no ACLs, Kafka restricts access to that resource by default. Only superusers can access it.

Run the following commands to set up permissions:

# Allow kafka-group2 to perform all operations on the cluster and topic test.
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
# Deny kafka-group1 all operations on topic test.
kafka-acls.sh --authorizer-properties zookeeper.connect=$KAFKA_ZOOKEEPER --add --deny-principal User:kafka-group1 --deny-host "*" --operation All --topic test
# Deny kafka-user1 read access on topic test.
kafka-acls.sh --authorizer-properties zookeeper.connect=$KAFKA_ZOOKEEPER --add --deny-principal User:kafka-user1 --deny-host "*" --operation Read --topic test
# Allow kafka-user2 read access on topic test.
kafka-acls.sh --authorizer-properties zookeeper.connect=$KAFKA_ZOOKEEPER --add --allow-principal User:kafka-user2 --allow-host "*" --operation Read --topic test

To view the current ACLs for the test topic:

kafka-acls.sh --authorizer-properties zookeeper.connect=$KAFKA_ZOOKEEPER --list --topic test

Expected output:

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: *

Step 5: Verify permissions

The following tests confirm that group-based permissions are applied correctly. A user's effective permissions are the combination of their individual ACLs and the ACLs of all groups they belong to.

kafka-user1 can write to the test topic (inherits Allow All from kafka-group2):

kafka-console-producer.sh --broker-list core-1-1:9092 --producer.config ./kafka-user1.properties --topic test

Write a few messages to verify:

>a
>b
>c
>d

kafka-user1 cannot read from the test topic (has an explicit Deny Read):

# Grant the consumer group permission first.
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

The read operation fails, confirming that the explicit Deny Read takes effect even though kafka-user1 belongs to kafka-group2.

kafka-user2 cannot write to the test topic (no write permission granted):

kafka-console-producer.sh --broker-list core-1-1:9092 --producer.config ./kafka-user2.properties --topic test

The write operation fails.

kafka-user2 can read from the test topic (has Allow Read):

# Grant the consumer group permission first.
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

Data is consumed successfully.