如何构建安全的Kafka集群

Kafka是由LinkedIn设计的一个高吞吐量、分布式、基于发布订阅模式的消息系统，使用Scala编写，它以可水平扩展、可靠性、异步通讯和高吞吐率等特性而被普遍使用。目前愈来愈多的开源分布式处理系统都支持与Kafka集成，其中Spark Streaming做为后端流引擎配合Kafka做为前端消息系统正成为当前流处理系统的主流架构之一。

然而，当下愈来愈多的安全漏洞、数据泄露等问题的爆发，安全正成为系统选型不得不考虑的问题，Kafka因为其安全机制的匮乏，也致使其在数据敏感行业的部署存在严重的安全隐患。本文将围绕Kafka，先介绍其总体架构和关键概念，再深刻分析其架构之中存在的安全问题，最后分享下Transwarp在Kafka安全性上所作的工做及其使用方法。

Kafka架构与安全

首先，咱们来了解下有关Kafka的几个基本概念：

Topic：Kafka把接收的消息按种类划分，每一个种类都称之为Topic，由惟一的Topic Name标识。

Producer：向Topic发布消息的进程称为Producer。

Consumer：从Topic订阅消息的进程称为Consumer。

Broker：Kafka集群包含一个或多个服务器，这种服务器被称为Broker。

Kafka的总体架构以下图所示，典型的Kafka集群包含一组发布消息的Producer，一组管理Topic的Broker，和一组订阅消息的Consumer。Topic能够有多个分区，每一个分区只存储于一个Broker。Producer能够按照必定的策略将消息划分给指定的分区，如简单的轮询各个分区或者按照特定字段的Hash值指定分区。Broker须要经过ZooKeeper记录集群的全部Broker、选举分区的Leader，记录Consumer的消费消息的偏移量，以及在Consumer Group发生变化时进行relalance. Broker接收和发送消息是被动的：由Producer主动发送消息，Consumer主动拉取消息。

然而，分析Kafka框架，咱们会发现如下严重的安全问题：

1.网络中的任何一台主机，均可以经过启动Broker进程而加入Kafka集群，可以接收Producer的消息，可以篡改消息并发送给Consumer。

2.网络中的任何一台主机，均可以启动恶意的Producer/Consumer链接到Broker，发送非法消息或拉取隐私消息数据。

3.Broker不支持链接到启用Kerberos认证的ZooKeeper集群，没有对存放在ZooKeeper上的数据设置权限。任意用户都可以直接访问ZooKeeper集群，对这些数据进行修改或删除。

4.Kafka中的Topic不支持设置访问控制列表，任意链接到Kafka集群的Consumer（或Producer）都能对任意Topic读取（或发送）消息。

随着Kafka应用场景愈来愈普遍，特别是一些数据隐私程度较高的领域（如道路交通的视频监控），上述安全问题的存在犹如一颗定时炸弹，一旦内网被黑客入侵或者内部出现恶意用户，全部的隐私数据（如车辆出行记录）都可以轻易地被窃取，而无需攻破Broker所在的服务器。

Kafka安全设计

基于上述分析，Transwarp从如下两个方面加强Kafka的安全性：

身份认证（Authentication）：设计并实现了基于Kerberos和基于IP的两种身份认证机制。前者为强身份认证，相比于后者具备更好的安全性，后者适用于IP地址可信的网络环境，相比于前者部署更为简便。

权限控制（Authorization）：设计并实现了Topic级别的权限模型。Topic的权限分为READ（从Topic拉取数据）、WRITE（向Topic中生产数据）、CREATE（建立Topic）和DELETE（删除Topic）。

基于Kerberos的身份机制以下图所示：

Broker启动时，须要使用配置文件中的身份和密钥文件向KDC（Kerberos服务器）认证，认证经过则加入Kafka集群，不然报错退出。

Producer（或Consumer）启动后须要通过以下步骤与Broker创建安全的Socket链接：

1.Producer向KDC认证身份，经过则获得TGT（票证请求票证），不然报错退出

2.Producer使用TGT向KDC请求Kafka服务，KDC验证TGT并向Producer返回SessionKey（会话密钥）和ServiceTicket（服务票证）

3.Producer使用SessionKey和ServiceTicket与Broker创建链接，Broker使用自身的密钥解密ServiceTicket，得到与Producer通讯的SessionKey，而后使用SessionKey验证Producer的身份，经过则创建链接，不然拒绝链接。

ZooKeeper须要启用Kerberos认证模式，保证Broker或Consumer与其的链接是安全的。

Topic的访问控制列表（ACL）存储于ZooKeeper中，存储节点的路径为/acl/<topic>/<user>，节点数据为R(ead)、W(rite)、C(reate)、D(elete)权限的集合，如/acl/transaction/jack节点的数据为RW，则表示用户jack可以对transaction这个topic进行读和写。
另外，kafka为特权用户，只有kafka用户可以赋予/取消权限。所以，ACL相关的ZooKeeper节点权限为kafka具备全部权限，其余用户不具备任何权限。

