我们在之前介绍了很多Kafka的基础和进阶的知识，他们中很多都是服务于功能，性能或者可靠性的。而现代开发中有一个点也越来越被关注，那就是安全性相关的内容，本文就来聊聊Kafka在Security这个方面都做了些什么。
概述
在我们开始介绍security之前，先来看看几个专有名词，知道了它们对我们后面内容的理解会很有帮助。
Authentication：这个是用来鉴别身份，也就是表明你是谁。就像军队中说我是司令，我是团长，我是小排长之类的。
Authorization：这个用来判断你这个身份可以做什么。比如说司令可以号令全军，团长只能管你这个团。
Encryption：这个是用来加密，防止被偷听和篡改的。就像战争片中的电报要加密一样，防止被敌方截取或者篡改。
Auditing：监察你做了什么或者你想做什么。有点像纪律委员会？
Quotas：控制你能使用多少资源。
下面我们来看一个典型的Kafka流程，然后基于这个流程来分析Kafka在各个步骤都做了什么来保证数据的安全。
用户A产生了一个record，先发送一个message到一个topic，当然这个写是发送到了leader的broker上。
Leader broker把这个数据写到了本地log文件中。
Follower从leader中把这个数据fetch到本地，并写入本地的文件中。
Leader broker通过Zookeeper更新一些数据，比如in-sync的replicas之类的。
另外一个用户B通过consumer来获取刚刚用户A写入的message。
还有一个内部的应用会从broker中获取所有人写入的数据来进行一些实时的分析。
那么我们来粗略看看整个这个流程中会涉及到哪些security相关的内容：
Client authenticity: 首先在用户A和broker之间建立连接的时候，broker要能够通过Authentication来判断这个连接的确是从用户A来创建的，而不是别人冒充的。
Server authenticity：在用户A真正开始发送message给broker之前，它需要能够判断接收数据的对象是真的我们想要发送的broker而不是一个冒充的接收者。
Data privacy：所有传输的数据，包括写入到磁盘中的数据，都应该加密，这样可以保证不会被人偷听或者窃取。
Date integrity：要能够保证数据的完整性，就是在一个不安全的网络上传输之后，假如数据被篡改了，我们能够知道。
Access Control：在leader broker把用户A发送的数据写入到log文件之前，我们需要判断用户A是有权限来写当前broker的，同样的在返回message给用户B之前，我们也需要判断用户B是有权限来读取当前broker的。
Availability：broker需要进行限流，也就是我们前面说的quotas，不能让一个用户把所有的资源全部都消耗了。
下面我们就来分别看看Kafka是如何做到上面提到的这些安全相关的功能的。
Security Protocols
Kafka的broker其实就是在一个或多个端口进行监听，然后允许client端来进行连接。这个监听就可以使用不同的安全配置来进行设置。Kafka目前支持四种security protocols，他们都是基于TLS或者SASL的。
PLAINTEXT：这个说白了就是什么都不做，没有任何authentication。一般来说在内部网络之间传输一些非敏感的数据可以使用。
SSL：支持可选的SSL client authentication。一般用于client和server之间是不安全网络，因为它同时也支持encryption。
SASL_PLAINTEXT：在PLAINTEXT的基础上加入了client的authentication。有些SASL也支持server authentication。但是它不支持encryption，所以一般用于内部网络。
SASL_SSL：就是支持client的authentication，也同时支持server的authentication，又支持encryption。算是最强的security支持了。
这个配置也还可以配置interbroker的交互，比如下面这个例子就是设置interbroker和内部listeners都是SSL，然后外面的listener设置成SASL_SSL。
listener.security.protocol.map=EXTERNAL:SASL_SSL,INTERNAL:SSL,BROKER:SSL
Authentication
我们前面也说了Kafka有client端的authentication和server端的authentication，目的也很明确就是用来表明用户A的确是用户A，连接的leader broker也的确是我想连接的broker。Kafka使用一个KafkaPrincipal的instance来代表用户的身份并且给相关的连接分配资源和quota。每个连接相对应的KafkaPrincipal都是在基于authentication protocol进行authentication的时候建立的。我们下面来具体看看：
SSL
当Kafka配置成SSL或者SASL_SSL的时候，安全传输层可以使用TLS，所以当一个连接是建立在TLS上的时候，TLS握手流程会进行authentication，交互密码参数，产生用于encryption的共享key。Client端是通过server端的数字证书来判断server是不是想要connect的server。类似的，server端也可以通过client端的数字证书来进行判断client端是否是那个client端。
SASL
SASL除了可以和TLSS来一起做authentication和encryption，它还有内置的各种SASL mechanism可以使用。Kafka目前支持下面这些SASL mechanism：
GSSAPI: 使用SASL/GSSAPI的时候可以支持Kerberos authentication，这样我们就可以和一些Kerberos server（AD/OpendLDAP等）来进行集成。
PLAIN：就是使用用户名和密码来进行authentication，需要从外面密码store中来进行验证。
SCRAM-SHA-256以及SCRAM-SHA-512：同样的用户名，密码authentication，只是不需要外面的密码store。
OAUTHBEAREER: 使用OAuth token来进行authentication。
关于Reauthentication，有些security的mechanism比如Kerberos和OAuth都是有时间限制的，Kafka是通过一个后台的login thread来获取新的credentials的，当然新的credentials也只在新的连接中使用，现有的连接还是会超时的。
Encryption
我们前面也谈到了Encryption其实就是保护数据的隐私和完整，上面提到的TLS可以提供安全的加密通道来保证数据的传输。但是这是不够的，我们还需要保证别的地方数据的安全，比如说哪怕有人可以直接访问物理磁盘，我们也要能保证磁盘中的数据是安全的。也就是说我们需要在数据的传输和保存两个方面都做好Encryption的操作才行。
我们以前也提到过Kafka的数据在传输的时候需要进行Serialization和deserialization，其实我们可以把Encryption的过程也加入这两者上面去，我们来看一个端到端的过程，如下图所示：
使用KafkaProducer发送一个message。
Producer使用KMS中的一个encryptionkey来encrypt这个message。
这个被encrypted的message发送到了broker，broker会直接保存这个encrypted的message到log 文件中（这样一来，即使可以物理访问磁盘，因为没有key，也无法解析具体的内容）
当有consumer来读这个数据的时候，Kafkabroker其实也是发送的这个Encrypted的message。
Consumer端再到KMS中拿到encryptionkey来decrypt这个message。
Authorization
Authorization是用来判断哪些操作是可以做的，哪些资源是可以访问的等等。Kafka会根据KafkaPrincipal来进行判断，比如说我们知道了这个request是从用户A发送过来的，就可以根据用户A的权限来判断这个request是否可以做。那这个判断是如何进行的呢？Kafka有一个内置的authorizer称之为AclAuthorizer，你可以直接使能：
这里的ACL就是access control lists，它其实是保存在ZooKeeper中，然后每个broker都有一个本地的cache，这样就可以提高性能，根据这个ACL的信息再结合KafkaPrincipal你就可以判断相关的请求和操作是否可以执行了。
当然你也可以自定义authorization来实现你想要进行判断的内容。
Auditing
Kafka可以通过配置来产生log4j的log用于auditing和debug。你可以通过log4j.properties来进行设置。
总结
至此，本文就把Kafka涉及的Security相关的内容介绍完毕了，希望对你能有所帮助。
你好，我是晓东。
赞和在看就是最大的支持
