Kafka Güvenliği: SSL ve SASL ile Kimlik Doğrulama

Yazar:komut.io ekibi
0 Yorum

Production ortamında Kafka cluster’ı kurup her şeyi düzgün çalıştırıyorsunuz, servisler birbirleriyle haberleşiyor, mesajlar akıyor… Peki ya güvenlik? Bir gün audit ekibi gelip “bu Kafka’ya kim bağlanabilir?” diye sorduğunda ne cevap vereceksiniz? Ben bu soruyu ilk aldığımda cluster’ımın plain text, authentication’sız çalıştığını fark etmiştim. O günden sonra Kafka güvenliğini hiç hafife almadım.

Bu yazıda SSL ve SASL mekanizmalarını, gerçek production senaryoları üzerinden ele alacağız. Sadece “şu komutu çalıştır” demeyeceğim, neyi neden yaptığımızı da konuşacağız.

Kafka Güvenlik Modeline Genel Bakış

Kafka’nın güvenlik katmanı üç temel bileşenden oluşuyor:

  • Şifreleme (Encryption): SSL/TLS ile veri transferini şifreleme
  • Kimlik doğrulama (Authentication): SASL mekanizmalarıyla kim olduğunu ispatlama
  • Yetkilendirme (Authorization): ACL’lerle ne yapabileceğini kontrol etme

Bu yazıda ilk iki konuya odaklanacağız. ACL konusu ayrı bir yazıyı hak ediyor.

Önce şunu belirteyim: SSL ve SASL birbirinin alternatifi değil, tamamlayıcısı. SSL transport katmanında şifreleme sağlıyor, SASL uygulama katmanında kimlik doğrulama yapıyor. İkisini birlikte kullanmak en güvenli yaklaşım.

SSL ile Şifreleme ve Kimlik Doğrulama

Certificate Authority Oluşturma

Her şey bir CA (Certificate Authority) ile başlıyor. Production ortamında kurumsal CA’nızı kullanmanız gerekiyor, ama test ortamı için kendi CA’nızı oluşturabilirsiniz.

# CA private key ve sertifikası oluştur
openssl req -new -x509 -keyout ca-key.pem -out ca-cert.pem -days 365 
  -subj "/C=TR/ST=Istanbul/L=Istanbul/O=SirketAdi/OU=DevOps/CN=kafka-ca" 
  -passout pass:ca-sifre-buraya

# Oluşturulan dosyaları doğrula
openssl x509 -in ca-cert.pem -text -noout | grep -E "Subject:|Validity"

Bu CA sertifikasını güvenli bir yerde saklayın. Yıllar sonra yeni broker ekleyeceğinizde, yeni client certificate’ı sign etmeniz gerekecek ve CA key olmadan bunu yapamazsınız. Bunu bir kez unutmuştum, çok acı bir deneyimdi.

Broker Keystore ve Truststore Oluşturma

Her broker için ayrı keystore oluşturmanız gerekiyor. Şimdi bunun neden önemli olduğunu açıklayayım: Bir broker’ın private key’i ele geçirildiğinde sadece o broker’ı revoke etmek istiyorsunuz, tüm cluster’ı değil.

#!/bin/bash
BROKER_HOSTNAME="kafka-broker-01"
KEYSTORE_PASSWORD="keystore-sifre"
KEY_PASSWORD="key-sifre"
VALIDITY_DAYS=365

# Broker için keystore oluştur
keytool -keystore kafka.server.keystore.jks 
  -alias $BROKER_HOSTNAME 
  -validity $VALIDITY_DAYS 
  -genkey 
  -keyalg RSA 
  -keysize 2048 
  -storepass $KEYSTORE_PASSWORD 
  -keypass $KEY_PASSWORD 
  -dname "CN=$BROKER_HOSTNAME,OU=Kafka,O=SirketAdi,L=Istanbul,ST=Istanbul,C=TR"

# CSR (Certificate Signing Request) oluştur
keytool -keystore kafka.server.keystore.jks 
  -alias $BROKER_HOSTNAME 
  -certreq 
  -file broker-cert-request.csr 
  -storepass $KEYSTORE_PASSWORD 
  -keypass $KEY_PASSWORD

