用户认证功能,是一个成熟组件不可或缺的功能。在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其他的类。usernamepassword则是用于集群内部broker的认证用的。

这里会让人疑惑的,应该是user_adminuser_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包都换一下。

以上,如果觉得有用,不妨点个赞吧~

11-24 04:14