Kafka SASL/PLAIN认证插件

Kafka SenLin 4年前 (2020-01-01) 508次浏览 已收录 0个评论

背景

我们的Kafka一直是没有安全机制地裸奔,最近需要给它升个级,加上认证和授权机制,认证机制是SASL/PLAIN,授权是Kafka自带的ACL授权。

SASL/PLAIN认证机制开启方法:

  1. 修改Kafka配置文件 server.properties 或者其它名字(CDH和Ambari的发行版可能文件名有所差异)
sasl.enabled.mechanisms = PLAIN
sasl.mechanism.inter.broker.protocol = PLAIN
security.inter.broker.protocol = SASL_PLAINTEXT
listeners = SASL_PLAINTEXT://localhost:9092
  1. 编辑 kafka_jaas.conf
KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    # broker之间互相认证的凭据
    username="admin"
    password="admin-secret"
    # 以下是用户信息
    user_admin="admin-secret"   # 用户admin的密码是admin-secret
    user_alice="alice-secret";  # 用户alice的密码是alice-secret
};

启动kafka的时候把 kafka_jaas.conf 的路径当作命令行参数传入:

-Djava.security.auth.login.config=/path/to/kafka_jaas.conf

ACL授权机制开启方法:

编辑Kafka配置文件,添加如下内容:

authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer

以上的配置方案除了没有使用SSL加密之外,还存在一个严重的缺陷:用户信息是通过静态配置文件的方式存储的,当对用户信息进行添加、删除和修改的时候都需要重启Kafka集群,而我们知道,作为消息中间件,Kafka的上下游与众多组件相连,重启可能造成数据丢失或重复,Kafka应当尽量避免重启。

解决方案

还好,Kafka允许用户为SASL/PLAIN认证机制提供了自定义的回调函数,根据KIP-86,如果不希望采用静态配置文件存储用户认证信息的话,只需要编写一个实现了 AuthenticateCallbackHandler 接口的类,然后在配置文件中指明这个类即可,指明的方法为在Kafka配置文件中添加如下内容:

listener.name.sasl_plaintext.plain.sasl.server.callback.handler.class=com.example.MyCallbackHandler

先看下 AuthenticateCallbackHandler 这个接口的默认实现,可以在github上看到:

public class PlainServerCallbackHandler implements AuthenticateCallbackHandler {

    private static final String JAAS_USER_PREFIX = "user_";
    private List<AppConfigurationEntry> jaasConfigEntries;

    @Override
    public void configure(Map<String, ?> configs, String mechanism, List<AppConfigurationEntry> jaasConfigEntries) {
        this.jaasConfigEntries = jaasConfigEntries;
    }

    @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);
        }
    }

    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 {
    }

}

这个接口有三个方法需要实现,分别是configure()、handle()和close(),configure() 进行初始化配置,close() 关闭打开的相关资源,handle() 处理用户的认证请求。我采用MySQL来存储用户信息,因此可以满足动态增删用户的需求,大概的实现为:

public class MySQLAuthCallback implements AuthenticateCallbackHandler {
    private static final Logger log = LoggerFactory.getLogger("plugin");
    private DruidDataSource dataSource = null;

    public MySQLAuthCallback() {
        this.dataSource = new DruidDataSource();
        this.dataSource.setDriverClassName("com.mysql.jdbc.Driver");
        this.dataSource.setUrl("jdbc:mysql://localhost:3306/kafka_user_info");
        this.dataSource.setUsername("kafka");
        this.dataSource.setPassword("kafka");
        this.dataSource.setInitialSize(5);
        this.dataSource.setMinIdle(1);
        this.dataSource.setMaxActive(10);
        this.dataSource.setPoolPreparedStatements(false);
    }

    public void configure(Map<String, ?> configs, String mechanism, List<AppConfigurationEntry> jaasConfigEntries) {
    }

    public void handle(Callback[] callbacks) throws 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);
        }
    }

    protected boolean authenticate(String username, char[] password) {
        return userExists(username, new String(password));
    }

    private boolean userExists(String username, String password) {
        Connection conn = null;
        PreparedStatement statement = null;
        ResultSet resultSet = null;
        try {
            conn = getConnection();
            statement = conn.prepareStatement("select * from users where username=? and password=?");
            statement.setString(1, username);
            statement.setString(2, password);
            resultSet = statement.executeQuery();
            while (resultSet.next()) {
                boolean b = username.equals(resultSet.getString("username")) &&
                        password.equals(resultSet.getString("password"));
                log.info("user {} authentication status: {}.", username, b);
                return b;
            }
        } catch (SQLException e) {
            log.info("sql exception occurred: {}", e.getMessage());

        } finally {
            // 按照打开顺序逆序关闭打开的资源
            }
        }
        return false;
    }

    private Connection getConnection() throws SQLException {
        // 通过datasource获取connection,省略了try/catch
        synchronized(dataSource) {
          return dataSource.getConnection();
        }
    }

    public void close() throws KafkaException {
    }
}

自定义类编写完成后后,将jar包拷贝到每个broker的CLASSPATH下,比如kafka的libs目录下。在MySQL中插入几条用户信息,然后尝试以这些用户的身份来连接Kafka(为方便起见,测试阶段可以先不要开启ACL),我们会发现,MySQL中的用户可以连接,而写在配置文件中的用户无法连接,说明这个插件达到了预期的效果,增删用户不需要重启了。

插件日志问题

在编写这个插件以及生产中使用这个插件的过程中,可能需要输出日志来帮助调试,因此还需要调整相关配置。

首先,在代码中创建Logger对象:

private static final Logger log = LoggerFactory.getLogger("plugin");

其次,修改Kafka的 log4j.properties 配置文件:

log4j.logger.plugin=DEBUG, pluginAppender
log4j.additivity.plugin=false

log4j.appender.pluginAppender=org.apache.log4j.DailyRollingFileAppender
log4j.appender.pluginAppender.DatePattern='.'yyyy-MM-dd-HH
log4j.appender.pluginAppender.File=/path/to/plugin.log
log4j.appender.pluginAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.pluginAppender.layout.ConversionPattern=[%d{ISO8601}] %p %m (%c)%n
log4j.appender.pluginAppender.MaxFileSize = 256MB
log4j.appender.pluginAppender.MaxBackupIndex = 20

注意代码和配置文件的联系在于getLogger的参数和log4j.logger后面的字符串需要相同。

欢迎交流讨论,吐槽建议。

勤学似春起之苗,不见其增,日有所长
辍学如磨刀之石,不见其损,日有所亏
关注【大数据学徒】,用技术干货助你日有所长


top8488大数据 , 版权所有丨如未注明 , 均为原创丨本网站采用BY-NC-SA协议进行授权
转载请注明原文链接:Kafka SASL/PLAIN认证插件
喜欢 (0)
[]
分享 (0)

您必须 登录 才能发表评论!