# CA ile imzala
openssl x509 -req 
  -CA ca-cert.pem 
  -CAkey ca-key.pem 
  -in broker-cert-request.csr 
  -out broker-cert-signed.crt 
  -days $VALIDITY_DAYS 
  -CAcreateserial 
  -passin pass:ca-sifre-buraya

# CA sertifikasını keystore'a import et
keytool -keystore kafka.server.keystore.jks 
  -alias CARoot 
  -import 
  -file ca-cert.pem 
  -storepass $KEYSTORE_PASSWORD 
  -noprompt

# İmzalı broker sertifikasını import et
keytool -keystore kafka.server.keystore.jks 
  -alias $BROKER_HOSTNAME 
  -import 
  -file broker-cert-signed.crt 
  -storepass $KEYSTORE_PASSWORD 
  -keypass $KEY_PASSWORD 
  -noprompt

# Truststore oluştur ve CA sertifikasını ekle
keytool -keystore kafka.server.truststore.jks 
  -alias CARoot 
  -import 
  -file ca-cert.pem 
  -storepass $KEYSTORE_PASSWORD 
  -noprompt

Bu script’i her broker için ayrı çalıştırmanız gerekiyor. Ben bunu Ansible ile otomatize ettim, üç broker’lı cluster’da elle yapmak hem zahmetli hem hata yapmaya açık.

Broker SSL Konfigürasyonu

# server.properties - SSL ayarları

# Listener tanımları
listeners=PLAINTEXT://0.0.0.0:9092,SSL://0.0.0.0:9093
advertised.listeners=PLAINTEXT://kafka-broker-01:9092,SSL://kafka-broker-01:9093

# SSL Keystore
ssl.keystore.location=/etc/kafka/ssl/kafka.server.keystore.jks
ssl.keystore.password=keystore-sifre
ssl.key.password=key-sifre

# SSL Truststore
ssl.truststore.location=/etc/kafka/ssl/kafka.server.truststore.jks
ssl.truststore.password=keystore-sifre

# SSL protokol ve cipher suite ayarları
ssl.enabled.protocols=TLSv1.2,TLSv1.3
ssl.protocol=TLSv1.3

# Client authentication - REQUIRED yaparsanız mutual TLS zorunlu olur
ssl.client.auth=required

# Hostname verification - production'da mutlaka açık bırakın
ssl.endpoint.identification.algorithm=HTTPS

ssl.client.auth=required ayarı mutual TLS’i (mTLS) zorunlu kılıyor. Bu sayede sadece broker değil, client’lar da sertifika ile kendini kanıtlamak zorunda. Eğer sadece şifreleme istiyorsanız none veya requested kullanabilirsiniz, ama kimlik doğrulama için required olmalı.

Client SSL Konfigürasyonu

# client.properties - Producer ve Consumer için

security.protocol=SSL
ssl.truststore.location=/etc/kafka/ssl/kafka.client.truststore.jks
ssl.truststore.password=client-truststore-sifre
ssl.keystore.location=/etc/kafka/ssl/kafka.client.keystore.jks
ssl.keystore.password=client-keystore-sifre
ssl.key.password=client-key-sifre
ssl.endpoint.identification.algorithm=HTTPS

Client için de benzer şekilde keystore ve truststore oluşturmanız gerekiyor. Client sertifikasını da CA ile imzalamanız şart, aksi halde broker reddediyor.

SASL ile Kimlik Doğrulama

SASL (Simple Authentication and Security Layer) birkaç farklı mekanizma sunuyor. Bunların her birinin farklı kullanım senaryoları var:

  • SASL/PLAIN: Kullanıcı adı ve şifre, basit ama şifre düz metin gidiyor (bu yüzden SSL ile kullanılmalı)
  • SASL/SCRAM-SHA-256 ve SASL/SCRAM-SHA-512: Challenge-response, şifre hash’lenerek gidiyor
  • SASL/GSSAPI (Kerberos): Enterprise ortamlar için, AD entegrasyonu
  • SASL/OAUTHBEARER: OAuth 2.0 token bazlı, modern mikroservis mimarileri için

SASL/SCRAM ile Kurulum

SCRAM, PLAIN’e göre çok daha güvenli. Şifreyi direkt göndermek yerine challenge-response mekanizması kullanıyor. Production’da Kerberos yoksa SCRAM-SHA-512 kullanmanızı tavsiye ederim.

Önce broker konfigürasyonu:

# server.properties - SASL/SCRAM ayarları

listeners=SASL_SSL://0.0.0.0:9093
advertised.listeners=SASL_SSL://kafka-broker-01:9093
listener.security.protocol.map=SASL_SSL:SASL_SSL

# SASL mekanizması
sasl.enabled.mechanisms=SCRAM-SHA-512
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512

# Inter-broker iletişim için SASL
inter.broker.listener.name=SASL_SSL

# SSL ayarları (önceki bölümden)
ssl.keystore.location=/etc/kafka/ssl/kafka.server.keystore.jks
ssl.keystore.password=keystore-sifre
ssl.key.password=key-sifre
ssl.truststore.location=/etc/kafka/ssl/kafka.server.truststore.jks
ssl.truststore.password=keystore-sifre
ssl.client.auth=required

Broker’ın kendi kimliğini doğrulaması için JAAS konfigürasyonu gerekiyor. Bunu ayrı bir dosyada tutuyorum:

# /etc/kafka/kafka_server_jaas.conf

KafkaServer {
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="kafka-broker"
    password="broker-inter-sifre";
};
# kafka-env.sh veya systemd service dosyasında JVM parametresi
export KAFKA_OPTS="-Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf"

SCRAM Kullanıcı Yönetimi

SCRAM kullanıcılarını Kafka’nın kendi araçlarıyla yönetiyoruz. Önemli bir nokta: bu kullanıcılar ZooKeeper’da (veya yeni versiyonlarda KRaft metadata’sında) saklanıyor.

# Admin kullanıcısı oluştur - bu işlemi broker ayağa kalkmadan önce yapın
kafka-configs.sh --zookeeper zookeeper:2181 
  --alter 
  --add-config 'SCRAM-SHA-512=[password=admin-sifre-buraya]' 
  --entity-type users 
  --entity-name admin

# Uygulama kullanıcısı oluştur
kafka-configs.sh --zookeeper zookeeper:2181 
  --alter 
  --add-config 'SCRAM-SHA-512=[password=app-sifre-buraya]' 
  --entity-type users 
  --entity-name order-service

# Monitoring kullanıcısı (read-only)
kafka-configs.sh --zookeeper zookeeper:2181 
  --alter 
  --add-config 'SCRAM-SHA-512=[password=monitoring-sifre]' 
  --entity-type users 
  --entity-name prometheus-consumer

# Kullanıcıları listele
kafka-configs.sh --zookeeper zookeeper:2181 
  --describe 
  --entity-type users

Kafka 3.x ve KRaft moduyla birlikte ZooKeeper bağımlılığı ortadan kalkıyor. KRaft modunda kullanıcı yönetimi için --bootstrap-server parametresini kullanıyorsunuz.

Client SASL/SCRAM Konfigürasyonu

# consumer.properties veya producer.properties

security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512

# SSL ayarları
ssl.truststore.location=/etc/kafka/ssl/kafka.client.truststore.jks
ssl.truststore.password=truststore-sifre

# SASL JAAS config - inline kullanım
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required 
  username="order-service" 
  password="app-sifre-buraya";

Java uygulamalarında JAAS config’i ayrı dosyada tutmak yerine inline olarak geçebiliyorsunuz. Bu özellikle Kubernetes ortamında Secret’tan environment variable olarak almak için çok kullanışlı.

SASL/GSSAPI (Kerberos) Entegrasyonu

Kurumsal ortamlarda Active Directory veya MIT Kerberos varsa, GSSAPI mekanizmasını kullanmak çok daha mantıklı. Merkezi kimlik yönetimi sağlıyor ve kullanıcı yönetimini Kafka dışında tutabiliyorsunuz.

# server.properties - Kerberos ayarları
sasl.enabled.mechanisms=GSSAPI
sasl.mechanism.inter.broker.protocol=GSSAPI
sasl.kerberos.service.name=kafka
# /etc/kafka/kafka_server_jaas.conf - Kerberos için

KafkaServer {
    com.sun.security.auth.module.Krb5LoginModule required
    useKeyTab=true
    storeKey=true
    keyTab="/etc/kafka/kafka.service.keytab"
    principal="kafka/[email protected]";
};

