Pulsar provides a way to use custom authentication and authorization mechanisms.

Authentication

Pulsar supports mutual TLS and Athenz authentication plugins. For how to use these authentication plugins, you can refer to the description in Security.

You can choose to use a custom authentication mechanism by providing the implementation in the form of two plugins. One plugin is for the Client library and the other plugin is for the Pulsar Broker to validate the credentials.

Client authentication plugin

For client library, you need to implement org.apache.pulsar.client.api.Authentication. By entering the command below you can pass this class when you create a Pulsar client:

  1. PulsarClient client = PulsarClient.builder()
  2.     .serviceUrl("pulsar://localhost:6650")
  3.     .authentication(new MyAuthentication())
  4.     .build();

You can use 2 interfaces to implement on the client side:

This in turn needs to provide the client credentials in the form of org.apache.pulsar.client.api.AuthenticationDataProvider. This leaves the chance to return different kinds of authentication token for different types of connection or by passing a certificate chain to use for TLS.

You can find examples for client authentication providers at:

Broker authentication plugin

On broker side, you need the corresponding plugin to validate the credentials that the client passes. Broker can support multiple authentication providers at the same time.

In conf/broker.conf you can choose to specify a list of valid providers:

  1. # Autentication provider name list, which is comma separated list of class names
  2. authenticationProviders=

To implement org.apache.pulsar.broker.authentication.AuthenticationProvider on one single interface:

  1. /**
  2.  * Provider of authentication mechanism
  3.  */
  4. public interface AuthenticationProvider extends Closeable {
  6.     /**
  7.      * Perform initialization for the authentication provider
  8.      *
  9.      * @param config
  10.      *            broker config object
  11.      * @throws IOException
  12.      *             if the initialization fails
  13.      */
  14.     void initialize(ServiceConfiguration config) throws IOException;
  16.     /**
  17.      * @return the authentication method name supported by this provider
  18.      */
  19.     String getAuthMethodName();
  21.     /**
  22.      * Validate the authentication for the given credentials with the specified authentication data
  23.      *
  24.      * @param authData
  25.      *            provider specific authentication data
  26.      * @return the "role" string for the authenticated connection, if the authentication was successful
  27.      * @throws AuthenticationException
  28.      *             if the credentials are not valid
  29.      */
  30.     String authenticate(AuthenticationDataSource authData) throws AuthenticationException;
  32. }

The following is the example for Broker authentication plugins:

Authorization

Authorization is the operation that checks whether a particular “role” or “principal” has a permission to perform a certain operation.

By default, Pulsar provides an embedded authorization, though configuring a different one through a plugin is also an alternative choice.

To provide a custom provider, you need to implement the org.apache.pulsar.broker.authorization.AuthorizationProvider interface, put this class in the Pulsar broker classpath and configure the class in conf/broker.conf:

  1. # Authorization provider fully qualified class-name
  2. authorizationProvider=org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider
  1. /**
  2.  * Provider of authorization mechanism
  3.  */
  4. public interface AuthorizationProvider extends Closeable {
  6.     /**
  7.      * Perform initialization for the authorization provider
  8.      *
  9.      * @param config
  10.      *            broker config object
  11.      * @param configCache
  12.      *            pulsar zk configuration cache service
  13.      * @throws IOException
  14.      *             if the initialization fails
  15.      */
  16.     void initialize(ServiceConfiguration conf, ConfigurationCacheService configCache) throws IOException;
  18.     /**
  19.      * Check if the specified role has permission to send messages to the specified fully qualified topic name.
  20.      *
  21.      * @param topicName
  22.      *            the fully qualified topic name associated with the topic.
  23.      * @param role
  24.      *            the app id used to send messages to the topic.
  25.      */
  26.     CompletableFuture<Boolean> canProduceAsync(TopicName topicName, String role,
  27.             AuthenticationDataSource authenticationData);
  29.     /**
  30.      * Check if the specified role has permission to receive messages from the specified fully qualified topic name.
  31.      *
  32.      * @param topicName
  33.      *            the fully qualified topic name associated with the topic.
  34.      * @param role
  35.      *            the app id used to receive messages from the topic.
  36.      * @param subscription
  37.      *            the subscription name defined by the client
  38.      */
  39.     CompletableFuture<Boolean> canConsumeAsync(TopicName topicName, String role,
  40.             AuthenticationDataSource authenticationData, String subscription);
  42.     /**
  43.      * Check whether the specified role can perform a lookup for the specified topic.
  44.      *
  45.      * For that the caller needs to have producer or consumer permission.
  46.      *
  47.      * @param topicName
  48.      * @param role
  49.      * @return
  50.      * @throws Exception
  51.      */
  52.     CompletableFuture<Boolean> canLookupAsync(TopicName topicName, String role,
  53.             AuthenticationDataSource authenticationData);
  55.     /**
  56.      *
  57.      * Grant authorization-action permission on a namespace to the given client
  58.      *
  59.      * @param namespace
  60.      * @param actions
  61.      * @param role
  62.      * @param authDataJson
  63.      *            additional authdata in json format
  64.      * @return CompletableFuture
  65.      * @completesWith <br/>
  66.      *                IllegalArgumentException when namespace not found<br/>
  67.      *                IllegalStateException when failed to grant permission
  68.      */
  69.     CompletableFuture<Void> grantPermissionAsync(NamespaceName namespace, Set<AuthAction> actions, String role,
  70.             String authDataJson);
  72.     /**
  73.      * Grant authorization-action permission on a topic to the given client
  74.      *
  75.      * @param topicName
  76.      * @param role
  77.      * @param authDataJson
  78.      *            additional authdata in json format
  79.      * @return CompletableFuture
  80.      * @completesWith <br/>
  81.      *                IllegalArgumentException when namespace not found<br/>
  82.      *                IllegalStateException when failed to grant permission
  83.      */
  84.     CompletableFuture<Void> grantPermissionAsync(TopicName topicName, Set<AuthAction> actions, String role,
  85.             String authDataJson);
  87. }