Authentication

This guide explains how a Kafka client proves its identity to the KubeMQ Kafka connector, and how the connector decides what that identity is allowed to do. The connector supports four ways to authenticate — SASL/PLAIN, SASL/SCRAM-SHA-256 or SCRAM-SHA-512, OAUTHBEARER against an OIDC provider, and mutual TLS — plus a Casbin-backed ACL that authorizes every Produce, Fetch, and group-coordinator request against the resolved identity.

Authentication mechanisms at a glance

SASL/PLAIN and SCRAM

SASL/PLAIN and SASL/SCRAM share the same credential store: Connectors.Kafka.Credentials, a list of {Username, Password} pairs. It's config-file/secret-only — there is no environment variable and no CRD field, so plan how you'll deliver it (a mounted config.yaml or a Secret-mounted file) before you turn SASL on. The moment Credentials is non-empty, the connector enforces SASL on every listener — plaintext (SASL_PLAINTEXT) and TLS (SASL_SSL) alike.

SaslMechanisms is an operator allow-list. Leave it empty and the connector offers all three password-based mechanisms — PLAIN, SCRAM-SHA-256, SCRAM-SHA-512; set it explicitly to restrict the handshake (for example ["SCRAM-SHA-256", "SCRAM-SHA-512"] to drop cleartext PLAIN). SCRAM's verifier is derived from each credential's password with PBKDF2, using ScramIterations (default 4096, the RFC 7677 minimum) — raise it for a slower, more brute-force-resistant boot-time derivation.

A client that offers a mechanism outside the allow-list, or authenticates with a wrong username/password, is closed with UNSUPPORTED_SASL_MECHANISM(33) or SASL_AUTHENTICATION_FAILED(58) respectively — neither is retried in place.

The wire config barely differs between PLAIN and SCRAM — only the mechanism name, and the credential derivation on the server side, change:

# SASL/PLAIN, plaintext transport
kcat -b localhost:9092 -L \
  -X security.protocol=SASL_PLAINTEXT \
  -X sasl.mechanisms=PLAIN \
  -X sasl.username=alice \
  -X sasl.password="$KAFKA_PASSWORD"

# SASL/SCRAM-SHA-256 — same flags, different mechanism
# (add -X security.protocol=SASL_SSL -X ssl.ca.location=... to run it over TLS)
kcat -b localhost:9092 -L \
  -X security.protocol=SASL_PLAINTEXT \
  -X sasl.mechanisms=SCRAM-SHA-256 \
  -X sasl.username=alice \
  -X sasl.password="$KAFKA_PASSWORD"

import (
    "os"

    "github.com/twmb/franz-go/pkg/kgo"
    "github.com/twmb/franz-go/pkg/sasl/plain"
    "github.com/twmb/franz-go/pkg/sasl/scram"
)

// SASL/PLAIN
cl, err := kgo.NewClient(
    kgo.SeedBrokers("localhost:9092"),
    kgo.SASL(plain.Auth{User: "alice", Pass: os.Getenv("KAFKA_PASSWORD")}.AsMechanism()),
)

// SASL/SCRAM-SHA-256 — swap the mechanism constructor, everything else is identical
cl, err = kgo.NewClient(
    kgo.SeedBrokers("localhost:9092"),
    kgo.SASL(scram.Auth{User: "alice", Pass: os.Getenv("KAFKA_PASSWORD")}.AsSha256Mechanism()),
)

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("security.protocol", "SASL_PLAINTEXT");
props.put("sasl.mechanism", "PLAIN"); // or "SCRAM-SHA-256" / "SCRAM-SHA-512"

String module = "org.apache.kafka.common.security.plain.PlainLoginModule";
// for SCRAM, use: "org.apache.kafka.common.security.scram.ScramLoginModule"
props.put("sasl.jaas.config", module + " required username=\"alice\" password=\""
    + System.getenv("KAFKA_PASSWORD") + "\";");

OAUTHBEARER — OIDC federated tokens

OAUTHBEARER activates the moment Connectors.Kafka.OAuthBearer.Issuer is non-empty — there's no separate enable flag. It's enforced only on the TLS listener (SASL_SSL, TlsPort); a client that offers OAUTHBEARER on the plaintext listener is refused UNSUPPORTED_SASL_MECHANISM(33), because a bearer token must never cross an unencrypted transport. Configure the TLS listener first — see TLS and mTLS.

The broker validates the token against your OIDC provider and takes the token's sub claim as the authenticated principal — the same Casbin gate as SASL/SCRAM. ClientID (the OAuth2 audience) is checked unless you set SkipClientIDCheck; the expiry, issuer, and signature checks are hard-rejected if you try to disable them — SkipExpiryCheck, SkipIssuerCheck, and InsecureSkipSignatureCheck must all stay false. SkipClientIDCheck is the one flag Kafka lets you set true, for IdPs that omit or vary the audience claim. Provider discovery runs in the background at boot, so the server never blocks startup on an unreachable IdP — until the provider is ready, OAUTHBEARER auth fails closed rather than silently accepting.

import (
    "context"
    "crypto/tls"

    "github.com/twmb/franz-go/pkg/kgo"
    "github.com/twmb/franz-go/pkg/sasl/oauth"
)

cl, err := kgo.NewClient(
    kgo.SeedBrokers("localhost:9093"), // OAUTHBEARER is TLS-only
    kgo.DialTLSConfig(&tls.Config{}),  // uses the OS trust store — load your CA into RootCAs for a private/dev cert (see TLS and mTLS)
    kgo.SASL(oauth.Oauth(func(ctx context.Context) (oauth.Auth, error) {
        token, err := fetchOIDCToken(ctx) // your IdP's client-credentials call
        return oauth.Auth{Token: token}, err
    })),
)

security.protocol=SASL_SSL
sasl.mechanism=OAUTHBEARER
sasl.login.callback.handler.class=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler
sasl.oauthbearer.token.endpoint.url=https://idp.example.com/oauth2/token
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  clientId="kafka-client" \
  clientSecret="${OIDC_CLIENT_SECRET}";

mTLS — certificate identity

mTLS resolves identity from the connection itself — no SASL exchange, no password. When the global Security block is in mTLS mode (a client CA is configured; see TLS and mTLS), the connector requires and verifies a client certificate at the TLS handshake and takes the verified chain's leaf certificate Subject.CommonName as the principal. An unverified or CN-less certificate never becomes a principal — the connection is treated as unauthenticated rather than trusting a client-asserted name.

Authorization — the ACL model

Once a principal is resolved, the connector maps every Kafka operation onto the same Casbin authorization engine every KubeMQ connector shares — see Security → Authorization. It never ingests real Kafka ACLs; access is authored directly as Casbin policy against (ClientID, resource, channel), where ClientID is the principal you just authenticated.

When server-wide Authorization is disabled, every request is allowed regardless of principal. When it's enabled, a request with no authenticated principal (no SASL, no mTLS) is denied outright — there's no anonymous fallback once ACL enforcement is on.

Quick reference

Related