Configure Mutual TLS (mTLS) authentication on Confluent Cloud

Prerequisites

Kafka or Confluent Platform client

Mutual TLS (mTLS)

  • A valid X.509 certificate chain PEM file, used to configure a trusted certificate authority in Confluent Cloud. The PEM file must meet the following requirements:

    • Minimum of one root or intermediate certificate.

    • Maximum of four certificates in the chain.

    • If using librdkafka TLS clients, the certificate chain must include the signing certificate of the client certificate; otherwise, the TLS handshake fails during mTLS authentication.

    • The file must be named using a .pem extension.

      Example: acme-internal-client-ca.pem

  • A PKCS12 or JKS keystore, used by the client application to authenticate to Confluent Cloud clusters. The keystore must contain the following:

    • A single client private key.

    • The client certificate.

    • The associated CA certificate chain that signs the client certificate.

    • Early Access only: The full certificate chain, including the root CA, must be included in the client keystore. For General Availability (GA), the certificate chain is automatically built if the certificate chain is not provided.

    • The file must be named using a .p12 or .jks extension.

      PKCS12 example: acme.client.keystore.p12

      JKS example: acme.client.keystore.jks

  • A client configuration file containing the keystore settings for TLS authentication. This file is used by the client application to authenticate to your Kafka clusters.

    PKCS12 example for client.properties

    security.protocol=SSL
ssl.keystore.location=path/to/<client-keystore-filename>.p12
ssl.keystore.type=PKCS12
ssl.keystore.password=<keystore password>
ssl.key.password=<key password>

    JKS example for client.properties

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

Confluent Cloud

Steps to configure mTLS authentication on Confluent Cloud

Follow these steps to configure mTLS authentication on Confluent Cloud. All of the examples are based on Acme, Inc. internal client applications.

Step 1 - Configure trusted certificate authority on Confluent Cloud

Prepare a JSON payload for the trusted client certificate authority (CA) chain that you want to configure for your Confluent Cloud organization. You are recommended to prepare a payload this way to preserve formatting of the certificate_chain as a string when configuring the certificate authority in Confluent Cloud.

Template

Replace the placeholders with your actual values.

json_payload=$(jq -n \
  --arg name "<Name of Certificate Authority>" \
  --arg description "<Description of Certificate Authority>" \
  --arg certificate_chain "$(cat path/to/<ca-chain-filename>.pem)" \
  --arg certificate_chain_filename "<ca-chain-filename>.pem" \
    '{
      display_name: $name,
      description: $description,
      certificate_chain: $certificate_chain,
      certificate_chain_filename: $certificate_chain_filename
    }');

Example

Here is an example of a JSON payload:

json_payload=$(jq -n \
  --arg name "Acme Internal Client CA" \
  --arg description "Acme certificate authority for internal client applications" \
  --arg certificate_chain "$(cat path/to/acme-internal-client-ca.pem)" \
  --arg certificate_chain_filename "acme-internal-client-ca.pem" \
    '{
      display_name: $name,
      description: $description,
      certificate_chain: $certificate_chain,
      certificate_chain_filename: $certificate_chain_filename
    }')

Here is the resulting JSON payload:

{
  "display_name": "Acme Internal Client CA",
  "description": "Acme certificate authority for internal client applications",
  "certificate_chain": "-----BEGIN CERTIFICATE-----\nMIICGzCCAaGgAwIBAgIQQdKd0XLq7qeAwSxs6S+HUjAKBggqhkjOPQQDAzBPMQsw\nCQYDVQQGEwJVUzEpMCcGA1UEChMgSW50ZXJuZXQgU2VjdXJpdHkgUmVzZWFyY2gg\nR3JvdXAxFTATBgNVBAMTDElTUkcgUm9vdCBYMjAeFw0yMDA5MDQwMDAwMDBaFw00\nMDA5MTcxNjAwMDBaME8xCzAJBgNVBAYTAlVTMSkwJwYDVQQKEyBJbnRlcm5ldCBT\nZWN1cml0eSBSZXNlYXJjaCBHcm91cDEVMBMGA1UEAxMMSVNSRyBSb290IFgyMHYw\nEAYHKoZIzj0CAQYFK4EEACIDYgAEzZvVn4CDCuwJSvMWSj5cz3es3mcFDR0HttwW\n+1qLFNvicWDEukWVEYmO6gbf9yoWHKS5xcUy4APgHoIYOIvXRdgKam7mAHf7AlF9\nItgKbppbd9/w+kHsOdx1ymgHDB/qo0IwQDAOBgNVHQ8BAf8EBAMCAQYwDwYDVR0T\nAQH/BAUwAwEB/zAdBgNVHQ4EFgQUfEKWrt5LSDv6kviejM9ti6lyN5UwCgYIKoZI\nzj0EAwMDaAAwZQIwe3lORlCEwkSHRhtFcP9Ymd70/aTSVaYgLXTWNLxBo1BfASdW\ntL4ndQavEi51mI38AjEAi/V3bNTIZargCyzuFJ0nN6T5U6VR5CmD1/iQMVtCnwr1\n/q4AaOeMSQ+2b1tbFfLn\n-----END CERTIFICATE-----",
  "certificate_chain_filename": "acme-internal-client-ca.pem"
}

Step 2 - Configure your trusted client certificate authority

Configure your trusted client certificate authority with Confluent Cloud REST APIs using the prepared payload.

For more information, see Certificate Authorities (iam/v2) [Confluent Cloud API].

Templates

Replace the placeholder values with your actual values.

