kafak權(quán)限管理

背景：最近公司因?yàn)橛玫脑品?wù)器，需要保證kafka的安全性熄诡。可喜的是kafka0.9開始诗力，已經(jīng)支持權(quán)限控制了凰浮。網(wǎng)上中文資料又少，特此基于kafka0.9苇本，記錄kafaka的權(quán)限控制 ( flume需要1.7及其以上才支持kafka的SSL認(rèn)證)袜茧。

下面各位看官跟著小二一起開始kafak權(quán)限認(rèn)證之旅吧！嘎嘎嘎瓣窄！

介紹：kafka權(quán)限控制整體可以分為三種類型：1.基于SSL(CDH 5.8不支持)2.基于Kerberos(此認(rèn)證一般基于CDH笛厦，本文不與討論)3.基于acl的 (CDH5.8中的kafka2.X不支持 )

本文主要基于apace版本的，實(shí)現(xiàn)1和3俺夕，也是用的最多的展開討論裳凸。

統(tǒng)一說明：在本文中&符號(hào)表示注釋

一，準(zhǔn)備工作組件分布kafka centos11劝贸，centos12姨谷，centos13zoopeeker centos11，centos12映九，centos13

二梦湘、在kafka集群任選一臺(tái)機(jī)子 ( 先介紹基于SSL的 )

密碼統(tǒng)一為123456

&Step 1 Generate SSL key and certificate for each Kafka broker
keytool -keystore server.keystore.jks -alias centos11 -validity 365 -genkey
?
%Step 2 Creating your own CA
openssl req -new -x509 -keyout ca-key -out ca-cert -days 365
keytool -keystore server.truststore.jks -alias CARoot -import -file ca-cert
keytool -keystore client.truststore.jks -alias CARoot -import -file ca-cert
?
&Step 3 Signing the certificate
keytool -keystore server.keystore.jks -alias centos11 -certreq -file cert-file
openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-file -out cert-signed -days 365 -CAcreateserial -passin pass:123456
keytool -keystore server.keystore.jks -alias CARoot -import -file ca-cert
keytool -keystore server.keystore.jks -alias centos11 -import -file cert-signed

三、其他的kafka集群

&機(jī)器centos13 centos12
keytool -keystore kafka.client.keystore.jks -alias centos13 -validity 365 -genkey
keytool -keystore kafka.client.keystore.jks -alias centos13 -certreq -file cert-file
cp cert-file cert-file-centos13

&centos11
scp ./ca* ce* server* root@centos13:/opt/kafka_2.10/

openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-file-centos13 -out cert-signed-centos13 -days 365 -CAcreateserial -passin pass:123456
keytool -keystore kafkacentos13.client.keystore.jks -alias CARoot -import -file ca-cert
keytool -keystore kafkacentos13.client.keystore.jks -alias centos13 -import -file cert-signed-centos13
?
?
?
rm -rf producer.properties
echo "bootstrap.servers=centos13:9093" >> producer.properties
echo "security.protocol=SSL" >> producer.properties
echo "ssl.truststore.location=/opt/kafka_2.10/kafkacentos12.client.keystore.jks">> producer.properties
echo "ssl.truststore.password=123456">> producer.properties
echo "ssl.keystore.location=/opt/kafka_2.10/server.keystore.jks">> producer.properties
echo "ssl.keystore.password=123456">> producer.properties
echo "ssl.key.password=123456">> producer.properties

4.驗(yàn)證：

openssl s_client -debug -connect localhost:9093 -tls1
?
output:
-----BEGIN CERTIFICATE-----
{variable sized random bytes}
-----END CERTIFICATE-----
subject=/C=US/ST=CA/L=Santa Clara/O=org/OU=org/CN=Sriharsha Chintalapani
issuer=/C=US/ST=CA/L=Santa Clara/O=org/OU=org/CN=kafka/emailAddress=test@test.com

5.使用：

