Apache Kafka 4.1.0 with Keycloak OAuth2 Authentication

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 통합 모드(broker+controller), 분할 아키텍처 아님
• librdkafka 클라이언트 중심 — confluent-kafka-python으로 테스트(URL 허용 목록 문제 없이 동작), Java 네이티브 클라이언트 아님
• 포괄적인 기술 문서 — 프로덕션 체크리스트, 문제 해결, 성능 튜닝, principal 매핑 세부 사항
• 더 깔끔한 인증서 관리 — 즉시 테스트 가능한 예제 인증서 포함
• 자동화된 Keycloak 설정 — audience 구성을 포함한 realm/client/mapper 스크립트 생성
• 동작하는 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 통합 broker+controller)
• 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/임의 파일 읽기 취약점을 수정하기 위해 URL 허용 목록(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/브로커 간)
                      ↔ 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 구성

Realm: kafka-realm

클라이언트

kafka-broker (기밀)

• 클라이언트 ID: kafka-broker
• 클라이언트 시크릿: setup-keycloak.sh가 자동 생성
• 용도: 브로커 간 OAuth 인증
• 매퍼:
  • Audience 매퍼: JWT aud 클레임에 kafka-broker 추가
  • 사용자 이름 매퍼: 토큰에 preferred_username 포함

kafka-producer (기밀)

• 클라이언트 ID: kafka-producer
• 클라이언트 시크릿: 자동 생성
• 용도: 외부 프로듀서 클라이언트
• Grant: client_credentials
• 매퍼: kafka-broker와 동일

kafka-consumer (기밀)

• 클라이언트 ID: kafka-consumer
• 클라이언트 시크릿: 자동 생성
• 용도: 외부 컨슈머 클라이언트
• Grant: 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: 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

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 .

예상 클레임:

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