Authentication

TLS, mTLS, SASL PLAIN, SCRAM, OAUTHBEARER and the built-in OIDC token provider.

Overview

krafka supports multiple security protocols:

Not supported: GSSAPI / Kerberos

krafka implements PLAIN, SCRAM-SHA-256/512, OAUTHBEARER and AWS MSK IAM. It does not implement GSSAPI, and this is a deliberate position rather than an oversight.

There is no mature pure-Rust GSSAPI implementation. Supporting it would mean linking libgssapi (and MIT Kerberos or Heimdal), which would break the crate's central promise — no C dependencies by default — for every user, in exchange for a mechanism most deployments do not use. Kerberos-integrated environments are therefore not served by this client today.

If you need Kerberos, rust-rdkafka supports it through librdkafka. If your broker offers OAUTHBEARER alongside Kerberos, that path works here and is the usual migration route.

Supported SASL Mechanisms

Security Protocol Selection

use krafka::auth::{AuthConfig, SecurityProtocol};

// Check what's configured
let config = AuthConfig::sasl_scram_sha256("user", "pass");
println!("Protocol: {}", config.security_protocol());
println!("Requires TLS: {}", config.requires_tls());
println!("Requires SASL: {}", config.requires_sasl());

Adding TLS to any mechanism

Every mechanism composes with TLS through one method, with_tls:

use krafka::auth::{AuthConfig, TlsConfig};

// SASL_SSL + SCRAM-SHA-512 — the default secured listener on Redpanda Cloud,
// Aiven, Instaclustr and most Strimzi installs.
let tls = TlsConfig::new().with_ca_cert("/etc/kafka/ca.pem");
let config = AuthConfig::sasl_scram_sha512("username", "password").with_tls(tls);

Prefer this to the per-mechanism _ssl constructors when you build the TLS configuration separately — it is the same thing, and it cannot be missing for the mechanism you happen to need. tests/builder_surface.rs asserts that every SaslMechanism is constructible under both SASL_PLAINTEXT and SASL_SSL from the public API alone.

SASL Authentication

SASL/PLAIN

Simple username/password authentication. Always use with TLS in production!

use krafka::auth::AuthConfig;

// Without TLS (development only!)
let config = AuthConfig::sasl_plain("username", "password")?;

// With TLS (recommended for production)
use krafka::auth::TlsConfig;
let config = AuthConfig::sasl_plain_ssl("username", "password", TlsConfig::new())?;

SASL/SCRAM-SHA-256

Challenge-response authentication with SHA-256 hashing. More secure than PLAIN.

use krafka::auth::{AuthConfig, TlsConfig};

// Without TLS (development only!)
let config = AuthConfig::sasl_scram_sha256("username", "password");

// With TLS (recommended for production)
let config = AuthConfig::sasl_scram_sha256_ssl("username", "password", TlsConfig::new());

SASL/SCRAM-SHA-512

Maximum security SCRAM authentication with SHA-512 hashing.

use krafka::auth::{AuthConfig, TlsConfig};

// Without TLS (development only!)
let config = AuthConfig::sasl_scram_sha512("username", "password");

// With TLS — this is what a managed Kafka offering almost always wants
let config = AuthConfig::sasl_scram_sha512_ssl("username", "password", TlsConfig::new());

Over TLS, SCRAM is additionally bound to the TLS session with tls-server-end-point channel binding (RFC 5929 §4.1) unless you turn it off with with_scram_channel_binding(false).

SCRAM Protocol Details

The SCRAM client implements RFC 5802 with:

• Salted Challenge-Response mechanism
• PBKDF2 key derivation with iteration count validation (4,096–1,000,000 range)
• HMAC signature verification
• Constant-time comparison via the subtle crate (timing-attack resistant)
• Automatic secret zeroization on drop (password, salted_password, server_signature)
• Debug output redacts the password as [REDACTED]

use krafka::auth::{ChannelBinding, ScramClient, ScramMechanism, ScramState};

// Create SCRAM client (no channel binding for SASL_PLAINTEXT)
let mut scram = ScramClient::new("alice", "secret", ScramMechanism::Sha256, ChannelBinding::None);
assert_eq!(scram.state(), ScramState::Initial);