构建安全的Kafka服务

首先，咱们为Broker启用Kerberos认证模式，配置文件为/etc/kafka/conf/server.properties，安全相关的参数以下所示：

其中，authentication参数表示认证模式，可选配置项为simple, kerberos和ipaddress，默认为simple。当认证模式为kerberos时，须要额外配置帐户属性principal和对应的密钥文件路径keytab.

认证模式为ipaddress时，Producer和Consumer建立时不须要作任何改变。而认证模式为kerberos时，须要预先建立好相应的principal和keytab，并使用API进行登陆，样例代码以下所示：

public class SecureProducer extends Thread {
private final kafka.javaapi.producer.Producer<Integer, String> producer;
private final String topic;
private final Properties props = new Properties();

public SecureProducer(String topic) {
 AuthenticationManager.setAuthMethod(“kerberos”);
    AuthenticationManager.login(“producer1”, “/etc/producer1.keytab”);
props.put(“serializer.class”, “kafka.serializer.StringEncoder”);
props.put(“metadata.broker.list”,

“172.16.1.190:9092,172.16.1.192:9092,172.16.1.193:9092”);
// Use random partitioner. Don’t need the key type. Just set it to Integer.
    // The message is of type String.
    producer = new kafka.javaapi.producer.Producer<Integer, String>(

new ProducerConfig(props));
this.topic = topic;
}

. . .

Topic权限管理

Topic的权限管理主要是经过AuthorizationManager这个类来完成的，其类结构以下图所示：

其中，resetPermission(user, Permissions, topic) 为重置user对topic的权限。

grant(user, Permissions, topic) 为赋予user对topic权限。

revoke(user, Permissions, topic) 为取消user对topic权限。

isPermitted(user, Permissions, topic) 为检查user对topic是否具备指定权限。

调用grant或revoke进行权限设置完成后，须要commit命令提交修改到ZooKeeper

Kerberos模式下，AuthorizationManager须要先使用AuthenticationManager.login方法登陆，与ZooKeeper创建安全的链接，再进行权限设置。示例代码以下所示：

public class AuthzTest {
public static void main(String[] args) {
Properties props = new Properties();
props.setProperty(“authentication”, “kerberos”);
props.setProperty(“zookeeper.connect”, “172.16.2.116:2181,172.16.2.117:2181,172.16.2.118:2181”);
props.setProperty(“principal”, “kafka/host1@TDH”);
props.setProperty(“keytab”, “/usr/lib/kafka/config/kafka.keytab”);
ZKConfig config = new ZKConfig(props);
AuthenticationManager.setAuthMethod(config.authentication());
AuthenticationManager.login(config.principal(), config.keytab());
AuthorizationManager authzManager = new AuthorizationManager(config);
// reset permission READ and WRITE to ip 172.16.1.87 on topic test
    authzManager.resetPermission(“172.16.1.87”,
new Permissions(Permissions.READ, Permissions.WRITE), “test”);
// grant permission WRITE to ip 172.16.1.87 on topic test
    authzManager.grant(“172.16.1.87”, new Permissions(Permissions.CREATE), “test”);
// revoke permission READ from ip 172.16.1.87 on topic test
    authzManager.revoke(“172.16.1.87”, new Permissions(Permissions.READ), “test”);
// commit the permission settings
    authzManager.commit();
authzManager.close();
}
}

 

ipaddress认证模式下，取消和赋予权限的操做以下所示：

public class AuthzTest {
public static void main(String[] args) {
Properties props = new Properties();
props.setProperty(“authentication”, “ipaddress”);
props.setProperty(“zookeeper.connect”,
“172.16.1.87:2181,172.16.1.88:2181,172.16.1.89:2181”);
ZKConfig config = new ZKConfig(props);
// new authorization manager
AuthorizationManager authzManager = new AuthorizationManager(config);
// reset permission READ and WRITE to ip 172.16.1.87 on topic test
authzManager.resetPermission(“172.16.1.87”,
new Permissions(Permissions.READ, Permissions.WRITE), “test”);
// grant permission WRITE to ip 172.16.1.87 on topic test
authzManager.grant(“172.16.1.87”, new Permissions(Permissions.CREATE), “test”);
// revoke permission READ from ip 172.16.1.87 on topic test
authzManager.revoke(“172.16.1.87”, new Permissions(Permissions.READ), “test”);
// commit the permission settings
authzManager.commit();
authzManager.close();
}
}

总结与展望

本文经过介绍Kafka现有架构，深刻挖掘其中存在的安全问题，并给出Transwarp在Kafka安全上所作的工做及其使用方式。然而，纵观Hadoop & Spark生态系统，安全功能还存在不少问题，各组件的权限系统独立混乱，缺乏集中易用的帐户管理系统。某些组件的权限管理还很不成熟，如Spark的调度器缺乏用户的概念，不能限制具体用户使用资源的多少。Transwarp基于开源版本，在安全方面已有至关多的积累，并持续改进开发，致力于为企业用户提供一个易用、高效、安全和稳定的基础数据平台。