1. 方案背景
Kafka security主要包含3大功能：认证(authentication)、信道加密(encryption)和授权(authorization)。信道加密就是为client到broker、broker到broker以及工具脚本与broker之间的数据传输配置SSL；认证机制主要是指配置SASL，而授权是通过ACL接口命令来完成的。
本方案主要是解决kafka和zk之前的权限认证问题，即ZK的数据安全性问题，而对于kafka本身的权限控制没有做说明。
ZooKeeper作为一个分布式协调框架，内部存储的都是一些分布式系统运行时状态的元数据。zookeeper本身提供了ACL机制，表示为scheme:id:permissions，第一个字段表示采用哪一种机制，第二个id表示用户，permissions表示相关权限（如只读，读写，管理等）。zookeeper提供了如下几种机制（scheme）：
- world: 它下面只有一个id, 叫anyone, world:anyone代表任何人，zookeeper中对所有人有权限的结点就是属于world:anyone的
- auth: 它不需要id, 只要是通过authentication的user都有权限（zookeeper支持通过kerberos来进行authencation, 也支持username/password形式的authentication)
- digest: 它对应的id为username:BASE64(SHA1(password))，它需要先通过username:password形式的authentication
- ip: 它对应的id为客户机的IP地址，设置的时候可以设置一个ip段，比如ip:192.168.1.0/16, 表示匹配前16个bit的IP段
- super: 在这种scheme情况下，对应的id拥有超级权限，可以做任何事情(cdrwa）
注意：ACL并无递归机制，任何一个znode创建后，都需要单独设置ACL，无法继承父节点的ACL设置
权限：Permission
对数据节点的控制操作权限分为五种：
- CREATE：创建节点以及子节点的权限
- DELETE：删除接单以及子节点的权限
- READ：数据节点的读取权限
- WRITE：数据节点的写权限
- ADMIN：数据节点的管理权限
2.方案目标
通过SASL授权认证+ACL的权限控制，对zookeeper的节点数据增加访问权限控制，只有通过授权认证的用户才可以访问相应的节点数据。
3.方案步骤
1. 创建zk_jaas.conf文件
创建zk_jaas.conf文件指定身份认证插件，可以放置在zk的conf目录下，设置登录到zookeeper的登录名与密码
Server {
org.apache.zookeeper.server.auth.DigestLoginModule required
username="admin"
password="bangsun2"
user_admin="bangsun2";
};
2. 修改zk的配置文件zookeeper.properties
修改zookeeper.properties，加入以下内容
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl
jaasLoginRenew=3600000
zookeeper.sasl.client=true
3. 修改kafka-run-class.sh
修改脚本，在脚本最末尾加入变量KAFKA_AUTH_OPTS，后续通过该变量传入jass认证文件的路径。截图如下：
4. 重写zk启动脚本zkStart.sh
使用zk启动脚本zookeeper-server-start.sh，加入KAFKA_AUTH_OPTS=zk_jaas.conf配置文件所在路径
export KAFKA_AUTH_OPTS="-Djava.security.auth.login.config=/home/plugins/kafka/kafka_2.10-0.10.2.0_jiami/config/zk_jaas.conf"
或者重写zk启动脚本如下：
之后启动zk服务，至此zk端配置SASL用户名密码认证已完成。
5. 创建kafka_jaas.conf
创建客户端认证文件kafka_jaas.conf，可放置于kafka的conf目录下。Client节用于与zookeeper通信时进行认证，用户名密码在上一步配置zookeeper时的zk_jaas.conf配置文件中定义了。
Client {
org.apache.zookeeper.server.auth.DigestLoginModule required
username="admin"
password="bangsun2";
};
6.修改kafka配置文件server.properties
增加如下配置:
zookeeper.set.acl=true
7.修改kafka启动脚本
使用kafka启动脚本kafka-server-start.sh ，加入KAFKA_AUTH_OPTS=kafka_jaas.conf配置文件所在路径
export KAFKA_AUTH_OPTS=" -Djava.security.auth.login.config=/home/plugins/kafka/kafka_2.10-0.10.2.0_jiami/config/kafka_jaas.conf"
8.启动kafka，查看ACL权限
启动kafka，通过zk的客户端工具zkCli.sh连接查看节点权限
./zkCli.sh -server 192.168.184.129:2181
通过命令可以看到zk原有的节点权限是world权限的，而通过kafka创建的节点只有sasl认证用户admin具有所有权限，一般未认证用户只有可读权限。
Zookeeper SASL身份验证允许匿名用户登录，如果身份验证失败，仍然允许连接和请求，对未经身份验证的用户或匿名用户将拒绝访问这些ACL。
https://issues.apache.org/jira/browse/ZOOKEEPER-1736
9. 控制未认证用户权限
到此步为止未认证用户具有Zk节点数据的可读权限，虽然zk节点数据为不敏感数据，但如果不想给未认证用户查看节点数据，目前只有给每个节点分别进行权限控制，父目录的权限无法限制子目录，所以每个目录都需要分别设置权限。
我们可以通过zk客户端工具zkCli.sh来设置节点权限，首先因为之前只有kafka的认证用户才有节点的操作权限，所以修改zkCli.sh增加权限认证：
"-Djava.security.auth.login.config=/home/plugins/kafka/kafka_2.10-0.10.2.0_jiami/config/kafka_jaas.conf"
连接zk节点
./zkCli.sh -server 192.168.184.129:2181
修改权限只有授权用户admin具有操作权限：
setAcl /brokers/topics/FrmsDSQueue sasl:admin:cdrwa
最后可以看到节点已经没有world的权限了。
4.测试验证
1. 未认证kafka节点无法获取zk节点信息，提示NOAUTH
2.账户密码认证错误将无法连接到zk节点