TLS Encryption

Encrypt communication between Dekaf and Kafka brokers using TLS.

Basic TLS

Enable TLS for encrypted connections:

using Dekaf;

var fileCertificateProducer = await Kafka.CreateProducer<string, string>()
    .WithBootstrapServers("kafka.example.com:9093")
    .UseTls()
    .BuildAsync();

This uses system CA certificates to validate the broker's certificate.

Custom CA Certificate

If your Kafka uses a private CA:

using Dekaf;

var tlsConfig = new TlsConfig
{
    CaCertificatePath = "/path/to/ca.crt"
};

var inMemoryCertificateProducer = await Kafka.CreateProducer<string, string>()
    .WithBootstrapServers("kafka.example.com:9093")
    .UseTls(tlsConfig)
    .BuildAsync();

Mutual TLS (mTLS)

For client certificate authentication:

using Dekaf;

// Using file paths
var certificateProducer = await Kafka.CreateProducer<string, string>()
    .WithBootstrapServers("kafka.example.com:9093")
    .UseMutualTls(
        caCertPath: "/path/to/ca.crt",
        clientCertPath: "/path/to/client.crt",
        clientKeyPath: "/path/to/client.key",
        keyPassword: "optional-password"
    )
    .BuildAsync();

// Using X509Certificate2
var clientCert = new X509Certificate2("client.pfx", "password");
var caCert = new X509Certificate2("ca.crt");

var producer = await Kafka.CreateProducer<string, string>()
    .WithBootstrapServers("kafka.example.com:9093")
    .UseMutualTls(clientCert, caCert)
    .BuildAsync();

TLS Configuration Options

using Dekaf;

var tlsConfig = new TlsConfig
{
    // CA certificate for server validation
    CaCertificatePath = "/path/to/ca.crt",

    // Client certificate for mTLS
    ClientCertificatePath = "/path/to/client.crt",
    ClientKeyPath = "/path/to/client.key",
    ClientKeyPassword = "password",

    // Validate the broker certificate chain
    ValidateServerCertificate = true,

    // Verify that the certificate matches the connection host name
    ValidateServerCertificateHostName = true,

    // Allowed TLS protocols
    EnabledSslProtocols = SslProtocols.Tls12 | SslProtocols.Tls13
};

var producer = await Kafka.CreateProducer<string, string>()
    .WithBootstrapServers("kafka.example.com:9093")
    .UseTls(tlsConfig)
    .BuildAsync();

Common Configurations

AWS MSK with IAM

using Dekaf;

var producer = await Kafka.CreateProducer<string, string>()
    .WithBootstrapServers("broker1.msk.us-east-1.amazonaws.com:9098")
    .UseTls()
    .WithAwsMskIam()
    .BuildAsync();

Confluent Cloud

using Dekaf;

var producer = await Kafka.CreateProducer<string, string>()
    .WithBootstrapServers("pkc-xxxxx.us-east-1.aws.confluent.cloud:9092")
    .UseTls()
    .WithSaslPlain(apiKey, apiSecret)
    .BuildAsync();

Troubleshooting

Certificate Validation Fails

AuthenticationException: The remote certificate was rejected

Solutions:

1. Ensure CA certificate is correct
2. Check certificate chain is complete
3. Verify server hostname matches certificate

Host Name Mismatch Only

If the broker certificate chain is trusted but the certificate host name does not match the address used by clients, disable only host-name verification:

var tlsConfig = new TlsConfig
{
    CaCertificatePath = "/path/to/ca.crt",
    ValidateServerCertificate = true,
    ValidateServerCertificateHostName = false
};

For Development Only

// ⚠️ NEVER use in production
var tlsConfig = new TlsConfig
{
    ValidateServerCertificate = false
};

Debug Certificate Issues

var cert = X509CertificateLoader.LoadCertificateFromFile("/path/to/ca.crt");
Console.WriteLine($"Subject: {cert.Subject}");
Console.WriteLine($"Issuer: {cert.Issuer}");

var tlsConfig = new TlsConfig
{
    CaCertificateObject = cert,
    ValidateServerCertificate = true
};