Authentication using mTLS

mTLS authentication overview

Mutual TLS (mTLS) is a mutual authentication mechanism. Not only servers have keys and certs that the client uses to verify the identity of servers, clients also have keys and certs that the server uses to verify the identity of clients.

The following figure illustrates how Pulsar processes mTLS authentication between clients and servers.

Enable mTLS authentication on brokers

To configure brokers to authenticate clients using mTLS, add the following parameters to the conf/broker.conf. If you use a standalone Pulsar, you need to add these parameters to the conf/standalone.conf file:

# enable authentication
authenticationEnabled=true
# set mTLS authentication provider
authenticationProviders=org.apache.pulsar.broker.authentication.AuthenticationProviderTls

# configure TLS for client to connect brokers
brokerClientTlsEnabled=true
brokerClientTrustCertsFilePath=/path/to/ca.cert.pem
brokerClientAuthenticationPlugin=org.apache.pulsar.client.impl.auth.AuthenticationTls
brokerClientAuthenticationParameters={"tlsCertFile":"/path/to/broker_client.cert.pem","tlsKeyFile":"/path/to/broker_client.key-pk8.pem"}

# configure TLS ports
brokerServicePortTls=6651
webServicePortTls=8081

# configure CA certificate
tlsTrustCertsFilePath=/path/to/ca.cert.pem
# configure server certificate
tlsCertificateFilePath=/path/to/server.cert.pem
# configure server's private key
tlsKeyFilePath=/path/to/server.key-pk8.pem

# enable mTLS
tlsRequireTrustedClientCertOnConnect=true
tlsAllowInsecureConnection=false

# TLS certificate refresh interval in seconds (0 disables background rotation)
tlsCertRefreshCheckDurationSec=300

Enable mTLS authentication on proxies

To configure proxies to authenticate clients using mTLS, add the following parameters to the conf/proxy.conf file.

# enable authentication
authenticationEnabled=true
# set mTLS authentication provider
authenticationProviders=org.apache.pulsar.broker.authentication.AuthenticationProviderTls

# configure TLS for the proxy to connect to brokers
tlsEnabledWithBroker=true
brokerClientTrustCertsFilePath=/path/to/ca.cert.pem
brokerClientAuthenticationPlugin=org.apache.pulsar.client.impl.auth.AuthenticationTls
brokerClientAuthenticationParameters={"tlsCertFile":"/path/to/proxy.cert.pem","tlsKeyFile":"/path/to/proxy.key-pk8.pem"}

# configure proxy TLS ports
servicePortTls=6651
webServicePortTls=8081

# configure CA certificate
tlsTrustCertsFilePath=/path/to/ca.cert.pem
# configure server certificate
tlsCertificateFilePath=/path/to/server.cert.pem
# configure server's private key
tlsKeyFilePath=/path/to/server.key-pk8.pem

# enable mTLS
tlsRequireTrustedClientCertOnConnect=true
tlsAllowInsecureConnection=false

On the brokers, configure the authenticated proxy role as a trusted proxyRoles entry and explicitly set authenticateOriginalAuthData=false. Pulsar enables original-client authentication by default, but the broker receives the proxy's certificate on its TLS connection rather than the client's certificate. The proxy authenticates the client and forwards its role; the broker still checks authorization for both roles. See Proxy roles.

Configure mTLS authentication in Pulsar clients

When using mTLS authentication, clients connect via TLS transport. Use https:// with the configured webServicePortTls (8081 in the examples above) and pulsar+ssl:// with the configured binary TLS port (6651 above). Certificates must match the hostnames used to connect; Java clients enable hostname verification by default.

Java

import org.apache.pulsar.client.api.PulsarClient;

PulsarClient client = PulsarClient.builder()
    .serviceUrl("pulsar+ssl://broker.example.com:6651/")
    .tlsTrustCertsFilePath("/path/to/ca.cert.pem")
    .authentication("org.apache.pulsar.client.impl.auth.AuthenticationTls",
                    "tlsCertFile:/path/to/client.cert.pem,tlsKeyFile:/path/to/client.key-pk8.pem")
    .build();

Python

from pulsar import Client, AuthenticationTLS

auth = AuthenticationTLS("/path/to/client.cert.pem", "/path/to/client.key-pk8.pem")
client = Client("pulsar+ssl://broker.example.com:6651/",
                tls_trust_certs_file_path="/path/to/ca.cert.pem",
                tls_allow_insecure_connection=False,
				authentication=auth)

C++

#include <pulsar/Client.h>

pulsar::ClientConfiguration config;
config.setUseTls(true);
config.setTlsTrustCertsFilePath("/path/to/ca.cert.pem");
config.setTlsAllowInsecureConnection(false);

pulsar::AuthenticationPtr auth = pulsar::AuthTls::create("/path/to/client.cert.pem",
                                                         "/path/to/client.key-pk8.pem")
config.setAuth(auth);

pulsar::Client client("pulsar+ssl://broker.example.com:6651/", config);

Node.js

const Pulsar = require('pulsar-client');

(async () => {
  const auth = new Pulsar.AuthenticationTls({
    certificatePath: '/path/to/client.cert.pem',
    privateKeyPath: '/path/to/client.key-pk8.pem',
  });

  const client = new Pulsar.Client({
    serviceUrl: 'pulsar+ssl://broker.example.com:6651/',
    authentication: auth,
    tlsTrustCertsFilePath: '/path/to/ca.cert.pem',
  });
})();

