Kafka Security with Mutual TLS and ACLs: A KRaft Lab, the Principal Trap, and the Commands That Still Exist

Revision note (2026-09-15). The earlier version of this article managed ACLs with kafka-acls.sh --authorizer-properties zookeeper.connect=..., used --broker-list with the console producer, and told readers to verify handshakes with journalctl | grep SSLHandshake. None of that works on a Kafka the reader can download today: ZooKeeper mode was removed in 4.0 and the option is gone, --broker-list is gone, and the broker log does not record successful handshakes. It also stated that Kafka requires .jks keystores (JKS, PKCS12 and PEM are all supported) and wrote ACLs for User:producer1 without the mapping rule that turns a certificate CN into that name, so the ACLs would never have matched. This version is rebuilt around a lab you can run.

What the lab is

One broker from the apache/kafka:4.3.1 image, KRaft combined mode, every listener on TLS, including the controller listener, so the broker authenticates to itself with its own certificate and no anonymous path exists. Client authentication is required (ssl.client.auth=required), the KRaft StandardAuthorizer is on, nothing is allowed without an ACL, and two identities are super users: broker and a lab admin. Clients are Java (kafka-clients 4.3.1) with certificates for producer1, consumer1, an intruder who has a valid certificate but no ACLs, and a rogue certificate signed by a CA the broker does not trust.

Tested versions: apache/kafka:4.3.1 (broker and CLI tools, OpenJDK 21 inside the image), kafka-clients 4.3.1 on Amazon Corretto 17.0.14, OpenSSL 3.6.3, Docker 27.4.0, Gradle 8.8. Lab: examples/kafka-security in the repository behind this site.

Certificates

A script generates everything with OpenSSL. The parts that matter:

# CA
openssl req -new -x509 -newkey rsa:2048 -nodes -sha256 -days 30 
  -keyout ca-key.pem -out ca-cert.pem -subj "/CN=dev-drunk-lab-ca"

# Broker certificate with a SubjectAltName. Java clients verify the hostname
# against the SAN (ssl.endpoint.identification.algorithm=https is the default).
cat > broker-ext.cnf <<CNF
subjectAltName=DNS:localhost,DNS:dd-kafka-a-security,IP:127.0.0.1
extendedKeyUsage=serverAuth,clientAuth
CNF
openssl req -new -newkey rsa:2048 -nodes -sha256 -keyout broker-key.pem -out broker.csr -subj "/CN=broker"
openssl x509 -req -in broker.csr -CA ca-cert.pem -CAkey ca-key.pem -CAcreateserial 
  -days 30 -sha256 -extfile broker-ext.cnf -out broker-cert.pem

# One client certificate per identity; the CN is what becomes the principal
openssl req -new -newkey rsa:2048 -nodes -sha256 -keyout producer1-key.pem -out producer1.csr -subj "/CN=producer1"
openssl x509 -req -in producer1.csr -CA ca-cert.pem -CAkey ca-key.pem -CAcreateserial -days 30 -sha256 -out producer1-cert.pem

# PKCS12 keystore; Kafka reads it directly, no JKS conversion
openssl pkcs12 -export -in producer1-cert.pem -inkey producer1-key.pem -certfile ca-cert.pem 
  -name producer1 -passout pass:labpass -out producer1.p12

# Truststore with only the CA
keytool -importcert -noprompt -alias lab-ca -file ca-cert.pem 
  -keystore truststore.p12 -storetype PKCS12 -storepass labpass

ssl.keystore.type accepts JKS, PKCS12 and PEM (Kafka configuration reference), and since Kafka 2.7 the key, chain and CA can be passed inline as PEM strings with no store file at all. The lab's broker and clients run on PKCS12; one test runs a client on inline PEM to prove the point:

props.put("security.protocol", "SSL");
props.put("ssl.keystore.type", "PEM");
props.put("ssl.keystore.key", Files.readString(Path.of("producer1-key.pem")));
props.put("ssl.keystore.certificate.chain", Files.readString(Path.of("producer1-cert.pem")));
props.put("ssl.truststore.type", "PEM");
props.put("ssl.truststore.certificates", Files.readString(Path.of("ca-cert.pem")));

Broker configuration