// Generate client-first message
let client_first = scram.client_first_message();
// -> "n,,n=alice,r=<nonce>"

// When using SASL_SSL, pass channel binding data to tie SCRAM to the TLS session:
// let cb_data = extract_tls_server_end_point(&tls_stream).unwrap();
// let mut scram = ScramClient::new("alice", "secret", ScramMechanism::Sha256,
//     ChannelBinding::TlsServerEndPoint(cb_data));
// -> client-first: "p=tls-server-end-point,,n=alice,r=<nonce>"

// Process server-first message
// scram.process_server_first(server_response)?;

// Generate client-final message
// let client_final = scram.client_final_message()?;

// Verify server-final
// scram.process_server_final(server_response)?;

SASL/OAUTHBEARER

OAuth 2.0 bearer token authentication per RFC 7628 and KIP-255.

use krafka::auth::{AuthConfig, OAuthBearerToken};

// Basic token authentication
let config = AuthConfig::sasl_oauthbearer("your-jwt-token-here");

// With TLS (recommended for production)
use krafka::auth::TlsConfig;
let config = AuthConfig::sasl_oauthbearer_ssl("your-jwt-token-here", TlsConfig::new());

With SASL Extensions

For providers like Confluent Cloud that require additional SASL extensions:

use krafka::auth::{AuthConfig, OAuthBearerToken};

// Create token with extensions
let token = OAuthBearerToken::new("your-jwt-token")
    .with_extension("logicalCluster", "lkc-abc123")
    .with_extension("identityPoolId", "pool-xyz789");

let config = AuthConfig::sasl_oauthbearer_token(token);

// Or with TLS
use krafka::auth::TlsConfig;
let config = AuthConfig::sasl_oauthbearer_token_ssl(
    OAuthBearerToken::new("your-jwt-token")
        .with_extension("logicalCluster", "lkc-abc123"),
    TlsConfig::new(),
);

Builder Convenience Methods

All client builders support shorthand .sasl_oauthbearer(token) and .sasl_oauthbearer_provider(provider) methods:

use krafka::auth::OAuthBearerToken;
use krafka::producer::Producer;
use krafka::consumer::Consumer;

// Static token
let producer = Producer::builder()
    .bootstrap_servers("broker:9093")
    .sasl_oauthbearer("your-jwt-token")
    .build()
    .await?;

// Token provider (recommended)
let consumer = Consumer::builder()
    .bootstrap_servers("broker:9093")
    .group_id("my-group")
    .sasl_oauthbearer_provider(|| async {
        let token = fetch_token_from_oauth_server().await?;
        Ok(OAuthBearerToken::new(token))
    })
    .build()
    .await?;

Automatic Token Refresh via Provider

For production use, implement the OAuthBearerTokenProvider trait so that krafka can fetch a fresh token on every new broker connection — including automatic reconnections. This eliminates the need to restart clients when tokens expire.

Closure provider (simplest)

use krafka::auth::{AuthConfig, OAuthBearerToken};

let config = AuthConfig::sasl_oauthbearer_provider(|| async {
    // Called on every new broker connection
    let jwt = my_oauth_client.get_access_token().await?;
    Ok(OAuthBearerToken::new(jwt))
});

Struct provider (when you need shared state)

Security Note: Wrap secrets like client_secret in zeroize::Zeroizing<String> so they are erased from memory on drop. This does not by itself prevent the secret from being exposed via Debug/Display if the containing struct is logged or derives Debug. Callers must still avoid logging secrets and should implement a redacted Debug for any struct that holds credentials (or otherwise ensure secret fields are never formatted).

use krafka::auth::{OAuthBearerToken, OAuthBearerTokenProvider};
use krafka::error::Result;
use std::future::Future;
use std::pin::Pin;
use zeroize::Zeroizing;

struct MyTokenProvider {
    client_id: String,
    client_secret: Zeroizing<String>,
    token_url: String,
}

impl OAuthBearerTokenProvider for MyTokenProvider {
    fn provide_token(
        &self,
    ) -> Pin<Box<dyn Future<Output = Result<OAuthBearerToken>> + Send + '_>> {
        Box::pin(async move {
            // Use your preferred HTTP client to fetch a token
            let jwt = fetch_oauth_token(
                &self.token_url,
                &self.client_id,
                &self.client_secret,
            ).await?;
            Ok(OAuthBearerToken::new(jwt))
        })
    }
}

