<a id="configure-mtls-rbac"></a>

# 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](../../authentication/mutual-tls/tls-principal-mapping.md#tls-principal-mapping).

## 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](../../authentication/mutual-tls/overview.md#tls-authentication)
* [Configure mTLS Authentication and RBAC for Kafka Brokers](../../../kafka/configure-mds/mutual-tls-auth-rbac.md#mutual-tls-auth-rbac)

<a id="enable-mtls-in-ak"></a>

#### 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. The certificate<br/>is validated if the client brings in a certificate for authentication. |
| `ssl.client.auth`        | `NONE`      | Client authentication with a certificate is not required.                                                                                        |
* 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.

```properties
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.

```properties
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.

```properties
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:

```properties
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

```properties
# 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

```properties
# 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.

```properties
# 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:

```properties
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.

```properties
# 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:

```properties
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:

```properties
# 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 Streams in C3
# Set confluent.controlcenter.streams.bootstrap.servers to an SASL_SSL listener that can accept an MDS token
confluent.controlcenter.streams.bootstrap.servers=Kafka:9092
confluent.controlcenter.streams.sasl.mechanism=OAUTHBEARER
confluent.controlcenter.streams.security.protocol=SASL_SSL
confluent.controlcenter.streams.sasl.login.callback.handler.class=io.confluent.kafka.clients.plugins.auth.token.TokenCertificateLoginCallbackHandler
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:

```properties
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.

```properties
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 Confluent Platform for Apache Flink

Confluent Manager for Apache Flink supports all available RBAC authentication options to authenticate
with the Confluent Platform metadata service. The following examples cover some authentication methods
for the metadata service.

### OAuth example

The following YAML file shows how you might configure mTLS authentication for CMF with Oauth authentication with MDS.

```yaml
# authorization-values.yaml
 cmf:
   authorization:
     mdsRestConfig:
       endpoint: #Mandatory: replace with the endpoint for the Metadata Service
       authentication:
         type: oauth
         config:
           confluent.metadata.http.auth.credentials.provider: OAUTHBEARER
           confluent.metadata.oauthbearer.token.endpoint.url: #URL of your configured IDP used to store the users
           confluent.metadata.oauthbearer.login.client.id: #Client id for the user talking to MDS, needs to be a user that can request permissions for all users
           confluent.metadata.oauthbearer.login.client.secret: #Client secret for the user talking to MDS
           # Required when MDS uses HTTPS: truststore to trust the MDS server certificate.
           confluent.metadata.ssl.truststore.location: #truststore path to trust the MDS server certificate
           confluent.metadata.ssl.truststore.password: #optional password if the truststore is password protected
```

When you make the `helm install` call, use the `-f` flag to pass the YAML file with the security information
like the following:

```none
helm upgrade --install cmf confluent/confluent-manager-for-apache-flink \
  -f mtls-values.yaml \
  -f authorization-values.yaml
```

#### NOTE
When the MDS endpoint uses HTTPS, you must include `confluent.metadata.ssl.truststore.location`
in the authentication configuration so that CMF can trust the MDS server certificate. You only need to
include `confluent.metadata.ssl.truststore.password` if the truststore is password-protected. These
settings are specific to the CMF-to-MDS connection and are separate from any server SSL configuration
on CMF itself.

### mTLS example

The following example shows how you configure mTLS authentication for CMF with mTLS authentication with MDS.

```yaml
# authorization-values.yaml
cmf:
  authorization:
    mdsRestConfig:
      endpoint: #Mandatory: replace with the endpoint for the Metadata Service
      authentication:
        type: mtls
        config:
          confluent.metadata.ssl.truststore.location: #truststore path containing certificates needed to talk to metadata service
          confluent.metadata.ssl.truststore.password: #optional password if truststore is encrypted
          confluent.metadata.ssl.keystore.location: #keystore path containing keys/certificates needed to talk to metadata service
          confluent.metadata.ssl.keystore.password: #optional password if keystore is encrypted
          confluent.metadata.ssl.key.password:
```

When you make the `helm install` call, use the `-f` flag to pass the YAML file with the security information
like the following:

```none
helm upgrade --install cmf confluent/confluent-manager-for-apache-flink \
  -f mtls-values.yaml \
  -f authorization-values.yaml
```

## 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](../../../kafka-rest/production-deployment/rest-proxy/security.md#kafkarest-security).

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. For an
explanation of how request forwarding works and details on configuring it for
RBAC, see [Request Forwarding from Follower to Leader in Confluent Platform](request-forwarding-mtls-rbac.md#request-forwarding-mtls-rbac).

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:

```properties
# 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:

```properties
# 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>
```

## Configure ksqlDB with mTLS and RBAC

This section outlines configuration that allows ksqlDB to use
mTLS for secure communication across various Confluent Platform components, alongside
RBAC through impersonation. Implementing mTLS ensures that all
communication is encrypted and mutually authenticated, while RBAC provides
fine-grained control over user permissions within ksqlDB.

Before proceeding, ensure you have a working Confluent Platform environment with Kafka,
Metadata Service (MDS), Schema Registry, and Connect already configured.
You should also have the necessary mTLS certificates (keystores and
truststores) generated and available for your ksqlDB server and clients.

### Grant impersonation permissions for ksqlDB user

To enable RBAC for impersonated users, grant the ksqlDB user
permissions to allow the impersonation flow. To do this, add the
ksqlDB user to the `confluent.metadata.server.impersonation.super.users`
property in your Kafka brokers’ `server.properties` file. This setting designates the
ksqlDB user as a “super user” for impersonation purposes, allowing it to
forward the original client’s identity to Kafka for authorization.

```properties
confluent.metadata.server.impersonation.super.users=User:<ksql-user>
```

### ksqlDB server-side configurations

Apply these configurations to the ksqlDB server’s properties file,
typically ksql-server.properties. They enable mTLS for inbound connections
to the ksqlDB server and outgoing connections to other Confluent Platform components, as
well as enabling impersonation token passing.

#### Enable mTLS on ksqlDB server

To secure communication to the ksqlDB server itself, configure it to require
mTLS for client connections. Additionally, define properties that pass the
impersonated token to Kafka, this token allows RBAC to perform based on the
impersonated user’s identity.

```properties
security.protocol=SSL
ssl.client.authentication=REQUIRED
ssl.key.password=<key-password>
ssl.keystore.location=<keystore-location>
ssl.keystore.password=<keystore-password>
ssl.principal.mapping.rules=DEFAULT
ssl.truststore.location=<truststore-location>
ssl.truststore.password=<truststore-password>
# to pass impersonation token to kafka
ksql.kafka.delegate.client.bootstrap.servers=<SASL_SSL kafka bootstrap server>
ksql.kafka.delegate.client.security.protocol=SASL_SSL
```

#### Enable mTLS from ksqlDB to MDS

Secure communication between ksqlDB and MDS by enabling mTLS. This
configuration ensures ksqlDB can retrieve metadata and perform operations that
interact with MDS.

```properties
confluent.metadata.bootstrap.server.urls=https://<mds-url>:<mds-port>
confluent.metadata.ssl.endpoint.identification.algorithm=HTTPS
confluent.metadata.ssl.truststore.location=<truststore-location>
confluent.metadata.ssl.truststore.password=<truststore-password>
confluent.metadata.ssl.keystore.location=<keystore-location>
confluent.metadata.ssl.keystore.password=<keystore-password>
confluent.metadata.ssl.key.password=<key-password>
```

#### Enable mTLS from ksqlDB to Schema Registry

Secure communication between ksqlDB and Schema Registry is vital for managing
schemas of data streams. Configure ksqlDB to use mTLS when connecting to
Schema Registry.

```properties
ksql.schema.registry.ssl.key.password=<key-password>
ksql.schema.registry.ssl.keystore.location=<keystore-location>
ksql.schema.registry.ssl.keystore.password=<keystore-password>
ksql.schema.registry.ssl.truststore.location=<truststore-location>
ksql.schema.registry.ssl.truststore.password=<truststore-password>
ksql.schema.registry.url=https://<schema-registry-hostname>:<schema-registry-port>
```

#### Enable mTLS from ksqlDB to Connect

If your ksqlDB queries interact with Connect for source or sink
connectors, configure the connection between ksqlDB and
Connect.

```properties
ksql.connect.url=https://<connect-host>:<connect-port>
ksql.connect.ssl.key.password=<key-password>
ksql.connect.ssl.keystore.location=<keystore-location>
ksql.connect.ssl.keystore.password=<keystore-password>
ksql.connect.ssl.truststore.location=<truststore-location>
ksql.connect.ssl.truststore.password=<truststore-password>
```

### ksqlDB client configurations

Once the ksqlDB server is configured for mTLS, configure the various clients
that interact with ksqlDB to use the appropriate certificates for mutual
authentication.

#### Java clients

When developing Java applications that connect to ksqlDB, configure the
`ClientOptions` object with the client’s keystore path and truststore,
along with their respective passwords. This configuration establishes a secure
mTLS connection.

```java
ClientOptions options = ClientOptions.create()
      // enable TLS when KSQL server is on https
      .setUseTls(true)
      .setHost(<ksqlDB-server-host>)
      .setPort(<ksqlDB-server-host-port>)
      .setTrustStore("<truststore-location>")
      .setTrustStorePassword("<truststore-password>")
      .setKeyStore("<keystore-location>")
      .setKeyStorePassword("<keystore-password>")
      .setKeyPassword("<key-password>")
      // set to true if hostname verification should be enabled
      .setVerifyHost(false)
      .setUseAlpn(true);
```

#### REST Clients

For REST API interactions with ksqlDB, configure clients such as Postman or
`curl` to pass the client certificates during HTTPS requests. This ensures
that the REST client can establish a mutually authenticated connection with the
ksqlDB server.

Just the certificates need to be passed while making a REST API call. Postman
can be used to test.

#### Confluent Control Center UI

Configure Confluent Control Center to connect to ksqlDB using mTLS. This involves specifying
the ksqlDB service URL with HTTPS and providing the necessary keystore and
truststore details for Confluent Control Center to authenticate with ksqlDB.

```properties
confluent.controlcenter.ksql.ksql_service.advertised.url=https://<ksql-host>:<ksql-port>
confluent.controlcenter.ksql.ksql_service.ssl.key.password=<key-password>
confluent.controlcenter.ksql.ksql_service.ssl.keystore.location=<keystore-location>
confluent.controlcenter.ksql.ksql_service.ssl.keystore.password=<keystore-password>
confluent.controlcenter.ksql.ksql_service.ssl.truststore.location=<truststore-location>
confluent.controlcenter.ksql.ksql_service.ssl.truststore.password=<truststore-password>
confluent.controlcenter.ksql.ksql_service.url=https://<ksql-host>:<ksql-port>
```

#### ksqlDB CLI

Configure the ksqlDB CLI to connect to a mTLS-enabled ksqlDB server. The
CLI extracts the principal from the provided client certificates for
authentication.

To connect using a configuration file:

```bash
ksql --config-file <ksql-cli.properties> https://<ksql-host>:<ksql-port>
```

The ksql-cli.properties file should contain the following mTLS configuration:

```properties
ssl.key.password=<key-password>
ssl.keystore.location=<keystore-location>
ssl.keystore.password=<keystore-password>
ssl.truststore.location=<truststore-location>
ssl.truststore.password=<truststore-password>
```

#### ksql-migrations tool

When utilizing the `ksql-migrations` tool for managing ksqlDB migrations,
it’s essential to configure it with the appropriate mTLS settings to ensure
secure communication with the ksqlDB server.

```properties
ksql.server.url=https://<ksql-host>:<ksql-port>
# Migrations metadata configs:
ksql.migrations.stream.name=MIGRATION_EVENTS_TEST
ksql.migrations.table.name=MIGRATION_SCHEMA_VERSIONS
ksql.migrations.stream.topic.name=ksql-service-idksql_MIGRATION_EVENTS
ksql.migrations.table.topic.name=ksql-service-idksql_MIGRATION_SCHEMA_VERSIONS
ksql.migrations.topic.replicas=1
# TLS configs:
ssl.key.password=<key-password>
ssl.keystore.location=<keystore-location>
ssl.keystore.password=<keystore-password>
ssl.truststore.location=<truststore-location>
ssl.truststore.password=<truststore-password>
ssl.alpn=true
# set to false if hostname verification should be enabled
ssl.verify.host=true
```