process.roles=broker,controller
node.id=1
controller.quorum.voters=1@localhost:19093
controller.listener.names=CONTROLLER
listeners=SSL://0.0.0.0:9093,CONTROLLER://0.0.0.0:19093
advertised.listeners=SSL://localhost:9093
listener.security.protocol.map=SSL:SSL,CONTROLLER:SSL
inter.broker.listener.name=SSL

ssl.keystore.type=PKCS12
ssl.keystore.location=/etc/kafka/secrets/broker.p12
ssl.keystore.password=labpass
ssl.key.password=labpass
ssl.truststore.type=PKCS12
ssl.truststore.location=/etc/kafka/secrets/truststore.p12
ssl.truststore.password=labpass
ssl.client.auth=required

# Without this the principal is the whole DN, "User:CN=producer1"
ssl.principal.mapping.rules=RULE:^CN=([^,]+).*$/$1/,DEFAULT

authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer
allow.everyone.if.no.acl.found=false
super.users=User:broker;User:admin

The file is mounted at /mnt/shared/config in the container, the secrets at /etc/kafka/secrets, and the broker starts in about two seconds. The earlier article's listeners=SSL://0.0.0.0:9093 with security.inter.broker.protocol=SSL and nothing else is a ZooKeeper-era fragment; a KRaft node needs process.roles, a controller listener and a quorum, and the authorizer class for KRaft is org.apache.kafka.metadata.authorizer.StandardAuthorizer.

The principal trap

With TLS authentication, the principal Kafka derives from a certificate is, by default, the full X.500 distinguished name. A certificate with subject CN=producer1 gives the principal User:CN=producer1, not User:producer1. An ACL written for User:producer1 never matches it.

The lab shows this from the broker's own log. principal-check.sh starts a second broker from the same configuration minus the mapping rule. That broker never finishes starting, because its own identity is now User:CN=broker, which is not the super user User:broker, and the controller rejects the broker's registration:

$ docker logs dd-kafka-a-security-nomap | grep -o "principal=User:[^,]*" | sort | uniq -c
  13 principal=User:CN=broker
$ docker logs dd-kafka-a-security-nomap | grep -o "ClusterAuthorizationException.*permission" | head -1
ClusterAuthorizationException: Request Request(... listenerName=ListenerName(CONTROLLER), securityProtocol=SSL ...) needs CLUSTER_ACTION permission

Either write ACLs and super.users with the full DN, or set ssl.principal.mapping.rules as above so the CN becomes the name. The lab does the latter, and every ACL below uses the short name.

ACLs with the CLI that exists

All ACL management goes through a broker (--bootstrap-server) or a controller (--bootstrap-controller); there is no ZooKeeper path. The admin identity is a super user, which is how the first ACL can be created at all:

K=/opt/kafka/bin
ADMIN="--bootstrap-server localhost:9093 --command-config /etc/kafka/secrets/admin.properties"

$K/kafka-topics.sh $ADMIN --create --topic my-secure-topic --partitions 1 --replication-factor 1
$K/kafka-acls.sh $ADMIN --add --allow-principal User:producer1 --operation Write --topic my-secure-topic
$K/kafka-acls.sh $ADMIN --add --allow-principal User:consumer1 --operation Read --topic my-secure-topic
$K/kafka-acls.sh $ADMIN --add --allow-principal User:consumer1 --operation Read --group my-consumer-group
$K/kafka-acls.sh $ADMIN --list

Recorded output of the last command:

Current ACLs for resource `ResourcePattern(resourceType=TOPIC, name=my-secure-topic, patternType=LITERAL)`:
	(principal=User:consumer1, host=*, operation=READ, permissionType=ALLOW)
	(principal=User:producer1, host=*, operation=WRITE, permissionType=ALLOW)

Current ACLs for resource `ResourcePattern(resourceType=GROUP, name=my-consumer-group, patternType=LITERAL)`:
	(principal=User:consumer1, host=*, operation=READ, permissionType=ALLOW)

admin.properties is an ordinary client configuration: security.protocol=SSL plus the keystore and truststore entries.

What the old commands print today

The lab replays the earlier article's commands verbatim against the 4.3.1 tools:

$ kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:producer1 --operation Write --topic my-secure-topic
authorizer-properties is not a recognized option

$ kafka-acls.sh --list
One of --bootstrap-server or --bootstrap-controller must be specified

$ kafka-console-producer.sh --broker-list localhost:9093 --topic my-secure-topic --producer.config ...
broker-list is not a recognized option

