Skip to main content

TLS Encryption

Encrypt communication between Dekaf and Kafka brokers using TLS.

Basic TLS​

Enable TLS for encrypted connections:

using Dekaf;

var producer = 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 producer = 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 producer = 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 tlsConfig = new TlsConfig
{
    CaCertificatePath = "/path/to/ca.crt",
    ServerCertificateValidationCallback = (sender, cert, chain, errors) =>
    {
        Console.WriteLine($"Subject: {cert.Subject}");
        Console.WriteLine($"Issuer: {cert.Issuer}");
        Console.WriteLine($"Errors: {errors}");
        return errors == SslPolicyErrors.None;
    }
};