// Use with any client builder
let consumer = Consumer::builder()
    .bootstrap_servers("broker:9093")
    .group_id("my-group")
    .sasl_oauthbearer_provider(MyTokenProvider {
        client_id: "my-app".into(),
        client_secret: Zeroizing::new("secret".into()),
        token_url: "https://auth.example.com/oauth/token".into(),
    })
    .build()
    .await?;

With TLS (production)

use krafka::auth::{AuthConfig, OAuthBearerToken, TlsConfig};

let config = AuthConfig::sasl_oauthbearer_provider_ssl(
    || async { Ok(OAuthBearerToken::new("fresh-jwt")) },
    TlsConfig::new(),
);

How it works: The provider is called once per broker connection. When the connection pool detects a disconnection and reconnects, the provider is called again — delivering a fresh token without any client restart. Implementations may cache tokens internally and only refresh when approaching expiry. Provider resolution is bounded by the configured request timeout (default 30 s) to prevent hung providers from stalling reconnection loops.

If OAuthBearerToken::with_lifetime_ms() is set, krafka rejects tokens that are already expired or within 30 seconds of expiry before starting the SASL handshake. This avoids avoidable broker-side failures caused by client/broker clock skew. Provider implementations should return a token with comfortably more than 30 seconds of remaining lifetime.

OAUTHBEARER Protocol Details

The implementation follows RFC 7628 GS2 framing:

• Initial response: n,,\x01auth=Bearer <token>[\x01key=value]*\x01\x01
• Server success: Empty response (0 bytes)
• Server error: JSON or text error message
• Security: Token zeroized on drop via zeroize crate
• Debug safety: Token redacted as [REDACTED] in Debug output
• Extensions: Arbitrary key-value pairs appended to the GS2 frame

Note: GSSAPI/Kerberos is not supported. It requires system Kerberos libraries via FFI, which is incompatible with krafka’s #![deny(unsafe_code)] policy. Use OAUTHBEARER or SCRAM as alternatives.

Built-in OIDC token provider (oauth-oidc)

Enable the oauth-oidc feature and krafka fetches access tokens itself, instead of you writing the OAuth client:

cargo add krafka --features oauth-oidc

Two ways to authenticate to the token endpoint:

Client secret:

use krafka::auth::{AuthConfig, oidc::{ClientCredentials, OidcTokenProvider}};
use std::time::Duration;

let provider = OidcTokenProvider::builder("https://idp.example.com/oauth2/token")
    .credentials(ClientCredentials::secret("my-client-id", "my-client-secret"))
    .scope("kafka:write")
    .request_timeout(Duration::from_secs(10))
    .build()?;

let auth = AuthConfig::sasl_oauthbearer_provider(provider);

Client assertion (KIP-1258) — stronger, because the credential on the wire is a short-lived signature rather than a long-lived shared secret, and the private key never leaves the workload:

use krafka::auth::oidc::{AssertionSource, ClientCredentials, OidcTokenProvider};

let provider = OidcTokenProvider::builder("https://idp.example.com/oauth2/token")
    .credentials(ClientCredentials::assertion(
        // Re-read on every token request, so a SPIFFE agent or Vault sidecar
        // can rotate the assertion without restarting the process.
        AssertionSource::File("/var/run/secrets/oauth/assertion.jwt".into()),
    ))
    .client_id("my-client-id")
    .build()?;

krafka does not sign the assertion

Signing needs RSA or ECDSA, and pinning a specific implementation on every user of a Kafka client is a supply-chain decision that belongs to the application. So the JWT is sourced, not produced:

Confluent Cloud SASL extensions

SASL extensions travel in the OAUTHBEARER exchange with Kafka, and are deliberately a different list from the token-request form parameters — the identity provider and the broker are different audiences:

OidcTokenProvider::builder("https://idp.example.com/oauth2/token")
    .credentials(ClientCredentials::secret("id", "secret"))
    .sasl_extension("logicalCluster", "lkc-123")   // sent to Kafka
    .sasl_extension("identityPoolId", "pool-456")  // sent to Kafka
    .form_parameter("audience", "kafka")           // sent to the identity provider
    .build()?;

