<a id="configure-cs-for-oauth"></a>

# Configure Confluent Server Brokers for OAuth Authentication in Confluent Platform

The following configurations are for Confluent Server brokers, which are the proprietary
Kafka brokers included in Confluent Platform Enterprise.

## Configure a listener

Configure a listener with SASL/OAUTHBEARER to use OAuth authentication. In the
following configurations, a listener named `external` is configured for the same.

#### NOTE
You can get the values of `<idp-jwks-endpoint>`, `<idp-token-endpoint>`,
and other parameters from the OIDC discovery endpoint, which is usually the
following URL, replacing `<YOUR_IDP_DOMAIN>` with the actual domain name:

```text
https://<YOUR_IDP_DOMAIN>/.well-known/openid-configuration
```

### Required configuration settings

```text
# Configure a new listener
listeners=..,EXTERNAL://:9095
advertised.listeners=..,EXTERNAL://<host>:9095
listener.security.protocol.map=...,EXTERNAL:SASL_SSL

# Add OAuth to the Kafka listener
listener.name.external.sasl.enabled.mechanisms=OAUTHBEARER
listener.name.external.sasl.oauthbearer.jwks.endpoint.url=<idp-jwks-endpoint>
listener.name.external.sasl.oauthbearer.expected.audience=<idp-audience>
listener.name.external.principal.builder.class=io.confluent.kafka.security.authenticator.OAuthKafkaPrincipalBuilder
confluent.oauth.groups.claim.name=groups
```

### Optional configuration settings

The following settings are optional and depend on your Confluent Platform configuration.

#### For non-RBAC configurations

If you are not using RBAC, use the following SASL callback validator and
corresponding JAAS configuration.

```text
listener.name.external.oauthbearer.sasl.server.callback.handler.class=org.apache.kafka.common.security.oauthbearer.OAuthBearerValidatorCallbackHandler
listener.name.external.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;
```

#### For RBAC configurations

If you are using role-based access control (RBAC), then you must use the following SASL callback validator
and corresponding JAAS configuration.

```text
listener.name.external.oauthbearer.sasl.server.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.CompositeBearerValidatorCallbackHandler
listener.name.external.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required unsecuredLoginStringClaim_sub="thePrincipalName" publicKeyPath="/var/ssl/private/public.pem";
```

#### IMPORTANT
Without the `CompositeBearerValidatorCallbackHandler`, most Confluent Control Center API calls
from the Confluent Control Center UI fail.

#### Use a custom `sub` claim name

If you want to use a custom `sub` claim name, add the following configuration
setting, replacing `<sub_claim_name>` with your actual name:

```text
listener.name.external.sasl.oauthbearer.sub.claim.name=<sub_claim_name>
```

## Configure interbroker communication (optional)

To use the SASL/OAUTHBEARER as the interbroker listener, add the following
configuration settings, replacing `<idp-token-endpoint>`, `<client-id>`,
`<client-secret>`, and `<scope>` with your actual values:

```text
inter.broker.listener.name=EXTERNAL
sasl.mechanism.inter.broker.protocol=OAUTHBEARER
listener.name.external.oauthbearer.sasl.login.callback.handler.class=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler
listener.name.external.sasl.oauthbearer.token.endpoint.url=<idp-token-endpoint>
listener.name.external.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
    clientId="<client-id>" \
    clientSecret="<client-secret>" \
    scope="<scope>";
```

#### NOTE
The `scope` parameter is optional, and required only when your identity
provider does not have a default `scope` or your `groups` claim is
linked to a `scope`.

### Configure a superuser

When you use SASL/OAUTHBEARER for the interbroker listener, ensure that the
principal in the authentication token of the client serves as the superuser by
adding the following configuration setting, replacing `<sub-claim-value>`
with your actual value:

```text
super.users= ...;User:<sub-claim-value>  # usually User:<client-id>
```

### Configure the metadata server

If you want to keep LDAP and IdP both for User store (Authentication source),
you should be using `LDAP_WITH_OAUTH`. This is advised to use while migrating
from one user store to another. For details on using `LDAP_WITH_OAUTH`, see
[Configure Metadata Service (MDS) for OAuth Authentication in Confluent Platform](configure-mds.md#configure-mds-for-oauth).

## Configure for KRaft mode

When Kafka is running in KRaft mode, add the following required configurations
on controller nodes because requests are forwarded to the controller.
`OAuthKafkaPrincipalBuilder` provides the needed group extraction on the
controller.

```properties
listener.name.external.principal.builder.class=io.confluent.kafka.security.authenticator.OAuthKafkaPrincipalBuilder
confluent.oauth.groups.claim.name=groups
```

<a id="configure-ak-clients-for-oauth"></a>

## Configure Kafka clients

Add the following configurations to your Kafka clients to use OAuth authentication.

```properties
sasl.mechanism=OAUTHBEARER
security.protocol=SASL_SSL
sasl.login.callback.handler.class=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler
sasl.oauthbearer.token.endpoint.url=<idp-token-endpoint>
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
   clientId="<client-id>" \
   clientSecret="<client-secret>" \
   scope="<scope>";
# scope is optional and required only when your identity provider does not
# have a default scope or your groups claim is linked to a scope.
```

For a Kafka console consumer based on the preceding configuration, you can use the
following `kafka-console-consumer` command, replacing `<host>` with the
actual host name.

```shell
./bin/kafka-console-consumer \
  --bootstrap-server <host>:9095 \
  --topic purchases \
  --consumer.config oauth-client.properties
```

For Kafka Java clients supporting SASL OAUTHBEARER, allow specific IdP endpoints by setting the following configuration property:

```properties
org.apache.kafka.sasl.oauthbearer.allowed.urls=<idp_jwks_url>,<idp_token_url>,...
```

This property specifies a comma-separated list of allowed IdP JWKS (JSON Web Key
Set) and token endpoint URLs. Use \* (asterisk) as the value to allow any endpoint.

```properties
org.apache.kafka.sasl.oauthbearer.allowed.urls=*
```

You should consult the specific Kafka client and IdP documentation for the
exact interpretation and security implications of such a broad setting.

Java applications should set this property as a JVM system property when
launching the application:

```bash
-Dorg.apache.kafka.sasl.oauthbearer.allowed.urls=<idp_jwks_url>,<idp_token_url>,...
```

For other clients (for example, Python, Go, .NET) that are built on
librdkafka, these clients use different property names and configuration
mechanisms. So, refer to specific client library documentation for the
equivalent OAuthBEARER configuration properties.

## Related content

* [Configure Truststores for TLS Handshake with Identity Providers](configure-truststore.md#configure-truststore-for-oauth)
* [Use OAuth/OIDC for Authentication in Confluent Platform](overview.md#oauth-oidc-authentication-overview)
* [Use SASL/OAUTHBEARER Authentication Between Confluent Server Brokers and Kafka Clients in Confluent Platform](../sasl/oauthbearer/overview.md#sasl-oauthbearer-authentication-overview)
* [Passwordless Authentication for Confluent Server and Schema Registry](../../../schema-registry/security/passwordless-auth.md#passwordless-auth)