Kafka の SSL/TLS 暗号化: クラスタ通信と証明書管理の要点

Kafka の管理者試験では、リスナー設計、相互認証(mTLS)、証明書の取り回し、そして検証・運用の手順が頻出です。ここでは公式ドキュメントに沿った安定概念だけに絞り、試験でも現場でも通用する形でまとめます。

前提として、Kafka は broker↔client と broker↔broker の両方を TLS で保護できます。ZooKeeper 併用型/ KRaft モードいずれでも基本原理は同じで、証明書の正当性、SAN、リスナーの整合性が鍵になります。

TLS の適用範囲とアーキテクチャの全体像

Kafka では、クライアント通信とブローカー間レプリケーションの双方に TLS を適用できます。試験/実務で混乱しやすいのは「どのリスナーがどの経路を担うか」と「証明書の信頼関係」です。

最低限、クライアント向けリスナーを SSL もしくは SASL_SSL とし、inter.broker.listener.name で指定する内部レプリケーション用リスナーも暗号化する方針を取ります。証明書はブローカーごとに固有の鍵と SAN を持たせ、共通の信頼できる CA で署名します。

• broker↔client: SSL もしくは SASL_SSL。mTLS を使うなら ssl.client.auth=required。
• broker↔broker: inter.broker.listener.name が指すリスナーを SSL または SASL_SSL に設定。
• SAN 必須: DNS 名や必要に応じて IP を Subject Alternative Name に含めること。CN のみは不可。
• 同一 CA でチェーンを揃える。中間 CA を truststore に含め忘れない。

Kafka TLS 配置の俯瞰

ブローカーごとの証明書要件メモ

# 各ブローカーで固有の鍵/証明書
# SAN に FQDN (例: broker1.example.com) と必要なら IP を含める
# すべて同じ信頼された CA で署名
# truststore には中間 CA を含めて完全なチェーンを構成

リスナー設計と inter.broker.listener の要点

Kafka は複数リスナーを持てます。クライアント向けとブローカー間向けを分け、後者を inter.broker.listener.name で指定します。どのリスナーがどのプロトコルかは listener.security.protocol.map で定義します。

advertised.listeners はクライアントが接続時に参照する公開名で、証明書の SAN と一致させるのが鉄則です。一致しないとホスト名検証で失敗します。

• listeners と advertised.listeners は両方設定。広告名は外部から解決可能な DNS を使う。
• inter.broker.listener.name は map のキー名を指定。例: INTERNAL。
• mTLS を要求するリスナーでは ssl.client.auth=required。
• ブローカー間を SSL にしたら、全ブローカーが同一ポリシーで揃っている必要がある。

server.properties の代表設定例（クライアントと内部で分離）

listeners=INTERNAL://0.0.0.0:9092,EXTERNAL://0.0.0.0:9093
advertised.listeners=INTERNAL://broker1.example.com:9092,EXTERNAL://broker1.example.com:9093
listener.security.protocol.map=INTERNAL:SSL,EXTERNAL:SSL
inter.broker.listener.name=INTERNAL

# TLS 共通設定（JKS/PKCS12 どちらでも可。近年は PEM もサポートされるが試験では JKS/PKCS12 が無難）
ssl.keystore.location=/var/private/ssl/kafka.broker1.p12
ssl.keystore.password=${FILE:/etc/kafka/keystore.pass}
ssl.keystore.type=PKCS12
ssl.truststore.location=/var/private/ssl/kafka.truststore.p12
ssl.truststore.password=${FILE:/etc/kafka/truststore.pass}
ssl.truststore.type=PKCS12

# クライアント向けリスナーで相互認証を要求する場合
ssl.client.auth=required

通信モード比較: TLS 単方向, mTLS, SASL_SSL

用途に応じて、証明書のみで暗号化する単方向 TLS、クライアント証明書も要求する mTLS、あるいは TLS 上に SASL（PLAIN/SCRAM/OAUTHBEARER など）を重ねる方式を選びます。

管理者試験では、どのモードが何を証明し、どのストアが必要か、どの設定が肝かを正しく選べることが問われます。

• 単方向 TLS: サーバ認証と暗号化。クライアントは truststore のみ。
• mTLS: 相互認証。クライアントにも keystore が必要で、ブローカー側で ssl.client.auth=required。
• SASL_SSL: TLS で暗号化しつつ、ID は SASL で与える。運用では SCRAM が一般的。

SASL_SSL（SCRAM）の最小追加設定（ブローカー側）

# 例: INTERNAL は SSL、EXTERNAL を SASL_SSL に
listener.security.protocol.map=INTERNAL:SSL,EXTERNAL:SASL_SSL
listeners=INTERNAL://0.0.0.0:9092,EXTERNAL://0.0.0.0:9094
advertised.listeners=INTERNAL://broker1.example.com:9092,EXTERNAL://broker1.example.com:9094
inter.broker.listener.name=INTERNAL

# SASL 関連
sasl.enabled.mechanisms=SCRAM-SHA-512
listener.name.external.scram-sha-512.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required;
# 注意: 実際のユーザ作成は kafka-configs.sh --alter --add-config 'SCRAM-SHA-512=[password=...]' などで実施

証明書管理: 発行、インポート、ローテーション

