Apache Kafka 4.1.0 + Keycloak OAuth2 認証

Keycloak 26.1.1 OAuth2/OIDC 認証を備えた本番環境対応の Apache Kafka 4.1.0（KRaft モード）。Strimzi Kafka イメージを使用。

このプロジェクトが kafka-oauth-keycloak-tls-demo より優れている理由

これは以前の POC を大幅に改善した進化版です：

• Strimzi OAuth 0.17.0（旧 1.0.0 から変更）- Strimzi Kafka 0.48.0 イメージに同梱された安定版
• カスタム Docker ビルド不要 - OAuth がプリインストールされた公式 Strimzi イメージを使用し、Dockerfile の複雑さを排除
• CVE-2025-27817 への対応 - URL 許可リスト制限と、Strimzi OAuth がそれを回避する理由を文書化
• シンプルなアーキテクチャ - 単一 KRaft 結合モード（ブローカー+コントローラー）、分割アーキテクチャではない
• librdkafka クライアントに焦点 - Java ネイティブクライアントではなく、confluent-kafka-python でテスト済み（URL 許可リストの問題なし）
• 包括的な技術ドキュメント - 本番環境チェックリスト、トラブルシューティング、パフォーマンスチューニング、プリンシパルマッピングの詳細
• よりクリーンな証明書管理 - 即時テスト用のサンプル証明書を同梱
• 自動化された Keycloak セットアップ - スクリプトによるレルム/クライアント/マッパー作成と audience 設定
• 動作する Python テストスイート - OAuth エンドツーエンドのメッセージ配信を検証
• 明示的な issuer URL の扱い - トークンエンドポイントと issuer 検証における内部/外部 URL の二重性を文書化

アーキテクチャ

• Kafka ディストリビューション: Strimzi Kafka イメージ 0.48.0（Apache Kafka 4.1.0 + Strimzi OAuth 0.17.0 プリインストール）
• Kafka バージョン: Apache Kafka 4.1.0（KRaft 結合ブローカー+コントローラー）
• OAuth ライブラリ: Strimzi Kafka OAuth 0.17.0（イメージに同梱、CVE-2025-27817 の URL 許可リスト制限を回避）
• OAuth プロバイダー: Keycloak 26.1.1
• セキュリティ: 外部クライアントは SASL_SSL（OAuth）、ブローカー間は PLAINTEXT、自己署名 CA による SSL

CVE-2025-27817 の背景

Apache Kafka 4.0.0+ では、SSRF/任意ファイル読み取りの脆弱性を修正するため、JVM システムプロパティとして URL 許可リスト（org.apache.kafka.sasl.oauthbearer.allowed.urls）が導入されました。これにより、ネイティブ Apache Kafka クライアントでの標準的な OAuth 利用が妨げられます。

解決策: Strimzi Kafka OAuth ライブラリはこの制限を実装していないため、Kafka 4.1.0 で OAuth 機能を利用できます。

前提条件

• Docker Compose
• Python 3.x + uv（テスト用）
• OpenSSL（証明書生成用）

クイックスタート

# SSL 証明書の生成
cd kafka-security
./generate-certs.sh
cd ..

# サービスの起動
docker compose up -d

# Keycloak の確認
curl http://localhost:8080/health/ready

# Keycloak レルムとクライアントのセットアップ
./scripts/setup-keycloak.sh

# OAuth プロデューサーのテスト
source ~/.venv/bin/activate
uv pip install confluent-kafka
python tests/quick_test.py

ネットワークトポロジー

keycloak:8080 (HTTP) ←→ kafka-broker:9093 (SASL_SSL/OAuth)
                      ↔ kafka-broker:19092 (PLAINTEXT/ブローカー間)
                      ↔ kafka-broker:29093 (PLAINTEXT/KRaft コントローラー)

SSL 設定

CA 構成

• ルート CA: kafka-security/ca-cert + ca-key
• ブローカーキーストア: kafka-security/broker/kafka.server.keystore.jks（サーバー証明書 + 秘密鍵を含む）
• ブローカートラストストア: kafka-security/broker/kafka.server.truststore.jks（CA 証明書を含む）
• パスワード: changeit（すべてのキーストア/トラストストア）

証明書の詳細

# ブローカー証明書
CN=kafka-broker
SAN=DNS:kafka-broker,DNS:localhost,IP:127.0.0.1

# 有効期間: 3650 日
# 鍵アルゴリズム: RSA 2048 ビット
# 署名アルゴリズム: SHA256withRSA

Keycloak OAuth 設定

レルム: kafka-realm

クライアント

kafka-broker（機密）

• クライアント ID: kafka-broker
• クライアントシークレット: setup-keycloak.sh によって自動生成
• 目的: ブローカー間 OAuth 認証
• マッパー:
  • Audience マッパー: JWT の aud クレームに kafka-broker を追加
  • ユーザー名マッパー: トークンに preferred_username を含める

kafka-producer（機密）

• クライアント ID: kafka-producer
• クライアントシークレット: 自動生成
• 目的: 外部プロデューサークライアント
• グラント: client_credentials
• マッパー: kafka-broker と同じ

kafka-consumer（機密）

• クライアント ID: kafka-consumer
• クライアントシークレット: 自動生成
• 目的: 外部コンシューマークライアント
• グラント: client_credentials
• マッパー: kafka-broker と同じ

トークンエンドポイント

