Posted inセキュリティ

本番環境におけるApache Kafkaセキュリティ対策：TLS暗号化、SASL/SCRAM、およびACLによる堅牢化

Security tutorial - IT technology blog
Security tutorial - IT technology blog

なぜ保護されていないKafkaクラスタは本番環境で破綻するのか

私たちのチームはこの教訓を手痛い形で学びました。ある火曜日の早朝、アウトバウンドのネットワークトラフィックがベースラインの15 MB/sから突如400 MB/s近くまで跳ね上がったのです。VPC内のステージング環境用Podが、ブローカーのデフォルトである平文ポート9092を参照して起動したことが原因でした。ブローカーに暗号化も認証もアクセス制御も施されていなかったため、その単一のテストコンテナがネットワーク上から平文のクレジットカード決済データを直接吸い上げ始めてしまったのです。

デフォルト状態のApache Kafkaは開発者の利便性を最優先に設定されています。すべてのメッセージは平文でネットワーク上を流れ、ポート9092にトラフィックをルーティングできるホストであれば、任意のトピックの読み取りや破損レコードの書き込み、さらにはパーティションのサイレント削除まで行えてしまいます。本番環境のデプロイを保護するには、以下の3つの防御策が不可欠です。

  • Transport Layer Security (TLS): ブローカー間およびクライアント・ブローカー間の通信を暗号化します。AES-NI命令セットを搭載した最近のCPUであれば、CPUオーバーヘッドは通常3%未満に抑えられます。
  • SASL/SCRAM-512: ソルト付きチャレンジレスポンスハッシュを使用してクライアントを認証します。平文パスワードのネットワーク送信を防ぎ、Kerberos/KDCインフラ運用の煩わしさを回避できます。
  • Access Control Lists (ACLs): 認証されたプリンシパルに対し、指定したトピックやコンシューマグループへの明示的な操作のみに権限を制限します。

ステップ 1: 認証局（CA）とキーストアの作成

Kafkaの設定ファイルを変更する前に、公開鍵暗号基盤（PKI）の準備が必要です。ここでは専用のプライベート認証局（CA）を立ち上げ、最新のJVMホスト名検証をパスできるようにサブジェクト代替名（SAN）を含んだ署名済みキーストアを発行します。

1. 内部認証局（プライベートCA）の作成

# 3年間有効な内部CAの秘密鍵とルート証明書を生成
openssl req -new -x509 -keyout ca-key -out ca-cert -days 1095 \
  -subj "/CN=Internal-Kafka-CA/OU=Infra/O=ItFromZero/C=US" \
  -nodes

2. SANを含むブローカー用キーストアの作成

各ブローカーノード（例: IP 10.0.1.10 の kafka1.internal）で以下の手順を実行します。最新のJavaバージョンではデフォルトでTLSホスト名検証が強制されるため、SAN拡張を省略するとクライアントハンドシェイクが失敗します。

# 1. SAN拡張を含むブローカー用キーストアを作成
keytool -keystore server.keystore.jks -alias kafka1.internal \
  -validity 365 -genkey -keyalg RSA -keysize 2048 \
  -storepass "KeystoreSecret123!" -keypass "KeystoreSecret123!" \
  -dname "CN=kafka1.internal,OU=Infra,O=ItFromZero,C=US" \
  -ext SAN=DNS:kafka1.internal,IP:10.0.1.10

# 2. 証明書署名要求（CSR）をエクスポート
keytool -keystore server.keystore.jks -alias kafka1.internal \
  -certreq -file cert-file -storepass "KeystoreSecret123!" \
  -ext SAN=DNS:kafka1.internal,IP:10.0.1.10

# 3. 内部CAを使用してCSRに署名
openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-file \
  -out cert-signed -days 365 -CAcreateserial \
  -extfile <(printf "subjectAltName=DNS:kafka1.internal,IP:10.0.1.10")

# 4. CA証明書をキーストアにインポート
keytool -keystore server.keystore.jks -alias CARoot \
  -import -noprompt -file ca-cert -storepass "KeystoreSecret123!"

# 5. 署名済み証明書をインポート
keytool -keystore server.keystore.jks -alias kafka1.internal \
  -import -noprompt -file cert-signed -storepass "KeystoreSecret123!"

3. ブローカーおよびクライアント用トラストストアの構築

ブローカーとプロデューサーは、クラスタ証明書に署名したルートCAを信頼する必要があります。

# CA証明書を含むトラストストアを作成
keytool -keystore server.truststore.jks -alias CARoot \
  -import -noprompt -file ca-cert -storepass "TruststoreSecret123!"

# クライアント実行環境用にコピーを配布
cp server.truststore.jks client.truststore.jks

ステップ 2: ブローカー設定の堅牢化

1. ブローカーリスナーの設定

ブローカーの設定ファイル（通常は /etc/kafka/server.properties）を編集します。ポート9092は完全に閉鎖し、ポート9093でSASL/SCRAM認証と組み合わせた相互TLS暗号化を設定します。

# リスナーとプロトコル
listeners=SASL_SSL://kafka1.internal:9093
advertised.listeners=SASL_SSL://kafka1.internal:9093
listener.security.protocol.map=SASL_SSL:SASL_SSL
inter.broker.listener.name=SASL_SSL
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512

# TLSキーストアとトラストストア
ssl.keystore.location=/var/private/ssl/server.keystore.jks
ssl.keystore.password=KeystoreSecret123!
ssl.key.password=KeystoreSecret123!
ssl.truststore.location=/var/private/ssl/server.truststore.jks
ssl.truststore.password=TruststoreSecret123!
ssl.endpoint.identification.algorithm=HTTPS

