Extending Authentication and Authorization in Pulsar

Pulsar 提供了实现自定义认证和授权的机制。

认证

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.

客户端认证插件

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:

这反过来需要以 org.apache.pulsar.client.api.AuthentationDataProvider 的形式提供客户端凭据。 这样就有机会为不同类型的连接返回不同类型的身份验证令牌，或者通过传递证书链来为TLS使用。

你可以找到客户端认证实现的一些例子：

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.

在配置文件conf/broker.conf中，你能够指定一个有效提供者列表：

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

在单个接口实现org.apache.pulsar.broker.authentication.AuthenticationProvider：

  1. /**
  2.  * 认证机制提供者
  3.  */
  4. public interface AuthenticationProvider extends Closeable {
  6.     /**
  7.      * 初始化认证提供者
  8.      *
  9.      * @param config
  10.      *            Broker 配置对象
  11.      * @throws IOException
  12.      *             如果初始化失败，则抛出异常
  13.      */
  14.     void initialize(ServiceConfiguration config) throws IOException;
  16.     /**
  17.      * @return 返回本认证提供者的名称
  18.      */
  19.     String getAuthMethodName();
  21.     /**
  22.      * 使用指定的身份验证数据验证给定凭据
  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. }

以下是 Broker 认证插件的示例：

授权

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.

要提供自定义提供者，你必须实现org.apache.pulsar.broker.authorization.AuthorizationProvider接口。并将该类放到 Pulsar 所用的 classpath中，并将类名配置到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.      * 检查指定角色是否具有从指定的完全限定主题接收消息的权限.
  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.      * 检查指定角色是否可以执行指定主题的查找。
  44.      *
  45.      * 要调用它必须拥有生产者或消费者的权限.
  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.      * 在命名空间级别授予客户端授权操作的权限。
  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.      * 在主题上授予客户端授权操作的权限。
  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. }