POST http://localhost:8080/realms/kafka-realm/protocol/openid-connect/token
Content-Type: application/x-www-form-urlencoded

grant_type=client_credentials
&client_id=kafka-producer
&client_secret=<secret>
&scope=profile email

JWT トークン構造

{
  "aud": ["kafka-broker", "account"],
  "iss": "http://localhost:8080/realms/kafka-realm",
  "azp": "kafka-producer",
  "preferred_username": "service-account-kafka-producer",
  "scope": "profile email"
}

Kafka 設定

KRaft モード（kraft-config.properties）

# ノード ID
node.id=1
process.roles=broker,controller
controller.quorum.voters=1@kafka-broker:29093

# リスナー
listeners=SASL_SSL://0.0.0.0:9093,PLAINTEXT://0.0.0.0:19092,CONTROLLER://0.0.0.0:29093
advertised.listeners=SASL_SSL://localhost:9093,PLAINTEXT://kafka-broker:19092
listener.security.protocol.map=SASL_SSL:SASL_SSL,PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
inter.broker.listener.name=PLAINTEXT
controller.listener.names=CONTROLLER

# SASL メカニズム
sasl.enabled.mechanisms=OAUTHBEARER

# Strimzi OAuth ハンドラー（SASL_SSL のリスナーごと）
listener.name.sasl_ssl.oauthbearer.sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler
listener.name.sasl_ssl.oauthbearer.sasl.server.callback.handler.class=io.strimzi.kafka.oauth.server.JaasServerOauthValidatorCallbackHandler

# JAAS による OAuth 設定
listener.name.sasl_ssl.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  oauth.client.id="kafka-broker" \
  oauth.client.secret="<secret>" \
  oauth.token.endpoint.uri="http://keycloak:8080/realms/kafka-realm/protocol/openid-connect/token" \
  oauth.valid.issuer.uri="http://localhost:8080/realms/kafka-realm" \
  oauth.jwks.endpoint.uri="http://keycloak:8080/realms/kafka-realm/protocol/openid-connect/certs" \
  oauth.username.claim="preferred_username";

主要な Strimzi OAuth パラメータ

• oauth.client.id: トークン取得用のクライアント識別子
• oauth.client.secret: トークン取得用のクライアントシークレット
• oauth.token.endpoint.uri: Keycloak トークンエンドポイント（ブローカーは内部ホスト名 keycloak:8080 を使用）
• oauth.valid.issuer.uri: 期待される JWT issuer（トークンの iss クレームと一致する必要があり、外部の localhost:8080 を使用）
• oauth.jwks.endpoint.uri: JWT 署名検証用の JWKS エンドポイント
• oauth.username.claim: プリンシパル抽出用の JWT クレーム

認可

authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer
super.users=User:kafka-broker;User:ANONYMOUS
allow.everyone.if.no.acl.found=true

注記: 現在はテスト用に許可設定。本番環境では ACL を使用すべきです。

クライアント設定

Python プロデューサー（confluent-kafka）

from confluent_kafka import Producer

conf = {
    'bootstrap.servers': 'localhost:9093',
    'security.protocol': 'SASL_SSL',
    'sasl.mechanisms': 'OAUTHBEARER',
    'sasl.oauthbearer.method': 'oidc',
    'sasl.oauthbearer.client.id': 'kafka-producer',
    'sasl.oauthbearer.client.secret': '<secret>',
    'sasl.oauthbearer.token.endpoint.url': 'http://localhost:8080/realms/kafka-realm/protocol/openid-connect/token',
    'ssl.ca.location': 'kafka-security/ca-cert',
    'ssl.endpoint.identification.algorithm': 'none',
}

producer = Producer(conf)
producer.produce('topic', b'message')
producer.flush()

Python コンシューマー（confluent-kafka）

from confluent_kafka import Consumer

conf = {
    'bootstrap.servers': 'localhost:9093',
    'group.id': 'test-group',
    'security.protocol': 'SASL_SSL',
    'sasl.mechanisms': 'OAUTHBEARER',
    'sasl.oauthbearer.method': 'oidc',
    'sasl.oauthbearer.client.id': 'kafka-consumer',
    'sasl.oauthbearer.client.secret': '<secret>',
    'sasl.oauthbearer.token.endpoint.url': 'http://localhost:8080/realms/kafka-realm/protocol/openid-connect/token',
    'ssl.ca.location': 'kafka-security/ca-cert',
    'ssl.endpoint.identification.algorithm': 'none',
    'auto.offset.reset': 'earliest',
}

consumer = Consumer(conf)
consumer.subscribe(['topic'])
while True:
    msg = consumer.poll(1.0)
    if msg: print(msg.value())

librdkafka が動作する理由

confluent-kafka-python は、sasl.oauthbearer.method=oidc で OAuth を実装する librdkafka（C ライブラリ）を使用します。この実装は、ネイティブ Apache Kafka Java クライアントをブロックする org.apache.kafka.sasl.oauthbearer.allowed.urls システムプロパティをチェックしません。

トラブルシューティング

OAuth トークンの検証

TOKEN=$(curl -s -X POST http://localhost:8080/realms/kafka-realm/protocol/openid-connect/token \
  -d "grant_type=client_credentials" \
  -d "client_id=kafka-producer" \
  -d "client_secret=<secret>" | jq -r .access_token)

echo $TOKEN | cut -d. -f2 | base64 -d 2>/dev/null | jq .