Apache Kafka 4.1.0 مع مصادقة Keycloak OAuth2

Apache Kafka 4.1.0 جاهز للإنتاج (وضع KRaft) مع مصادقة Keycloak 26.1.1 OAuth2/OIDC باستخدام صورة Strimzi Kafka.

لماذا هذا المشروع مقارنة بـ kafka-oauth-keycloak-tls-demo

هذا تطوير للإثبات السابق للمفهوم مع تحسينات كبيرة:

Strimzi OAuth 0.17.0 (مقابل 1.0.0) - إصدار إنتاج مستقر مدمج في صورة Strimzi Kafka 0.48.0
لا حاجة لبناء Docker مخصص - يستخدم صورة Strimzi الرسمية مع OAuth مثبت مسبقًا، مما يلغي تعقيد Dockerfile
الوعي بـ CVE-2025-27817 - يوثق قيود قائمة عناوين URL ولماذا يتجاوزها Strimzi OAuth
بنية مبسطة - وضع KRaft المدمج الفردي (broker+controller)، وليس بنية مقسمة
التركيز على عميل librdkafka - تم اختباره مع confluent-kafka-python (يعمل بدون مشاكل قائمة عناوين URL)، وليس عملاء Java الأصليين
توثيق تقني شامل - قائمة فحص الإنتاج، استكشاف الأخطاء وإصلاحها، ضبط الأداء، تفاصيل تعيين principal
إدارة شهادات أنظف - شهادات أمثلة مرفقة للاختبار الفوري
إعداد Keycloak آلي - إنشاء realm/client/mapper بنصوص برمجية مع تكوين الجمهور
مجموعة اختبار Python عاملة - تتحقق من تسليم الرسائل الشامل عبر OAuth
معالجة صريحة لعنوان URL للمُصدر - يوثق ازدواجية العنوان الداخلي مقابل الخارجي لنقطة نهاية الرمز مقابل التحقق من المُصدر

البنية

توزيع Kafka: صورة Strimzi Kafka 0.48.0 (تتضمن Apache Kafka 4.1.0 + Strimzi OAuth 0.17.0 مثبتة مسبقًا)
إصدار Kafka: Apache Kafka 4.1.0 (KRaft مدمج broker+controller)
مكتبة OAuth: Strimzi Kafka OAuth 0.17.0 (مدمجة في الصورة، تتجاوز قيد قائمة عناوين URL الخاص بـ CVE-2025-27817)
مزود OAuth: Keycloak 26.1.1
الأمان: SASL_SSL (OAuth) للعملاء الخارجيين، PLAINTEXT للاتصال بين الوسطاء، SSL مع CA ذاتي التوقيع

سياق CVE-2025-27817

قدم Apache Kafka 4.0.0+ قائمة عناوين URL (org.apache.kafka.sasl.oauthbearer.allowed.urls) كخاصية نظام JVM لإصلاح ثغرة SSRF/قراءة الملفات التعسفية. هذا يكسر استخدام OAuth القياسي في عملاء Apache Kafka الأصليين.

الحل: مكتبة Strimzi Kafka OAuth لا تنفذ هذا القيد، مما يتيح وظيفة OAuth مع Kafka 4.1.0.

المتطلبات الأساسية

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

# إعداد realm وعملاء 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/inter-broker)
                      ↔ kafka-broker:29093 (PLAINTEXT/KRaft controller)

تكوين 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-bit
# خوارزمية التوقيع: SHA256withRSA

تكوين Keycloak OAuth

Realm: kafka-realm

العملاء

kafka-broker (سري)

معرف العميل: kafka-broker
سر العميل: يتم توليده تلقائيًا بواسطة setup-keycloak.sh
الغرض: مصادقة OAuth بين الوسطاء
أدوات التعيين:
- أداة تعيين الجمهور: تضيف kafka-broker إلى مطالبة aud في JWT
- أداة تعيين اسم المستخدم: تتضمن preferred_username في الرمز

kafka-producer (سري)

معرف العميل: kafka-producer
سر العميل: يتم توليده تلقائيًا
الغرض: عملاء المنتجين الخارجيين
المنح: client_credentials
أدوات التعيين: نفس kafka-broker

kafka-consumer (سري)

معرف العميل: 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)

# هوية العقدة
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

# تكوين OAuth عبر JAAS
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 المتوقع (يجب أن يطابق مطالبة iss في الرمز، يستخدم localhost:8080 الخارجي)
oauth.jwks.endpoint.uri: نقطة نهاية JWKS للتحقق من توقيع JWT
oauth.username.claim: مطالبة JWT لاستخراج principal

التفويض

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

ملاحظة: حاليًا متساهل للاختبار. يجب استخدام ACLs في الإنتاج.

تكوين العميل

منتج 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

kafka-keycloak-oauth

الأكثر شعبية