bin/kafka-console-consumer.sh --bootstrap-server kafka2:9093 --topic test --new-consumer --consumer.config config/producer.properties
?
bin/kafka-console-producer.sh --broker-list centos11:9093 --topic test --producer.config ./config/producer.properties
?
bin/kafka-console-consumer.sh --bootstrap-server centos11:9093 --topic test --new-consumer --consumer.config ./config/producer.properties
?
bin/kafka-console-consumer.sh --bootstrap-server centos13:9093 --topic test --new-consumer --consumer.config ./config/producer.properties --from-beginning

6.基于ACL

server.properties中加配置

allow.everyone.if.no.acl.found=true
super.users=User:root
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
principal.builder.class=org.apache.kafka.common.security.auth.DefaultPrincipalBuilder

7.ACL的簡(jiǎn)單使用：

bin/kafka-acls.sh --authorizer-properties zookeeper.connect=centos11:2181 --add --allow-principal User:Bob --allow-principal User:Alice --allow-host 198.51.100.0 --allow-host 198.51.100.1 --operation Read --operation Write --topic test
?
bin/kafka-acls.sh --authorizer-properties zookeeper.connect=centos11:2181 --remove --allow-principal User:Bob --allow-principal User:Alice --allow-host 198.51.100.0 --allow-host 198.51.100.1 --operation Read --operation Write --topic test

8.Java Demo需要將server.keystore.jks、client.truststore.jks從任一臺(tái)機(jī)器上拷貝下來即可践叠。

SSL is supported only for the new Kafka Producer and Consumer, the older API is not supported

ConsumerDemo

package xmhd.examples;
?
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.config.SslConfigs;
import java.util.Arrays;
import java.util.Properties;
/**

Created by shengjk1.
blog address :http://blog.csdn.net/jsjsjs1789
生產(chǎn)者可以保證權(quán)限認(rèn)證
SSL is supported only for the new Kafka Producer and Consumer, the older API is not supported
/
public class ConsumerZbdba {
public static void main(String[] args) {
// new ConsumerZbdba("test").start();// 使用kafka集群中創(chuàng)建好的主題 test
Properties props = new Properties();
/ 定義kakfa 服務(wù)的地址言缤，不需要將所有broker指定上 /
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "centos11:9093;centos13:9093;centos12:9093");
/ 制定consumer group /
props.put("group.id", "test");
props.put("auto.offset.reset","earliest");
/ 是否自動(dòng)確認(rèn)offset /
// props.put("enable.auto.commit", "true");
// props.put(ProducerConfig.CLIENT_ID_CONFIG, "myApiKey");
props.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, "F:\server.keystore.jks");
props.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, "123456");
props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, "F:\client.truststore.jks");
props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "123456");
props.put(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, "JKS");
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
//
?
?
/// 自動(dòng)確認(rèn)offset的時(shí)間間隔 /
// props.put("auto.commit.interval.ms", "1000");
// props.put("session.timeout.ms", "30000");
/ key的序列化類 /
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
/ value的序列化類 /
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
/ 定義consumer /
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
/ 消費(fèi)者訂閱的topic, 可同時(shí)訂閱多個(gè) /
consumer.subscribe(Arrays.asList("test"));
?
/ 讀取數(shù)據(jù)，讀取超時(shí)時(shí)間為100ms */
while (true) {
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records)
System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value()+"\n");
}
}
}

ProducerDemo

package xmhd.examples;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.config.SslConfigs;
import java.util.Properties;
/**

Created by shengjk1.
blog address :http://blog.csdn.net/jsjsjs1789
生產(chǎn)者可以保證權(quán)限認(rèn)證
*/
public class ProducerZbdba {
public static void main(String[] args) {
Properties producerProps = new Properties();
producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "centos12:9093");
// producerProps.put(ProducerConfig.CLIENT_ID_CONFIG, "myApiKey");
producerProps.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, "F:\server.keystore.jks");
producerProps.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, "123456");
producerProps.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, "F:\client.truststore.jks");
producerProps.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "123456");
producerProps.put(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, "JKS");
producerProps.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
KafkaProducer producer = new KafkaProducer(producerProps);
for(int i = 0; i < 100; i++)
producer.send(new ProducerRecord<String, String>("test", Integer.toString(i), Integer.toString(i)));
System.out.println("test");
producer.close();
}
}

