Apache Kafka 4.1.0 with Keycloak OAuth2 Authentication

Production-ready Apache Kafka 4.1.0 (KRaft mode) with Keycloak 26.1.1 OAuth2/OIDC authentication using Strimzi Kafka image.

Why This Project vs kafka-oauth-keycloak-tls-demo

This is an evolution of the previous POC with significant improvements:

• Strimzi OAuth 0.17.0 (vs 1.0.0) - stable production version bundled in Strimzi Kafka 0.48.0 image
• No custom Docker build required - uses official Strimzi image with OAuth pre-installed, eliminates Dockerfile complexity
• CVE-2025-27817 awareness - documents URL allowlist restriction and why Strimzi OAuth bypasses it
• Simplified architecture - single KRaft combined mode (broker+controller), not split architecture
• librdkafka client focus - tested with confluent-kafka-python (works without URL allowlist issues), not Java native clients
• Comprehensive technical documentation - production checklist, troubleshooting, performance tuning, principal mapping details
• Cleaner certificate management - included example certificates for immediate testing
• Automated Keycloak setup - scripted realm/client/mapper creation with audience configuration
• Working Python test suite - validates OAuth end-to-end message delivery
• Explicit issuer URL handling - documents internal vs external URL duality for token endpoint vs issuer validation

Architecture

• Kafka Distribution: Strimzi Kafka image 0.48.0 (includes Apache Kafka 4.1.0 + Strimzi OAuth 0.17.0 pre-bundled)
• Kafka Version: Apache Kafka 4.1.0 (KRaft combined broker+controller)
• OAuth Library: Strimzi Kafka OAuth 0.17.0 (bundled in image, bypasses CVE-2025-27817 URL allowlist restriction)
• OAuth Provider: Keycloak 26.1.1
• Security: SASL_SSL (OAuth) for external clients, PLAINTEXT for inter-broker, SSL with self-signed CA

CVE-2025-27817 Context

Apache Kafka 4.0.0+ introduced URL allowlist (org.apache.kafka.sasl.oauthbearer.allowed.urls) as JVM system property to fix SSRF/arbitrary file read vulnerability. This breaks standard OAuth usage in native Apache Kafka clients.

Solution: Strimzi Kafka OAuth library doesn't implement this restriction, enabling OAuth functionality with Kafka 4.1.0.

Prerequisites

• Docker Compose
• Python 3.x with uv (for testing)
• OpenSSL (for certificate generation)

Quick Start

# Generate SSL certificates
cd kafka-security
./generate-certs.sh
cd ..

# Start services
docker compose up -d

# Verify Keycloak
curl http://localhost:8080/health/ready

# Setup Keycloak realm and clients
./scripts/setup-keycloak.sh

# Test OAuth producer
source ~/.venv/bin/activate
uv pip install confluent-kafka
python tests/quick_test.py

Network Topology

keycloak:8080 (HTTP) ←→ kafka-broker:9093 (SASL_SSL/OAuth)
                      ↔ kafka-broker:19092 (PLAINTEXT/inter-broker)
                      ↔ kafka-broker:29093 (PLAINTEXT/KRaft controller)

SSL Configuration

CA Structure

• Root CA: kafka-security/ca-cert + ca-key
• Broker Keystore: kafka-security/broker/kafka.server.keystore.jks (contains server cert + private key)
• Broker Truststore: kafka-security/broker/kafka.server.truststore.jks (contains CA cert)
• Password: changeit (all keystores/truststores)

Certificate Details

# Broker certificate
CN=kafka-broker
SAN=DNS:kafka-broker,DNS:localhost,IP:127.0.0.1

# Validity: 3650 days
# Key algorithm: RSA 2048-bit
# Signature algorithm: SHA256withRSA

Keycloak OAuth Configuration

Realm: kafka-realm

Clients

kafka-broker (confidential)

• Client ID: kafka-broker
• Client Secret: Auto-generated by setup-keycloak.sh
• Purpose: Broker inter-broker OAuth authentication
• Mappers:
  • Audience mapper: adds kafka-broker to JWT aud claim
  • Username mapper: includes preferred_username in token

kafka-producer (confidential)

• Client ID: kafka-producer
• Client Secret: Auto-generated
• Purpose: External producer clients
• Grant: client_credentials
• Mappers: Same as kafka-broker

kafka-consumer (confidential)

• Client ID: kafka-consumer
• Client Secret: Auto-generated
• Purpose: External consumer clients
• Grant: client_credentials
• Mappers: Same as kafka-broker

Token Endpoint

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 Token Structure

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

Kafka Configuration

KRaft Mode (kraft-config.properties)

# Node identity
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 mechanism
sasl.enabled.mechanisms=OAUTHBEARER

# Strimzi OAuth handlers (per-listener for 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 configuration via 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";

Key Strimzi OAuth Parameters

• oauth.client.id: Client identifier for token acquisition
• oauth.client.secret: Client secret for token acquisition
• oauth.token.endpoint.uri: Keycloak token endpoint (broker uses internal hostname keycloak:8080)
• oauth.valid.issuer.uri: Expected JWT issuer (must match token iss claim, uses external localhost:8080)
• oauth.jwks.endpoint.uri: JWKS endpoint for JWT signature validation
• oauth.username.claim: JWT claim for principal extraction

Authorization

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

Note: Currently permissive for testing. Production should use ACLs.

Client Configuration

Python Producer (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 Consumer (confluent-kafka)

from confluent_kafka import Consumer