文章摘要

本文整理自 2022 年 8 月 Apache Pulsar Meetup 上傅騰題為《Apache Pulsar 企業級安全實踐》的分享。數據安全已經成為企業的一項重要競爭優勢。傅騰針對集群環境分享 Apache Pulsar 的認證和授權實現，講述企業如何構建安全的 Apache Pulsar 集群并打造滿足個性化要求的 Pulsar 認證或授權插件，以滿足不同的企業安全需求。

作者介紹

傅騰，StreamNative 技術支持工程師，有著 9 年電信運營商行業大數據相關經驗，熟悉大數據的平臺架構、建設、安全和維護，對實時數倉、云原生和 AI 具有平臺類產品建設經驗。

多種安全組合

Apache Pulsar 提供了多種安全組合，包括認證（支持全鏈路組件認證）、授權（ACL）、傳輸加密（TLS 和 mTLS）以及端到端加密（僅生產者和消費者可加解密數據），企業可根據實際環境的安全需求來搭配使用。

全鏈路可信

上圖為全鏈路可信場景。該場景通常見于開放集群或內部測試、功能驗證等。該場景中 Broker、Bookie 均開放，生產者明文寫入數據。?

內網可信

上面兩圖皆為內網可信場景。其中，第一種設計會做傳輸層加密和 Broker 認證授權，內網依舊開放。第二種設計將傳輸層加密放在外部與 Proxy 之間，Proxy 也要做認證。

上圖為內網可信場景中的第三種設計，即常見的負載均衡場景，外部到負載均衡器做傳輸層加密，后者將數據交給需要認證的 Proxy，再流向 Broker。

內網不可信

在內網不可信場景中，內部節點可能因為各種原因有安全隱患，這就需要在每一傳輸層都開啟加密和認證，在 Broker 開啟授權。該場景常見于企業跨部門協作、與子公司協作、與外部溝通等情況。

服務不可信（端到端加密）

服務不可信場景一般是指第三方（如云廠商）提供 Pulsar 服務，為此所有鏈路都要端到端加密，且只有生產者和消費者才能獲取原始數據。

易擴展的安全框架

Apache Pulsar 有著簡潔、非常容易擴展的安全框架。企業可方便地自定義安全插件，包括可插拔式的認證和授權、易集成的代碼結構、同時支持多種認證的認證鏈、認證緩存和檢測機制。

對于可插拔式的認證和授權：

• ? Pulsar 服務端（如 Broker、Proxy）會對客戶端做身份認證，并記錄一個 role 作為后續授權的客戶端身份標識。該 role 可視為用戶實際登錄的 token。

  • ??默認支持的認證插件：TLS、Athenz、Kerberos、JWT、OAuth2、Basic。

  • ??支持鏈式認證，同時支持多個認證源。服務端會將所有 Provider 緩存到本地并初始化，根據客戶端傳來的標識類型尋找對應的 Provider 做認證。鏈式認證在云環境中很常見。配置 authenticationProviders：

authenticationProviders=org.apache.pulsar.broker.authentication.AuthenticationProviderToken,org.apache.pulsar.broker.authentication.AuthenticationProviderBasic

• ? Pulsar 支持兩種鑒權插件：

  • ? AuthorizationProvider：

authorizationProvider=org.apache.pulsar.broker.authorization.AuthenticationProviderToken

• ? ?? MultiRolesTokenAuthorizationProvider，僅適用于認證插件是 JWT 的認證方式：

authorizationProvider=org.apache.pulsar.broker.authorization.MultiRolesTokenAuthorizationProvider

認證的觸發時機發生在客戶端建立連接時：

1.客戶端將一個 CommandConnect 發送給 Broker；

