一、
生成server broker證書倉庫：
# 生成server證書倉庫，alias指明一個公鑰私鑰對的入口名 keytool -keystore server.jks -alias server -validity 365 -genkey -keyalg RSA -storetype pkcs12 echo 01 > serial.txt touch index.txt
二、
生成自簽名證書（ca、根證書）：
# ca.cfg參考kafka官方文檔 openssl req -x509 -config ca.cfg -newkey rsa:4096 -sha256 -nodes -out cacert.pem -outform PEM
ca.cfg參考官網給出的：
HOME = . RANDFILE = $ENV::HOME/.rnd #################################################################### [ ca ] default_ca = CA_default # The default ca section [ CA_default ] base_dir = . certificate = $base_dir/cacert.pem # The CA certifcate private_key = $base_dir/cakey.pem # The CA private key new_certs_dir = $base_dir # Location for new certs after signing database = $base_dir/index.txt # Database index file serial = $base_dir/serial.txt # The current serial number default_days = 1000 # How long to certify for default_crl_days = 30 # How long before next CRL default_md = sha256 # Use public key default MD preserve = no # Keep passed DN ordering x509_extensions = ca_extensions # The extensions to add to the cert email_in_dn = no # Don't concat the email in the DN copy_extensions = copy # Required to copy SANs from CSR to cert #################################################################### [ req ] default_bits = 4096 default_keyfile = cakey.pem distinguished_name = ca_distinguished_name x509_extensions = ca_extensions string_mask = utf8only #################################################################### [ ca_distinguished_name ] countryName = Country Name (2 letter code) countryName_default = DE stateOrProvinceName = State or Province Name (full name) stateOrProvinceName_default = Test Province localityName = Locality Name (eg, city) localityName_default = Test Town organizationName = Organization Name (eg, company) organizationName_default = Test Company organizationalUnitName = Organizational Unit (eg, division) organizationalUnitName_default = Test Unit commonName = Common Name (e.g. server FQDN or YOUR name) commonName_default = Test Name emailAddress = Email Address emailAddress_default = test@test.com #################################################################### [ ca_extensions ] subjectKeyIdentifier = hash authorityKeyIdentifier = keyid:always, issuer basicConstraints = critical, CA:true keyUsage = keyCertSign, cRLSign #################################################################### [ signing_policy ] countryName = optional stateOrProvinceName = optional localityName = optional organizationName = optional organizationalUnitName = optional commonName = supplied emailAddress = optional #################################################################### [ signing_req ] subjectKeyIdentifier = hash authorityKeyIdentifier = keyid,issuer basicConstraints = CA:FALSE keyUsage = digitalSignature, keyEncipherment
有了ca后，生成服務端ca證書倉庫，並把ca加到服務端ca倉庫中：
keytool -keystore server.ca.jks -alias ca -import -file cacert.pem
三、
簽名服務端證書：
# 生成服務端簽名請求 keytool -certreq -keystore server.jks -alias server -file server.csr # 用ca簽名server.csr。policy、extensions 這些從ca.cfg里面的配置取 openssl ca -config ca.cfg -policy signing_policy -extensions signing_req -out server.pem -infiles server.csr # 把服務端證書加入證書倉庫 # 注意要先導入cacert再導入server證書，否則報：keytool error: java.lang.Exception: Failed to establish chain from reply # alias用server這個入口，與之前的秘鑰配成一對 keytool -keystore server.jks -alias ca -import -file cacert.pem keytool -keystore server.jks -alias server -import -file server.pem
注意：在填寫COMMON NAME（Common Name (e.g. server FQDN or YOUR name)）的時候，應該與服務端的域名一致，例如本例中就是local.kafka
否則客戶端連接的時候會報java.security.cert.CertificateException: No name match
四、
客戶端證書部分：
# 生成客戶端證書倉庫 keytool -keystore client.jks -alias client -validity 365 -genkey -keyalg RSA -storetype pkcs12 # 生成客戶端證書簽名請求 keytool -certreq -keystore client.jks -alias client -file client.csr
把client.csr扔給服務端（本案例是把ca放在服務端保存的），讓ca簽名：
# 在服務端執行 openssl ca -config ca.cfg -policy signing_policy -extensions signing_req -out client.pem -infiles client.csr
簽好名后，把ca、客戶端證書扔回給客戶端：
# 回到客戶端執行 # 把客戶端證書加入客戶端倉庫 # 仍要注意先加入ca再加入客戶端證書 # alias用client這個入口，配成公鑰私鑰對 keytool -keystore client.jks -alias ca -import -file cacert.pem keytool -keystore client.jks -alias client -import -file client.pem # 生成客戶端ca倉庫 keytool -keystore client.ca.jks -alias ca -import -file cacert.pem
五、
服務端。server.properties配置：
listeners=PLAINTEXT://0.0.0.0:9092,SSL://0.0.0.0:9093 #0.0.0.0表示接受任何ip的請求 advertised.listeners=PLAINTEXT://local.kafka:9092,SSL://local.kafka:9093 #hosts里面用local.kafka這個域名來訪問這個broker ssl.keystore.location=/usr/local/src/kafka_2.13-2.6.0/config/ssl/server.jks #證書倉庫，放了server的公鑰私鑰對、ca ssl.keystore.password=YOURPASSWORD ssl.key.password=YOURPASSWORD ssl.truststore.location=/usr/local/src/kafka_2.13-2.6.0/config/ssl/server.ca.jks #CA書倉庫，專門放ca ssl.truststore.password=YOURPASSWORD ssl.client.auth=required #可選，具體參考kafka手冊 ssl.enable.protocol=TLSv1.2 #可選，指定tls版本，具體參考kafka手冊 ssl.keystore.type=JKS #可選，說明keystore類型 ssl.truststore.type=JKS
客戶端。使用comsumer作為例子，新建consumer-ssl.properties：
security.protocol=SSL ssl.truststore.location=/usr/local/src/kafka_2.13-2.6.0/config/ssl/client.ca.jks ssl.truststore.password=YOURPASSWORD ssl.keystore.location=/usr/local/src/kafka_2.13-2.6.0/config/ssl/client.jks ssl.keystore.password=YOURPASSWORD ssl.key.password=YOURPASSWORD ssl.enabled.protocols=TLSv1.2 ssl.keystore.type=JKS ssl.truststore.type=JKS
./kafka-console-consumer.sh .... --consumer.config consumer-ssl.properties 進行加密消費