Configure Mutual TLS (mTLS) authentication on Confluent Cloud¶
Prerequisites¶
Kafka or Confluent Platform client¶
- Java or librdkafka Kafka clients. For supported versions, see Build Kafka Client Applications on Confluent Cloud.
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¶
- A Confluent Cloud principal with OrganizationAdmin role binding. The principal requires permission to use the certificate authority APIs.
- A Dedicated cluster.
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.