Behaviour worth knowing

• https is required. A plain-http token endpoint is rejected at build time: the request carries a client credential and the response carries an access token.
• Errors name the cause. RFC 6749 §5.2 bodies are parsed, so a failure reads invalid_client: unknown client id rather than HTTP 400.
• expires_in drives refresh. The token store caches until the token nears expiry; a response without expires_in falls back to the bounded unknown-expiry schedule rather than being cached forever.
• Secrets are redacted in Debug and zeroized on drop. That applies throughout the stack, including the protocol layer: SaslAuthenticateRequest and SaslAuthenticateResponse report a byte count rather than their auth_bytes, which for SASL/PLAIN is \0username\0password in cleartext and for OAUTHBEARER is the bearer token verbatim. just secret-debug fails CI if a credential-bearing type ever derives Debug again.

TLS/SSL Encryption

Crypto backend

krafka's TLS is rustls, which needs a crypto backend. ring is the default; rustls-aws-lc-rs selects aws-lc-rs, preferable on AWS Graviton and in FIPS-oriented deployments:

cargo add krafka --no-default-features --features rustls-aws-lc-rs,compression

The two features are additive. Cargo features cannot be made mutually exclusive without breaking dependency graphs where two crates each pick a different backend, so enabling rustls-aws-lc-rs on top of the default ring (or building with --all-features) is a supported configuration: aws-lc-rs wins deterministically. krafka selects the provider explicitly on every path, including certificate verification, rather than letting rustls infer it from crate features — inference panics when the features are ambiguous.

To pick the backend for the whole process, including krafka, install one before opening any connection:

rustls::crypto::aws_lc_rs::default_provider()
    .install_default()
    .expect("crypto provider already installed");

Basic TLS

Use Mozilla's root certificates for server verification:

use krafka::auth::{AuthConfig, TlsConfig};

let config = AuthConfig::ssl(TlsConfig::new());

Custom CA Certificate

For self-signed or private CA certificates:

use krafka::auth::TlsConfig;

let tls_config = TlsConfig::new()
    .with_ca_cert("/path/to/ca.pem");

with_ca_cert() pins the trust store to the provided CA bundle — the default WebPKI (Mozilla) roots are not loaded. This matches the Java Kafka client (ssl.truststore.location) and librdkafka (ssl.ca.location).

Native Platform Trust Stores

By default, krafka uses compiled-in webpki-roots. To use the operating system trust store on macOS, Windows, or Linux, enable the native-tls-roots feature and opt in explicitly:

cargo add krafka --features native-tls-roots

use krafka::auth::TlsConfig;

let tls_config = TlsConfig::new()
    .with_native_roots();

You can combine with_native_roots() and with_ca_cert() to trust both platform roots and an additional private CA bundle.

Mutual TLS (mTLS)

Client certificate authentication:

use krafka::auth::TlsConfig;

let tls_config = TlsConfig::new()
    .with_ca_cert("/path/to/ca.pem")
    .with_client_cert("/path/to/client.pem", "/path/to/client-key.pem");

SNI Hostname

For servers behind load balancers or proxies:

use krafka::auth::TlsConfig;

let mut tls_config = TlsConfig::new();
tls_config.sni_hostname = Some("kafka.example.com".to_string());

Certificate rotation (KIP-1288)

Certificates rotated on disk by cert-manager, Vault or an SDS sidecar are picked up without restarting the process. Apache Kafka added this to the Java client in 4.2 (KIP-1288); krafka offers both an event-driven and an unattended path.

Event-driven — call refresh_tls() when your watcher fires. Available on Producer, Consumer, AdminClient and KrafkaClient:

// inotify fired, the secret volume was remounted, the sidecar signalled …
producer.refresh_tls().await?;

Unattended — reload on a timer:

use krafka::network::TransportConfig;
use std::time::Duration;

let transport = TransportConfig::builder()
    .tls_reload_interval(Some(Duration::from_secs(3600)))
    .build()?;

Both paths behave the same way:

• Existing TLS sessions are unaffected. They keep the connector they handshaked with and are replaced naturally as connections cycle. Only connections opened after a successful reload use the new material.
• A failed reload keeps the old certificates. Catching a half-written PEM mid-rotation logs a warning and changes nothing, so a non-atomic rotation converges on the next attempt rather than breaking every new connection in between.
• No-op without TLS. Nothing on disk to reload.

Using a KrafkaClient shares one pool across all its producers, consumers and admin clients, so one refresh_tls() there rotates certificates for all of them.

Skip Verification (Development Only)

Never use in production!

use krafka::auth::TlsConfig;

let tls_config = TlsConfig::insecure();

AWS MSK IAM Authentication

For AWS Managed Streaming for Apache Kafka using IAM authentication:

Binary Size Note: The aws-msk feature adds the AWS SDK, which increases binary size by approximately 2-3 MB (release build). If binary size is critical, use AwsMskIamCredentials::from_env() which works without the aws-msk feature.

From Environment Variables (Recommended)

The simplest approach is to load credentials from environment variables:

use krafka::auth::{AuthConfig, AwsMskIamCredentials};

// Load from AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKEN, AWS_REGION
let creds = AwsMskIamCredentials::from_env()?;
let config = AuthConfig::aws_msk_iam_with_credentials(creds);

Environment variables used:

• AWS_ACCESS_KEY_ID - Required
• AWS_SECRET_ACCESS_KEY - Required
• AWS_SESSION_TOKEN - Optional (for temporary credentials)
• AWS_REGION or AWS_DEFAULT_REGION - Required

When the region comes from your own configuration

If the keys live in the environment but the region comes from a config file or a secret manager, use from_env_with_region, which neither reads nor requires AWS_REGION:

use krafka::auth::{AuthConfig, AwsMskIamCredentials};

let creds = AwsMskIamCredentials::from_env_with_region(configured_region)?;
let config = AuthConfig::aws_msk_iam_with_credentials(creds);

To re-region a credential you already have, use with_region:

let creds = AwsMskIamCredentials::from_env()?.with_region("eu-central-1");

Do not rebuild the credential through new to change one field: secret_access_key and session_token are deliberately unreadable, so rebuilding silently drops the session token. Every deployment using an assumed role, an EC2/ECS instance profile or an EKS web identity then fails SigV4 verification at connect time, with an error that never mentions the token.

From AWS SDK Default Chain (Recommended for Production)

For production deployments on EC2, ECS, Lambda, or EKS, use the AWS SDK default chain:

use krafka::auth::{AuthConfig, AwsMskIamCredentials};

// Requires the `aws-msk` feature:
//   cargo add krafka --features aws-msk

// Loads from (in order):
// 1. Environment variables
// 2. Shared credentials file (~/.aws/credentials)
// 3. IAM role for EC2/ECS/Lambda
// 4. Web identity token (for EKS)
let creds = AwsMskIamCredentials::from_default_chain("us-east-1").await?;
let config = AuthConfig::aws_msk_iam_with_credentials(creds);

With Explicit Credentials (Development Only)

For development or testing, you can provide credentials directly:

use krafka::auth::AuthConfig;

// With permanent credentials (avoid in production!)
let config = AuthConfig::aws_msk_iam(
    "AKIAIOSFODNN7EXAMPLE",
    "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
    "us-east-1",
);

// With temporary credentials (session token)
use krafka::auth::AwsMskIamCredentials;

let creds = AwsMskIamCredentials::new(
    "AKIAIOSFODNN7EXAMPLE",
    "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
    "us-east-1",
)
.with_session_token("session-token-here");

Using SecureConnectionConfig with MSK IAM

use krafka::network::SecureConnectionConfig;

let config = SecureConnectionConfig::builder()
    .client_id("msk-client")
    .aws_msk_iam("AKID", "secret", "us-east-1")
    .build();

Automatic Credential Refresh (Recommended)

For production workloads using temporary credentials (STS, IRSA, ECS task role, EC2 instance profile), use a credential provider so that credentials are automatically refreshed on every broker reconnection:

use krafka::auth::{AuthConfig, AwsMskIamCredentials};

