[toc] 使用者認證功能，是一個成熟元件不可或缺的功能。在0.9版本以前kafka是沒有使用者認證模組的（或者說只有SSL），好在kafka0.9版本以後逐漸釋出了多種使用者認證功能，彌補了這一缺陷（這裡僅介紹SASL）。 本篇會先介紹當前kafka的四種認證方式，然後過一遍部署SASL/PLAIN認證功能的流程。最後再介紹如如何使用kafka2.x新推出的callback api，對SASL/PLAIN功能進行二次開發。 # kafka 2.x使用者認證方式小結 需要先明確的一點是，使用者認證和許可權控制是兩碼事。**使用者認證是確認這個使用者能否訪問當前的系統，而許可權控制是控制使用者對當前系統中各種資源的訪問許可權。**使用者認證就是今天要講的內容，而kafka的許可權控制，則是對應`bin/kafka-acls.sh`工具所提供的一系列功能，這裡不詳細展開。 標題特地說明kafka2.x是因為kafka2.0的時候推出一種新的使用者認證方式，SASL/OAUTHBEARER，在此前的版本是不存在這個東西的。那麼加上這個之後，kafka目前共有4種常見的認證方式。 - SASL/GSSAPI（kerberos）：kafka0.9版本推出，即藉助kerberos實現使用者認證，如果公司恰好有kerberos環境，那麼用這個是比較合適的。 - SASL/PLAIN：kafka0.10推出，非常簡單，簡單得有些雞肋，不建議生產環境使用，除非對這個功能二次開發，這也是我後面要講的。 - SASL/SCRAM：kafka0.10推出，全名Salted Challenge Response Authentication Mechanism，為解決SASL/PLAIN的不足而生，缺點可能是某些客戶端並不支援這種方式認證登陸（使用比較複雜）。 - SASL/OAUTHBEARER：kafka2.0推出，實現較為複雜，目前業內應該較少實踐。 其實除了上述四種使用者認證功能之外，還有一個叫Delegation Token的東西。這個東西說一個輕量級的工具，是對現有SASL的一個補充，能夠提高使用者認證的效能（主要針對Kerberos的認證方式）。算是比較高階的用法，一般也用不到，所以也不會多介紹，有興趣可以看這裡[Authentication using Delegation Tokens](https://kafka.apache.org/documentation/#security_delegation_token)。 **SASL/GSSAPI** 如果已經有kerberos的環境，那麼會比較適合使用這種方式，只需要讓管理員分配好principal和對應的keytab，然後在配置中新增對應的選項就可以了。需要注意的是，一般採用這種方案的話，zookeeper也需要配置kerberos認證。 **SASL/PLAIN** 這種方式其實就是一個使用者名稱/密碼的認證方式，不過它有很多缺陷，比如使用者名稱密碼是儲存在檔案中，不能動態新增，明文等等！這些特性決定了它比較雞肋，但好處是足夠簡單，這使得我們可以方便地對它進行二次開發。本篇文章後續會介紹SASL/PLAIN的部署方式和二次開發的例子（基於kafka2.x）。 **SASL/SCRAM** 針對PLAIN方式的不足而提供的另一種認證方式。這種方式的使用者名稱/密碼是儲存中zookeeper的，因此能夠支援動態新增使用者。該種認證方式還會使用sha256或sha512對密碼加密，安全性相對會高一些。 而且配置起來和SASL/PLAIN差不多同樣簡單，新增使用者/密碼的命令官網也有提供，個人比較推薦使用這種方式。不過有些客戶端是不支援這個方式認證登陸的，比如python的kafka客戶端，這點需要提前調研好。 具體的部署方法官網或網上有很多，這裡不多介紹，貼下官網的[Authentication using SASL/SCRAM](https://kafka.apache.org/documentation/#security_sasl_scram)。 **SASL/OAUTHBEARER** SASL/OAUTHBEARER是基於OAUTH2.0的一個新的認證框架，這裡先說下什麼是OAUTH吧，引用維基百科。 > OAuth是一個開放標準，允許使用者讓第三方應用訪問該使用者在某一網站上儲存的私密的資源（如照片，視訊，聯絡人列表），而無需將使用者名稱和密碼提供給第三方應用。而 OAUTH2.0算是OAUTH的一個加強版。 說白了，SASL/OAUTHBEARER就是一套讓使用者使用第三方認證工具認證的標準，通常是需要自己實現一些token認證和建立的介面，所以會比較繁瑣。 詳情可以通過這個kip瞭解[KIP-255](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=75968876) 說了這麼多，接下來就說實戰了，先介紹下如何配置SASL/PLAIN。 # SASL/PLAIN例項（配置及客戶端） ### broker配置 **kafka_server_jaas.conf** 這裡簡單介紹下SASL/PLAIN的部署方式，另外除了SASL/OAUTHBEARER，其他幾種應該也是類似的部署方式，基本都是大同小異。 PS：本配置版本適用於kafka2.x，且無需配置zk認證 kafka的使用者認證，是基於java的jaas。所以我們需要先新增jaas服務端的配置檔案。在`kafka_home/config/kafka_server_jaas.conf`中新增以下配置資訊： ``` KafkaServer { org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin-secret" user_admin="admin-secret" user_alice="alice-secret"; }; ``` 注意最後一個屬性後面需要加封號！配置是不難理解的，第一行指定`PlainLoginModule`，算是宣告這是一個SASL/PLAIN的認證型別，如果是其他的，那麼就需要reqired其他的類。`username`和`password`則是用於叢集內部broker的認證用的。 這裡會讓人疑惑的，應該是`user_admin`和`user_alice`這兩個屬性了。這個其實是用來定義使用者名稱和密碼的，形式是這樣：user_userName=password。所以這裡其實是定義了使用者admin和使用者alice到密碼。 這一點可以在原始碼的`PlainServerCallbackHandler`類中找到對應的資訊，kafka原始碼中顯示，對使用者認證的時候，就會到jaas配置檔案中，通過user_username屬性獲取對應username使用者的密碼，再進行校驗。當然這樣也導致了該配置檔案只有重啟才會生效，即無法動態新增使用者。 說回來，寫完配置後，需要在kafka的配置中新增jaas檔案的路徑。在`kafka_home/bin/kafka-run-class.sh`中，找到下面的配置，修改KAFKA_OPTS到配置資訊。 ``` # Generic jvm settings you want to add if [ -z "$KAFKA_OPTS" ]; then KAFKA_OPTS="" fi ``` 將上述到KAFKA_OPTS修改為 > KAFKA_OPTS="-Djava.security.auth.login.config=kafka_home/config/kafka_server_jaas.conf" **server.properties** 然後修改`kafka_home/config/server.properties`： ``` listeners=SASL_PLAINTEXT://host.name:port security.inter.broker.protocol=SASL_PLAINTEXT sasl.mechanism.inter.broker.protocol=PLAIN sasl.enabled.mechanisms=PLAIN ``` 其中*SASL_PLAINTEXT*的意思，是明文傳輸的意思，如果是SSL，那麼應該是SASL_SSL。 這樣就算是配置好kafka broker了，接下來啟動kafka，觀察輸出日誌，沒有錯誤一般就沒問題了。 ### 客戶端配置 以producer為例，只需要在`kafka_home/config/producer.properties`中新增jaas認證資訊，以及使用者名稱密碼： ``` sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \ username="alice" \ password="alice-secret"; security.protocol=SASL_SSL sasl.mechanism=PLAIN ``` 然後使用console producer驗證： ``` bin/kafka-console-producer.sh --broker-list kafka:9092 --topic test --producer.config config/producer.properties ``` 一般能夠傳送資料就說明部署完成了～ # 自定義SASL/PLAIN認證（二次開發） 前面小節介紹了kafka sasl_plain的部署方式，但這種方式的諸多弊病決定了它並不適合用於生產環境。這裡我們先介紹kafka2的新認證介面，然後演示下如何使用新的api自定義。 ### kafka2新的callback介面介紹 這一api提出的背景，是因為最開始的api（即`SaslServer`），不方便對使用者認證進行拓展。這個問題在開發SASL/SCRAM功能的時候尤其突出。按官方的說法，要新增SASL/SCRAM功能，需要重寫`SaslServer`類。 所以官方重寫了這方面的功能，使用回撥的方式實現了這部分的功能模組。使得開發者可以方便得對使用者認證模組進行拓展或修改。 並且新增加了四個自定義認證的配置，分別是： - sasl客戶端類：sasl.client.callback.handler.class - sasl服務端類：sasl.server.callback.handler.class - login類：sasl.login.class - login回撥類：sasl.login.callback.handler.class 這幾個配置預設都是null，需要填寫的內容是自定義的類的路徑+名稱。我們這次只需要關注sasl服務端類的配置，即`sasl.server.callback.handler.class`。 這部分的內容具體是在[KIP-86](https://cwiki.apache.org/confluence/display/KAFKA/KIP-86%3A+Configurable+SASL+callback+handlers#KIP86:ConfigurableSASLcallbackhandlers-SASL/SCRAMServerCallbacks)。 ### 自定義sasl/plain功能 先詳細介紹下`sasl.server.callback.handler.class`配置。這個配置在使用的時候，需要以小寫方式指定SASL的型別。舉個例子，如果是SASL_PLAINTEXT，那麼就需要這樣： > listener.name.sasl_plaintext.plain.sasl.server.callback.handler.class=com.example.CustomPlainCallbackHandler 即以**listener.name.sasl_plaintext.plain.sasl**開頭。然後在kafka中，SASL_PLAINTEXT預設實現的callback handler是`PlainServerCallbackHandler`，實現了`AuthenticateCallbackHandler`介面。這個的邏輯其實還蠻簡單的，我們可以看看它重點的方法和程式碼。 ``` public class PlainServerCallbackHandler implements AuthenticateCallbackHandler { private static final String JAAS_USER_PREFIX = "user_"; //jaas配置資訊，初始化一次，這就是為什麼plain無法新增使用者 private List jaasConfigEntries; @Override public void configure(Map configs, String mechanism, List jaasConfigEntries) { this.jaasConfigEntries = jaasConfigEntries; } //核心類，獲取使用者密碼後，呼叫authenticate方法 @Override public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException { String username = null; for (Callback callback: callbacks) { if (callback instanceof NameCallback) username = ((NameCallback) callback).getDefaultName(); else if (callback instanceof PlainAuthenticateCallback) { PlainAuthenticateCallback plainCallback = (PlainAuthenticateCallback) callback; boolean authenticated = authenticate(username, plainCallback.password()); plainCallback.authenticated(authenticated); } else throw new UnsupportedCallbackException(callback); } } //使用者密碼是通過獲取jaas檔案的屬性，屬性名就是JAAS_USER_PREFIX變數當前綴+username protected boolean authenticate(String username, char[] password) throws IOException { if (username == null) return false; else { String expectedPassword = JaasContext.configEntryOption(jaasConfigEntries, JAAS_USER_PREFIX + username, PlainLoginModule.class.getName()); return expectedPassword != null && Arrays.equals(password, expectedPassword.toCharArray()); } } @Override public void close() throws KafkaException { } } ``` 前面說plain方式不支援動態新增使用者，user_username驗證密碼，看程式碼就一清二楚。既然知道這個後，那要自定義校驗邏輯就很簡單了。 只需要繼承`PlainServerCallbackHandler`這個類，然後重寫`authenticate`方法實現自己的邏輯就實現自定義了。 比如我想讓使用者名稱和密碼相同的就驗證通過，那麼可以這樣： ``` public class MyPlainServerCallbackHandler extends PlainServerCallbackHandler{ @Override protected boolean authenticate(String username, char[] password) throws IOException { if (username == null) return false; else { return expectedPassword != null && Arrays.equals(password, username.toCharArray()); } } } ``` 然後中server.properpose中新增server callback資訊，就可以了 ``` listener.name.sasl_plaintext.plain.sasl.server.callback.handler.class=com.example.MyPlainServerCallbackHandler ``` **對了，幾得重新編譯打包，替換掉kafka-client掉jar包**，如果修改了一些全域性資訊（比如build.gradle引入新的依賴），那最好kafka全套jar包都換一下。 以上，如果覺得有用，不妨點個