Go

client, err := pulsar.NewClient(ClientOptions{
		URL:                   "pulsar+ssl://broker.example.com:6651/",
		TLSTrustCertsFilePath: "/path/to/ca.cert.pem",
		Authentication:        pulsar.NewAuthenticationTLS("/path/to/client.cert.pem", "/path/to/client.key-pk8.pem"),
	})

C#

var clientCertificate = new X509Certificate2("admin.pfx");
var client = PulsarClient.Builder()
                         .AuthenticateUsingClientCertificate(clientCertificate)
                         .Build();

Configure mTLS authentication in CLI tools

Command-line tools like pulsar-admin, pulsar-perf, and pulsar-client use the conf/client.conf config file in a Pulsar installation.

To use mTLS authentication with the CLI tools of Pulsar, you need to add the following parameters to the conf/client.conf file, alongside the configurations to enable mTLS encryption:

webServiceUrl=https://broker.example.com:8081/
brokerServiceUrl=pulsar+ssl://broker.example.com:6651/
authPlugin=org.apache.pulsar.client.impl.auth.AuthenticationTls
authParams=tlsCertFile:/path/to/admin.cert.pem,tlsKeyFile:/path/to/admin.key-pk8.pem

Configure mTLS authentication with KeyStore

Apache Pulsar supports TLS encryption and mTLS authentication between clients and Apache Pulsar service. By default, it uses PEM format file configuration.

To configure mTLS authentication with KeyStore, complete the following steps.

Step 1: Configure brokers

Configure the broker.conf file as follows.

# Configuration to enable authentication
authenticationEnabled=true
authenticationProviders=org.apache.pulsar.broker.authentication.AuthenticationProviderTls

# Enable KeyStore type
tlsEnabledWithKeyStore=true

# key store
tlsKeyStoreType=JKS
tlsKeyStore=/var/private/tls/broker.keystore.jks
tlsKeyStorePassword=brokerpw

# trust store
tlsTrustStoreType=JKS
tlsTrustStore=/var/private/tls/broker.truststore.jks
tlsTrustStorePassword=brokerpw

# internal client/admin-client config
brokerClientTlsEnabled=true
brokerClientTlsEnabledWithKeyStore=true
brokerClientTlsTrustStoreType=JKS
brokerClientTlsTrustStore=/var/private/tls/client.truststore.jks
brokerClientTlsTrustStorePassword=clientpw
# internal auth config
brokerClientAuthenticationPlugin=org.apache.pulsar.client.impl.auth.AuthenticationKeyStoreTls
brokerClientAuthenticationParameters={"keyStoreType":"JKS","keyStorePath":"/var/private/tls/client.keystore.jks","keyStorePassword":"clientpw"}

tlsRequireTrustedClientCertOnConnect=true
tlsAllowInsecureConnection=false

Step 2: Configure clients

Besides configuring TLS encryption, you need to configure the KeyStore, which contains a valid CN as client role, for clients.

For example:

1. for Command-line tools like pulsar-admin, pulsar-perf, and pulsar-client, set the conf/client.conf file in a Pulsar installation.

   webServiceUrl=https://broker.example.com:8443/
   brokerServiceUrl=pulsar+ssl://broker.example.com:6651/
   useKeyStoreTls=true
   tlsTrustStoreType=JKS
   tlsTrustStorePath=/var/private/tls/client.truststore.jks
   tlsTrustStorePassword=clientpw
   authPlugin=org.apache.pulsar.client.impl.auth.AuthenticationKeyStoreTls
   authParams={"keyStoreType":"JKS","keyStorePath":"/var/private/tls/client.keystore.jks","keyStorePassword":"clientpw"}

2. for Java client

   import org.apache.pulsar.client.api.PulsarClient;
   
   PulsarClient client = PulsarClient.builder()
       .serviceUrl("pulsar+ssl://broker.example.com:6651/")
       .useKeyStoreTls(true)
       .tlsTrustStorePath("/var/private/tls/client.truststore.jks")
       .tlsTrustStorePassword("clientpw")
       .allowTlsInsecureConnection(false)
       .enableTlsHostnameVerification(true)
       .authentication(
               "org.apache.pulsar.client.impl.auth.AuthenticationKeyStoreTls",
               "keyStoreType:JKS,keyStorePath:/var/private/tls/client.keystore.jks,keyStorePassword:clientpw")
       .build();

3. for Java admin client

       PulsarAdmin amdin = PulsarAdmin.builder().serviceHttpUrl("https://broker.example.com:8443")
           .useKeyStoreTls(true)
           .tlsTrustStorePath("/var/private/tls/client.truststore.jks")
           .tlsTrustStorePassword("clientpw")
           .allowTlsInsecureConnection(false)
           .enableTlsHostnameVerification(true)
           .authentication(
                  "org.apache.pulsar.client.impl.auth.AuthenticationKeyStoreTls",
                  "keyStoreType:JKS,keyStorePath:/var/private/tls/client.keystore.jks,keyStorePassword:clientpw")
           .build();

note

Configure tlsTrustStorePath when you set useKeyStoreTls to true.