// With a closure (requires `aws-msk` feature for from_default_chain)
let config = AuthConfig::aws_msk_iam_provider(|| async {
    AwsMskIamCredentials::from_default_chain("us-east-1").await
});

// Or implement AwsMskIamCredentialProvider for custom logic
use krafka::auth::AwsMskIamCredentialProvider;

struct MyCredentialProvider;
impl AwsMskIamCredentialProvider for MyCredentialProvider {
    fn provide_credentials(
        &self,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = krafka::error::Result<AwsMskIamCredentials>> + Send + '_>> {
        Box::pin(async {
            // Custom credential loading logic
            AwsMskIamCredentials::from_env()
        })
    }
}

let config = AuthConfig::aws_msk_iam_provider(MyCredentialProvider);

The provider pattern mirrors OAUTHBEARER's sasl_oauthbearer_provider(). The SecureConnectionConfig builder also supports it:

use krafka::network::SecureConnectionConfig;
use krafka::auth::AwsMskIamCredentials;

let config = SecureConnectionConfig::builder()
    .client_id("msk-client")
    .aws_msk_iam_provider(|| async {
        AwsMskIamCredentials::from_default_chain("us-east-1").await
    })
    .build();

Direct MskIamAuthenticator Usage

For low-level control over the authentication process:

use krafka::auth::{AwsMskIamCredentials, MskIamAuthenticator};

let creds = AwsMskIamCredentials::new("AKID", "secret", "us-east-1");
let authenticator = MskIamAuthenticator::new(&creds, "broker.kafka.us-east-1.amazonaws.com")?;

// Generate signed authentication payload
let payload = authenticator.create_auth_payload();
// -> JSON with AWS Signature v4 signed request

MSK IAM Protocol Details

The implementation uses AWS Signature v4 signing:

• Service Name: kafka-cluster
• Action: kafka-cluster:Connect
• Payload Format: JSON with signed headers
• TLS Required: Always uses SASL_SSL (TLS is mandatory)
• Region-Aware: Credentials are scoped to AWS region
• Clock Skew: Authentication uses the system clock. On recognized SigV4 clock-skew failures, reconnects apply a best-effort correction capped at +/-300 seconds; larger drift should be fixed with NTP or host time sync.

Configuration Options

TlsConfig

ALPN Protocol Negotiation

Some environments (service meshes, load balancers like Envoy or AWS ALB) require ALPN for protocol multiplexing. Use with_kafka_alpn() as a convenience or with_alpn_protocols() for custom protocols:

use krafka::auth::TlsConfig;

// Advertise "kafka" ALPN protocol
let tls = TlsConfig::new().with_kafka_alpn();

// Or custom protocols
let tls = TlsConfig::new().with_alpn_protocols(vec![b"kafka".to_vec()]);

AuthConfig

Plus one method that applies to all of them:

The _ssl constructors are shorthand for with_tls. Reach for with_tls when you build the TlsConfig separately or when you want one code path that handles every mechanism.

From environment variables

AuthConfig::from_env builds any of the above from the standard Kafka environment variables. It is a convenience for applications whose configuration is the environment; a library embedder resolving credentials from a secret manager should build the AuthConfig directly.

AWS_MSK_IAM takes its credentials from AwsMskIamCredentials::from_env(). Setting only one half of the client-certificate pair is an error rather than a silently ignored setting. There is deliberately no environment variable that disables certificate verification — use TlsConfig::insecure() in code if you need that.

Client Authentication

All krafka clients — AdminClient, Producer, TransactionalProducer, and Consumer — support the same authentication methods through dedicated builder methods. Authentication is wired end-to-end: TLS upgrade and SASL handshake happen automatically during connection establishment.

Admin Client

use krafka::AdminClient;

// SASL/PLAIN
let admin = AdminClient::builder()
    .client_id("admin-client")
    .bootstrap_servers("broker:9092")
    .sasl_plain("username", "password")
    .build();

// SASL/SCRAM-SHA-256
let admin = AdminClient::builder()
    .bootstrap_servers("broker:9092")
    .sasl_scram_sha256("username", "password")
    .build();

// SASL/SCRAM-SHA-512
let admin = AdminClient::builder()
    .bootstrap_servers("broker:9092")
    .sasl_scram_sha512("username", "password")
    .build();