2.AuthenticationService 從 CommandConnect 中可以獲取認證方法類型(AuthMethodName）；

3.AuthenticationService 再根據緩存的認證插件，調用 authenticate 方法進行認證。

4.注意：緩存認證憑證和認證的過期檢查（AuthChallenge，默認值 authenticationRefreshCheckSeconds=60s）

Pulsar 的權限體系中，支持的鑒權級別分別是：

• ? Broker，支持 superuser；

• ? Tenant，支持 tenant admin；

• ? Namespace，支持生產、消費和 Functions；

• ? Topics，支持生產和消費。

插件實現有幾點需要注意：initialize() 可以獲取 Broker 級別的配置信息和管理元數據；每次鑒權需要先檢查傳進來的用戶是否是 superuser 和 tenant admin，再檢查具體的權限。還要注意，只有 superuser 或者 tenant admin 才能為用戶賦權。

案例介紹

案例一：使用 JWT 做認證

JSON Web Token（RFC-7519）是 Web 服務中常用的一種認證方案，簡稱 JWT 認證。JWT 基于一個 token 字符串來識別用戶身份。該 token 有著嚴格的三層結構：

• ? Header 層：JSON 串，用于指定簽名算法（Pulsar 中默認使用 HS256），使用 base64url 編碼。

• ? Payload 層：JSON 串，用于指定用戶名標識（Pulsar 中定義為 sub）和過期時間，使用 base64url 編碼。

• ? Signature 層：使用特定算法，將 Header 和 Payload 加一個 secret 值編碼而來。簽名保證了 token 不被中途篡改。

需要注意前兩層是透明的，可以反解出來。

基于 JWT 的特性，在 Pulsar 中使用需要注意幾點：

1.使用 JWT 可以做認證和授權，但數據仍然處于暴露狀態，所以在安全要求較高的環境建議開啟 TLS 對數據傳輸加密（會犧牲一小部分性能），以進一步鞏固安全性。

2.JWT 最重要的特性是不需要借助第三方服務（例如 Kerberos 之于 KDC），憑借 JWT 內容本身就能驗證 token 是否有效。這也帶來一個問題，即 token 一旦簽發，在有效期間將會一直有效，無法撤回。因此對于執行某些重要操作的 token，有效期要盡量設置短。

3.Pulsar 支持兩種 JWT 加密方式，即使用對稱秘鑰（secret key）和使用非對稱密鑰（private/public key），選擇其中一個即可。

4.注意 token 的配置容易出錯（例如需要操作 pulsar-admin 但僅給了 test-user 的 token），建議配置后可使用 validate 命令做驗證。

5.Pulsar Broker 會緩存客戶端的認證信息，并會在一個固定時間（默認 60 秒）來檢查每個連接的認證是否過期，這個刷新時間可以配置 broker.conf 中的 authenticationRefreshCheckSeconds 參數來自定義。

6.注意 token 參數的配置既可使用字符串形式，也可使用文件形式，可擇優選擇。

字符串形式 brokerClientAuthenticationParameters={"token":"eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJ0ZXN0LXVzZXIifQ.9OHgE9ZUDeBTZs7nSMEFIuGNEX18FLR3qvy8mqxSxXw"} 從文件讀取 brokerClientAuthenticationParameters={"file":"///path/to/proxy-token.txt"}

7.使用 bin/pulsar tokens show 命令可以查看 token 的 header 和 payload：

bin/pulsar?tokens?show?-i?eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJ0ZXN0LXVzZXIiLCJleHAiOjE2NTY3NzYwOTh9.awbp6DreQwUyV8UCkYyOGXCFbfo4ZoV-dofXYTnFXO8{"alg":"HS256"} --- {"sub":"test-user","exp":1656776098}

8.使用 bin/pulsar tokens validate 命令可以用 secret key 或者 public key 驗證 token:

bin/pulsar?tokens?validate?-pk??my-public.key?-i?"eyJhbGciOiJSUzI1NiJ9.eyJzdWIiOiJhZG1pbiJ9.ijp-Qw4JDn1aOQbYy4g4YGBbXYIgLA9lCVrnP-heEtPCdDq11_c-9pQdQwc6RdphvlSfoj50qwL5OtmFPysDuF2caSYzSV1kWRWN-tFzrt-04_LRN-vlgb6D06aWubVFJQBC4DyS-INrYqbXETuxpO4PI9lB6lLXo6px-SD5YJzQmcYwi2hmQedEWszlGPDYi_hDG9SeDYmnMpXTtPU3BcjaDcg9fO6PlHdbnLwq2MfByeIj-VS6EVhKUdaG4kU2EJf5uq2591JJAL5HHiuTZRSFD6YbRXuYqQriw4RtnYWSvSeVMMbcL-JzcSJblNbMmIOdiez43MPYFPTB7TMr8g"{sub=admin}

案例二：使用 Kerberos 做認證

Kerberos 是大數據領域常用的認證方案，以簡單易用和穩定性著稱。Pulsar 使用 Java 的 JAAS 機制來支持通過 Kerberos 做身份認證。JAAS 中一個用戶信息對應一個 section 。對于 Kerberos 認證而言，一個用戶信息中最重要的是 principle 和 keytab，現在可以方便地封裝到一個 section 里，最后將這些用戶信息拼裝到 jaas.conf 文件中。

例如：

SectionName?{com.sun.security.auth.module.Krb5LoginModule?requiredusekeyTab=truestorekey=trueuseTicketCache=falsekeyTab="/etc/security/keytabs/pulsarbroker.keytab"principal="broker/localhost@EXAMPLE.COM". };AnotherSectionName{... };

SectionName 是指定了用戶名標識，內部封裝了一個 Kerberos 用戶信息。Pulsar 使用 Kerberos 認證時，從配置上而言，就是告知進程該以哪個 section 的身份來啟動程序。以 broker 為例：

1.在 broker.conf 中指定服務使用的 section 。

2.而 section 信息是在 pulsar_env.sh 中通過指定 jaas.conf 文件來加載獲得的。

使用 Kerberos 做認證時有幾點需要注意：

1.Broker service principal 的 {hostname} 和 advertisedAddress 保持一致。例如 Broker service 使用的 Principal 是 "broker/172.17.0.7@SNIO"，則 advertisedAddress 也必須配置為相同的 IP：172.17.0.7。因為客戶端從 TGT Server 獲取到的 Broker 服務端 Principal 中的 hostname 部分是 advertisedAddress 定義的 IP，可從 KDC 日志（/var/log/krb5kdc.log）查看到問題。Broker 服務會涉及兩個 Principal，一個是 Broker 服務注冊的 Principal，另一個是 TGT Server 返回的 Principal。所以如果兩個 Principal 內部的 IP 不一致，用戶會永遠無法訪問到服務，畢竟用戶一次只能攜帶一個服務端Principal 信息。

2.服務端 Principal 命名規則注意：

Broker 的 Principal 命名格式示例： #正確示范?broker/host1@MY.REALM #錯誤示例?broker@MY.REALM?pulsarbroker/host1@MY.REALM

• ??必須包含完整的三部分：service/hostname@REALM。

• ??service 命名部分，pulsar 里建議使用關鍵字如?broker、proxy?（其他會報 warn）。

• ??主機地址部分為了防止多塊網卡的 IP 配置問題，均建議使用 hostname。但如果沒有 DNS 解析環境，請參考第一條。可使用 IP，保證服務 IP 和 advertisedAddressIP 一致。

• ? REALM 域名建議使用大寫，注意不要復制錯。

3.在配置 FQDN 解析時，Kerberos 所有主機都需要支持 FQDN 解析（少量機器可在 /etc/hosts 配置所有主機 DNS 信息）。

案例三：自定義一個鑒權插件

這里以一個鑒權插件為例介紹可插拔的認證和授權實踐。案例背景是我們需要界面化的鑒權過程，為此選用大數據領域常用的鑒權組件 Ranger 進行界面化權限管理。具體的開發流程是：

1.根據 Pulsar 現有的權限體系，結合 Ranger 的權限結構，將 Pulsar 注冊為 Ranger 的一個服務。

2.在鑒權插件實現 Pulsar 鑒權接口，并在鑒權方法（尤其是初始化方法）中創建 Ranger Client 連接 Ranger 服務做初始化操作。每一個鑒權點都要讀取配置好的策略信息進行鑒權。

編寫 Ranger 服務定義文件時分為幾步：

1.定義資源（包括租戶、命名空間和 topic）

2.定義權限類型（如生產和消費）

Pulsar 中的鑒權接口代碼如下：

@Override public?CompletableFuture<Boolean>?allowTopicOperationAsync(TopicName?topicName,String?role,TopicOperation?operation,AuthenticationDataSourceauthData)?{if?(log.isDebugEnabled?())?{log.debug?("'Check?allowTopicOperationAsync?({}]?on?(t}].",?operation.name?(),?topicName);}return?validateTenantAdminAccess(topicName.getTenant(),role,authData).thenCompose(isSuperUserOrAdmin?->?{if?(log.?isDebugEnabled())?{log.debug?("Verify?if?role?(}?is?allowed?to?(}?to?topic?{}:?isSuperUserOrAdmin=(}",role,?operation,?topicName,?isSuperUserOrAdmin);}if?(isSuperUserOrAdmin){return?CompletableFuture.completedFuture(true);}?else?{switch?(operation)?{case?LOOKUP:case?GET?STATS:case?GET?METADATA:return?canLookupAsync(topicName,?role,?authData);case?PRODUCE:return?canProduceAsync(topicName,?role,?authData);case?GET?SUBSCRIPTIONS:case?CONSUME:case?SUBSCRIBE:case?UNSUBSCRIBE:case?SKIP:case?EXPIRE?MESSAGES:case?PEEK?MESSAGES:case?RESET_CURSOR:case?GET_BACKLOG_SIZE:case?SET_REPLICATED_SUBSCRIPTION_STATUS:case?GET_REPLICATED_SUBSCRIPTION_STATUS:return?canConsumeAsync(topicName,?role,?authData,?authData.getSubscription());case?TERMINATE:case?COMPACT:case?OFFLOAD:case?UNLOAD:case?ADD_BUNDLE_RANGE:case?GET_BUNDLE_RANGE:case?DELETE_BUNDLE_RANGE:return?CompletableFuture.completedFuture(false);default:return?FutureUtil.failedFuture(new?IllegalStateException("TopicOperation?["?+?operation.name()?+?"]?is?not?supported.'))?:}}}); }

Ranger 方法的代碼如下：

@Override public?CompletableFuture<Boolean>?canProduceAsync?(TopicName?topicName,?String?role,AuthenticationDataSource?authenticationData)?{CompletableFuture<Boolean>?future?=?new?CompletableFuture@();RangerAccessResourceImpl?resource?=?new?RangerAccessResourceImpl();resource.setValue(KEY_TENANT, topicName.getTenant())；resource.setValue(KEY_NAMESPACE,?topicName.getNamespacePortion());resource.setValue(KEY_TOPIC,?topicName.getLocalName().split("-partition-")?[0]);//resource.setValue(KEY_TAG,?"*");RangerAccessRequestImplrequest?=?new?RangerAccessRequestImpl();request.setAccessType(AuthAction.produce.name());request.setUser(role);request-setResource(resource);request.setAction?(AuthAction.produce.name?()?)?;try?{RangerAccessResult?result?=?rangerPlugin.?isAccessAllowed?(request);loq-info("request--->{}",?request);log.info("result--->{}",?result);if?(result.getIsAllowed())?{future.?complete?(true);}?else?{String?errMsg?=?String.format?("User?'%s'?doesn't?have?produce?access?to?%s,?matched?policy?id?=?%d",request.getUser(),?topicName.toString(),?result.getPolicyId());log-error(errMsg);future.completeExceptionally?(new?Exception(errMsg));}}?catch?(Exception?e)?{//access?allowed?in?abnormal?situationlog.error("User?(]?encounter?exception?in?(}?produce?authorization?step."request.getUser(),?topicName.?toString(),?e);future.?complete?(true);}return?future; }

最終完成效果如下：

Q & A

Q：JWT 認證讀取本地 token 文件時是否實時？是否需要重啟 Proxy 或 Broker？

A：Client 是被動讀取 token 文件的。客戶端 token 文件變更不影響服務端，即不需要重啟。

Q：生產者是否支持在 HTTP 發送消息過程中在連接時認證？

A：Pulsar 使用 Pulsar 協議發送消息，在第一次連接的時候發起認證。對于管理側如 pulsar-admin 使用 HTTP 協議，而 HTTP 協議每次連接都會發起認證。

Q：對外鑒權時，是否只需配置 Proxy 而不管 Broker？

A：鑒權是由 Broker 來做的，因此需要配置 Broker。

Q：請具體介紹使用 JWT 時 Broker 對客戶端認證信息的緩存策略。

A：有線程專門負責周期性檢測所緩存的客戶端認證是否過期。如果發現過期，就發 auth challenge 命令給客戶端，客戶端接收到 auth challenge 命令后會讀取 token 文件發送給 Broker；如果檢測到更新后的 token 文件，認證就能繼續通過，Broker 端重新緩存認證信息。如果客戶端在下次檢查周期內沒有返回認證數據，就會關閉連接。

▼?關注「Apache Pulsar」，獲取干貨與動態

👇加入?Apache Pulsar?中文交流群?👇

點擊閱讀原文，給 Pulsar 點贊

總結

以上是生活随笔為你收集整理的技术探究｜Apache Pulsar 认证与鉴权实践指南的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。