用户认证功能，是一个成熟组件不可或缺的功能。在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。
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。
SASL/OAUTHBEARER
SASL/OAUTHBEARER是基于OAUTH2.0的一个新的认证框架，这里先说下什么是OAUTH吧，引用维基百科。
说白了，SASL/OAUTHBEARER就是一套让用户使用第三方认证工具认证的标准，通常是需要自己实现一些token认证和创建的接口，所以会比较繁琐。
详情可以通过这个kip了解KIP-255
说了这么多，接下来就说实战了，先介绍下如何配置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修改为
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。
自定义sasl/plain功能
先详细介绍下sasl.server.callback.handler.class配置。这个配置在使用的时候，需要以小写方式指定SASL的类型。举个例子，如果是SASL_PLAINTEXT，那么就需要这样：
即以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<AppConfigurationEntry> jaasConfigEntries;
@Override
public void configure(Map<String, ?> configs, String mechanism, List<AppConfigurationEntry> 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包都换一下。
以上，如果觉得有用，不妨点个赞吧～