Producer

use krafka::producer::Producer;

// SASL/PLAIN
let producer = Producer::builder()
    .bootstrap_servers("broker:9092")
    .sasl_plain("username", "password")
    .build()
    .await?;

// SASL/SCRAM-SHA-256
let producer = Producer::builder()
    .bootstrap_servers("broker:9092")
    .sasl_scram_sha256("username", "password")
    .build()
    .await?;

// SASL/SCRAM-SHA-512
let producer = Producer::builder()
    .bootstrap_servers("broker:9092")
    .sasl_scram_sha512("username", "password")
    .build()
    .await?;

Consumer

use krafka::consumer::Consumer;

// SASL/PLAIN
let consumer = Consumer::builder()
    .bootstrap_servers("broker:9092")
    .group_id("my-group")
    .sasl_plain("username", "password")
    .build()
    .await?;

// SASL/SCRAM-SHA-256
let consumer = Consumer::builder()
    .bootstrap_servers("broker:9092")
    .group_id("my-group")
    .sasl_scram_sha256("username", "password")
    .build()
    .await?;

// SASL/SCRAM-SHA-512
let consumer = Consumer::builder()
    .bootstrap_servers("broker:9092")
    .group_id("my-group")
    .sasl_scram_sha512("username", "password")
    .build()
    .await?;

Transactional Producer

use krafka::producer::TransactionalProducer;

// SASL/PLAIN
let producer = TransactionalProducer::builder()
    .bootstrap_servers("broker:9092")
    .transactional_id("my-txn-id")
    .sasl_plain("username", "password")
    .build()
    .await?;

// SASL/SCRAM-SHA-256
let producer = TransactionalProducer::builder()
    .bootstrap_servers("broker:9092")
    .transactional_id("my-txn-id")
    .sasl_scram_sha256("username", "password")
    .build()
    .await?;

// SASL/SCRAM-SHA-512
let producer = TransactionalProducer::builder()
    .bootstrap_servers("broker:9092")
    .transactional_id("my-txn-id")
    .sasl_scram_sha512("username", "password")
    .build()
    .await?;

Generic AuthConfig

For advanced configurations or AWS MSK IAM, use .auth() on any builder:

use krafka::AdminClient;
use krafka::producer::{Producer, TransactionalProducer};
use krafka::consumer::Consumer;
use krafka::auth::AuthConfig;

let auth = AuthConfig::aws_msk_iam("access_key", "secret_key", "us-east-1");

// Works on all client types
let admin = AdminClient::builder()
    .bootstrap_servers("broker:9092")
    .auth(auth.clone())
    .build();

let producer = Producer::builder()
    .bootstrap_servers("broker:9092")
    .auth(auth.clone())
    .build()
    .await?;

let txn_producer = TransactionalProducer::builder()
    .bootstrap_servers("broker:9092")
    .transactional_id("my-txn-id")
    .auth(auth.clone())
    .build()
    .await?;

let consumer = Consumer::builder()
    .bootstrap_servers("broker:9092")
    .group_id("my-group")
    .auth(auth)
    .build()
    .await?;

Session Reauthentication (KIP-368)

krafka supports KIP-368 session lifetime tracking. When a broker reports a session lifetime via SaslAuthenticateResponse v1, krafka tracks the expiry and proactively replaces the connection before the session expires.

How It Works

1. During SASL handshake, the broker may include a session_lifetime_ms value in its v1 response.
2. If non-zero, krafka calculates a reauthentication deadline at a randomised point between 85% and 95% of the lifetime. The jitter prevents a thundering-herd where many connections to the same broker all expire simultaneously.
3. When the connection pool serves a connection request, it checks is_usable() — which verifies the connection is both alive and not past its reauthentication deadline.
4. Expired-session connections are transparently replaced with a fresh connection that performs a new SASL handshake.

This behaviour matches the Java Kafka client and is fully automatic — no client configuration is required. It works with all SASL mechanisms and is especially important for OAUTHBEARER, where tokens have a natural expiry.

Security Best Practices

