How to block creating/deleting kafka topic by unauthorized users?

Question

I have setup Kafka and zookeeper authentication with SASL+ACL and Kafka to producer and consumer by SSL two way authentication including encryption.

By enabling SASL and ACL between Kafka and zookeeper it doesn't allow to login unauthorized Kafka broker to the zookeeper cluster. But still, topic creation and deletion can be done without any restrictions.

zookeeper.properties

dataDir=/x02/lsesv2-s/data/Zookeeper

clientPort=15300

tickTime=2000

initLimit=10

syncLimit=5


authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider

requireClientAuthScheme=sasl

jaasLoginRenew=3600000

quorum.auth.enableSasl=true

quorum.auth.learnerRequireSasl=true

quorum.auth.serverRequireSasl=true

quorum.auth.learner.loginContext=QuorumLearner

quorum.auth.server.loginContext=QuorumServer

server.1=172.25.33.12:15302:15301
server.2=172.25.33.13:15302:15301
server.3=172.25.33.11:15302:15301

zookeeper_jaas.conf

Server {
        org.apache.zookeeper.server.auth.DigestLoginModule required
        username="admin"
        password="abc123"
        user_admin="abc123";
};

QuorumServer {
        org.apache.zookeeper.server.auth.DigestLoginModule required
        user_admin="abc123";
};

QuorumLearner {
        org.apache.zookeeper.server.auth.DigestLoginModule required
        username="admin"
        password="abc123";
};

Set ACL by below code

 final CountDownLatch connectedSignal = new CountDownLatch(1);
        String connect = "localhost:15300";
        ZooKeeper zooKeeper = null;

        try
        {
            String userName = "admin";
            String password = "mit123";

            zooKeeper = new ZooKeeper(connect, 5000, we ->
            {
                if (we.getState() == Watcher.Event.KeeperState.SyncConnected)
                {
                    connectedSignal.countDown();
                }
            });

            connectedSignal.await();

            zooKeeper.addAuthInfo("digest", (userName + ":" + password).getBytes());

            final String aclString = "auth:" + userName + ":" + password + ":" + "cdrwa" + 
            ",sasl:" + userName + ":" + "cdrwa";

            zooKeeper.setACL("/", parseACLs(aclString), -1);

        } finally
        {
            if (zooKeeper != null)
            {
                zooKeeper.close();
            }
        }

Above code is working and below is the result after executing the code.

Welcome to ZooKeeper!
JLine support is disabled

WATCHER::

WatchedEvent state:SyncConnected type:None path:null
getAcl /
'sasl,'admin
: cdrwa
'digest,'admin:oiasY+rmnmmK9mec8kpnvv281HE=
: cdrwa

Instead of server.properties file I have overridden Kafka properties when it is started. *

Kafka properties

kafka/bin/kafka-server-start.sh /x02/lsesv2-s/current/kafka/config/server.properties 
--override broker.id=1 
--override zookeeper.connect=10g-flton-onl01:15300,10g-flton-onl02:15300,10g-flton-nor02:15300 
--override num.network.threads=16 
--override num.io.threads=16 
--override socket.send.buffer.bytes=10240000
--override socket.receive.buffer.bytes=10240000 
--override log.dirs=/x02/lsesv2-s/data/Kafka 
--override offsets.topic.replication.factor=1 
--override min.insync.replicas=1 
--override inter.broker.listener.name=INTERNAL 
--override listeners=INTERNAL://10g-flton-onl01:15307 
--override advertised.listeners=INTERNAL://10g-flton-onl01:15307 
--override listener.security.protocol.map=INTERNAL:SSL 
--override security.protocol=SSL 
--override ssl.client.auth=required 
--override ssl.key.password=abc123 
--override ssl.keystore.location=configs/MHV/kafka.server.keystore.jks 
--override ssl.keystore.password=abc123 
--override ssl.truststore.location=configs/MHV/kafka.server.truststore.jks 
--override ssl.truststore.password=abc123 
--override ssl.endpoint.identification.algorithm=

Kafka to producer/consumer authentication works fine and zookeeper to kafka authentication is also working fine. But still, topic creation and deletion can be done by unauthorized users too.

Topic creation

kafka/bin/kafka-topics.sh --create --zookeeper localhost:15300 --replication-factor 3 --partitions 8 --topic test

Topic deletion

kafka/bin/kafka-topics.sh --zookeeper localhost:15300 --delete --topic test

Note: I didn't set -Djava.security.auth.login.config=kafka_server_jaas.conf when creating or deleting topics. So this operation should be restricted. But actually, it doesn't.

Help me with topic creation and deletion for only authorized users.

perkss perkss · Accepted Answer · 2019-09-28T10:27:25

It seems this is the required property from testing locally.

 KAFKA_ZOOKEEPER_SET_ACL: "true"

For the Confluent images or maps directly too.

zookeeper.set.acl

Reference

Also as stated at Kafka 101 Confluent

the metadata stored in ZooKeeper is such that only brokers will be able to modify the corresponding znodes, but znodes are world readable.

Because we configured ZooKeeper to require SASL authentication, we need to set the java.security.auth.login.config system property while starting the kafka-topics tool:

A code example and docker-compose file is shown here

How to block creating/deleting kafka topic by unauthorized users?

1 Answers