8.1 flume1.7 的配置 (基于kafka SSL認(rèn)證)

tier1.sources = source1
tier1.channels = channel1
tier1.sinks = sink1
?
?
tier1.sources.source1.type = exec
tier1.sources.source1.command = tail -F -n+1 /opt/scan.log
tier1.sources.source1.channels = channel1
?
tier1.channels.channel1.type = memory
tier1.sinks.sink1.type = org.apache.flume.sink.kafka.KafkaSink
tier1.sinks.sink1.topic = test
tier1.sinks.sink1.kafka.bootstrap.servers = centos11:9093,centos12:9093,centos13:9093
tier1.channels.channel1.kafka.bootstrap.servers = centos11:9093,centos12:9093,centos13:9093
tier1.sinks.sink1.requiredAcks = 1
tier1.sinks.sink1.batchSize = 100
?
?
tier1.sinks.sink1.kafka.producer.security.protocol = SSL
tier1.sinks.sink1.kafka.producer.ssl.truststore.type=JKS
tier1.sinks.sink1.kafka.producer.ssl.truststore.location = /opt/kafka_2.10/server.truststore.jks
tier1.sinks.sink1.kafka.producer.ssl.truststore.password =123456
tier1.sinks.sink1.kafka.producer.security.protocol = SSL
tier1.sinks.sink1.kafka.producer.ssl.keystore.location = /opt/kafka_2.10/server.keystore.jks
tier1.sinks.sink1.kafka.producer.ssl.keystore.password =123456
&tier1.sinks.sink1.kafka.producer.ssl.endpoint.identification.algorithm = HTTPS
?
?
tier1.sinks.sink1.channel = channel1
tier1.channels.channel1.capacity = 100
?
?
tier1.channels.channel1.kafka.producer.security.protocol = SSL
tier1.channels.channel1.kafka.producer.ssl.truststore.type=JKS
tier1.channels.channel1.kafka.producer.ssl.truststore.location = /opt/kafka_2.10/server.truststore.jks
tier1.channels.channel1.kafka.producer.ssl.truststore.password =123456
tier1.channels.channel1.kafka.producer.security.protocol = SSL
tier1.channels.channel1.kafka.producer.ssl.keystore.location = /opt/kafka_2.10/server.keystore.jks
tier1.channels.channel1.kafka.producer.ssl.keystore.password =123456
tier1.channels.channel1.kafka.producer.ssl.endpoint.identification.algorithm = HTTPS
?
?
tier1.channels.channel1.kafka.consumer.security.protocol = SSL
tier1.channels.channel1.kafka.consumer.ssl.truststore.location = /opt/kafka_2.10/server.truststore.jks
tier1.channels.channel1.kafka.consumer.ssl.truststore.password =123456
tier1.channels.channel1.kafka.consumer.security.protocol = SSL
tier1.channels.channel1.kafka.consumer.ssl.keystore.location = /opt/kafka_2.10/server.truststore.jks
tier1.channels.channel1.kafka.consumer.ssl.keystore.password =123456
tier1.channels.channel1.kafka.consumer.ssl.endpoint.identification.algorithm = HTTPS

9.kafak Server.properties最終版的

三臺(tái)機(jī)子需保證一樣禁灼，centos11為機(jī)器名管挟，根據(jù)需要自行修改