HTTP Basic authentication using Authorization header
auth=$(echo -n '<username>:<password>' | base64)
curl -X POST \
  -H "Authorization: Basic $auth" \
  -H 'Content-Type:application/json' \
  -d "$json_payload" https://api.confluent.cloud/iam/v2/certificate-authorities
Bearer authentication using Authorization header
curl -X POST \
  -H "Authorization: Bearer <your_token>" \
  -H 'Content-Type:application/json' \
  -d "$json_payload" https://api.confluent.cloud/iam/v2/certificate-authorities

Step 3 - Create certificate identity pools for granular access control

Create an identity pool for your certificate authority. You must configure a valid filter that your client certificates match. For testing, true can be used as the filter.

For more information, see:

Template

Replace placeholder values with your actual values.

curl -X POST \
  -H "Authorization: <Basic or Bearer Auth>" \
  -H 'Content-Type: application/json' \
  -d '{"display_name": "<certificate pool name>",
    "description": "<description>",
    "filter":"<filter>",
    "external_identifier": "<external identifier>"}' \
  https://api.confluent.cloud/iam/v2/certificate-authorities/<certificate_authority_id>/identity-pools

The external_identifier field is used to determine the claim to match on. The following values are supported:

CN Common Name
DN Distinguished Name
SAN Subject Alternative Name
Serial Serial Number
SHA1 SHA-1 fingerprint

Example

The following example shows the use of HTTP Basic Authentication with an API key and a filter (CN==\"acme-client\") that is applied for the CN external identifier:

# Set the API key in an environment variable
export API_KEY='your_actual_api_key'

# Encode the API key for use in the curl command.
# The -n option is used to remove the newline character.
# For username:password, use the API key as the username and append
# the colon before encoding, even though there no password is used.
export ENCODED_API_KEY=$(echo -n "$API_KEY:" | base64)

# Use the environment variable in the curl command
curl -X POST \
  -H "Authorization: Basic $ENCODED_API_KEY" \
  -H 'Content-Type: application/json' \
  -d '{"display_name": "Acme Client Pool",
     "description": "Certificate identity pool for Acme client certificates",
     "filter":"CN==\"acme-client\"",
     "external_identifier": "CN"}' \
  https://api.confluent.cloud/iam/v2/certificate-authorities/<certificate_authority_id>/identity-pools

Step 4 - Assign role bindings for your certificate identity pools

Assign role bindings at the desired scope for your certificate identity pools.

For more information, see:

Template

curl -X POST -H "Authorization: <Basic or Bearer Auth>" \
  -H 'content-type: application/json' \
  -d '{"principal":"User:<pool-id>",
    "role_name":"<role name>",
    "crn_pattern":"<crn pattern>"}' \
  https://api.confluent.cloud/iam/v2/role-bindings

Example

This example below grants CloudClusterAdmin (cloudclusteradmin) to the certificate identity pool Acme Client Pool in the Dedicated cluster.

curl -X POST -H "Authorization: Bearer <your_token>" \
  -H 'content-type: application/json' \
  -d '{"principal":"User:Acme Client Pool",
    "role_name":"CloudClusterAdmin",
    "crn_pattern":"crn://confluent.cloud/organization=fc5ba16d-661d-474c-85df-c2a1ed26032c/environment=env-pqr45/cloud-cluster=lkc_a1b2c"}' \
  https://api.confluent.cloud/iam/v2/role-bindings

Step 5 - Verify and use mTLS authentication

After configuring a trusted certificate chain as your client certificate authority and a corresponding identity pool principal with sufficient permissions, a valid client certificate can be used for authentication and authorization.

Unlike OAuth/OIDC, the certificate identity pool does not need to be specified in the client configuration. The identity pool is automatically mapped based on the identity pool filters.

  • If multiple identity pools are mapped, the client receives a union-of-permissions of all the mapped identity pools.

  • If a client certificate matches multiple identity pools with different external identifier claims, the external identifier is prioritized as follows:

    Priority External identifier Maximum length
    1 CN 255 characters
    2 DN 255 characters
    3 Serial number  
    4 SAN 255 characters
    5 SHA-1 fingerprint  

    If the CN, DN, or SAN exceeds 255 characters, the values used are truncated to 255 characters.

To verify that the client can authenticate to the Dedicated Kafka cluster, you can use the client configuration file to list topics in the cluster. Here’s an example of how to use the configured mTLS authentication with a Kafka command-line tool.

Example

To run this example:

  • Replace <bootstrap URL> with the bootstrap URL for your Dedicated Kafka cluster.
  • The client.properties file is the client configuration file that you created in Step 3. It should contain the following settings:

PKCS12 example for client.properties

security.protocol=SSL
ssl.keystore.location=path/to/acme.client.keystore.p12
ssl.keystore.type=PKCS12
ssl.keystore.password=<keystore password>
ssl.key.password=<key password>

JKS example for client.properties

security.protocol=SSL
ssl.keystore.location=path/to/acme.client.keystore.jks
ssl.keystore.type=JKS
ssl.keystore.password=<keystore password>
ssl.key.password=<key password>

# list topics in the cluster
bin/kafka-topics.sh --bootstrap-server <bootstrap URL> \
  --command-config client.properties \
  --list

This should return a list of topics in the cluster. Note that you didn’t need to specify the certificate identity pools in the client configuration. Multiple identity pools can be selected based on the filters in the identity pools that evaluate to true for the client certificate. The correct identity pools are automatically selected based on the client certificate in the keystore.

Related content