技术探究｜Apache Pulsar 认证与鉴权实践指南
文章摘要
本文整理自 2022 年 8 月 Apache Pulsar Meetup 上傅騰題為《Apache Pulsar 企業(yè)級安全實踐》的分享。數(shù)據(jù)安全已經(jīng)成為企業(yè)的一項重要競爭優(yōu)勢。傅騰針對集群環(huán)境分享 Apache Pulsar 的認證和授權(quán)實現(xiàn)，講述企業(yè)如何構(gòu)建安全的 Apache Pulsar 集群并打造滿足個性化要求的 Pulsar 認證或授權(quán)插件，以滿足不同的企業(yè)安全需求。
作者介紹
傅騰，StreamNative 技術(shù)支持工程師，有著 9 年電信運營商行業(yè)大數(shù)據(jù)相關(guān)經(jīng)驗，熟悉大數(shù)據(jù)的平臺架構(gòu)、建設(shè)、安全和維護，對實時數(shù)倉、云原生和 AI 具有平臺類產(chǎn)品建設(shè)經(jīng)驗。
多種安全組合
Apache Pulsar 提供了多種安全組合，包括認證（支持全鏈路組件認證）、授權(quán)（ACL）、傳輸加密（TLS 和 mTLS）以及端到端加密（僅生產(chǎn)者和消費者可加解密數(shù)據(jù)），企業(yè)可根據(jù)實際環(huán)境的安全需求來搭配使用。
全鏈路可信
上圖為全鏈路可信場景。該場景通常見于開放集群或內(nèi)部測試、功能驗證等。該場景中 Broker、Bookie 均開放，生產(chǎn)者明文寫入數(shù)據(jù)。?
內(nèi)網(wǎng)可信
上面兩圖皆為內(nèi)網(wǎng)可信場景。其中，第一種設(shè)計會做傳輸層加密和 Broker 認證授權(quán)，內(nèi)網(wǎng)依舊開放。第二種設(shè)計將傳輸層加密放在外部與 Proxy 之間，Proxy 也要做認證。
上圖為內(nèi)網(wǎng)可信場景中的第三種設(shè)計，即常見的負載均衡場景，外部到負載均衡器做傳輸層加密，后者將數(shù)據(jù)交給需要認證的 Proxy，再流向 Broker。
內(nèi)網(wǎng)不可信
在內(nèi)網(wǎng)不可信場景中，內(nèi)部節(jié)點可能因為各種原因有安全隱患，這就需要在每一傳輸層都開啟加密和認證，在 Broker 開啟授權(quán)。該場景常見于企業(yè)跨部門協(xié)作、與子公司協(xié)作、與外部溝通等情況。
服務(wù)不可信（端到端加密）
服務(wù)不可信場景一般是指第三方（如云廠商）提供 Pulsar 服務(wù)，為此所有鏈路都要端到端加密，且只有生產(chǎn)者和消費者才能獲取原始數(shù)據(jù)。
易擴展的安全框架
Apache Pulsar 有著簡潔、非常容易擴展的安全框架。企業(yè)可方便地自定義安全插件，包括可插拔式的認證和授權(quán)、易集成的代碼結(jié)構(gòu)、同時支持多種認證的認證鏈、認證緩存和檢測機制。
對于可插拔式的認證和授權(quán)：
? Pulsar 服務(wù)端（如 Broker、Proxy）會對客戶端做身份認證，并記錄一個 role 作為后續(xù)授權(quán)的客戶端身份標識。該 role 可視為用戶實際登錄的 token。
??默認支持的認證插件：TLS、Athenz、Kerberos、JWT、OAuth2、Basic。
??支持鏈式認證，同時支持多個認證源。服務(wù)端會將所有 Provider 緩存到本地并初始化，根據(jù)客戶端傳來的標識類型尋找對應(yīng)的 Provider 做認證。鏈式認證在云環(huán)境中很常見。配置 authenticationProviders：
? Pulsar 支持兩種鑒權(quán)插件：
? AuthorizationProvider：
? ?? MultiRolesTokenAuthorizationProvider，僅適用于認證插件是 JWT 的認證方式：
認證的觸發(fā)時機發(fā)生在客戶端建立連接時：
1.客戶端將一個 CommandConnect 發(fā)送給 Broker；
2.AuthenticationService 從 CommandConnect 中可以獲取認證方法類型(AuthMethodName）；
3.AuthenticationService 再根據(jù)緩存的認證插件，調(diào)用 authenticate 方法進行認證。
4.注意：緩存認證憑證和認證的過期檢查（AuthChallenge，默認值 authenticationRefreshCheckSeconds=60s）
Pulsar 的權(quán)限體系中，支持的鑒權(quán)級別分別是：
? Broker，支持 superuser；
? Tenant，支持 tenant admin；
? Namespace，支持生產(chǎn)、消費和 Functions；
? Topics，支持生產(chǎn)和消費。
插件實現(xiàn)有幾點需要注意：initialize() 可以獲取 Broker 級別的配置信息和管理元數(shù)據(jù)；每次鑒權(quán)需要先檢查傳進來的用戶是否是 superuser 和 tenant admin，再檢查具體的權(quán)限。還要注意，只有 superuser 或者 tenant admin 才能為用戶賦權(quán)。
案例介紹
案例一：使用 JWT 做認證
JSON Web Token（RFC-7519）是 Web 服務(wù)中常用的一種認證方案，簡稱 JWT 認證。JWT 基于一個 token 字符串來識別用戶身份。該 token 有著嚴格的三層結(jié)構(gòu)：
? Header 層：JSON 串，用于指定簽名算法（Pulsar 中默認使用 HS256），使用 base64url 編碼。
? Payload 層：JSON 串，用于指定用戶名標識（Pulsar 中定義為 sub）和過期時間，使用 base64url 編碼。
? Signature 層：使用特定算法，將 Header 和 Payload 加一個 secret 值編碼而來。簽名保證了 token 不被中途篡改。
需要注意前兩層是透明的，可以反解出來。
基于 JWT 的特性，在 Pulsar 中使用需要注意幾點：
1.使用 JWT 可以做認證和授權(quán)，但數(shù)據(jù)仍然處于暴露狀態(tài)，所以在安全要求較高的環(huán)境建議開啟 TLS 對數(shù)據(jù)傳輸加密（會犧牲一小部分性能），以進一步鞏固安全性。
2.JWT 最重要的特性是不需要借助第三方服務(wù)（例如 Kerberos 之于 KDC），憑借 JWT 內(nèi)容本身就能驗證 token 是否有效。這也帶來一個問題，即 token 一旦簽發(fā)，在有效期間將會一直有效，無法撤回。因此對于執(zhí)行某些重要操作的 token，有效期要盡量設(shè)置短。
3.Pulsar 支持兩種 JWT 加密方式，即使用對稱秘鑰（secret key）和使用非對稱密鑰（private/public key），選擇其中一個即可。
4.注意 token 的配置容易出錯（例如需要操作 pulsar-admin 但僅給了 test-user 的 token），建議配置后可使用 validate 命令做驗證。
5.Pulsar Broker 會緩存客戶端的認證信息，并會在一個固定時間（默認 60 秒）來檢查每個連接的認證是否過期，這個刷新時間可以配置 broker.conf 中的 authenticationRefreshCheckSeconds 參數(shù)來自定義。
6.注意 token 參數(shù)的配置既可使用字符串形式，也可使用文件形式，可擇優(yōu)選擇。
字符串形式 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 是大數(shù)據(jù)領(lǐng)域常用的認證方案，以簡單易用和穩(wěn)定性著稱。Pulsar 使用 Java 的 JAAS 機制來支持通過 Kerberos 做身份認證。JAAS 中一個用戶信息對應(yīng)一個 section 。對于 Kerberos 認證而言，一個用戶信息中最重要的是 principle 和 keytab，現(xiàn)在可以方便地封裝到一個 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 是指定了用戶名標識，內(nèi)部封裝了一個 Kerberos 用戶信息。Pulsar 使用 Kerberos 認證時，從配置上而言，就是告知進程該以哪個 section 的身份來啟動程序。以 broker 為例：
1.在 broker.conf 中指定服務(wù)使用的 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 服務(wù)端 Principal 中的 hostname 部分是 advertisedAddress 定義的 IP，可從 KDC 日志（/var/log/krb5kdc.log）查看到問題。Broker 服務(wù)會涉及兩個 Principal，一個是 Broker 服務(wù)注冊的 Principal，另一個是 TGT Server 返回的 Principal。所以如果兩個 Principal 內(nèi)部的 IP 不一致，用戶會永遠無法訪問到服務(wù)，畢竟用戶一次只能攜帶一個服務(wù)端Principal 信息。
2.服務(wù)端 Principal 命名規(guī)則注意：
Broker 的 Principal 命名格式示例： #正確示范?broker/host1@MY.REALM #錯誤示例?broker@MY.REALM?pulsarbroker/host1@MY.REALM??必須包含完整的三部分：service/hostname@REALM。
??service 命名部分，pulsar 里建議使用關(guān)鍵字如?broker、proxy?（其他會報 warn）。
??主機地址部分為了防止多塊網(wǎng)卡的 IP 配置問題，均建議使用 hostname。但如果沒有 DNS 解析環(huán)境，請參考第一條。可使用 IP，保證服務(wù) IP 和 advertisedAddressIP 一致。
? REALM 域名建議使用大寫，注意不要復制錯。
3.在配置 FQDN 解析時，Kerberos 所有主機都需要支持 FQDN 解析（少量機器可在 /etc/hosts 配置所有主機 DNS 信息）。
案例三：自定義一個鑒權(quán)插件
這里以一個鑒權(quán)插件為例介紹可插拔的認證和授權(quán)實踐。案例背景是我們需要界面化的鑒權(quán)過程，為此選用大數(shù)據(jù)領(lǐng)域常用的鑒權(quán)組件 Ranger 進行界面化權(quán)限管理。具體的開發(fā)流程是：
1.根據(jù) Pulsar 現(xiàn)有的權(quán)限體系，結(jié)合 Ranger 的權(quán)限結(jié)構(gòu)，將 Pulsar 注冊為 Ranger 的一個服務(wù)。
2.在鑒權(quán)插件實現(xiàn) Pulsar 鑒權(quán)接口，并在鑒權(quán)方法（尤其是初始化方法）中創(chuàng)建 Ranger Client 連接 Ranger 服務(wù)做初始化操作。每一個鑒權(quán)點都要讀取配置好的策略信息進行鑒權(quán)。
編寫 Ranger 服務(wù)定義文件時分為幾步：
1.定義資源（包括租戶、命名空間和 topic）
2.定義權(quán)限類型（如生產(chǎn)和消費）
Pulsar 中的鑒權(quán)接口代碼如下：
@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 文件變更不影響服務(wù)端，即不需要重啟。
Q：生產(chǎn)者是否支持在 HTTP 發(fā)送消息過程中在連接時認證？
A：Pulsar 使用 Pulsar 協(xié)議發(fā)送消息，在第一次連接的時候發(fā)起認證。對于管理側(cè)如 pulsar-admin 使用 HTTP 協(xié)議，而 HTTP 協(xié)議每次連接都會發(fā)起認證。
Q：對外鑒權(quán)時，是否只需配置 Proxy 而不管 Broker？
A：鑒權(quán)是由 Broker 來做的，因此需要配置 Broker。
Q：請具體介紹使用 JWT 時 Broker 對客戶端認證信息的緩存策略。
A：有線程專門負責周期性檢測所緩存的客戶端認證是否過期。如果發(fā)現(xiàn)過期，就發(fā) auth challenge 命令給客戶端，客戶端接收到 auth challenge 命令后會讀取 token 文件發(fā)送給 Broker；如果檢測到更新后的 token 文件，認證就能繼續(xù)通過，Broker 端重新緩存認證信息。如果客戶端在下次檢查周期內(nèi)沒有返回認證數(shù)據(jù)，就會關(guān)閉連接。
▼?關(guān)注「Apache Pulsar」，獲取干貨與動態(tài)
👇加入?Apache Pulsar?中文交流群?👇
點擊閱讀原文，給 Pulsar 點贊
總結(jié)
以上是生活随笔為你收集整理的技术探究｜Apache Pulsar 认证与鉴权实践指南的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 小程序源码：2022强大的修复版趣味心理
- 下一篇： python 变量大小，进程和内存信息