Apache Pulsar 安全扩展指南：自定义认证与授权插件开发实战

消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载导读Pulsar 内置了 mutual TLS、Athenz、Token 等多种认证插件与一套嵌入式授权实现但真实的生产环境中企业往往需要对接自有的身份系统如统一的 OAuth、内部 CA 或自研权限中心。本文以 Apache Pulsar 官方文档 security-extending.md 为骨架结合当前仓库源码完整讲解如何通过实现插件接口来扩展 Pulsar 的认证Authentication与授权Authorization机制客户端侧需要实现什么接口、Proxy/Broker 侧如何配置与校验、授权插件又该覆盖哪些判定方法。读完本文你将能够从零开发一套可插拔的 Pulsar 安全插件并理解其与 Broker 配置、内置实现的对应关系。一、扩展机制总览两套插件体系Pulsar 的安全体系分为两层分别对应两个独立的扩展点认证Authentication回答你是谁。Pulsar 通过两个插件协同完成认证——一个插件运行在Client 库负责携带并提交凭据另一个插件运行在Pulsar Proxy / Pulsar Broker负责校验凭据并产出 role/principal。授权Authorization回答你能做什么。它检查某个 role 是否有权限执行生产、消费、lookup、grant 等操作。文档明确强调了一个关键设计差异Authentication 插件同时适用于 Proxy 与 Broker而 Authorization 插件只在 Broker 上生效当开启授权时Proxy 仅会对 role 做一些简单的授权检查。这一点在配置与架构选型时需要特别注意。二、扩展认证客户端认证插件2.1 需要实现的两个客户端接口在客户端侧认证逻辑由两个接口共同组成定义于 pulsar-client-api/src/main/java/org/apache/pulsar/client/api/org.apache.pulsar.client.api.Authentication认证插件的主入口负责描述认证方式、初始化并提供认证数据。它继承自Closeable, Serializable核心方法包括getAuthMethodName()返回该认证方法的标识如tls、tokengetAuthData()/getAuthData(String brokerHostName)获取要发送给 Broker 的认证数据后者支持针对不同目标 Broker 返回不同凭据例如多集群场景configure(MapString, String authParams)以参数键值对配置插件start()初始化插件。源码中还提供了authenticationStage、newRequestHeader等默认方法用于 HTTP 类认证如基于 header 的挑战-响应流程。org.apache.pulsar.client.api.AuthenticationDataProvider提供具体凭据数据的载体其默认方法全部返回null/false子类按需覆写。接口中按协议类型分组TLS 相关hasDataForTls()、getTlsCertificates()、getTlsPrivateKey()、getTlsCerificateFilePath()、getTlsPrivateKeyFilePath()、getTlsTrustStoreStream()、getTlsKeyStoreParams()HTTP 相关hasDataForHttp()、getHttpAuthType()、getHttpHeaders()Pulsar 协议Command相关hasDataFromCommand()、getCommandData()、以及用于 SASL 等双向认证的authenticate(AuthData data)。正是由于AuthenticationDataProvider是一个独立的凭据容器插件可以为不同类型的连接TLS 连接、HTTP 管理接口、二进制协议返回不同的认证凭据——例如 TLS 连接返回证书链HTTP 请求则返回相应的认证 header。2.2 在创建 Client 时挂载认证插件实现Authentication后在创建 PulsarClient 时通过authentication(...)传入实例即可完整用法见原文档中的示例PulsarClient client PulsarClient.builder() .serviceUrl(pulsar://localhost:6650) .authentication(new MyAuthentication()) .build();2.3 仓库内可参考的客户端认证实现当前仓库在 pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/ 下提供了大量可直接参照的内置实现AuthenticationTls.javaAuthenticationDataTls.javamutual TLS 认证通过getTlsCertificates()/getTlsPrivateKey()提供证书链AuthenticationToken.javaAuthenticationDataToken.javaToken 认证凭据通过getCommandData()/getHttpHeaders()携带AuthenticationBasic.javaAuthenticationDataBasic.javaBasic 认证AuthenticationKeyStoreTls.javaAuthenticationDataKeyStoreTls.java基于 KeyStore 的 TLS 认证AuthenticationDisabled.java、AuthenticationDataNull.java空认证占位实现。Athenz 认证的客户端实现位于 pulsar-client-auth-athenz/src/main/java/org/apache/pulsar/client/impl/auth。开发自有插件时直接对照这些类覆写相应方法即可无需从零摸索。三、扩展认证Proxy / Broker 侧认证插件3.1 配置认证提供者列表客户端发送凭据后需要由 Proxy/Broker 侧的插件完成校验。Broker 与 Proxy 可同时挂载多个认证提供者通过conf/broker.confProxy 对应conf/proxy.conf中以逗号分隔的类名列表指定# Authentication provider name list, which is comma separated list of class names authenticationProviders配套开关见 conf/broker.conf# Enable authentication authenticationEnabledfalse # Interval of time for checking for expired authentication credentials authenticationRefreshCheckSeconds60 # Role names that are treated as super-user, meaning they will be able to do all admin # operations and publish/consume from all topics superUserRoles # Authentication settings of the broker itself. Used when the broker connects to other brokers, # either in same or other clusters brokerClientAuthenticationPlugin brokerClientAuthenticationParameters注意开启认证前必须先将authenticationEnabled置为truesuperUserRoles用于声明拥有全部权限的超级用户角色列表。3.2 实现AuthenticationProvider接口Broker/Proxy 侧插件只需实现单接口org.apache.pulsar.broker.authentication.AuthenticationProvider源码位于 pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProvider.java核心方法如下/** * Provider of authentication mechanism */ public interface AuthenticationProvider extends Closeable { /** * Perform initialization for the authentication provider * * param config broker config object * throws IOException if the initialization fails */ void initialize(ServiceConfiguration config) throws IOException; /** * return the authentication method name supported by this provider */ String getAuthMethodName(); /** * Validate the authentication for the given credentials with the specified authentication data * * param authData provider specific authentication data * return the role string for the authenticated connection, if the authentication was successful * throws AuthenticationException if the credentials are not valid */ String authenticate(AuthenticationDataSource authData) throws AuthenticationException; }authenticate(...)的返回值就是该连接对应的role 字符串后续授权阶段会基于这个 role 做权限判定凭据非法时抛出AuthenticationException。此外该接口在仓库中的实现还提供了若干可覆写的默认方法用于更复杂的场景newAuthState(AuthData, SocketAddress, SSLSession)创建认证状态机默认返回OneStageAuthenticationState用于单阶段认证SASL 等多阶段双向认证可自行实现AuthenticationStateauthenticateHttpRequest(HttpServletRequest, HttpServletResponse)处理 HTTP 管理接口/WebSocket 等 HTTP 类请求的认证返回是否继续执行后续 filter 链incrementFailureMetric(Enum?)向AuthenticationMetrics上报认证失败指标。3.3 仓库内可参考的 Broker 侧实现Mutual TLSAuthenticationProviderTlspulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderTls.javaAthenzAuthenticationProviderAthenzpulsar-broker-auth-athenz/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderAthenz.javaToken / BasicAuthenticationProviderToken、AuthenticationProviderBasic同样位于pulsar-broker-common的 authentication 包下。同目录下的AuthenticationProviderList.java、AuthenticationService.java展示了 Broker 如何按配置加载并串联多个认证提供者可作为理解多认证提供者并存机制的入口。四、扩展授权实现AuthorizationProvider4.1 授权与认证的分工授权检查的是某个 role/principal 是否有权执行特定操作。Pulsar 默认内置嵌入式授权实现PulsarAuthorizationProvider同时允许通过插件替换为自有实现。文档特别提示Authentication 插件同时用于 Proxy 与 Broker而 Authorization 插件仅设计在 Broker 上使用Proxy 只在开启授权时对 role 做简单检查。4.2 配置自定义授权提供者将实现类放入 Broker classpath并在conf/broker.conf中替换authorizationProvider配置项# Authorization provider fully qualified class-name authorizationProviderorg.apache.pulsar.broker.authorization.PulsarAuthorizationProvider同时需要开启授权开关# Enforce authorization authorizationEnabledfalse # Allow wildcard matching in authorization # (wildcard matching only applicable if wildcard-char: # * presents at first or last position eg: *.pulsar.service, pulsar.service.*) authorizationAllowWildcardsMatchingfalse4.3AuthorizationProvider接口全解析接口定义位于 pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java。原文档给出了该接口的核心方法骨架仓库中的最新版本在此基础上又补充了函数操作、租户/命名空间/主题策略操作等大量默认方法。以下是核心判定方法/** * Provider of authorization mechanism */ public interface AuthorizationProvider extends Closeable { /** * Perform initialization for the authorization provider * * param conf broker config object * param configCache pulsar zk configuration cache service * throws IOException if the initialization fails */ void initialize(ServiceConfiguration conf, ConfigurationCacheService configCache) throws IOException; /** * Check if the specified role has permission to send messages to the specified fully qualified topic name. */ CompletableFutureBoolean canProduceAsync(TopicName topicName, String role, AuthenticationDataSource authenticationData); /** * Check if the specified role has permission to receive messages from the specified fully qualified topic name. */ CompletableFutureBoolean canConsumeAsync(TopicName topicName, String role, AuthenticationDataSource authenticationData, String subscription); /** * Check whether the specified role can perform a lookup for the specified topic. * For that the caller needs to have producer or consumer permission. */ CompletableFutureBoolean canLookupAsync(TopicName topicName, String role, AuthenticationDataSource authenticationData); /** * Grant authorization-action permission on a namespace to the given client. * completesWith IllegalArgumentException when namespace not found * completesWith IllegalStateException when failed to grant permission */ CompletableFutureVoid grantPermissionAsync(NamespaceName namespace, SetAuthAction actions, String role, String authDataJson); /** * Grant authorization-action permission on a topic to the given client. */ CompletableFutureVoid grantPermissionAsync(TopicName topicName, SetAuthAction actions, String role, String authDataJson); }全部方法均返回CompletableFuture即授权判定是异步的便于接入外部权限系统如远程权限服务、数据库查询而不会阻塞 Broker 线程。4.4 默认实现与可扩展的判定方法默认授权提供者PulsarAuthorizationProvider位于 pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java它基于存储在 metadata store如 ZooKeeper中的租户/命名空间/主题权限策略进行判定。从仓库源码看现代版本的AuthorizationProvider还扩展了以下默认方法默认抛出不支持的异常可按需覆写isSuperUser(String role, AuthenticationDataSource, ServiceConfiguration)超级用户判定默认依据superUserRoles配置isTenantAdmin(...)租户管理员判定allowFunctionOpsAsync/allowSourceOpsAsync/allowSinkOpsAsync命名空间内 Function/Source/Sink 操作授权allowTenantOperationAsync/allowNamespaceOperationAsync/allowNamespacePolicyOperationAsync租户、命名空间及策略级操作授权allowTopicOperationAsync/allowTopicPolicyOperationAsync主题及主题策略级操作授权grantSubscriptionPermissionAsync/revokeSubscriptionPermissionAsync订阅级管理权限的授予与撤销。对于自定义实现多数场景只需覆写canProduceAsync、canConsumeAsync、canLookupAsync与grantPermissionAsync等核心方法若需要全面接管 Pulsar 的权限模型则应将上述操作级方法一并实现避免相关请求被默认实现拒绝。五、开发与验证建议对照内置实现起步客户端插件参照 pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/Broker 侧插件参照pulsar-broker-common的 authentication 与 authorization 包复制其骨架后替换为自有认证/判定逻辑。角色贯穿始终认证插件返回的 role 字符串会传递给授权插件的每个判定方法设计时务必保证两套插件对角色语义理解一致。按环境选择配置位置Broker 使用conf/broker.confProxy 使用conf/proxy.conf两类配置文件的字段结构一致。从配置到代码的验证链路可通过authenticationEnabledtrue、authenticationProviders自定义类开启认证后用pulsar-client或官方测试验证authenticate返回值与授权判定结果是否符合预期。结语Pulsar 的认证与授权扩展点非常清晰客户端侧一对接口AuthenticationAuthenticationDataProvider负责携带凭据服务端单接口AuthenticationProvider负责校验凭据授权侧AuthorizationProvider负责基于 role 做异步权限判定。将原文档中的接口骨架与仓库内的AuthenticationProviderTls、AuthenticationProviderAthenz、PulsarAuthorizationProvider等真实实现对照阅读即可快速掌握自定义安全插件的全部要点将 Pulsar 无缝接入企业既有的身份与权限体系。赞分享消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载相关推荐3步快速获取中小学电子课本tchMaterial-parser免费下载工具终极指南3步快速获取中小学电子课本tchMaterial parser免费下载工具终极指南 作为一名教育工作者或家长您是否经常需要在国家中小学智慧教育平台上查找电子网页爬虫教育Apache Pulsar 安全扩展实战自定义 Authentication 与 Authorization 插件开发指南Apache Pulsar 安全扩展实战自定义 Authentication 与 Authorization 插件开发指南 Apache Pulsar 的安全消息队列后端流处理深入 Apache Pulsar 安全扩展自定义 Authentication 与 Authorization 插件开发指南深入 Apache Pulsar 安全扩展自定义 Authentication 与 Authorization 插件开发指南 本文以 Apache Pulsa消息队列后端流处理上一篇掌握Jupyter魔术命令提升Deep Learning with Python效率的7个实用技巧下一篇Webnovel Writer 37 个内置题材模板总览从修仙到规则怪谈全覆盖创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考