# Authentication (/connectors/kafka/how-to/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.

<Callout type="info">
  On a stock dev broker, `Connectors.Kafka.Credentials` is empty and no SASL mechanism is enforced,
  so the runnable examples across these docs connect with no credentials at all. Configure a
  credential store (below) to turn SASL on; TLS/mTLS is a separate, additive setting — see
  [TLS and mTLS](/connectors/kafka/how-to/tls-and-mtls).
</Callout>

## Authentication mechanisms at a glance [#authentication-mechanisms-at-a-glance]

| Mechanism                          | Activated by                                                        | Credential                                                | Principal                              |
| ---------------------------------- | ------------------------------------------------------------------- | --------------------------------------------------------- | -------------------------------------- |
| SASL/PLAIN                         | `Connectors.Kafka.Credentials` non-empty                            | username + password, checked against the credential store | the matched username                   |
| SASL/SCRAM-SHA-256 / SCRAM-SHA-512 | `Credentials` non-empty, mechanism allowed                          | RFC 5802/7677 salted challenge-response                   | the matched username                   |
| SASL/OAUTHBEARER                   | `OAuthBearer.Issuer` set + `OAUTHBEARER` in `SaslMechanisms`        | an OIDC bearer token, validated on the TLS listener only  | the token's `sub` claim                |
| mTLS                               | the global `Security` block in **mTLS** mode, and SASL not required | a verified client certificate                             | the certificate's `Subject.CommonName` |

## SASL/PLAIN and SCRAM [#saslplain-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:

<Tabs groupId="language" items="['kcat', 'Go', 'Java']">
  <Tab value="kcat">
    ```bash
    # 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"
    ```
  </Tab>

  <Tab value="Go">
    ```go
    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()),
    )
    ```
  </Tab>

  <Tab value="Java">
    ```java
    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") + "\";");
    ```
  </Tab>
</Tabs>

## OAUTHBEARER — OIDC federated tokens [#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](/connectors/kafka/how-to/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.

<Tabs groupId="language" items="['Go', 'Java']">
  <Tab value="Go">
    ```go
    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
        })),
    )
    ```
  </Tab>

  <Tab value="Java">
    ```properties
    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}";
    ```
  </Tab>
</Tabs>

## mTLS — certificate identity [#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](/connectors/kafka/how-to/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.

<Callout type="warn">
  **mTLS and SASL don't stack.** There's exactly one authenticated principal per connection. The
  certificate CN is used only when SASL is **not** required (`Credentials` empty); if a listener
  has both `Credentials` and mTLS configured, the SASL identity wins and the CN is never
  consulted.
</Callout>

## Authorization — the ACL model [#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](/configure/reference/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.

| Operation(s)                                                                                              | Required access        | Denied with                                                 |
| --------------------------------------------------------------------------------------------------------- | ---------------------- | ----------------------------------------------------------- |
| `Produce`, `OffsetCommit`(8), `OffsetDelete`(47), `DeleteGroups`(42)                                      | **Write**              | `TOPIC_AUTHORIZATION_FAILED` / `GROUP_AUTHORIZATION_FAILED` |
| `Fetch`, `ListOffsets`, `OffsetFetch`, `JoinGroup`/`SyncGroup`/`Heartbeat`/`LeaveGroup`, `DescribeGroups` | **Read**               | same                                                        |
| `AddOffsetsToTxn`(25) / `TxnOffsetCommit`(28) — the transactional offset-commit route                     | **Write** on the group | `GROUP_AUTHORIZATION_FAILED`                                |

<Callout type="warn">
  **Migrating an EOS producer from real Kafka?** `AddOffsetsToTxn`/`TxnOffsetCommit` require
  **Group WRITE** here — Apache Kafka itself only requires Group **Read** for that route. Grant
  your transactional principal Group Write, not the Kafka-default Group Read, or its first offset
  commit inside a transaction fails with `GROUP_AUTHORIZATION_FAILED`. See
  [Transactions & EOS](/connectors/kafka/how-to/transactions).
</Callout>

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 [#quick-reference]

| You want…                           | Do this                                                                                                                          |
| ----------------------------------- | -------------------------------------------------------------------------------------------------------------------------------- |
| Clone-and-run on a stock dev broker | No credentials — leave `Credentials` empty                                                                                       |
| Username/password authentication    | SASL/PLAIN or SCRAM-SHA-256/512 — configure `Credentials`                                                                        |
| Federate identity to your own IdP   | OAUTHBEARER — set `OAuthBearer.Issuer`, enable the TLS listener                                                                  |
| Certificate identity, no password   | mTLS — put the `Security` block in mTLS mode, leave `Credentials` empty                                                          |
| Diagnose a rejected connection      | `UNSUPPORTED_SASL_MECHANISM(33)` (bad mechanism / OAUTHBEARER on plaintext) or `SASL_AUTHENTICATION_FAILED(58)` (bad credential) |

## Related [#related]

<Cards>
  <Card title="TLS and mTLS" href="/connectors/kafka/how-to/tls-and-mtls" description="Secure the Kafka connector with TLS — the 9093 encrypted listener, server certificates, and mutual TLS where the client certificate's common name becomes the authenticated principal." />

  <Card title="Configuration" href="/connectors/kafka/concepts/configuration" description="How the Kafka connector is enabled, ported, and secured — the opt-in CONNECTORS_KAFKA_ENABLE flag and the 9092/9093 listeners." />

  <Card title="Kafka settings reference" href="/configure/reference/connectors#kafka" description="The full field-by-field Connectors.Kafka settings — Credentials, ScramIterations, SaslMechanisms, and the OAUTHBEARER block." />
</Cards>