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 से क्यों बेहतर है

यह पिछले POC का विकास है जिसमें महत्वपूर्ण सुधार हैं:

• Strimzi OAuth 0.17.0 (बनाम 1.0.0) - Strimzi Kafka 0.48.0 इमेज में बंडल किया गया स्थिर प्रोडक्शन संस्करण
• कोई कस्टम Docker बिल्ड आवश्यक नहीं - OAuth पूर्व-स्थापित के साथ आधिकारिक Strimzi इमेज का उपयोग करता है, Dockerfile जटिलता को समाप्त करता है
• CVE-2025-27817 जागरूकता - URL allowlist प्रतिबंध और Strimzi OAuth इसे क्यों बायपास करता है, इसका दस्तावेज़ीकरण
• सरलीकृत आर्किटेक्चर - एकल KRaft संयुक्त मोड (broker+controller), विभाजित आर्किटेक्चर नहीं
• librdkafka क्लाइंट फोकस - confluent-kafka-python के साथ परीक्षण किया गया (URL allowlist समस्याओं के बिना काम करता है), Java नेटिव क्लाइंट नहीं
• व्यापक तकनीकी दस्तावेज़ीकरण - प्रोडक्शन चेकलिस्ट, समस्या निवारण, प्रदर्शन ट्यूनिंग, principal मैपिंग विवरण
• स्वच्छ प्रमाणपत्र प्रबंधन - तत्काल परीक्षण के लिए शामिल उदाहरण प्रमाणपत्र
• स्वचालित Keycloak सेटअप - audience कॉन्फ़िगरेशन के साथ स्क्रिप्टेड realm/client/mapper निर्माण
• कार्यशील Python परीक्षण सूट - एंड-टू-एंड OAuth संदेश वितरण को मान्य करता है
• स्पष्ट issuer URL हैंडलिंग - token endpoint बनाम 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 संयुक्त broker+controller)
• OAuth लाइब्रेरी: Strimzi Kafka OAuth 0.17.0 (इमेज में बंडल, CVE-2025-27817 URL allowlist प्रतिबंध को बायपास करता है)
• OAuth प्रदाता: Keycloak 26.1.1
• सुरक्षा: बाहरी क्लाइंट के लिए SASL_SSL (OAuth), inter-broker के लिए PLAINTEXT, self-signed CA के साथ SSL

CVE-2025-27817 संदर्भ

Apache Kafka 4.0.0+ ने SSRF/मनमाना फ़ाइल पढ़ने की भेद्यता को ठीक करने के लिए URL allowlist (org.apache.kafka.sasl.oauthbearer.allowed.urls) को JVM सिस्टम प्रॉपर्टी के रूप में पेश किया। यह नेटिव 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 realm और क्लाइंट सेटअप करें
./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
• Broker Keystore: kafka-security/broker/kafka.server.keystore.jks (सर्वर प्रमाणपत्र + निजी कुंजी शामिल)
• Broker Truststore: kafka-security/broker/kafka.server.truststore.jks (CA प्रमाणपत्र शामिल)
• पासवर्ड: changeit (सभी keystores/truststores)

प्रमाणपत्र विवरण

# Broker प्रमाणपत्र
CN=kafka-broker
SAN=DNS:kafka-broker,DNS:localhost,IP:127.0.0.1

# वैधता: 3650 दिन
# कुंजी एल्गोरिथ्म: RSA 2048-बिट
# हस्ताक्षर एल्गोरिथ्म: SHA256withRSA

Keycloak OAuth कॉन्फ़िगरेशन

Realm: kafka-realm

क्लाइंट

kafka-broker (confidential)

• क्लाइंट ID: kafka-broker
• क्लाइंट सीक्रेट: setup-keycloak.sh द्वारा स्वतः-जनित
• उद्देश्य: Broker inter-broker OAuth प्रमाणीकरण
• Mappers:
  • Audience mapper: JWT aud दावे में kafka-broker जोड़ता है
  • Username mapper: टोकन में preferred_username शामिल करता है

kafka-producer (confidential)

• क्लाइंट ID: kafka-producer
• क्लाइंट सीक्रेट: स्वतः-जनित
• उद्देश्य: बाहरी प्रोड्यूसर क्लाइंट
• Grant: client_credentials
• Mappers: kafka-broker के समान

kafka-consumer (confidential)

• क्लाइंट ID: kafka-consumer
• क्लाइंट सीक्रेट: स्वतः-जनित
• उद्देश्य: बाहरी कंज़्यूमर क्लाइंट
• Grant: client_credentials
• Mappers: 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
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)
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 टोकन एंडपॉइंट (broker आंतरिक होस्टनाम keycloak:8080 का उपयोग करता है)
• oauth.valid.issuer.uri: अपेक्षित JWT issuer (टोकन iss दावे से मेल खाना चाहिए, बाहरी localhost:8080 का उपयोग करता है)
• oauth.jwks.endpoint.uri: JWT हस्ताक्षर सत्यापन के लिए JWKS एंडपॉइंट
• oauth.username.claim: principal निष्कर्षण के लिए 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