文档章节

如何构建安全的Kafka集群

varchard
 varchard
发布于 2017/04/06 18:02
字数 2163
阅读 30
收藏 0

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

然而,分析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的身份机制如下图所示:

6

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,安全相关的参数如下所示:

7

其中,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这个类来完成的,其类结构如下图所示:

Kafka

其中,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基于开源版本,在安全方面已有相当多的积累,并持续改进开发,致力于为企业用户提供一个易用、高效、安全和稳定的基础数据平台。

本文转载自:http://www.36dsj.com/archives/29526

varchard
粉丝 6
博文 58
码字总数 19458
作品 0
程序员
私信 提问
福利丨这也许是目前市面上最权威的Kafka学习指南!

Kafka,不是村上春树的《海边的卡夫卡》,不是《变形记》的作者弗兰兹·卡夫卡。程序员世界里的 Kafka,有三分之一的世界 500 强公司都在使用它发布和订阅、存储及实时地处理大规模流数据。 ...

DBAplus社群
2018/01/15
0
0
​福利丨又一波大数据新书推介,来做第一个吃螃蟹的人吧!

又到了每周发福利的时刻! 延续上期主题,本期给大家带来3本@图灵教育的大数据新书推荐 小编云: 经典读物固然能让收获有所保障 可新书却给人更多阅读的新鲜体验 来吧,一起做第一个吃螃蟹的...

DBAplus社群
2017/12/20
0
0
分布式的、高吞吐量、高可扩展性消息队列服务Kafka商业化发布!

摘要:消息队列Kafka是一个分布式的、高吞吐量、高可扩展性消息队列服务,广泛用于日志收集、监控数据聚合、流式数据处理、在线和离线分析等,是大数据生态中不可或缺的产品之一,阿里云提供...

萌萌怪兽
2018/07/30
0
0
kafka-manager安装与使用

kafka-manager安装与使用 www.blogways.net2018-01-071 阅读 安装kafka 目 录 1 kafka manager简介 2 kafka manager安装 3 kafka manager使用 一、kafka manager简介 为了简化开发者和服务工...

www.blogways.net
2018/01/07
0
0
Kafka不停机,如何无感知迁移ZooKeeper集群?

Kafka 在 Yelp 的应用十分广泛,Yelp 每天通过各种集群发送数十亿条消息,在这背后,Kafka 使用 Zookeeper 完成各种分布式协调任务。 因为Yelp 非常依赖 Kafka,那么问题来了,它是否可以在不...

微笑向暖wx
02/25
28
0

没有更多内容

加载失败,请刷新页面

加载更多

Spring Boot + Mybatis-Plus 集成与使用(二)

前言: 本章节介绍MyBatis-Puls的CRUD使用。在开始之前,先简单讲解下上章节关于Spring Boot是如何自动配置MyBatis-Plus。 一、自动配置 当Spring Boot应用从主方法main()启动后,首先加载S...

伴学编程
昨天
7
0
用最通俗的方法讲spring [一] ──── AOP

@[TOC](用最通俗的方法讲spring [一] ──── AOP) 写这个系列的目的(可以跳过不看) 自己写这个系列的目的,是因为自己是个比较笨的人,我曾一度怀疑自己的智商不适合干编程这个行业.因为在我...

小贼贼子
昨天
7
0
Flutter系列之在 macOS 上安装和配置 Flutter 开发环境

本文为Flutter开发环境在macOS下安装全过程: 一、系统配置要求 想要安装并运行 Flutter,你的开发环境需要最低满足以下要求: 操作系统:macOS(64位) 磁盘空间:700 MB(不包含 IDE 或其余...

過愙
昨天
6
0
OSChina 周六乱弹 —— 早上儿子问我他是怎么来的

Osc乱弹歌单(2019)请戳(这里) 【今日歌曲】 @凉小生 :#今日歌曲推荐# 少点戾气,愿你和这个世界温柔以待。中岛美嘉的单曲《僕が死のうと思ったのは (曾经我也想过一了百了)》 《僕が死の...

小小编辑
昨天
2.7K
16
Excption与Error包结构,OOM 你遇到过哪些情况,SOF 你遇到过哪些情况

Throwable 是 Java 中所有错误与异常的超类,Throwable 包含两个子类,Error 与 Exception 。用于指示发生了异常情况。 Java 抛出的 Throwable 可以分成三种类型。 被检查异常(checked Exc...

Garphy
昨天
42
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部