Configure mTLS RBAC in Confluent Platform

Starting with Confluent Platform 7.8, you can use mutual TLS (mTLS) authentication with role-based access control (RBAC) authorization. The sections below cover configurations for use of mTLS with RBAC.

Note

Using Confluent for Kubernetes (CFK) or Ansible Playbooks for Confluent Platform for the deployment and management of mTLS RBAC is highly recommended.

Note the following:

• A trust store and keystore need to be configured.
• Metadata Service (MDS) must be configured. Depending on whether there is an identity provider for fetching users and client identity, a user store needs to be configured at the Metadata Service (MDS) service.
• For every server, a mandatory or optional certificate, token-based authentication, and authorization need to be configured.
• For Confluent REST Proxy principal propagation, secure impersonation with protected users and impersonated users need to be configured.
• When using a multi-node setup for a Confluent Platform service, request forwarding from follower to leader using mTLS needs to be configured. This involves providing protected users and impersonated users for request forwarding with secure impersonation.
• Principal mapping rules need to be configured for extracting the principal to use for RBAC. For details, see Use Principal Mapping in Confluent Platform.

Configure RBAC with mTLS identity

For Confluent Platform services, the following configurations are required for RBAC with mTLS identity across the Confluent Platform cluster.

Configure Confluent Server brokers

Kafka clients can use mTLS (encryption plus authentication) with the following security protocols: SSL and SASL_SSL.

These configurations are mostly the same as those for Apache Kafka® and a Kafka cluster running in Confluent Platform.

To use mTLS for the interbroker listener, additional configuration is needed. For details, refer to the following:

• Use TLS Authentication in Confluent Platform
• Configure mTLS Authentication and RBAC for Kafka Brokers

Enable mTLS in Kafka

Add the following configurations to enable mTLS in Kafka.

Listener configuration	Value	Description
ssl.client.auth	REQUIRED	Client authentication with certificate is required
ssl.client.auth	REQUESTED	Client authentication with a certificate is optional. If the certificate is validated if the client brings in a certificate for authentication.
ssl.client.auth	NONE	Client authentication with a certificate is not mandatory. If the client brings in a certificate, it is validated.

• For Kafka, the Anonymous user principal is used for authorization when no certificate is available from the client in REQUESTED mode or NONE mode and no other additional token-based authentication is set up.
• When using REQUESTED mode, client certificates are optional and additional authentications, like SASL/OAUTHBEARER, work when configured. The REQUESTED mode is recommended only for ensuring backward compatibility during migration from a prior authentication mechanism to mTLS.
• Also, when using REQUESTED mode, if a client brings an invalid certificate or an expired certificate, authentication would fail.
• For the MDS service, when NONE is configured for client certificate-based authentication, authentication with a backing user store on OAuth, LDAP, or a backing file store is mandatory.

When deploying RBAC with mTLS, the behavior and configuration options above are available for use across all other Confluent Platform components, including Schema Registry, Connect, and Confluent REST Proxy.

Configure Kafka to use mTLS only

Add the following configurations to enable mTLS only for Kafka.

listeners=..,EXTERNAL://:9096
listener.security.protocol.map=...,EXTERNAL:SSL

listener.name.external.ssl.client.auth=required
listener.name.external.ssl.truststore.location=/path/to/server.truststore.jks
listener.name.external.ssl.truststore.password=<truststore-password>
listener.name.external.ssl.keystore.location=/path/to/server.keystore.jks
listener.name.external.ssl.keystore.password=<keystore-password>
listener.name.external.ssl.key.password=<key-password>

# Optional settings and their default values
listener.name.external.ssl.endpoint.identification.algorithm=HTTPS
listener.name.external.ssl.truststore.type=JKS
listener.name.external.ssl.provider=null
listener.name.external.ssl.cipher.suites=null
listener.name.external.ssl.enabled.protocols=TLSv1.2,TLSv1.3

# Optional - For enhanced security with impersonation
listener.name.external.principal.builder.class=io.confluent.kafka.security.authenticator.OAuthKafkaPrincipalBuilder

Configure mTLS with SASL

Add the following configurations to enable mTLS with SASL for Kafka.

listeners=..,EXTERNAL://:9096
listener.security.protocol.map=...,EXTERNAL:SASL_SSL

# mTLS configs
listener.name.external.ssl.client.auth=required
listener.name.external.ssl.truststore.location=/path/to/server.truststore.jks
listener.name.external.ssl.truststore.password=<truststore-password>
listener.name.external.ssl.keystore.location=/path/to/server.keystore.jks
listener.name.external.ssl.keystore.password=<keystore-password>
listener.name.external.ssl.key.password=<key-password>

