一、簡介
Kafka作為一個分布式的發(fā)布-訂閱消息系統(tǒng)，在日常項目中被頻繁使用，通常情況下無論是生產(chǎn)者還是消費者只要訂閱Topic后，即可進行消息的發(fā)送和接收。而kafka在0.9.0.0版本后添加了身份認證和權(quán)限控制兩種安全服務(wù)，本文主要介紹在實際項目使用過程中遇到第三方kafka需身份認證時如何解決，以及對可能會碰到的問題進行總結(jié)。
二、原理介紹
Kafka身份認證主要分為以下幾種：
（1）客戶端與broker之間的連接認證
（2）broker與broker之間的連接認證
（3）broker與zookeeper之間的連接認證
日常項目中，無論是生產(chǎn)者還是消費者，我們都是作為客戶端與kafka進行交互，因此使用的最多的是客戶端與broker之間的連接認證。圖1是客戶端與服務(wù)端broker之間的認證過程圖，客戶端提交認證數(shù)據(jù)，服務(wù)端會根據(jù)認證數(shù)據(jù)對當(dāng)前客戶端進行身份校驗，校驗成功后的客戶端即可成功登錄kafka，進行后續(xù)操作。
圖1 客戶端與broker之間認證過程圖
目前Kafka提供了SASL、SSL、Delegation Tokem三種安全認證機制，而SASL認證又分為了以下幾種方式：
（1）基于Kerberos的GSSAPI
SASL-GSSAPI提供了一種非常安全的身份驗證方法，但使用前提是企業(yè)中有Kerberos基礎(chǔ)，一般使用隨機密碼的keytab認證方式，密碼是加密的，在0.9版本中引入，目前是企業(yè)中使用最多的認證方式。
（2）SASL-PLAIN
SASL-PLAIN方式是一個經(jīng)典的用戶名/密碼的認證方式，其中用戶名和密碼是以明文形式保存在服務(wù)端的JAAS配置文件中的，當(dāng)客戶端使用PLAIN模式進行認證時，密碼是明文傳輸?shù)?，因此安全性較低，但好處是足夠簡單，方便我們對其進行二次開發(fā)，在0.10版本引入。
（3）SASL-SCRAM
SASL-SCRAM是針對SASL-PLAIN方式的不足而提供的另一種認證方式，它將用戶名/密碼存儲在zookeeper中，并且可以通過腳本動態(tài)增減用戶，當(dāng)客戶端使用SCRAM模式進行認證時，密碼會經(jīng)過SHA-256或SHA-512哈希加密后傳輸?shù)椒?wù)器，因此安全性較高，在0.10.2版本中引入。
對Kafka集群來說，要想實現(xiàn)完整的安全模式，首先為集群中的每臺機器生成密鑰和證書是第一步，其次利用SASL對客戶端進行身份驗證是第二步，最后對不同客戶端進行讀寫操作的授權(quán)是第三步，這些步驟即可以單獨運作也可以同時運作，從而提高kafka集群的安全性。
三、具體實現(xiàn)
本文主要介紹作為kafka生產(chǎn)者，如何基于Kerberos進行身份認證給第三方kafka發(fā)送數(shù)據(jù)。
Kerberos主要由三個部分組成：密鑰分發(fā)中心Key Distribution Center（即KDC）、客戶端Client、服務(wù)端Service，大致關(guān)系圖如下圖2所示，其中KDC是實現(xiàn)身份認證的核心組件，其包含三個部分：
- Kerberos Database：儲存用戶密碼以及其他信息
- Authentication Service(AS)：進行用戶身份信息驗證，為客戶端提供Ticket Granting Tickets(TGT)
- Ticket Granting Service(TGS)：驗證TGT，為客戶端提供Service Tickets
我們作為生產(chǎn)者向第三方kafka發(fā)送數(shù)據(jù)，因此需要第三方提供以下安全認證文件：
- 用戶名principle：標識客戶端的用戶身份，也即用于登錄的用戶名
- 指定用戶名對應(yīng)的秘鑰文件xx.keytab：存儲了用戶的加密密碼
- 指定安全認證的服務(wù)配置文件krb5.conf：客戶端根據(jù)該文件中的信息去訪問KDC
獲取以上安全認證文件后，即可編寫java代碼連接第三方kafka，步驟如下：
1、將安全認證文件xx.keytab和krb5.conf放置于某一路徑下，確保后續(xù)java代碼可進行讀取
2、添加kafka配置文件，開啟安全模式認證，其中kerberos.path是第一步中認證文件所在的目錄
3、修改Kafka生產(chǎn)者配置，開啟安全連接
4、調(diào)用認證工具類進行登錄認證
LoginUtil認證工具類的核心是根據(jù)第一步中提供的安全認證文件自動生成jaas配置文件，該文件是kafka安全模式下認證的核心。代碼如下：
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
/**
* @ProjectName: stdp-security-demo
* @Package:
* @ClassName: LoginUtil
* @Author: stdp
* @Description: ${description}
*/
public class LoginUtil {
public enum Module {
KAFKA("KafkaClient"), ZOOKEEPER("Client");
private String name;
Module(String name) {
this.name = name;
}
public String getName() {
return name;
}
}
private static final Logger LOGGER = LoggerFactory.getLogger(LoginUtil.class);
/**
* line operator string
*/
private static final String LINE_SEPARATOR = System.getProperty("line.separator");
/**
* jaas file postfix
*/
private static final String JAAS_POSTFIX = ".jaas.conf";
private static final String JAVA_SECURITY_KRB5_CONF_KEY = "java.security.krb5.conf";
public static final String JAVA_SECURITY_LOGIN_CONF_KEY = "java.security.auth.login.config";
private static final String ZOOKEEPER_SERVER_PRINCIPAL_KEY = "zookeeper.server.principal";
private static final boolean IS_IBM_JDK = System.getProperty("java.vendor").contains("IBM");
/**
* oracle jdk login module
*/
private static final String SUN_LOGIN_MODULE = "com.sun.security.auth.module.Krb5LoginModule required";
public synchronized static void login(String userPrincipal, String userKeytabPath, String krb5ConfPath)
throws IOException
{
// 1.check input parameters
if ((userPrincipal == null) || (userPrincipal.length() <= 0))
{
LOGGER.error("input userPrincipal is invalid.");
throw new IOException("input userPrincipal is invalid.");
}
if ((userKeytabPath == null) || (userKeytabPath.length() <= 0))
{
LOGGER.error("input userKeytabPath is invalid.");
throw new IOException("input userKeytabPath is invalid.");
}
if ((krb5ConfPath == null) || (krb5ConfPath.length() <= 0))
{
LOGGER.error("input krb5ConfPath is invalid.");
throw new IOException("input krb5ConfPath is invalid.");
}
// 2.check file exsits
File userKeytabFile = new File(userKeytabPath);
if (!userKeytabFile.exists())
{
LOGGER.error("userKeytabFile(" + userKeytabFile.getAbsolutePath() + ") does not exsit.");
throw new IOException("userKeytabFile(" + userKeytabFile.getAbsolutePath() + ") does not exsit.");
}
if (!userKeytabFile.isFile())
{
LOGGER.error("userKeytabFile(" + userKeytabFile.getAbsolutePath() + ") is not a file.");
throw new IOException("userKeytabFile(" + userKeytabFile.getAbsolutePath() + ") is not a file.");
}
File krb5ConfFile = new File(krb5ConfPath);
if (!krb5ConfFile.exists())
{
LOGGER.error("krb5ConfFile(" + krb5ConfFile.getAbsolutePath() + ") does not exsit.");
throw new IOException("krb5ConfFile(" + krb5ConfFile.getAbsolutePath() + ") does not exsit.");
}
if (!krb5ConfFile.isFile())
{
LOGGER.error("krb5ConfFile(" + krb5ConfFile.getAbsolutePath() + ") is not a file.");
throw new IOException("krb5ConfFile(" + krb5ConfFile.getAbsolutePath() + ") is not a file.");
}
// 3.set and check krb5config
setKrb5Config(krb5ConfFile.getAbsolutePath());
// LOGGER.info("check zookeeper server Principal =============================================");
setZookeeperServerPrincipal(userPrincipal);
// LOGGER.info("check jaas.conf +++++++++++++++++++++++++++++++++++++++++++++++++");
setJaasFile(userPrincipal,userKeytabPath);
LOGGER.info("Login success!!!!!!!!!!!!!!");
}
public static void setKrb5Config(String krb5ConfigFile) throws IOException {
System.setProperty(JAVA_SECURITY_KRB5_CONF_KEY,krb5ConfigFile);
String ret = System.getProperty(JAVA_SECURITY_KRB5_CONF_KEY);
if (ret == null) {
LOGGER.error(JAVA_SECURITY_KRB5_CONF_KEY + " is null.");
throw new IOException(JAVA_SECURITY_KRB5_CONF_KEY + " is null.");
}
if (!ret.equals(krb5ConfigFile)){
LOGGER.error(JAVA_SECURITY_KRB5_CONF_KEY + " is " + ret + " is not " + krb5ConfigFile + ".");
throw new IOException(JAVA_SECURITY_KRB5_CONF_KEY + " is " + ret + " is not " + krb5ConfigFile + ".");
}
}
public static void setJaasFile(String userPrincipal,String userKeytabPath) throws IOException {
String jaasPath = new File(System.getProperty("java.io.tmpdir")) + File.separator + System.getProperty("user.name") + JAAS_POSTFIX;
LOGGER.info("jaasPath = {}",jaasPath);
//windows路徑下分隔符替換
jaasPath = jaasPath.replace("\\","\\\\");
userKeytabPath = userKeytabPath.replace("\\","\\\\");
//刪除jaas文件
deleteJaasFile(jaasPath);
writeJaasFile(jaasPath,userPrincipal,userKeytabPath);
System.setProperty(JAVA_SECURITY_LOGIN_CONF_KEY,jaasPath);
}
private static void deleteJaasFile(String jaasPath) throws IOException {
File jaasFile = new File(jaasPath);
if (jaasFile.exists()){
if (!jaasFile.delete()){
throw new IOException("failed to delete exists jaas file.");
}
}
}
private static void writeJaasFile(String jaasPath,String userPrincipal,String userKeytabPath) throws IOException {
FileWriter writer = new FileWriter(new File(jaasPath));
try{
writer.write(getJaasConfContext(userPrincipal,userKeytabPath));
writer.flush();
}catch (IOException e){
throw new IOException("Failed to create jaas.conf File.");
}finally {
writer.close();
}
}
private static String getJaasConfContext(String userPrincipal,String userKeytabPath) throws IOException{
Module[] allModule = Module.values();
StringBuffer builder = new StringBuffer();
for (Module module: allModule){
String serviceName = null;
if ("Client".equals(module.getName())){
serviceName = "zookeeper";
}else if ("KafkaClient".equals(module.getName())){
serviceName = "kafka";
}
builder.append(getModuleContext(userPrincipal,userKeytabPath,module,serviceName));
}
return builder.toString();
}
private static String getModuleContext(String userPrincipal,String userKeytabPath,Module module,String serviceName) throws IOException {
StringBuffer builder = new StringBuffer();
if (IS_IBM_JDK){
builder.append(module.getName()).append(" {").append(LINE_SEPARATOR);
builder.append("credsType=both").append(LINE_SEPARATOR);
builder.append("principal=\"" + userPrincipal.trim() + "\"").append(LINE_SEPARATOR);
builder.append("useKeytab=\"" + userKeytabPath + "\"").append(LINE_SEPARATOR);
builder.append("serviceName=\""+serviceName + "\"").append(LINE_SEPARATOR);
builder.append("debug=true;").append(LINE_SEPARATOR);
builder.append("};").append(LINE_SEPARATOR);
}else {
builder.append(module.getName()).append(" {").append(LINE_SEPARATOR);
builder.append(SUN_LOGIN_MODULE).append(LINE_SEPARATOR);
builder.append("useKeyTab=true").append(LINE_SEPARATOR);
builder.append("keyTab=\"" + userKeytabPath + "\"").append(LINE_SEPARATOR);
builder.append("principal=\"" + userPrincipal.trim() + "\"").append(LINE_SEPARATOR);
builder.append("serviceName=\""+serviceName + "\"").append(LINE_SEPARATOR);
builder.append("useTicketCache=false").append(LINE_SEPARATOR);
builder.append("storeKey=true").append(LINE_SEPARATOR);
builder.append("debug=true;").append(LINE_SEPARATOR);
builder.append("};").append(LINE_SEPARATOR);
}
return builder.toString();
}
public static void setZookeeperServerPrincipal(String zkServerPrincipal) throws IOException {
System.setProperty(ZOOKEEPER_SERVER_PRINCIPAL_KEY,zkServerPrincipal);
String ret = System.getProperty(ZOOKEEPER_SERVER_PRINCIPAL_KEY);
if (ret == null) {
LOGGER.error(ZOOKEEPER_SERVER_PRINCIPAL_KEY + " is null.");
throw new IOException(ZOOKEEPER_SERVER_PRINCIPAL_KEY + " is null.");
}
if (!ret.equals(zkServerPrincipal)){
LOGGER.error(ZOOKEEPER_SERVER_PRINCIPAL_KEY + " is " + ret + " is not " + zkServerPrincipal + ".");
throw new IOException(ZOOKEEPER_SERVER_PRINCIPAL_KEY + " is " + ret + " is not " + zkServerPrincipal + ".");
}
}
}
經(jīng)過以上四步的配置，啟動項目后即可自動連接kafka進行身份校驗，若登錄成功，會輸出如下提示信息：Login success，并且會將生成的jaas文件路徑打印出來。
四、常見問題
1、認證文件找不到
這是因為步驟1中kerberos.path配置有問題，檢查path路徑下是否存在認證文件keytab和krb5.conf。
2、?principal和keytab不匹配
不同的用戶名對應(yīng)不同的密碼，在身份校驗時，需保證用戶名principle和密碼keytab的一致性，否則無法驗證通過。而principal和keytab不匹配可能存在以下兩種場景：
- ?配置文件中出現(xiàn)問題：檢查kerberos.principle和kerberos.keytab中的用戶名（即hkjj）是否一致。
- ?檢查生成的jaas文件中用戶名和配置的用戶名是否相同
如果步驟1檢查沒用問題，則可根據(jù)日志中輸出的jaas文件路徑查看自動生成的jaas文件中的principal和配置文件中的kerberos.principle是否一致。比如我的這個項目中，就是由于現(xiàn)場技術(shù)配置kerberos.principle時后面多打了一個空格，導(dǎo)致自動生成的jaas文件中的principle后多一個空格，因此和keytab認證失敗。
為了徹底解決這個誤打空格的問題，可以直接修改認證工具類LoginUtil，在生成jaas文件的principle時去掉可能存在的空格。
3、用戶密碼keytab更新，導(dǎo)致出現(xiàn)checksum failed
這是由于principal對應(yīng)的密碼修改了，但是程序中使用的還是舊的密碼，就會出現(xiàn)這個問題。解決辦法是找第三方提供principal對應(yīng)的最新的密碼文件keytab。
4、jaas文件找不到
該問題是由于找不到jaas.conf 這個文件導(dǎo)致的，而基于kerberos認證時一般不會出現(xiàn)，這是因為kerberos認證時jaas文件是由LoginUtil工具類根據(jù)安全認證文件自動生成并且存儲在指定路徑下的。
該問題通常出現(xiàn)在SASL-PLAIN方式的認證中，因為該方式需要添加一個配置參數(shù)java.security.auth.login.config來標識jaas文件的路徑，如果文件路徑出錯則會報以上錯誤。
五、總結(jié)
在kafka身份認證的過程中，需要的principal，keytab，ServiceName等信息均配置在jaas文件中，因此保證認證的服務(wù)可以讀取到正確的文件及正確的配置是kafka安全模式下認證的核心。文章來源：http://www.zghlxwxcb.cn/news/detail-850175.html
基于kerberos認證時，可根據(jù)安全認證文件自動生成jaas配置文件，從而保證了密碼加密傳輸，相比于SASL-PLAIN模式更具安全性，并且認證實現(xiàn)過程也較為簡單。文章來源地址http://www.zghlxwxcb.cn/news/detail-850175.html
到了這里，關(guān)于Kafka安全模式之身份認證的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
