• kafka 2.x用户认证方式小结
• SASL/PLAIN实例（配置及客户端）
  • • broker配置
    • 客户端配置
• 自定义SASL/PLAIN认证（二次开发）
  • • kafka2新的callback接口介绍
    • 自定义sasl/plain功能

用户认证功能，是一个成熟组件不可或缺的功能。在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吧，引用维基百科。

OAuth是一个开放标准，允许用户让第三方应用访问该用户在某一网站上存储的私密的资源（如照片，视频，联系人列表），而无需将用户名和密码提供给第三方应用。而 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修改为

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。

自定义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包都换一下。

以上，如果觉得有用，不妨点个赞吧～