# SASL - OAuth configs
listener.name.external.sasl.enabled.mechanisms=OAUTHBEARER
listener.name.external.sasl.oauthbearer.jwks.endpoint.url=<jwks-endpoint>
listener.name.external.sasl.oauthbearer.expected.audience=<expected-audience>
listener.name.external.principal.builder.class=io.confluent.kafka.security.authenticator.OAuthKafkaPrincipalBuilder

# Optional - For enhanced security with impersonation
listener.name.external.principal.builder.class=io.confluent.kafka.security.authenticator.OAuthKafkaPrincipalBuilder

Map principal from certificates

When using mTLS, the authentication principal is constructed from client certificates based on principal mapping rules. The principal is used for checking authorization with RBAC, similar to authorization with an ACL.

listener.name.external.ssl.principal.mapping.rules = RULE:.*CN=([a-zA-Z0-9.-]*).*$/$1/,DEFAULT

Example:

Subject/DN: C=IN, ST=KA, O=CFLT, CN=client1.confluent.io Final Principal Name: client1.confluent.io

Configure Kafka clients

To use Kafka clients with mTLS, set the following configurations, replacing the placeholders with your actual values:

security.protocol=SSL
ssl.truststore.location=/path/to/client.truststore.jks
ssl.truststore.password=<truststore-password>
ssl.keystore.location=/path/to/client.keystore.jks
ssl.keystore.password=<keystore-password>
ssl.key.password=<key-password>

If there is additional SASL authentication, corresponding SASL client configurations need to be provided above.

Configure MDS service

To configure the MDS service for mTLS, the following configurations are required.

Enable HTTPS

# Specify HTTPS Listeners for MDS
confluent.metadata.server.listeners=https://0.0.0.0:8090
confluent.metadata.server.advertised.listeners=https://<localhost>:8090

# SSL configs
confluent.metadata.server.ssl.keystore.location=/path/to/keystore.jks
confluent.metadata.server.ssl.keystore.password=<keystore-password>
confluent.metadata.server.ssl.key.password=<key-password>
confluent.metadata.server.ssl.truststore.location=/path/to/truststore.jks
confluent.metadata.server.ssl.truststore.password=<truststore-password>

Enable mTLS

# set user store to NONE for mTLS only
confluent.metadata.server.user.store=NONE

# support token authentication on the MDS side
confluent.metadata.server.authentication.method=BEARER

# Enable mTLS
confluent.metadata.server.ssl.client.authentication=REQUIRED

# Principal mapping for AuthNZ
confluent.metadata.server.auth.ssl.principal.mapping.rules=
  RULE:.*CN=([a-zA-Z0-9.-]*).*$/$1/,DEFAULT

Note

Even when using mTLS without any token authentication from external clients, token support must be enabled in MDS for request forwarding, principal propagation, and other cases of internal communication.

Enable impersonation

For components relying on impersonation, for example REST Proxy, the following configurations are required in MDS.

# These users can impersonate a target user.
confluent.metadata.server.impersonation.super.users=User:<rp-principal-name>;User:<sr-principal-name>;User:<connect-principal-name>;User:<other-super-user>

# These users cannot be impersonated for security reasons.
confluent.metadata.server.impersonation.protected.users=User:<rp-principal-name>;User:<other-super-user>

Configure MDS client

All Confluent Platform services that act as clients to MDS need to set MDS client configuration properties for mTLS authentication. The configuration for this is similar to standard TLS/SSL settings and a reference can be found here. Ensure the following properties are correctly set:

confluent.metadata.ssl.truststore.location
confluent.metadata.ssl.truststore.password
confluent.metadata.ssl.keystore.location
confluent.metadata.ssl.keystore.password
confluent.metadata.ssl.key.password

When using mTLS only for authentication at the MDS service without any LDAP or OAuth-based authentication on the cluster, the following configurations are not required for connecting to MDS.

# Credentials to use with the MDS
confluent.metadata.basic.auth.user.info=<username>:<password>
confluent.metadata.http.auth.credentials.provider=BASIC

Configure Schema Registry

To configure Schema Registry for mTLS, the following configurations are required, replacing the placeholders with your actual values:

listeners=https://<hostname>:8081
ssl.truststore.location=/path/to/truststore.jks
ssl.truststore.password=<truststore-password>
ssl.keystore.location=/path/to/keystore.jks
ssl.keystore.password=<keystore-password>
ssl.key.password=<key-password>
ssl.client.authentication=REQUIRED
ssl.protocol=SSL
ssl.principal.mapping.rules =
  RULE:.*CN=([a-zA-Z0-9.-]*).*$/$1/,DEFAULT