$ kafka-topics.sh --zookeeper localhost:2181 --list
zookeeper is not a recognized option

$ ls /opt/kafka/bin | grep -c zookeeper
0

The working console commands are kafka-console-producer.sh --bootstrap-server localhost:9093 --topic my-secure-topic --producer.config producer1.properties and kafka-console-consumer.sh --bootstrap-server ... --consumer.config consumer1.properties. Both tools print a warning that --producer.config / --consumer.config are deprecated in favor of --command-config.

Eight client scenarios

The Java tests connect to the live listener. Recorded results, one line per test:

[lab] connected as admin over mutual TLS, cluster NNVNhkThQOa3QWv2nD3f6g
[lab] producer1 wrote my-secure-topic-0@2
[lab] consumer1 read hello-2716026620246458
[lab] producer1 via PEM wrote offset 3
[lab] intruder produce -> TopicAuthorizationException: Not authorized to access topics: [my-secure-topic]
[lab] intruder listTopics -> []
[lab] producer1 listTopics -> [my-secure-topic] (Write implies Describe)
[lab] consumer1 in an unlisted group -> GroupAuthorizationException: Not authorized to access group: some-other-group
[lab] rogue CA -> SslAuthenticationException: Failed to process post-handshake messages
    Caused by: javax.net.ssl.SSLHandshakeException: Received fatal alert: certificate_required
[lab] no client cert -> SslAuthenticationException: Failed to process post-handshake messages
    Caused by: javax.net.ssl.SSLHandshakeException: Received fatal alert: certificate_required
[lab] PLAINTEXT on 9093 -> TimeoutException: Timed out waiting for a node assignment. Call: listNodes

What each line means for troubleshooting:

A valid certificate is not permission. intruder passes TLS and is stopped by the authorizer, with TopicAuthorizationException on produce and an empty listTopics() (topics you cannot Describe are hidden from metadata). The broker records the decision in kafka-authorizer.log: Principal = User:intruder is Denied operation = DESCRIBE ... on resource = Topic:LITERAL:my-secure-topic ... based on rule DefaultDeny.
Reading needs two ACLs. consumer1 has Read on the topic and on my-consumer-group; in any other group poll() throws GroupAuthorizationException.
A forged CN buys nothing. The rogue keystore says CN=producer1 but is signed by a CA the broker does not trust. The failure is the same certificate_required alert as sending no certificate at all: the broker announces the acceptable CA names during the handshake, the Java client finds no matching certificate and sends none, and ssl.client.auth=required ends the connection.
A plaintext client on a TLS port gets no protocol error, only silence and then a timeout. If your client "hangs" against a secured cluster, check security.protocol before anything else.

Verifying the handshake

The broker log is the wrong place to look. After many successful handshakes:

$ docker logs dd-kafka-a-security 2>&1 | grep -c -i SSLHandshake
0

Use openssl s_client from the outside; it shows the certificate chain the broker presents, the CA names it will accept from clients, and the negotiated protocol:

$ openssl s_client -connect localhost:9093 -CAfile ca-cert.pem -cert producer1-cert.pem -key producer1-key.pem </dev/null
subject=CN=broker
issuer=CN=dev-drunk-lab-ca
Acceptable client certificate CA names
CN=dev-drunk-lab-ca
Protocol: TLSv1.3
Verify return code: 0 (ok)

Failures show up on the client side as SslAuthenticationException with the alert name in the cause, as in the test output above, and denied operations show up in the broker's authorizer log.

What this does not cover

SASL (SCRAM-SHA-512, OAUTHBEARER, Kerberos). Mutual TLS is one way to authenticate; if you need identities that are not certificates, SASL over TLS is the other, and it was not tested here.
Certificate rotation and the dynamic reload of ssl.keystore.location through kafka-configs.sh.
Hostname verification failures. The broker certificate carries a correct SAN; a wrong SAN was not exercised.
Multi-broker clusters, non-Java clients, and the CPU cost of TLS, which was not measured.

Reproduce it

cd examples/kafka-security
./gen-certs.sh
./start-broker.sh          # apache/kafka:4.3.1 on localhost:9093
./acls.sh
gradle test --no-daemon    # 8 tests
./article-commands.sh      # the old commands and their errors
./principal-check.sh       # the principal without the mapping rule
./stop-broker.sh