broker.id=0
############################# Socket Server Settings #############################
&這一點(diǎn)可能需要特別的注意,PLAINTEXT注釋掉之后，一些基本的kafka腳本都不在能用了
&listeners=PLAINTEXT://centos11:9092,SSL://centos11:9093
listeners=SSL://centos11:9093
advertised.listeners=SSL://centos11:9093
ssl.keystore.location=/opt/kafka_2.10/server.keystore.jks
ssl.keystore.password=123456
ssl.key.password=123456
ssl.truststore.location=/opt/kafka_2.10/server.truststore.jks
ssl.truststore.password=123456
ssl.client.auth=required
ssl.enabled.protocols=TLSv1.2,TLSv1.1,TLSv1
ssl.keystore.type=JKS
ssl.truststore.type=JKS
security.inter.broker.protocol=SSL
?
&acl
allow.everyone.if.no.acl.found=true
super.users=User:root
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
principal.builder.class=org.apache.kafka.common.security.auth.DefaultPrincipalBuilder
?
host.name=centos11
advertised.host.name=centos11
?
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
############################# Log Basics #############################
log.dirs=/opt/kafka-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
############################# Zookeeper #############################
zookeeper.connect=centos11:2181,centos12:2181,centos13:2181
zookeeper.connection.timeout.ms=6000</pre>

kafka producer.propertiescentos11為機(jī)器名弄捕，根據(jù)需求自行修改

bootstrap.servers=centos11:9093
security.protocol=SSL
ssl.truststore.location=/opt/kafka_2.10/client.truststore.jks
ssl.truststore.password=123456
ssl.keystore.location=/opt/kafka_2.10/server.keystore.jks
ssl.keystore.password=123456
ssl.key.password=123456</pre>

10僻孝、參考網(wǎng)址：具體細(xì)節(jié)可參考官網(wǎng)！

http://kafka.apache.org/090/documentation.html#security_authz http://www.cloudera.com/documentation/kafka/latest/topics/kafka_security.html http://blog.csdn.net/zbdba/article/details/52458654

11.通信協(xié)議的支持情況

image

12.擴(kuò)展閱讀：

關(guān)于SSL原理 http://www.linuxde.net/2012/03/8301.html

最后編輯于：2018.05.07 13:04:15

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者