Configure Confluent Control Center

Confluent Control Center to MDS and other Confluent Platform services like REST Proxy, Confluent Server, Schema Registry, and Connect can be enabled to work with mTLS when RBAC is enabled.

Confluent Control Center to MDS communication uses base TLS/SSL configurations for Confluent Control Center (and not confluent.metadata.ssl.* configurations).

To configure Confluent Control Center for mTLS, the following configurations are required, replacing the placeholders with your actual values:

# C3 configs
confluent.controlcenter.rest.listeners=https://0.0.0.0:9021
confluent.controlcenter.rest.authentication.method=BEARER
confluent.controlcenter.rest.ssl.key.password=<keystore-password>
confluent.controlcenter.rest.ssl.keystore.location=<keystore-location>
confluent.controlcenter.rest.ssl.keystore.password=<keystore-password>
confluent.controlcenter.rest.ssl.truststore.location=<truststore-location>
confluent.controlcenter.rest.ssl.truststore.password=<truststore-password>

# SSL configs for Stream (Interceptor) Monitoring
confluent.monitoring.interceptor.security.protocol=SSL
confluent.monitoring.interceptor.ssl.key.password=<key-password>
confluent.monitoring.interceptor.ssl.keystore.location=<keystore-location>
confluent.monitoring.interceptor.ssl.keystore.password=<keystore-password>
confluent.monitoring.interceptor.ssl.truststore.location=<truststore-location>
confluent.monitoring.interceptor.ssl.truststore.password=<truststore-password>

# SSL configs for Streams in C3
confluent.controlcenter.streams.security.protocol=SSL
confluent.controlcenter.streams.ssl.key.password=<key-password>
confluent.controlcenter.streams.ssl.keystore.location=<keystore-location>
confluent.controlcenter.streams.ssl.keystore.password=<keystore-password>
confluent.controlcenter.streams.ssl.truststore.location=<truststore-location>
confluent.controlcenter.streams.ssl.truststore.password=<truststore-password>

# configs for C3 to SR
confluent.controlcenter.schema.registry.schema.registry.ssl.truststore.location=<truststore-location>
confluent.controlcenter.schema.registry.schema.registry.ssl.truststore.password=<truststore-password>
confluent.controlcenter.schema.registry.schema.registry.ssl.keystore.location=<keystore-location>
confluent.controlcenter.schema.registry.schema.registry.ssl.keystore.password=<keystore-password>
confluent.controlcenter.schema.registry.schema.registry.ssl.key.password=<key-password>

# Configs for C3 to Connect
confluent.controlcenter.connect.connect.ssl.truststore.location=<truststore-location>
confluent.controlcenter.connect.connect.ssl.truststore.password=<truststore-password>
confluent.controlcenter.connect.connect.ssl.keystore.location=<keystore-location>
confluent.controlcenter.connect.connect.ssl.keystore.password=<keystore-password>
confluent.controlcenter.connect.connect.ssl.key.password=<key-password>

Configure a single proxy server store

As an alternative to configuring the Control Center for each component, you can configure a single proxy server store. To configure a proxy store, set the following configurations, replacing the placeholders with your actual values:

confluent.controlcenter.rest.proxy.ssl.truststore.location=<truststore-location>
confluent.controlcenter.rest.proxy.ssl.truststore.password=<password>
confluent.controlcenter.rest.proxy.ssl.keystore.location=<keystore-location>
confluent.controlcenter.rest.proxy.ssl.keystore.password=<password>
confluent.controlcenter.rest.proxy.ssl.key.password=<password>

Configure Kafka Connect

The following configurations are required for Kafka Connect worker operations, like group coordination and internal topic management when Confluent Server brokers require client authentication. Replace the placeholders with your actual values.

bootstrap.servers=<kafka-host>:9096
security.protocol=SSL
ssl.truststore.location=/path/to/truststore.jks
ssl.truststore.password=<truststore-password>
ssl.keystore.location=/path/to/keystore.jks
ssl.keystore.password=<keystore-password>
ssl.key.password=<key-password>

producer.security.protocol=SSL
producer.ssl.truststore.location=/path/to/truststore.jks
producer.ssl.truststore.password=<truststore-password>
producer.ssl.keystore.location=/path/to/keystore.jks
producer.ssl.keystore.password=<keystore-password>
producer.ssl.key.password=<key-password>

consumer.security.protocol=SSL
consumer.ssl.truststore.location=/path/to/truststore.jks
consumer.ssl.truststore.password=<truststore-password>
consumer.ssl.keystore.location=/path/to/keystore.jks
consumer.ssl.keystore.password=<keystore-password>
consumer.ssl.key.password=<key-password>

