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 configurations below, 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:

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

Required configuration settings

# 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.

listener.name.external.oauthbearer.sasl.server.callback.handler.class=org.apache.kafka.common.security.oauthbearer.secured.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.

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:

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:

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.secured.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:

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 other. For details on using LDAP_WITH_OAUTH, see Configure Metadata Service (MDS) for OAuth Authentication in Confluent Platform.

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.

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

Configure Kafka clients

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

sasl.mechanism=OAUTHBEARER
security.protocol=SASL_SSL
sasl.login.callback.handler.class=org.apache.kafka.common.security.oauthbearer.secured.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 doesn't have a default scope or your groups claim is linked to a scope.

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

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

Related content