# 認証設定
sasl.enabled.mechanisms=SCRAM-SHA-512

# 認可設定（KRaft StandardAuthorizer / Kafka 3.x以降）
authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer
allow.everyone.if.no.acl.found=false
super.users=User:admin

2. ブローカーJAASファイルの設定

ブローカーはJava Authentication and Authorization Service (JAAS)を通じて着信クライアントのハンドシェイクを認証します。このファイルを /etc/kafka/kafka_server_jaas.conf として保存します。本番クラスタを運用するにあたり、強固なパスワードの生成は必須条件です。サーバー用パスワードには toolcraft.app/ja/tools/security/password-generator のパスワードジェネレーターを活用しています。ブラウザ上ですべて完結し、ネットワーク経由でデータが送信されないため、オフライン環境のインフラ構築でも安心して利用できます。

KafkaServer {
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="admin"
    password="e9V#kP9$mR2@zQ7!xT5";
};

Client {
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="admin"
    password="e9V#kP9$mR2@zQ7!xT5";
};

デーモンを起動する前に、KafkaプロセスがこのJAASファイルを参照するように環境変数を設定します。

export KAFKA_OPTS="-Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf"
systemctl restart kafka

3. SCRAM-SHA-512によるユーザー認証情報の登録

ブローカーがTLSでリスニングを開始したら、クラスタのメタデータにユーザー認証情報を直接登録します。管理者用の認証情報ファイル（/tmp/admin-bootstrap.properties）を使用してブートストラップを行います。

# クラスタ管理者ユーザーを登録
kafka-configs.sh --bootstrap-server kafka1.internal:9093 \
  --entity-type users --entity-name admin \
  --alter --add-config 'SCRAM-SHA-512=[password=e9V#kP9$mR2@zQ7!xT5]' \
  --command-config /tmp/admin-bootstrap.properties

# アプリケーション用サービスアカウントを登録
kafka-configs.sh --bootstrap-server kafka1.internal:9093 \
  --entity-type users --entity-name payments-worker \
  --alter --add-config 'SCRAM-SHA-512=[password=uP4!vL9$xN2@bM7^kQ1]' \
  --command-config /tmp/admin-bootstrap.properties

4. ACLによるスコープを絞った権限付与

allow.everyone.if.no.acl.found=false が有効になっているため、デフォルトでは全トピックへのアクセスが拒否されます。payments-worker に対して、送信イベントの書き込みと受信タスクの読み取りに必要な最小限の権限を付与します。

# payments-workerに対してトピック 'payment-events' への書き込みを許可
kafka-acls.sh --bootstrap-server kafka1.internal:9093 \
  --command-config /tmp/admin-bootstrap.properties \
  --add --allow-principal User:payments-worker \
  --operation Write --operation Describe \
  --topic payment-events \
  --resource-pattern-type literal

# payments-workerに対して 'payments-group' 内での 'settlement-tasks' のコンシュームを許可
kafka-acls.sh --bootstrap-server kafka1.internal:9093 \
  --command-config /tmp/admin-bootstrap.properties \
  --add --allow-principal User:payments-worker \
  --operation Read --operation Describe \
  --topic settlement-tasks \
  --group payments-group

ステップ 3: 検証とオブザーバビリティ

変更を本番環境へ反映する前に、実際のプロデューサーコマンドを用いてセキュリティ境界を検証します。

1. 認可されたトラフィックの検証

client-payments.properties にクライアント設定を定義します。

security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
ssl.truststore.location=/var/private/ssl/client.truststore.jks
ssl.truststore.password=TruststoreSecret123!
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
    username="payments-worker" \
    password="uP4!vL9$xN2@bM7^kQ1";

許可されたトピックへテストメッセージを送信します。

echo '{"tx_id": "98412", "amount": 120.50}' | kafka-console-producer.sh \
  --bootstrap-server kafka1.internal:9093 \
  --topic payment-events \
  --producer.config client-payments.properties

2. 不正なリクエストが拒否されることの確認

権限が付与されていないトピック（例: audit-logs）へのメッセージ送信が正しくブロックされることを確認します。

echo 'test event' | kafka-console-producer.sh \
  --bootstrap-server kafka1.internal:9093 \
  --topic audit-logs \
  --producer.config client-payments.properties

ブローカーは即座にバッチを拒否します。

[2026-11-07 02:45:12,104] ERROR Error when sending message to topic audit-logs with key: null
org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [audit-logs]

3. Prometheusで監視すべき本番メトリクス

不正アクセスの試行や証明書の有効期限切れを検知するため、KafkaのJMXメトリクスを監視対象として公開します。

  • kafka.server:type=RequestMetrics,name=ErrorsPerSec,error=TOPIC_AUTHORIZATION_FAILED: 0 req/secを超える値が継続して発生している場合、認証情報の誤設定や不正なネットワーク探索スキャンが行われている可能性を示唆します。
  • kafka.server:type=RequestMetrics,name=ErrorsPerSec,error=SASL_AUTHENTICATION_FAILED: 急激なスパイクの発生は、ブルートフォース攻撃や期限切れのサービスパスワードによる認証試行を示しています。
  • ssl-certificate-expiration-date: ブローカー証明書の有効期限切れによる予期せぬTLSハンドシェイク障害を防ぐため、有効期限の30日前にアラートを発報するように設定します。

TLS、SCRAM-512、そして最小権限のACLを適用することで、無防備なイベントバスからコンプライアンス監査にも耐えうる堅牢なメッセージング基盤へと強化できます。

Share:
Tags: