Dynamic Secret Rotation for Kafka Clients using Vault and Kubernetes CSI — code snippets

snippet-1.sh

# Enable the PKI secrets engine at a specific path for Kafka client certs
vault secrets enable -path=kafka-pki pki

# Tune the secrets engine to support up to 30 days max TTL
vault secrets tune -max-lease-ttl=720h kafka-pki

# Generate the root CA certificate for Kafka clients
vault write -format=json kafka-pki/root/generate/internal \
    common_name="Kafka Client Root CA" \
    ttl=87600h > root_ca.json

# Configure the URLs for CRL distribution and Authority Information Access (AIA)
vault write kafka-pki/config/urls \
    issuing_certificates="http://vault.vault.svc.cluster.local:8200/v1/kafka-pki/ca" \
    crl_distribution_points="http://vault.vault.svc.cluster.local:8200/v1/kafka-pki/crl"

# Create a role for kafka clients that allows generating certs with a 24-hour default TTL
vault write kafka-pki/roles/kafka-client \
    allowed_domains="kafka-client,svc.cluster.local" \
    allow_subdomains=true \
    max_ttl=72h \
    ttl=24h \
    allow_any_name=true \
    key_type="rsa" \
    key_bits=2048

snippet-2.sh

# Enable Kubernetes authentication
vault auth enable kubernetes

# Configure the auth method to communicate with the local K8s API server
vault write auth/kubernetes/config \
    kubernetes_host="https://kubernetes.default.svc.cluster.local:443"

# Create a policy allowing access to the PKI issue endpoint
vault policy write kafka-client-policy - <<EOF
path "kafka-pki/issue/kafka-client" {
  capabilities = ["update"]
}
EOF

# Bind the policy to the application service account
vault write auth/kubernetes/role/kafka-client-role \
    bound_service_account_names=kafka-client-sa \
    bound_service_account_namespaces=production \
    policies=kafka-client-policy \
    ttl=1h

snippet-3.yaml

apiVersion: secrets-store.csi.x-k8s.io/v1
kind: SecretProviderClass
metadata:
  name: kafka-client-vault-certs
  namespace: production
spec:
  provider: vault
  parameters:
    usePodServiceAccount: "true"
    roleName: "kafka-client-role"
    vaultAddress: "http://vault.vault.svc.cluster.local:8200"
    # To prevent generating mismatched certs, we issue them in a single call to Vault.
    # The Vault CSI provider allows extracting multiple keys from a single secret response.
    objects: |
      - objectName: "kafka-certs"
        secretPath: "kafka-pki/issue/kafka-client"
        secretArgs:
          common_name: "payment-service.production.svc.cluster.local"
          ttl: "24h"
        fileMap:
          certificate: "client.crt"
          private_key: "client.key"
          issuing_ca: "ca.crt"

snippet-4.yaml

apiVersion: apps/v1
kind: Deployment
metadata:
  name: payment-service
  namespace: production
  labels:
    app: payment-service
spec:
  replicas: 3
  selector:
    matchLabels:
      app: payment-service
  template:
    metadata:
      labels:
        app: payment-service
    spec:
      serviceAccountName: kafka-client-sa
      containers:
        - name: payment-service
          image: gcr.io/my-production-cluster/payment-service:v1.2.0
          volumeMounts:
            - name: kafka-certs-volume
              mountPath: "/var/run/secrets/kafka"
              readOnly: true
          resources:
            limits:
              cpu: "1"
              memory: "1Gi"
            requests:
              cpu: "500m"
              memory: "512Mi"
      volumes:
        - name: kafka-certs-volume
          csi:
            driver: secrets-store.csi.k8s.io
            readOnly: true
            volumeAttributes:
              secretProviderClass: "kafka-client-vault-certs"

snippet-5.go

package kafka

import (
	"crypto/tls"
	"crypto/x509"
	"os"
	"sync"
	"sync/atomic"
	"time"

	"github.com/fsnotify/fsnotify"
	"github.com/rs/zerolog/log"
)

type DynamicTLSProvider struct {
	certFile string
	keyFile  string
	caFile   string

	mu          sync.RWMutex
	clientCert  *tls.Certificate
	caCertPool  *x509.CertPool
	hasCerts    atomic.Bool
}

func NewDynamicTLSProvider(certFile, keyFile, caFile string) (*DynamicTLSProvider, error) {
	p := &DynamicTLSProvider{
		certFile: certFile,
		keyFile:  keyFile,
		caFile:   caFile,
	}

	if err := p.reload(); err != nil {
		return nil, err
	}

	go p.watchFiles()
	return p, nil
}

func (p *DynamicTLSProvider) reload() error {
	certPEM, err := os.ReadFile(p.certFile)
	if err != nil {
		return err
	}

	keyPEM, err := os.ReadFile(p.keyFile)
	if err != nil {
		return err
	}

	caPEM, err := os.ReadFile(p.caFile)
	if err != nil {
		return err
	}

	cert, err := tls.X509KeyPair(certPEM, keyPEM)
	if err != nil {
		return err
	}

	caPool := x509.NewCertPool()
	if !caPool.AppendCertsFromPEM(caPEM) {
		return os.ErrInvalid
	}

	p.mu.Lock()
	p.clientCert = &cert
	p.caCertPool = caPool
	p.mu.Unlock()
	p.hasCerts.Store(true)

	log.Info().Msg("Successfully reloaded Kafka TLS certificates from disk")
	return nil
}

func (p *DynamicTLSProvider) GetClientCertificate(info *tls.CertificateRequestInfo) (*tls.Certificate, error) {
	p.mu.RLock()
	defer p.mu.RUnlock()
	return p.clientCert, nil
}

func (p *DynamicTLSProvider) GetTLSConfig() *tls.Config {
	return &tls.Config{
		GetClientCertificate: p.GetClientCertificate,
		RootCAs:              p.getCAPool(),
		MinVersion:           tls.VersionTLS13,
	}
}

func (p *DynamicTLSProvider) getCAPool() *x509.CertPool {
	p.mu.RLock()
	defer p.mu.RUnlock()
	return p.caCertPool
}

func (p *DynamicTLSProvider) watchFiles() {
	watcher, err := fsnotify.NewWatcher()
	if err != nil {
		log.Error().Err(err).Msg("Failed to initialize file watcher")
		return
	}
	defer watcher.Close()

	// In Kubernetes CSI, files are symlinks. We watch the directory
	// containing the files to catch symlink updates.
	dir := "/var/run/secrets/kafka"
	if err := watcher.Add(dir); err != nil {
		log.Error().Err(err).Msg("Failed to watch Kafka certs directory")
		return
	}

	// Rate limit reloads to avoid thundering herd on file modifications
	cooldown := 5 * time.Second
	var lastReload time.Time

	for {
		select {
		case event, ok := <-watcher.Events:
			if !ok {
				return
			}
			// Kubernetes CSI driver rotates files by changing the symlink target.
			// This generates a Write event on the directory/symlink itself.
			if event.Has(fsnotify.Write) || event.Has(fsnotify.Create) {
				if time.Since(lastReload) > cooldown {
					log.Info().Str("event", event.Name).Msg("Detected certificate file change, reloading...")
					if err := p.reload(); err != nil {
						log.Error().Err(err).Msg("Failed to hot-reload certificates")
					} else {
						lastReload = time.Now()
					}
				}
			}
		case err, ok := <-watcher.Errors:
			if !ok {
				return
			}
			log.Error().Err(err).Msg("File watcher error occurred")
		}
	}
}

snippet-6.go

package kafka

import (
	"context"
	"strings"
	"time"

	"github.com/rs/zerolog/log"
	"github.com/twmb/franz-go/pkg/kgo"
)

func NewKafkaClient(brokers string, tlsProvider *DynamicTLSProvider) (*kgo.Client, error) {
	brokerList := strings.Split(brokers, ",")
	
	opts := []kgo.Opt{
		kgo.SeedBrokers(brokerList...),
		kgo.DialTLSConfig(tlsProvider.GetTLSConfig()),
		kgo.ConnTimeout(10 * time.Second),
		kgo.RequestTimeout(30 * time.Second),
		kgo.AllowAutoTopicCreation(),
	}

	client, err := kgo.NewClient(opts...)
	if err != nil {
		return nil, err
	}

	// Ping the cluster to ensure connectivity
	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()
	
	if err := client.Ping(ctx); err != nil {
		client.Close()
		return nil, err
	}

	log.Info().Msg("Kafka client initialized successfully with dynamic mTLS")
	return client, nil
}

snippet-7.txt

package com.production.kafka;

import org.apache.kafka.common.security.auth.SslEngineFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.net.ssl.*;
import java.io.File;
import java.security.KeyStore;
import java.security.SecureRandom;
import java.util.Map;
import java.util.Set;

public class DynamicSslEngineFactory implements SslEngineFactory {
    private static final Logger log = LoggerFactory.getLogger(DynamicSslEngineFactory.class);