Kerberos kurulumu biraz daha karmaşık, keytab dosyalarının yönetimi ve bilet yenileme mekanizmaları ayrıca dikkat gerektiriyor. Ama büyük ölçekli enterprise ortamlarda bu zahmete değiyor.

Gerçek Dünya Senaryosu: Multi-Tenant Kafka Cluster

Şirkette farklı takımların aynı Kafka cluster’ını kullandığı bir senaryoyu ele alalım. Order servisi, inventory servisi ve analytics ekibi aynı cluster’da ama birbirlerinin topic’lerine erişmemeleri gerekiyor.

Önce kullanıcı ve sertifikaları oluşturuyoruz:

#!/bin/bash
# create-client-certs.sh

# Kullanım: ./create-client-certs.sh order-service
SERVICE_NAME=$1
CA_CERT="ca-cert.pem"
CA_KEY="ca-key.pem"
CA_PASS="ca-sifre-buraya"

# Client keystore
keytool -keystore kafka.$SERVICE_NAME.keystore.jks 
  -alias $SERVICE_NAME 
  -validity 365 
  -genkey 
  -keyalg RSA 
  -keysize 2048 
  -storepass client-sifre 
  -keypass client-sifre 
  -dname "CN=$SERVICE_NAME,OU=Microservices,O=SirketAdi,L=Istanbul,ST=Istanbul,C=TR"

# CSR
keytool -keystore kafka.$SERVICE_NAME.keystore.jks 
  -alias $SERVICE_NAME 
  -certreq 
  -file $SERVICE_NAME.csr 
  -storepass client-sifre

# İmzala
openssl x509 -req 
  -CA $CA_CERT 
  -CAkey $CA_KEY 
  -in $SERVICE_NAME.csr 
  -out $SERVICE_NAME-signed.crt 
  -days 365 
  -CAcreateserial 
  -passin pass:$CA_PASS

# Keystore'a ekle
keytool -keystore kafka.$SERVICE_NAME.keystore.jks 
  -alias CARoot 
  -import 
  -file $CA_CERT 
  -storepass client-sifre 
  -noprompt

keytool -keystore kafka.$SERVICE_NAME.keystore.jks 
  -alias $SERVICE_NAME 
  -import 
  -file $SERVICE_NAME-signed.crt 
  -storepass client-sifre 
  -noprompt

echo "$SERVICE_NAME için sertifikalar oluşturuldu."

Bu script ile her servis için ayrı sertifika oluşturup ardından ACL kurallarıyla hangi servise hangi topic üzerinde ne yetkisi olduğunu tanımlayabiliyorsunuz. CN değeri Kafka’da principal olarak kullanılıyor, ACL’lerde bu değeri kullanıyorsunuz.

SSL Bağlantısını Test Etme

Konfigürasyonu yaptıktan sonra test etmek kritik. Ben her zaman önce openssl s_client ile test ediyorum, Kafka araçlarını ikinci adım olarak kullanıyorum.

# SSL bağlantısını test et
openssl s_client 
  -connect kafka-broker-01:9093 
  -CAfile ca-cert.pem 
  -cert client-cert.pem 
  -key client-key.pem

# Başarılı bağlantıda "Verify return code: 0 (ok)" görmelisiniz

# Kafka'nın kendi araçları ile test
kafka-console-producer.sh 
  --broker-list kafka-broker-01:9093 
  --topic test-topic 
  --producer.config /etc/kafka/client-ssl-scram.properties

# Consumer test
kafka-console-consumer.sh 
  --bootstrap-server kafka-broker-01:9093 
  --topic test-topic 
  --from-beginning 
  --consumer.config /etc/kafka/client-ssl-scram.properties

Eğer bağlantı başarısız olursa, Kafka log’larını /var/log/kafka/server.log üzerinden takip edin. SSL hataları genellikle çok detaylı mesaj veriyor, SSLHANDSHAKE ile başlayan satırlara bakın.

Sertifika Yenileme Süreci

Production’da en sık unutulan şey sertifika expiry yönetimi. Ben bunu acı bir şekilde öğrendim, gece 2’de sertifikası dolan broker yüzünden cluster yarim saat ayağa kalkmadı.

Şimdi her cluster için bir monitoring job’ı çalıştırıyorum:

