一、背景
在我们部署完kafka之后，虽然我们已经可以“肆意”的用kafka了，但是在一个大公司的实际生产环境中，kafka集群往往十分庞大，每个使用者都应该只关心自己所负责的Topic，并且对其他人所使用的Topic没有权限。这样一来可以将资源隔离开来，二来可以防止误操作。
在权限控制之前，我们必须要启用的就是用户认证，没有用户，自然没有权限一说了。
二、kafka启用kerberos认证
2.1 在KDC中添加kafka用户，并生成keytab
|
新建kfaka用户 |
kadmin.local -q ‘addprinc -randkey kafka/{hostname}@{REALM}’ |
|
生成keytab |
kadmin.local -q “ktadd -k /etc/security/keytabs/{keytabname}.keytab kafka/{hostname}@{REALM}” |
注意：
1、如果之前zookeeper没有启用kerberos，这里也要启用zookeeper的kerberos
2、如果之前在CM中启用了kerberos，我们可以直接从CM中获取keytab，但是注意keytab一定要保持最新的，否则认证不会通过，keytab的位置是：
/var/run/cloudera-scm-agent/process/****-kafka-KAFKA_BROKER/kafka.keytab
/var/run/cloudera-scm-agent/process/****-zookeeper-server/zookeeper.keytab
2.2 修改server.properties
|
//修改这一句 listeners=SASL_PLAINTEXT://host.name:port //新增以下 authorizer.class.name = kafka.security.auth.SimpleAclAuthorizer security.inter.broker.protocol=SASL_PLAINTEXT sasl.mechanism.inter.broker.protocol=GSSAPI sasl.enabled.mechanisms=GSSAPI sasl.kerberos.service.name=kafka super.users=User:kafka |
2.3 新建kafka_server.jaass
|
KafkaServer { com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true serviceName=”kafka” keyTab=”/etc/keytab/kafka_122.keytab” //替换为自己的keytab所在位置 principal=”kafka/{hostname}@{REALM}”;//替换为自己的keytab所对应principle };
// Zookeeper client authentication，因为卡夫卡使用过程中会和zookeeper进行交互 Client { com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true serviceName=”zookeeper” keyTab=”/etc/keytab/kafka_122.keytab” //替换为自己的keytab所在位置 principal=”kafka/{hostname}@{REALM}”;//替换为自己的keytab所对应principle }; |
2.4 修改启动脚本
|
export KAFKA_OPTS=”-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/root/kafka/config/kafka_server.jaas” //刚才的kafka_server.jaas位置 |
2.5重启broker
|
bin/kafka-server-stop.sh bin/kafka-server-start.sh |
2.6 客户端启用kerberos
在broker启用kerberos之后，如果我们后续需要在命令行界面进行操作，及consumer与producer操作，我们需要在这些客户端也配置上kerberos
2.6.1新增kafka_client.jaas
|
KafkaClient { com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab=”/etc/keytab/kafka_122.keytab” serviceName=”kafka” principal=”kafka/{hostname}@{REALM}”; };
// Zookeeper client authentication Client { com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true serviceName=”zookeeper” keyTab=”/etc/keytab/kafka_122.keytab” principal=”kafka/{hostname}@{REALM}”; };
|
2.6.2配置生效
|
当前会话生效： export KAFKA_OPTS=”-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/opt/kafka/config/kafka_client.jaas” |
|
配置到环境变量中 vim /etc/profile 增加 export KAFKA_OPTS=”-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/opt/kafka/config/kafka_client.jaas” |
2.6.3若有用到consumer或者producer，在consumer.properties或producer.properties中增加
|
security.protocol=SASL_PLAINTEXT sasl.mechanism=GSSAPI sasl.kerberos.service.name=kafka |
三、kafka启用acl（参考http://orchome.com/185）
Kafka认证管理CLI（和其他的CLI脚本）可以在bin目录中找到。CLI脚本名是kafka-acls.sh。启用之前，需要在server.properties里添加这句:
|
allow.everyone.if.no.acl.found=false |
以下列出了所有脚本支持的选项：
|
选项 |
描述 |
默认 |
类型选择 |
|
–add |
添加一个acl |
|
Action |
|
–remove |
移除一个acl |
|
Action |
|
–list |
列出acl |
|
Action |
|
–authorizer |
authorizer的完全限定类名 |
kafka.security.auth.SimpleAclAuthorizer |
Configuration |
|
–authorizer-properties |
key=val，传给authorizer进行初始化，例如：zookeeper.connect=localhost:2181 |
|
Configuration |
|
–cluster |
指定集群作为资源。 |
|
Resource |
|
–topic [topic-name] |
指定topic作为资源。 |
|
Resource |
|
–group [group-name] |
指定 consumer-group 作为资源。 |
|
Resource |
|
-allow-principal |
添加到允许访问的ACL中，Principal是PrincipalType:name格式。 |
|
Principal |
|
–deny-principal |
添加到拒绝访问的ACL中，Principal是PrincipalType:name格式。 |
|
Principal |
|
–allow-host |
–allow-principal中的principal的IP地址允许访问。 |
如果–allow-principal指定的默认值是*，则意味着指定“所有主机” |
Host |
|
–deny-host |
允许或拒绝的操作。 |
ALL |
Operation |
|
–operation |
–deny-principal中的principals的IP地址拒绝访问。 |
如果 –deny-principal指定的默认值是 * 则意味着指定 “所有主机” |
Host |
|
–producer |
为producer角色添加/删除acl。生成acl，允许在topic上WRITE, DESCRIBE和CREATE集群。 |
|
Convenience |
|
–consumer |
为consumer role添加/删除acl，生成acl，允许在topic上READ, DESCRIBE 和 consumer-group上READ。 |
|
Convenience |
|
–force |
假设所有操作都是yes，规避提示 |
|
Convenience |
常用命令举例：因为kafka的acl信息是存在zookeeper上的，所以需要提供zookeeper.connect参数，并且acl的存在与否与资源的存在与否无关。
最后的资源可以是topic，也可是cluster，也可以是consumer-group
|
给Bob和Alice从198.168.159.0（1）对test读写的权限 |
bin/kafka-acls.sh –authorizer-properties zookeeper.connect=localhost:2181 –add –allow-principal User:Bob –allow-principal User:Alice –allow-host 198.168.159.0 –allow-host 198.168.159.1 –operation Read –operation Write –topic test |
|
只拒绝BadBob从198.168.159.3对test的读权限 |
bin/kafka-acls.sh –authorizer-properties zookeeper.connect=localhost:2181 –add –allow-principal User:* –allow-host * –deny-principal User:BadBob –deny-host 198.168.159.3 –operation Read –topic test |
|
列出test的所有权限 |
bin/kafka-acls.sh –authorizer-properties zookeeper.connect=localhost:2181 –list –topic test |
|
删除权限 |
bin/kafka-acls.sh –authorizer-properties zookeeper.connect=localhost:2181 –remove –allow-principal User:* –allow-host * –deny-principal User:BadBob –deny-host 198.168.159.3 –operation Read –topic test |
|
给生产者Bob对test的生产权限 |
bin/kafka-acls.sh –authorizer-properties zookeeper.connect=localhost:2181 –add –allow-principal User:Bob –producer –topic test |
|
给消费者Bob对test的消费权限 |
bin/kafka-acls.sh –authorizer-properties zookeeper.connect=localhost:2181 –add –allow-principal User:Bob –consumer –topic test –group Group-1 |
四、MirrorMaker的跨域同步
4.1 修改kerberos配置
|
添加互信principle |
kadmin.local下操作 addprinc krbtgt/{REALMA}@{REALMB} addprinc krbtgt/{REALMB}@{REALMA} |
|
修改krb5.conf |
[realms]//realms 里配上两个域的信息 HADOOP.SPADE.COM = { kdc = hb21-bd-cm-130-61:88 admin_server = hb21-bd-cm-130-61:749 } HADOOP.TEST.COM = { kdc = tk-dba-hadoop-152:88 admin_server = tk-dba-hadoop-152:749 } [domain_realm] //domain_realm 配上域名和主机名的映射，有多少机器就要配多少 tk-dba-hadoop-154 = HADOOP.TEST.COM hb21-dba-kfk-130-120 = HADOOP.SPADE.COM
[capaths] //capaths 配上互信的域的映射 HADOOP.SAPDE.COM ={ HADOOP.TEST.COM = . } HADOOP.TEST.COM={ HADOOP.SPADE.COM = . }
|
4.2 修改broker配置
添加sasl.kerberos.principal.to.local.rules属性
|
sasl.kerberos.principal.to.local.rules=RULE:[1:$1@$0](.*@\HADOOP.TEST.COM$)s/@\HADOOP.TEST.COM$//,RULE:[2:$1@$0](.*@\HADOOP.TEST.COM$)s/@\HADOOP.TEST.COM$//,RULE:[1:$1@$0](.*@\HADOOP.SPADE.COM$)s/@\HADOOP.SPADE.COM$//,RULE:[2:$1@$0](.*@\HADOOP.SPADE.COM$)s/@\HADOOP.SPADE.COM$//,DEFAULT |
4.3 验证互信是否成功
从域B中复制出keytab到域A的机器中，然后在A中使用该keytab，配置jaas文件，导入环境变量中。用该keytab操作集群A或者集群B中的topic，能正常写入数据即为成功。
五、启用kerberos之后的平滑过度期
生产环境启用kerberos之后，为了给业务向的consumer和producer一个平滑的接入认证系统的缓冲时间，这段时间我们可以给kafka启用两个监听端口，一个是需要kerberos认证的端口，一个不需要认证的端口。让他们共同存在，同时服务。
5.1 增加监听端口
|
修改server.properties |
listeners=SASL_PLAINTEXT://10.21.130.120:9092,PLAINTEXT://10.21.130.120:9093 allow.everyone.if.no.acl.found=false |
5.2 添加ANONYMOUS用户的访问权限
|
bin/kafka-acls.sh –add –authorizer-properties zookeeper.connect={host:port/childpath} –allow-principal User:ANONYMOUS –allow-host * –operation All –topic {topicname} |
5.3 测试不同认证方式共存成功与否
|
删除jaas环境变量 |
unset {变量名} |
|
producer测试 |
bin/kafka-console-producer.sh –broker-list {host}:9093 –topic{topicname} |
六、启用zookeeper的acl同步
kafka的bin目录下的zookeeper-security-migration.sh，可以将kafka的权限，遍历赋给zookeeper中每个子节点，然后分别设置acl，因为zookeeper的acl是仅对当前节点生效，对其下节点不生效的，单独赋权限很麻烦。zookeeper-security-migration.sh解决了这个问题。
|
修改server.properties,增加 zookeeper.set.acl=true |
|
重启kafka集群（批量重启或滚动重启） |
|
启动zookeeper-security-migration.sh脚本,secure设置同步，unsecure取消同步 bin/zookeeper-security-migration –zookeeper.acl=secure –zookeeper.connect={host}:{port}/{path} |