人面猴
序言：七十年代末守谓，一起剝皮案震驚了整個(gè)濱河市穿铆，隨后出現(xiàn)的幾起案子，更是在濱河造成了極大的恐慌斋荞，老刑警劉巖荞雏，帶你破解...
沈念sama閱讀 221,635評(píng)論 6贊 515
死咒
序言：濱河連續(xù)發(fā)生了三起死亡事件，死亡現(xiàn)場(chǎng)離奇詭異平酿，居然都是意外死亡凤优，警方通過查閱死者的電腦和手機(jī)，發(fā)現(xiàn)死者居然都...
沈念sama閱讀 94,543評(píng)論 3贊 399
救了他兩次的神仙讓他今天三更去死
文/潘曉璐我一進(jìn)店門蜈彼，熙熙樓的掌柜王于貴愁眉苦臉地迎上來筑辨，“玉大人，你說我怎么就攤上這事幸逆」髟” “怎么了？”我有些...
開封第一講書人閱讀 168,083評(píng)論 0贊 360
道士緝兇錄：失蹤的賣姜人
文/不壞的土叔我叫張陵还绘，是天一觀的道長(zhǎng)楚昭。經(jīng)常有香客問我，道長(zhǎng)拍顷，這世上最難降的妖魔是什么哪替？我笑而不...
開封第一講書人閱讀 59,640評(píng)論 1贊 296
?港島之戀（遺憾婚禮）
正文為了忘掉前任，我火速辦了婚禮菇怀，結(jié)果婚禮上凭舶，老公的妹妹穿的比我還像新娘。我一直安慰自己爱沟，他們只是感情好帅霜，可當(dāng)我...
茶點(diǎn)故事閱讀 68,640評(píng)論 6贊 397
惡毒庶女頂嫁案：這布局不是一般人想出來的
文/花漫我一把揭開白布。她就那樣靜靜地躺著呼伸，像睡著了一般身冀。火紅的嫁衣襯著肌膚如雪钝尸。梳的紋絲不亂的頭發(fā)上，一...
開封第一講書人閱讀 52,262評(píng)論 1贊 308
城市分裂傳說
那天搂根，我揣著相機(jī)與錄音珍促，去河邊找鬼。笑死剩愧，一個(gè)胖子當(dāng)著我的面吹牛猪叙，可吹牛的內(nèi)容都是我干的。我是一名探鬼主播仁卷，決...
沈念sama閱讀 40,833評(píng)論 3贊 421
雙鴛鴦連環(huán)套：你想象不到人心有多黑
文/蒼蘭香墨我猛地睜開眼穴翩，長(zhǎng)吁一口氣：“原來是場(chǎng)噩夢(mèng)啊……” “哼！你這毒婦竟也來了锦积？” 一聲冷哼從身側(cè)響起芒帕，我...
開封第一講書人閱讀 39,736評(píng)論 0贊 276
萬榮殺人案實(shí)錄
序言：老撾萬榮一對(duì)情侶失蹤，失蹤者是張志新（化名）和其女友劉穎丰介，沒想到半個(gè)月后背蟆，有當(dāng)?shù)厝嗽跇淞掷锇l(fā)現(xiàn)了一具尸體，經(jīng)...
沈念sama閱讀 46,280評(píng)論 1贊 319
?護(hù)林員之死
正文獨(dú)居荒郊野嶺守林人離奇死亡哮幢，尸身上長(zhǎng)有42處帶血的膿包…… 初始之章·張勛以下內(nèi)容為張勛視角年9月15日...
茶點(diǎn)故事閱讀 38,369評(píng)論 3贊 340
?白月光啟示錄
正文我和宋清朗相戀三年淆储，在試婚紗的時(shí)候發(fā)現(xiàn)自己被綠了。大學(xué)時(shí)的朋友給我發(fā)了我未婚夫和他白月光在一起吃飯的照片家浇。...
茶點(diǎn)故事閱讀 40,503評(píng)論 1贊 352
活死人
序言：一個(gè)原本活蹦亂跳的男人離奇死亡，死狀恐怖碴裙，靈堂內(nèi)的尸體忽然破棺而出钢悲，到底是詐尸還是另有隱情，我是刑警寧澤舔株，帶...
沈念sama閱讀 36,185評(píng)論 5贊 350
?日本核電站爆炸內(nèi)幕
正文年R本政府宣布莺琳，位于F島的核電站，受9級(jí)特大地震影響载慈，放射性物質(zhì)發(fā)生泄漏惭等。R本人自食惡果不足惜，卻給世界環(huán)境...
茶點(diǎn)故事閱讀 41,870評(píng)論 3贊 333
男人毒藥：我在死后第九天來索命
文/蒙蒙一办铡、第九天我趴在偏房一處隱蔽的房頂上張望辞做。院中可真熱鬧，春花似錦寡具、人聲如沸秤茅。這莊子的主人今日做“春日...
開封第一講書人閱讀 32,340評(píng)論 0贊 24
一樁弒父案童叠，背后竟有這般陰謀
文/蒼蘭香墨我抬頭看了看天上的太陽框喳。三九已至，卻和暖如春，著一層夾襖步出監(jiān)牢的瞬間五垮，已是汗流浹背乍惊。一陣腳步聲響...
開封第一講書人閱讀 33,460評(píng)論 1贊 272
情欲美人皮
我被黑心中介騙來泰國打工，沒想到剛下飛機(jī)就差點(diǎn)兒被人妖公主榨干…… 1. 我叫王不留放仗，地道東北人润绎。一個(gè)月前我還...
沈念sama閱讀 48,909評(píng)論 3贊 376
代替公主和親
正文我出身青樓，卻偏偏與公主長(zhǎng)得像匙监，于是被迫代替她去往敵國和親凡橱。傳聞我的和親對(duì)象是個(gè)殘疾皇子，可洞房花燭夜當(dāng)晚...
茶點(diǎn)故事閱讀 45,512評(píng)論 2贊 359

kafak權(quán)限管理

推薦閱讀更多精彩內(nèi)容