listeners.https.ssl.client.auth=required
listeners.https.ssl.truststore.location=/path/to/truststore.jks
listeners.https.ssl.truststore.password=<truststore-password>
listeners.https.ssl.keystore.location=/path/to/keystore.jks
listeners.https.ssl.keystore.password=<keystore-password>
listeners.https.ssl.key.password=<key-password>

To allow for request forwarding from follower to leader on mTLS, Kafka Connect workers need to be configured on the secure impersonation super user list on MDS.

Configure REST Proxy request forwarding over mTLS with RBAC

REST Proxy requires elevated permissions for request forwarding its client identity to Kafka. This elevated access allows REST Proxy to securely forward client requests without having access to certificates of its clients. For details on REST Proxy security and principal propagation, see Deploy Secure Standalone REST Proxy in Confluent Platform.

Impersonation must be enabled at MDS for REST Proxy. You also need to configure the allowed list of users for impersonation and protected user list.

All services can connect to Kafka on an TLS listener with mTLS RBAC. For REST Proxy, an mTLS RBAC SASL_SSL listener is required.

Enable mTLS in REST Proxy

To enable mTLS in REST Proxy, you can either configure a standalone REST Proxy or configure REST Proxy embedded in Confluent Platform services.

Configure Standalone REST Proxy

All internal clients (for example, monitoring client and license client) connect to Kafka over the TLS listener. The REST Proxy client itself, however, connects to Kafka over the SASL_SSL listener in order for the impersonation flow to work. Add the following configurations to enable security in REST Proxy, replacing the placeholders with your actual values:

# Enable security in RP
kafka.rest.resource.extension.class=io.confluent.kafkarest.security.KafkaRestSecurityResourceExtension
public.key.path=<path-to-public.pem>
rest.servlet.initializor.classes=io.confluent.common.security.jetty.initializer.AuthenticationHandler

# Enabling mTLS on RP server
ssl.client.authentication=REQUIRED
ssl.key.password=<key-password>
ssl.keystore.location=/path/to/keystore.jks
ssl.keystore.password=<keystore-password>
ssl.truststore.location=/path/to/truststore.jks
ssl.truststore.password=<truststore-password>

# RP connecting to MDS over mTLS
confluent.metadata.bootstrap.server.urls=<mds-url>:<mds-port>
confluent.metadata.ssl.truststore.location=/path/to/truststore.jks
confluent.metadata.ssl.truststore.password=<truststore-password>
confluent.metadata.ssl.keystore.location=/path/to/keystore.jks
confluent.metadata.ssl.keystore.password=<keystore-password>
confluent.metadata.ssl.key.password=<key-password>

# RP Client connects over SASL_SSL for impersonation flow
client.bootstrap.servers=<KAFKA_SASL_SSL_LISTENER>
client.sasl.mechanism=OAUTHBEARER
client.security.protocol=SASL_SSL
client.ssl.key.password=<key-password>
client.ssl.keystore.location=/path/to/keystore.jks
client.ssl.keystore.password=<keystore-password>
client.ssl.truststore.location=/path/to/truststore.jks

Configure Embedded REST Proxy

To configure REST Proxy embedded in Confluent Platform services, the following configurations are required, replacing the placeholders with your actual values:

# Existing non-mTLS configs to enable ERP and security
kafka.rest.bootstrap.servers=<kafka-bootstrap-server>
kafka.rest.enable=true
kafka.rest.kafka.rest.resource.extension.class=io.confluent.kafkarest.security.KafkaRestSecurityResourceExtension
kafka.rest.public.key.path=<path-to-public.pem>
kafka.rest.rest.servlet.initializor.classes=io.confluent.common.security.jetty.initializer.AuthenticationHandler

# Protocol used to communicate with brokers.
kafka.rest.client.security.protocol=SASL_SSL

# SSL configs used to connect to broker
kafka.rest.client.ssl.key.password=<keystore-password>
kafka.rest.client.ssl.keystore.location=<keystore-location>
kafka.rest.client.ssl.keystore.password=<keystore-password>
kafka.rest.client.ssl.truststore.location=<truststore-location>
kafka.rest.client.ssl.truststore.password=<truststore-password>

# SSL configs for connecting to MDS
kafka.rest.confluent.metadata.bootstrap.server.urls=<mds-url>:<mds-port>
kafka.rest.confluent.metadata.ssl.truststore.location=<truststore-location>
kafka.rest.confluent.metadata.ssl.truststore.password=<truststore-password>
kafka.rest.confluent.metadata.ssl.keystore.location=<keystore-location>
kafka.rest.confluent.metadata.ssl.keystore.password=<keystore-password>