各ブローカーは固有の秘密鍵と証明書を持ち、信頼できる CA による署名を受けます。SAN に FQDN を必ず含め、advertised.listeners と一致させます。中間 CA を含む完全なチェーンを truststore に格納することが重要です。

ローテーションはローリングで行い、先に truststore に新 CA を追加→新証明書を配布→各ノードを段階的に再起動の順が安全です。パスワードやファイル権限は最小権限で管理します。

• 形式: JKS または PKCS12 が安定。Kafka は近年 PEM もサポートするが、試験では JKS/PKCS12 前提で覚えると安全。
• SAN 不備やホスト名不一致は最頻出の障害要因。
• 有効期限監視を自動化し、期限切れでのダウンを回避。

keytool/openssl による代表的な手順

# 1) キーストア作成と鍵・CSR 生成（PKCS12）
keytool -genkeypair -alias broker1 -keyalg RSA -keystore kafka.broker1.p12 -storetype PKCS12 -storepass changeit -keypass changeit -dname "CN=broker1.example.com, OU=Kafka, O=Example, L=Tokyo, C=JP" -ext SAN=DNS:broker1.example.com,IP:10.0.0.11
keytool -certreq -alias broker1 -keystore kafka.broker1.p12 -storepass changeit -file broker1.csr -ext SAN=DNS:broker1.example.com,IP:10.0.0.11

# 2) CA で署名し、CA ルート/中間と一緒にインポート
keytool -importcert -alias CARoot -file ca-root.crt -keystore kafka.broker1.p12 -storepass changeit -noprompt
keytool -importcert -alias CAInt -file ca-int.crt -keystore kafka.broker1.p12 -storepass changeit -noprompt
keytool -importcert -alias broker1 -file broker1.signed.crt -keystore kafka.broker1.p12 -storepass changeit

# 3) truststore 作成（CA チェーンを格納）
keytool -importcert -alias CARoot -file ca-root.crt -keystore kafka.truststore.p12 -storetype PKCS12 -storepass changeit -noprompt
keytool -importcert -alias CAInt -file ca-int.crt -keystore kafka.truststore.p12 -storetype PKCS12 -storepass changeit -noprompt

# 4) 検証
keytool -list -v -keystore kafka.broker1.p12 -storepass changeit | grep -E "Alias name|Owner|Issuer|SubjectAlternativeName"

クライアント設定と検証の勘所

クライアントの最小設定は security.protocol と truststore の指定です。mTLS を要求するブローカーに接続する場合は、クライアント側にも keystore を設定します。ホスト名検証はデフォルトで有効（ssl.endpoint.identification.algorithm=HTTPS）で、証明書の SAN と接続先名が一致している必要があります。

接続性の一次切り分けには openssl s_client が有効です。証明書チェーンやホスト名検証の失敗を即座に確認できます。

• プロパティファイルに機密を平文で置かない。可能なら外部秘匿ストアを利用。
• 負荷試験時はキーストアのロックや DNS 解決失敗で偽陽性を出さないように注意。
• クライアントライブラリの TLS バージョン/暗号スイートの互換性も確認。

client.properties の例と接続検証

# client.properties（単方向 TLS）
bootstrap.servers=broker1.example.com:9093,broker2.example.com:9093
security.protocol=SSL
ssl.truststore.location=/etc/client/kafka.truststore.p12
ssl.truststore.password=${FILE:/etc/client/trust.pass}
ssl.truststore.type=PKCS12
ssl.endpoint.identification.algorithm=HTTPS

# mTLS で必要な追加
ssl.keystore.location=/etc/client/client.p12
ssl.keystore.password=${FILE:/etc/client/key.pass}
ssl.keystore.type=PKCS12

# openssl で事前検証（SNI とホスト名検証の観点で FQDN を使う）
openssl s_client -connect broker1.example.com:9093 -servername broker1.example.com -showcerts

トラブルシューティングと試験の落とし穴

最頻出の失敗は SAN/ホスト名不一致、チェーン不完全、inter.broker.listener.name の不整合、証明書期限切れ、パスワード不一致です。ログには javax.net.ssl.SSLHandshakeException、certificate_unknown、handshake_failure などが出ます。

ACL と証明書主体名の対応も試験ポイントです。mTLS で証明された DN を短縮して主体名にマッピングする場合は ssl.principal.mapping.rules を使います。

• advertised.listeners のホストと証明書 SAN が一致しているか確認。
• 中間 CA を truststore に入れ忘れていないか。
• inter.broker.listener.name が SSL/SASL_SSL を指しているか、全ブローカーで統一されているか。
• 期限切れ間近の証明書はローリングで計画的に更新。
• ZooKeeper 併用環境では zk との TLS 設定も別途必要。KRaft ではコントローラ/ブローカー間 TLS を確認。

よく使う確認コマンドと設定

# ブローカーの SSL ハンドシェイクエラーを抽出
grep -E "SSLHandshakeException|handshake_failure|certificate_unknown" /var/log/kafka/server.log

# 証明書の SAN を確認
keytool -list -v -keystore /var/private/ssl/kafka.broker1.p12 -storepass changeit | grep -A2 "SubjectAlternativeName"

# mTLS で DN から主体名を抽出するマッピング例（CN を user とする）
ssl.principal.mapping.rules=RULE:^CN=(.*?),.*$/$1/,DEFAULT

問題で確認

よくある質問