oriolrius/kafka-keycloak-oauthApache Kafka 4.1.0 (KRaft) Keycloak OAuth2 प्रमाणीकरण के साथ Strimzi का उपयोग करते हुए - CVE-2025-27817 URL अनुमति-सूची प्रतिबंध को बायपास करता है
प्रोडक्शन-रेडी Apache Kafka 4.1.0 (KRaft मोड) Keycloak 26.1.1 OAuth2/OIDC प्रमाणीकरण के साथ, Strimzi Kafka इमेज का उपयोग करते हुए।
यह पिछले POC का विकास है जिसमें महत्वपूर्ण सुधार हैं:
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 कार्यक्षमता सक्षम होती है।
# 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)
kafka-security/ca-cert + ca-keykafka-security/broker/kafka.server.keystore.jks (सर्वर प्रमाणपत्र + निजी कुंजी शामिल)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
kafka-broker (confidential)
kafka-brokersetup-keycloak.sh द्वारा स्वतः-जनितaud दावे में kafka-broker जोड़ता हैpreferred_username शामिल करता हैkafka-producer (confidential)
kafka-producerclient_credentialskafka-consumer (confidential)
kafka-consumerclient_credentialsPOST 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
{
"aud": ["kafka-broker", "account"],
"iss": "http://localhost:8080/realms/kafka-realm",
"azp": "kafka-producer",
"preferred_username": "service-account-kafka-producer",
"scope": "profile email"
}
# नोड पहचान
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";
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 का उपयोग करना चाहिए।
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()
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())
confluent-kafka-python librdkafka (C लाइब्रेरी) का उपयोग करता है जो sasl.oauthbearer.method=oidc के माध्यम से OAuth लागू करती है। यह कार्यान्वयन org.apache.kafka.sasl.oauthbearer.allowed.urls सिस्टम प्रॉपर्टी की जाँच नहीं करता है जो नेटिव Apache Kafka Java क्लाइंट को ब्लॉक करती है।
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 .
अपेक्षित दावे:
{
"aud": ["kafka-broker", "account"],
"iss": "http://localhost:8080/realms/kafka-realm",
"azp": "kafka-producer",
"preferred_username": "service-account-kafka-producer"
}
docker logs kafka-broker 2>&1 | grep -E "Strimzi|JWTSignatureValidator|OAUTHBEARER"
अपेक्षित:
[io.strimzi.kafka.oauth.validator.JWTSignatureValidator] JWKS keys change detected
docker exec kafka-broker netstat -tlnp | grep java
अपेक्षित:
tcp6 0.0.0.0:9093 LISTEN (SASL_SSL)
tcp6 0.0.0.0:19092 LISTEN (PLAINTEXT)
tcp6 0.0.0.0:29093 LISTEN (CONTROLLER)
docker exec kafka-broker cat /var/lib/kafka/data/meta.properties
अपेक्षित:
version=1
cluster.id=kafka-cluster-01
node.id=1
समस्या: {"status":"invalid_token"}
oauth.jwks.endpoint.uri broker कंटेनर से पहुंच योग्य हैdocker exec kafka-broker curl http://keycloak:8080/realms/kafka-realm/protocol/openid-connect/certsसमस्या: Token audience mismatch
aud दावे में kafka-broker शामिल नहीं है./scripts/setup-keycloak.sh चलाएँaud दावे में kafka-broker शामिल हैसमस्या: Token issuer mismatch
iss oauth.valid.issuer.uri से मेल नहीं खाताoauth.valid.issuer.uri=http://localhost:8080/realms/kafka-realm (बाहरी होस्टनाम)http://keycloak:8080 का उपयोग करता है लेकिन http://localhost:8080 issuer के विरुद्ध सत्यापित करता हैसमस्या: नेटिव Java Kafka क्लाइंट URL allowlist त्रुटि के साथ विफल होते हैं
Keycloak से JWT टोकन की समाप्ति 5 मिनट है। Strimzi OAuth स्वचालित रूप से रीफ्रेश संभालता है:
oauth.refresh.token: उपयोग नहीं किया गया (client_credentials grant)sasl.oauthbearer.jwks.endpoint.refresh.ms=3600000 # 1 घंटा
sasl.oauthbearer.jwks.endpoint.retry.backoff.ms=100
sasl.oauthbearer.jwks.endpoint.retry.backoff.max.ms=10000
connections.max.idle.ms=600000
connection.failed.authentication.delay.ms=1000
ssl.endpoint.identification.algorithm=https अपडेट करें (none हटाएँ)allow.everyone.if.no.acl.found=true हटाएँ)kafka-acls --bootstrap-server localhost:9093 \
--command-config admin.properties \
--add --allow-principal User:kafka-producer \
--operation Write --topic '*'
oauth.token.endpoint.uri और oauth.jwks.endpoint.uri को HTTPS URL में अपडेट करें.
├── docker-compose.yml # ऑर्केस्ट्रेशन
├── .env # सीक्रेट (gitignored)
├── kafka-config/
│ ├── kraft-config.properties # Kafka broker कॉन्फ़िगरेशन
│ ├── producer.properties # प्रोड्यूसर OAuth कॉन्फ़िग (CLI टूल्स के लिए)
│ └── consumer.properties # कंज़्यूमर OAuth कॉन्फ़िग (CLI टूल्स के लिए)
├── kafka-security/
│ ├── generate-certs.sh # SSL प्रमाणपत्र जनरेटर
│ ├── ca-cert # रूट CA प्रमाणपत्र
│ ├── ca-key # रूट CA निजी कुंजी
│ └── broker/
│ ├── kafka.server.keystore.jks
│ └── kafka.server.truststore.jks
├── scripts/
│ └── setup-keycloak.sh # Keycloak realm/client सेटअप
└── tests/
└── quick_test.py # OAuth सत्यापन परीक्षण
आधिकारिक Apache Kafka इमेज के बजाय Strimzi Kafka इमेज (quay.io/strimzi/kafka:0.48.0-kafka-4.1.0) का उपयोग किया जाता है क्योंकि:
io.strimzi.kafka.oauth.*)इमेज विवरण:
Broker कॉन्फ़िगरेशन में दो URL हैं:
oauth.token.endpoint.uri=http://keycloak:8080/... (आंतरिक Docker नेटवर्क)oauth.valid.issuer.uri=http://localhost:8080/... (बाहरी, JWT iss दावे से मेल खाता है)ऐसा इसलिए है क्योंकि:
Broker JWT preferred_username दावे से principal निकालता है:
service-account-kafka-producer → User:service-account-kafka-producer
ACL प्राधिकरण के लिए इस principal को संदर्भित करते हैं।
| घटक | संस्करण | नोट्स |
|---|---|---|
| Apache Kafka | 4.1.0 | KRaft मोड (कोई ZooKeeper नहीं) |
| Strimzi Kafka इमेज | 0.48.0 | Docker इमेज: quay.io/strimzi/kafka:0.48.0-kafka-4.1.0 |
| Strimzi OAuth लाइब्रेरी | 0.17.0 | Strimzi Kafka 0.48.0 इमेज में पूर्व-बंडल |
| Keycloak | 26.1.1 | नवीनतम LTS |
| librdkafka | 2.12.0+ | OIDC OAuth समर्थन |
| confluent-kafka-python | 2.12.0+ | librdkafka संस्करण से मेल खाता है |