Amazon Managed Streaming for?Apache Kafka?(Amazon MSK)是亞馬遜云推出的一個(gè)完全托管唤蔗、高度可用且安全的Apache Kafka服務(wù)。本文講述了如何在MSK中啟用用戶(hù)名和密碼的方式來(lái)完成認(rèn)證淌喻，同時(shí)也提及了ACL的配置。因?yàn)檫@部分內(nèi)容和自建的Kafka集群有些不同雀摘，所以想通過(guò)本文來(lái)做個(gè)演示裸删。本文步驟基本上按照MSK官方文檔，只是增加了截圖和例子阵赠，方便理解涯塔。
AmazonMSK使用SASL/SCRAM (Simple Authentication andSecurity Layer/ Salted Challenge Response Mechanism)來(lái)完成用戶(hù)名和密碼驗(yàn)證，它的前提是客戶(hù)端和代理之間需要使用TLS加密清蚀。所以在創(chuàng)建MSK集群時(shí)匕荸，需要選擇 客戶(hù)端和代理之間只允許TLS。此選項(xiàng)無(wú)法在創(chuàng)建后修改枷邪，所以如果你選擇了其他選項(xiàng)榛搔，必須要重新創(chuàng)建欣的集群。
在創(chuàng)建過(guò)程中選擇傳輸加密东揣，客戶(hù)端和代理之間 僅允許經(jīng)TLS加密的流量践惑。
身份驗(yàn)證選擇SASL/SCRAM身份驗(yàn)證。
AmazonMSK使用Amazon Secret Manager來(lái)存儲(chǔ)MSK所用的用戶(hù)名和密碼救斑。在設(shè)置Secret Manager之前童本，先要在AmazonKMS（Key Management Service）中創(chuàng)建一個(gè)客戶(hù)托管的Key。KMS是Amazon托管的密鑰管理服務(wù)脸候。
創(chuàng)建過(guò)程采用默認(rèn)選項(xiàng)即可穷娱，如果有特殊權(quán)限要求可以后續(xù)再改绑蔫。起名例如 kafka
4.? ? ?創(chuàng)建SecretsManager密鑰
AmazonSecrets Manager是一項(xiàng)密碼管理服務(wù)，可以把敏感的密碼等信息存放其中泵额，無(wú)須在代碼中嵌入配深，只要通過(guò)API調(diào)用就可以取得，避免在代碼中泄露嫁盲。
AmazonMSK通過(guò)Secrets Manager來(lái)存儲(chǔ)用戶(hù)名和密碼篓叶。在創(chuàng)建密碼時(shí)，注意密鑰名稱(chēng)必須具有前綴AmazonMSK_羞秤。
在Secrets Manager中選擇 其他類(lèi)型的密碼缸托，將用戶(hù)名和密碼填入，可以通過(guò)JSON格式瘾蛋，也可以通過(guò)Key/Value方式俐镐。加密的密鑰要選擇上面創(chuàng)建的KMS密鑰（非常重要）。
例如：
{
?"username": "alice",
?"password": "alice-secret"
}
等MSK集群創(chuàng)建完成后哺哼，就可以把密碼關(guān)聯(lián)到MSK集群了佩抹。
現(xiàn)在就可以來(lái)通過(guò)用戶(hù)名和密碼來(lái)連接MSK集群了。
請(qǐng)按照以下步驟來(lái)完成：
6.1.? ? ? ?運(yùn)行以下命令取董，同時(shí)替換?ClusterArn?為MSK集群的 Amazon 資源名稱(chēng) （ARN）：
aws kafka describe-cluster --cluster-arn "ClusterArn"
ClusterArn可以在MSK的集群界面查到：
運(yùn)行命令之后得到該集群的詳細(xì)信息：
從命令的 JSON 結(jié)果中棍苹，保存與名為“ZookeeperConnectString”的字符串關(guān)聯(lián)的值。
6.2.? ?在客戶(hù)端計(jì)算機(jī)上下載相應(yīng)版本的Kafka程序茵汰，具體鏈接可以查看https://kafka.apache.org/downloads枢里。
例如：
wget https://archive.apache.org/dist/kafka/2.6.1/kafka_2.13-2.6.1.tgz
下載完后解壓。
6.3.? ? ? ? 在客戶(hù)端計(jì)算機(jī)上kafka程序的?bin?目錄中運(yùn)行以下命令來(lái)在集群中創(chuàng)建主題经窖，替換?ZookeeperConnectString?為之前記錄的字符串坡垫。如果集群的節(jié)點(diǎn)數(shù)小于3個(gè)，replication-factor要設(shè)置小于3.
./kafka-topics.sh --create
--zookeeper ZookeeperConnectString--replication-factor 3 --partitions 1 --topicExampleTopicName
6.4.? ? ? ? 在客戶(hù)端計(jì)算機(jī)上画侣，創(chuàng)建一個(gè) JAAS 配置文件冰悠。例如，對(duì)于用戶(hù)?alice配乱，創(chuàng)建一個(gè)名為?users_jaas.conf?的文件溉卓，該文件包含以下內(nèi)容：
KafkaClient {
??org.apache.kafka.common.security.scram.ScramLoginModule required
?? username="alice"
?? password="alice-secret";
};
6.5.? ? ? ? 將此 JAAS 配置文件設(shè)置為 KAFKA_OPTS 環(huán)境變量：
export KAFKA_OPTS=-Djava.security.auth.login.config=<path-to-jaas-file>/users_jaas.conf
6.6.? ? ? ? 將 JDK 密鑰庫(kù)文件從 JVM 復(fù)制到用戶(hù)所需目錄。JDK的目錄根據(jù)安裝版本不同可能不同搬泥。
cp /usr/lib/jvm/java-1.8.0-openjdk-1.8.0.272.b10-1.amzn2.0.1.x86_64/jre/lib/security/cacerts?<path-to-keystore-file>/kafka.client.truststore.jks
6.7.? ? ? ?使用以下內(nèi)容創(chuàng)建名為?client_sasl.properties?的客戶(hù)端屬性文件桑寨。此文件定義 SASL 機(jī)制和協(xié)議，將被用來(lái)訪問(wèn)Kafka集群忿檩。MSK當(dāng)前只支持SCRAM-SHA-512認(rèn)證尉尾。
security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
ssl.truststore.location=<path-to-keystore-file>/kafka.client.truststore.jks
6.8. 接下來(lái)要取得MSK的代理（Broker）地址，替換?ClusterArn?替換為MSK集群的 Amazon 資源名稱(chēng) （ARN）：
aws kafka get-bootstrap-brokers
--cluster-arn ClusterArn
從命令的 JSON 結(jié)果中燥透，保存與名為“BootstrapBrokerStringSaslScram”的字符串關(guān)聯(lián)的值沙咏。
6.9.? ?現(xiàn)在可以通過(guò)Producer在Topic中生產(chǎn)數(shù)據(jù)了辨图。在Kafka客戶(hù)端的?bin?目錄中運(yùn)行以下命令，并將?BootstrapBrokerStringSaslScram?替換為上一個(gè)命令時(shí)獲取的值肢藐。
./kafka-console-producer.sh --broker-list BootstrapBrokerStringSaslScram --topic
ExampleTopicName --producer.config client_sasl.properties
輸入任意消息：
6.10.? ? ? ? ? ? ? ? 打開(kāi)另一窗口來(lái)通過(guò)Consumer消費(fèi)該Topic中的消息故河，在Kafka客戶(hù)端的?bin?目錄中運(yùn)行以下命令，并將?BootstrapBrokerStringSaslScram?替換為之前獲取的值吆豹。
./kafka-console-consumer.sh --bootstrap-server BootstrapBrokerStringSaslScram --topic
ExampleTopicName --from-beginning --consumer.config client_sasl.properties
可以看到之前鍵入的消息：
上文演示了單個(gè)用戶(hù)的過(guò)程鱼的，如果需要?jiǎng)?chuàng)建其他用戶(hù)，請(qǐng)重復(fù)步驟3-4來(lái)創(chuàng)建Secrets Manager的密鑰痘煤，并關(guān)聯(lián)到MSK集群凑阶。需要注意：
一個(gè)Amazon MSK集群最多可以有1000個(gè)用戶(hù)。
與Amazon MSK集群關(guān)聯(lián)的密鑰的名稱(chēng)必須具有前綴AmazonMSK_速勇。
ACL可以用來(lái)控制生產(chǎn)者（Producer）晌砾，消費(fèi)者（Consumer）和管理員（admin）對(duì)于Topic的權(quán)限控制坎拐。
Apache Kafka ACLs的格式為“Principal P is [Allowed/Denied] Operation O From Host Hon any Resource R matching ResourcePattern RP”烦磁。如果RP與特定資源R不匹配，則R沒(méi)有關(guān)聯(lián)的ACLs哼勇，因此不允許除超級(jí)用戶(hù)之外的任何人訪問(wèn)R都伪。
默認(rèn)情況下，AmazonMSK會(huì)將allow.everyone.if.no.acl.found設(shè)置為true积担。這意味著陨晶，如果沒(méi)有顯式設(shè)置ACL，則所有委托人都可以訪問(wèn)此資源帝璧。只有啟用了ACL先誉，才能限制只有授權(quán)的委托人才能訪問(wèn)它。
1.? 首先將MSK的代理添加到ACL表的烁，以允許它們讀取已配置ACL的所有主題褐耳。這可以保證代理（Broker節(jié)點(diǎn)）可以正常同步。
替換?ZooKeeper-Connection-String?替換為上文取得的連接字符串渴庆。
替換Distinguished-Name?替換為集群中任一代理的名字铃芦，然后將此第一個(gè)點(diǎn)之前的字符串替換為星號(hào) （*）。
bin/kafka-acls.sh --authorizer-properties zookeeper.connect=ZooKeeper-Connection-String --add
--allow-principal "User:CN=Distinguished-Name"
--operation Read --group=* --topic Topic-Name
?
8.2. 要授予某個(gè)用戶(hù)對(duì)主題的讀訪問(wèn)權(quán)襟雷，請(qǐng)?jiān)诳蛻?hù)端計(jì)算機(jī)上運(yùn)行以下命令刃滓。
bin/kafka-acls.sh --authorizer-properties zookeeper.connect=ZooKeeper-Connection-String --add
--allow-principal "User:CN=Distinguished-Name"
--operation Read --group=* --topic Topic-Name
例如
8.3.? ? ? 要授予對(duì)主題的寫(xiě)訪問(wèn)權(quán)，請(qǐng)?jiān)诳蛻?hù)端計(jì)算機(jī)上運(yùn)行以下命令
bin/kafka-acls.sh --authorizer-properties zookeeper.connect=ZooKeeper-Connection-String--add
--allow-principal "User:CN=Distinguished-Name" --operationWrite --topic Topic-Name
例如
8.4.? ? ? ? 如果使用未授權(quán)的用戶(hù)耸弄，將會(huì)出現(xiàn)沒(méi)有授權(quán)的錯(cuò)誤咧虎，無(wú)法讀取Topic內(nèi)容。
9. 結(jié)束
本文演示了如何在Amazon MSK中啟用用戶(hù)名和密碼驗(yàn)證以及簡(jiǎn)單的ACL設(shè)置计呈，其它未涉及的內(nèi)容以及更多信息砰诵，請(qǐng)參考AmazonWeb Services的MSK官方文檔僚饭。如有疑問(wèn)，歡迎探討胧砰。