1. Always use TLS in production - Use SASL_SSL instead of SASL_PLAINTEXT
2. Prefer SCRAM over PLAIN - SCRAM provides challenge-response security
3. Use mTLS for strongest authentication - Client certificates are harder to steal
4. Store credentials securely - Use environment variables or secrets managers
5. Rotate credentials regularly - Especially for long-running applications
6. Verify certificates in production - Never use TlsConfig::insecure() in production
7. Automatic secret zeroization - All credential types (ScramClient, MskIamAuthenticator, PlainCredentials, ScramCredentials, OAuthBearerToken) zeroize secrets on drop to prevent memory leaks. SASL PLAIN auth bytes are wrapped in Zeroizing<Vec<u8>> and automatically zeroized after being sent on the wire.
8. Debug safety - All credential types redact secrets in Debug output, so tracing::debug!("{:?}", auth) is safe to use
9. Cleartext warning - Using SASL_PLAINTEXT with PLAIN emits a tracing::warn! alerting that credentials will be sent in cleartext

Secure Connection Configuration

For integrated TLS and SASL configuration, use SecureConnectionConfig:

use krafka::network::SecureConnectionConfig;
use krafka::auth::TlsConfig;
use std::time::Duration;

let config = SecureConnectionConfig::builder()
    .client_id("my-app")
    .connect_timeout(Duration::from_secs(10))
    .sasl_scram_sha256("username", "password")
    .tls(TlsConfig::new())
    .build();

SaslAuthenticator

For handling SASL handshakes, use SaslAuthenticator:

use krafka::network::SaslAuthenticator;
use krafka::auth::{AuthConfig, ChannelBinding};

let auth = AuthConfig::sasl_scram_sha256("user", "pass");
let mut authenticator = SaslAuthenticator::new(&auth, ChannelBinding::None).unwrap();

// Get mechanism name for SASL handshake
let mechanism = authenticator.mechanism_name(); // "SCRAM-SHA-256"

// Get initial authentication bytes
let initial = authenticator.initial_response()?;

// Process server challenges
// let response = authenticator.process_challenge(&server_bytes)?;

// Check completion
if authenticator.is_complete() {
    println!("Authentication successful!");
}

When using OAUTHBEARER with a token provider, resolve the provider before creating the authenticator:

use krafka::network::SaslAuthenticator;
use krafka::auth::{AuthConfig, OAuthBearerToken};

let auth = AuthConfig::sasl_oauthbearer_provider(|| async {
    Ok(OAuthBearerToken::new("fresh-jwt"))
});

// Resolve the provider to get a config with the token set
let resolved = auth.resolve_provider_to_token().await?;
let auth = resolved.as_ref().unwrap_or(&auth);
let mut authenticator = SaslAuthenticator::new(auth, ChannelBinding::None).unwrap();

Example: Production Configuration

use krafka::auth::{AuthConfig, TlsConfig};
use krafka::producer::Producer;
use krafka::consumer::Consumer;
use std::env;

fn production_auth_config() -> AuthConfig {
    let username = env::var("KAFKA_USER").expect("KAFKA_USER required");
    let password = env::var("KAFKA_PASSWORD").expect("KAFKA_PASSWORD required");
    
    let tls_config = TlsConfig::new()
        .with_ca_cert("/etc/ssl/certs/kafka-ca.pem");
    
    // SCRAM-SHA-512 over TLS
    AuthConfig {
        security_protocol: krafka::auth::SecurityProtocol::SaslSsl,
        sasl_mechanism: Some(krafka::auth::SaslMechanism::ScramSha512),
        scram_credentials: Some(krafka::auth::ScramCredentials::new(username, password)),
        tls_config: Some(tls_config),
        ..Default::default()
    }
}

// Use with any client
async fn create_clients() {
    let auth = production_auth_config();

    let producer = Producer::builder()
        .bootstrap_servers("kafka.prod.example.com:9093")
        .auth(auth.clone())
        .build()
        .await
        .unwrap();

    let consumer = Consumer::builder()
        .bootstrap_servers("kafka.prod.example.com:9093")
        .group_id("prod-group")
        .auth(auth)
        .build()
        .await
        .unwrap();
}

Next Steps

• Producer Guide - Configure authenticated producers
• Consumer Guide - Configure authenticated consumers
• Configuration Reference - All connection options