    private String truststorePath;
    private String keystorePath;
    private String keystorePassword = "temporary-password";

    private SSLContext sslContext;
    private long lastLoadedTime = 0;

    @Override
    public void configure(Map<String, ?> configs) {
        this.truststorePath = (String) configs.get("ssl.truststore.location");
        this.keystorePath = (String) configs.get("ssl.keystore.location");
        
        try {
            this.sslContext = createSSLContext();
            this.lastLoadedTime = System.currentTimeMillis();
        } catch (Exception e) {
            throw new RuntimeException("Failed to initialize SSLContext", e);
        }
    }

    private SSLContext createSSLContext() throws Exception {
        // Load in-memory keystores using our PEM reader
        KeyStore keyStore = PemReader.loadKeyStore(keystorePath, keystorePath.replace(".crt", ".key"), keystorePassword);
        KeyStore trustStore = KeyStore.getInstance("PKCS12");
        trustStore.load(null, null);
        
        byte[] caBytes = java.nio.file.Files.readAllBytes(java.nio.file.Paths.get(truststorePath));
        java.security.cert.CertificateFactory cf = java.security.cert.CertificateFactory.getInstance("X.509");
        java.security.cert.Certificate caCert = cf.generateCertificate(new java.io.ByteArrayInputStream(caBytes));
        trustStore.setCertificateEntry("kafka-ca", caCert);

        KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
        kmf.init(keyStore, keystorePassword.toCharArray());

        TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
        tmf.init(trustStore);

        SSLContext context = SSLContext.getInstance("TLSv1.3");
        context.init(kmf.getKeyManagers(), tmf.getTrustManagers(), new SecureRandom());
        return context;
    }

    private synchronized void maybeReload() {
        File keystoreFile = new File(keystorePath);
        if (keystoreFile.lastModified() > lastLoadedTime) {
            log.info("Detected keystore modification. Reloading SSL context...");
            try {
                this.sslContext = createSSLContext();
                this.lastLoadedTime = System.currentTimeMillis();
                log.info("Successfully reloaded SSL context.");
            } catch (Exception e) {
                log.error("Failed to reload SSL context. Reusing old context.", e);
            }
        }
    }

    @Override
    public SSLEngine createSslEngine(String peerHost, int peerPort, String endpointIdentificationAlgorithm) {
        maybeReload();
        SSLEngine engine = sslContext.createSSLEngine(peerHost, peerPort);
        engine.setUseClientMode(true);
        return engine;
    }

    @Override
    public SSLParameters.Builder sslParametersBuilder() {
        maybeReload();
        SSLParameters parameters = sslContext.getDefaultSSLParameters();
        return new SSLParameters.Builder(parameters);
    }

    @Override
    public Set<String> reconfigurableConfigs() {
        return Set.of("ssl.truststore.location", "ssl.keystore.location");
    }

    @Override
    public void close() {}
}

snippet-8.txt

package com.production.kafka;

import java.io.ByteArrayInputStream;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.security.KeyFactory;
import java.security.KeyStore;
import java.security.PrivateKey;
import java.security.cert.Certificate;
import java.security.cert.CertificateFactory;
import java.security.spec.PKCS8EncodedKeySpec;
import java.util.Base64;

public class PemReader {
    public static KeyStore loadKeyStore(String certPath, String keyPath, String password) throws Exception {
        byte[] certBytes = Files.readAllBytes(Paths.get(certPath));
        byte[] keyBytes = Files.readAllBytes(Paths.get(keyPath));

        // Load Certificate
        CertificateFactory cf = CertificateFactory.getInstance("X.509");
        Certificate cert = cf.generateCertificate(new ByteArrayInputStream(certBytes));

        // Load Private Key (assuming PKCS8 format)
        String privateKeyPEM = new String(keyBytes)
                .replace("-----BEGIN PRIVATE KEY-----", "")
                .replaceAll("\\s", "")
                .replace("-----END PRIVATE KEY-----", "");
        byte[] decodedKey = Base64.getDecoder().decode(privateKeyPEM);
        PKCS8EncodedKeySpec spec = new PKCS8EncodedKeySpec(decodedKey);
        KeyFactory kf = KeyFactory.getInstance("RSA");
        PrivateKey privateKey = kf.generatePrivate(spec);

        // Put into in-memory KeyStore
        KeyStore keyStore = KeyStore.getInstance("PKCS12");
        keyStore.load(null, null);
        keyStore.setKeyEntry("kafka-client", privateKey, password.toCharArray(), new Certificate[]{cert});
        return keyStore;
    }
}