#!/bin/bash
# check-cert-expiry.sh - Cron ile her gün çalıştırın

KEYSTORE="/etc/kafka/ssl/kafka.server.keystore.jks"
KEYSTORE_PASS="keystore-sifre"
WARNING_DAYS=30

# Sertifika bitiş tarihini al
EXPIRY=$(keytool -list -v 
  -keystore $KEYSTORE 
  -storepass $KEYSTORE_PASS 
  -alias $(hostname) 2>/dev/null | 
  grep "Valid until" | 
  awk '{print $NF}')

# Gün hesapla
TODAY=$(date +%s)
EXPIRY_DATE=$(date -d "$EXPIRY" +%s 2>/dev/null || date -j -f "%a %b %d %T %Z %Y" "$EXPIRY" +%s)
DAYS_LEFT=$(( ($EXPIRY_DATE - $TODAY) / 86400 ))

if [ $DAYS_LEFT -le $WARNING_DAYS ]; then
  echo "UYARI: $(hostname) sertifikası $DAYS_LEFT gün sonra doluyor!"
  # Buraya Slack webhook, email veya PagerDuty entegrasyonu ekleyin
  curl -s -X POST "$SLACK_WEBHOOK_URL" 
    -H 'Content-type: application/json' 
    --data "{"text":"KAFKA SSL UYARISI: $(hostname) sertifikası $DAYS_LEFT gün sonra doluyor!"}"
fi

Bu script’i her broker’da cron’a ekleyin. 30 gün önceden uyarı almanız, yenileme işlemi için yeterli zaman sağlıyor.

Yaygın Hatalar ve Çözümleri

Kafka SSL/SASL konfigürasyonunda en sık karşılaştığım sorunlar:

  • SSL handshake hatası: Genellikle CN mismatch veya CA sertifikasının truststore’a eklenmemesinden kaynaklanıyor. ssl.endpoint.identification.algorithm boş bırakmak geçici çözüm ama production’da bu şekilde bırakmayın.
  • SCRAM kullanıcısı bulunamıyor: Kullanıcıyı ZooKeeper’a ekledikten sonra broker’ı restart etmeniz gerekmiyor ama ZooKeeper bağlantısının sağlıklı olduğundan emin olun.
  • Inter-broker authentication hatası: Broker’ın kendi kullanıcı bilgilerini JAAS config’de doğru tanımlamadıysanız broker’lar birbirleriyle konuşamıyor. inter.broker.listener.name ile sasl.mechanism.inter.broker.protocol uyumlu olmalı.
  • JVM TLS versiyonu uyumsuzluğu: Eski JVM’lerde TLSv1.3 desteği yok. ssl.enabled.protocols=TLSv1.2 ile sınırlayabilirsiniz ama Java 11+ kullanmanızı tavsiye ederim.
  • Keystore ve truststore şifre karışıklığı: ssl.keystore.password ve ssl.key.password farklı şeyler. Keystore container’ın şifresi, key password ise içindeki özel anahtarın şifresi. Bunları karıştırmak çok zaman kaybettiriyor.

Sonuç

Kafka güvenliğini baştan düzgün kurmak sonradan düzeltmekten çok daha kolay. SSL ile şifreleme, SASL/SCRAM veya Kerberos ile kimlik doğrulama, ve bunların üzerine ACL ile yetkilendirme yapılandırıldığında gerçekten sağlam bir güvenlik katmanı elde ediyorsunuz.

Birkaç pratik öneri ile bitirelim: Geliştirme ortamında bile güvenliği aktif tutun, production’a geçince alışkanlık sorunu yaşamayın. Sertifika yönetimini otomatize edin, manuel süreçler hata üretiyor. SASL mekanizması seçerken kurumunuzda ne kullanıldığına bakın, Kerberos altyapısı varsa onu kullanın. Son olarak, güvenlik konfigürasyonunu kod olarak yönetin (Ansible, Terraform), wikiye not almak yetmiyor.

Bir sonraki yazıda Kafka ACL’leri ve ZooKeeper güvenliğini ele alacağım. O konuda da paylaşmak istediğiniz senaryolar varsa yorumlara yazın.

Bir yanıt yazın

E-posta adresiniz yayınlanmayacak. Gerekli alanlar * ile işaretlenmişlerdir