Hi Ashish, I'm glad you were able to get it working! :)
Reply to your questions below... On Mon, Mar 31, 2025, at 10:19 PM, ashish sood wrote: > Hi Kirk, > > I managed to get it running finally. > > > *Server.properties* > > listeners=SASL_PLAINTEXT://:9093 > advertised.listeners=SASL_PLAINTEXT://xxxxxx:9093 > sasl.enabled.mechanisms=OAUTHBEARER > sasl.oauthbearer.expected.audience=https://myprovider.com > oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule > required; > inter.broker.listener.name=SASL_PLAINTEXT > sasl.mechanism.inter.broker.protocol=OAUTHBEARER > listener.name.sasl_plaintext.oauthbearer.sasl.login.callback.handler.class=org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler > listener.name.sasl_plaintext.oauthbearer.sasl.server.callback.handler.class=org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerValidatorCallbackHandler > sasl.oauthbearer.token.endpoint.url= > https://xxxxxxxxxxx.auth0.com/oauth/token > sasl.oauthbearer.jwks.endpoint.url= > https://xxxxxxxxxxxxx.auth0.com/.well-known/jwks.json > > *jaas.config* > KafkaServer { > org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule > required > clientId="xxxxxxxxxxxxxxxxxxxxxx" > clientSecret="xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx" > token.endpoint.uri="xxxxxxxxxxxxxxxxxxxauth0.com/oauth/token"; > }; > > Although I have a conceptual doubt if you could please help. How can we > leverage the scope coming in the token ? For example - If the token has > scope set to *kafka.read*, the client should have read access to a specific > topic. Do we need to write a custom class for this OR the existing classes > (OAuthBearerValidatorCallbackHandler/OAuthBearerLoginCallbackHandler) can > help achieve this ? This would unfortunately require a custom validator callback handler, yes. > Currently I am using the Client ID in the access token in the ACL to allow > read/write permissions on topics. Are you wanting to use Auth0 for both authentication and authorization instead of authentication via Auth0 and authorization via ACLs? Thanks, Kirk > > Thanks & Regards > Ashish > > On Tue, Apr 1, 2025 at 6:07 AM Kirk True <k...@kirktrue.pro> wrote: > > > Hi Ashish, > > > > In your stack trace I see it's invoking > > org.apache.kafka.common.security.oauthbearer.internals.unsecured.OAuthBearerUnsecuredLoginCallbackHandler, > > so something in your configuration seems amiss. > > > > If you can capture the AdminClientConfig output (with sensitive stuff > > redacted, obvs), that would be helpful. > > > > Thanks, > > Kirk > > > > On Thu, Mar 20, 2025, at 3:55 AM, ashish sood wrote: > > > Hi Kirk, > > > > > > Thanks for checking. > > > > > > I am trying to setup a Kafka cluster with end-to-end oauth (i.e. Kafka - > > > Kafka communication within a cluster & clients to Kafka broker). I was > > able > > > to get my broker started without errors with below config however I am > > now > > > unable to create topics with below error. > > > > > > *Current config* > > > *jaas.config* > > > > > > KafkaServer { > > > org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule > > > required > > > clientId="<xxxxxxxx>" > > > clientSecret="<xxxxxxxx>" > > > audience="https://myprovider.com"; > > > token.endpoint.uri="https://xxxxxxxx/oauth/token"; > > > scope="kafka.read kafka.write"; > > > }; > > > > > > *server.properties* > > > listeners=SASL_PLAINTEXT://:9093 > > > advertised.listeners=SASL_PLAINTEXT://<>:9093 > > > sasl.enabled.mechanisms=OAUTHBEARER > > > sasl.oauthbearer.expected.audience=https://myprovider.com > > > > > oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule > > > required; > > > inter.broker.listener.name=SASL_PLAINTEXT > > > sasl.mechanism.inter.broker.protocol=OAUTHBEARER > > > > > listener.name.sasl_plaintext.oauthbearer.sasl.login.callback.handler.class=org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler > > > > > listener.name.sasl_plaintext.oauthbearer.sasl.server.callback.handler.class=org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerValidatorCallbackHandler > > > sasl.oauthbearer.token.endpoint.url=https://<xxxxxxxx>/oauth/token > > > sasl.oauthbearer.jwks.endpoint.url=https:// > > <xxxxxxxx>/.well-known/jwks.json > > > > > > *ERROR WHILE CREATING TOPIC* > > > > > > This is very strange because when I check the fetch the token manually > > via > > > curl and check it , I clearly see the "sub" field populated with value > > > <clientid@clients> > > > > > > > > > * ERROR No principal name in JWT claim: sub > > > > > (org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule)java.io.IOException: > > > No principal name in JWT claim: sub* > > > at > > > > > org.apache.kafka.common.security.oauthbearer.internals.unsecured.OAuthBearerUnsecuredLoginCallbackHandler.handle(OAuthBearerUnsecuredLoginCallbackHandler.java:165) > > > at > > > > > org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule.identifyToken(OAuthBearerLoginModule.java:316) > > > at > > > > > org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule.login(OAuthBearerLoginModule.java:301) > > > at > > > > > java.base/javax.security.auth.login.LoginContext.invoke(LoginContext.java:754) > > > at > > > > > java.base/javax.security.auth.login.LoginContext$4.run(LoginContext.java:678) > > > at > > > > > java.base/javax.security.auth.login.LoginContext$4.run(LoginContext.java:676) > > > at > > > > > java.base/java.security.AccessController.doPrivileged(AccessController.java:714) > > > at > > > > > java.base/javax.security.auth.login.LoginContext.invokePriv(LoginContext.java:676) > > > at > > > > > java.base/javax.security.auth.login.LoginContext.login(LoginContext.java:587) > > > at > > > > > org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin.login(ExpiringCredentialRefreshingLogin.java:204) > > > at > > > > > org.apache.kafka.common.security.oauthbearer.internals.OAuthBearerRefreshingLogin.login(OAuthBearerRefreshingLogin.java:150) > > > at > > > > > org.apache.kafka.common.security.authenticator.LoginManager.<init>(LoginManager.java:62) > > > at > > > > > org.apache.kafka.common.security.authenticator.LoginManager.acquireLoginManager(LoginManager.java:105) > > > at > > > > > org.apache.kafka.common.network.SaslChannelBuilder.configure(SaslChannelBuilder.java:170) > > > at > > > > > org.apache.kafka.common.network.ChannelBuilders.create(ChannelBuilders.java:192) > > > at > > > > > org.apache.kafka.common.network.ChannelBuilders.clientChannelBuilder(ChannelBuilders.java:81) > > > at > > > > > org.apache.kafka.clients.ClientUtils.createChannelBuilder(ClientUtils.java:119) > > > at > > > > > org.apache.kafka.clients.ClientUtils.createNetworkClient(ClientUtils.java:223) > > > at > > > > > org.apache.kafka.clients.ClientUtils.createNetworkClient(ClientUtils.java:189) > > > at > > > > > org.apache.kafka.clients.admin.KafkaAdminClient.createInternal(KafkaAdminClient.java:525) > > > at > > > > > org.apache.kafka.clients.admin.KafkaAdminClient.createInternal(KafkaAdminClient.java:492) > > > at org.apache.kafka.clients.admin.Admin.create(Admin.java:137) > > > at > > > > > org.apache.kafka.tools.TopicCommand$TopicService.createAdminClient(TopicCommand.java:437) > > > at > > > > > org.apache.kafka.tools.TopicCommand$TopicService.<init>(TopicCommand.java:426) > > > at > > org.apache.kafka.tools.TopicCommand.execute(TopicCommand.java:98) > > > at > > > org.apache.kafka.tools.TopicCommand.mainNoExit(TopicCommand.java:87) > > > at org.apache.kafka.tools.TopicCommand.main(TopicCommand.java:82) > > > > > > > > > > > > > > > > > > Regards > > > Ashish Sood > > > > > > On Thu, Mar 20, 2025 at 12:15 AM Kirk True <k...@kirktrue.pro> wrote: > > > > > > > Hi Ashish, > > > > > > > > Are you using OAuth for client->broker communication, inter-broker > > > > communication, or both? > > > > > > > > Based on the server.properties configuration that was shared, it looks > > > > like the configuration is attempting to set up inter-broker > > communication > > > > using OAuth. > > > > > > > > For a broker to *retrieve* tokens , it needs to have this > > configuration: > > > > > > > > > > > > > > listener.name.SASL_PLAINTEXT.oauthbearer.sasl.login.callback.handler.class=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler > > > > > > > > For a broker to *validate* tokens, it needs to have this configuration: > > > > > > > > > > > > > > listener.name.SASL_PLAINTEXT.oauthbearer.sasl.server.callback.handler.class=org.apache.kafka.common.security.oauthbearer.OAuthBearerValidatorCallbackHandler > > > > > > > > Then the SASL configs would need to be included too: > > > > > > > > > > listener.name.SASL_PLAINTEXT.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule > > > > required \ > > > > clientId="XXXXXXXXXXXXXXXXXX" > > > > clientSecret="XXXXXXXXXXXXXXXXXX" > > > > audience="https://myprovider.com"; > > > > serviceName="kafka" > > > > scope="kafka.read kafka.write"; > > > > > > > > If possible, please share any non-sensitive logs. > > > > > > > > Thanks, > > > > Kirk > > > > > > > > On Wed, Mar 19, 2025, at 3:41 AM, ashish sood wrote: > > > > > Hi All, > > > > > > > > > > I am setting up oauth for my Kafka broker. I have set up an account > > on > > > > Auth0 for the same and set up an application and API. > > > > > > > > > > With the below config in the server.properties and Jaas.config file I > > > > keep getting invalid token. Although if I generate a manual token via > > curl > > > > it works fine. Also Auth0 logs show successful generation of the token, > > > > still the Kafka shows error. Any suggestions to resolve this issue > > would be > > > > appreciated. > > > > > > > > > > *Server.properties* > > > > > listeners=SASL_PLAINTEXT://:9093 > > > > > advertised.listeners=SASL_PLAINTEXT://<XXXXXX>:9093 > > > > > sasl.enabled.mechanisms=OAUTHBEARER > > > > > sasl.oauthbearer.jwks.endpoint.url=https://XXXXXXXXX/oauth/token < > > > > https://xxxxxxxxx/oauth/token> > > > > > sasl.oauthbearer.expected.audience=https://myprovider.com > > > > > > > > > > > oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule > > > > required; > > > > > > > > > > > listener.name.sasl_plaintext.oauthbearer.sasl.login.callback.handler.class=org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler > > > > > confluent.oauth.groups.claim.name=groups > > > > > inter.broker.listener.name=SASL_PLAINTEXT > > > > > sasl.mechanism.inter.broker.protocol=OAUTHBEARER > > > > > super.users=User:<ClientID> > > > > > sasl.oauthbearer.token.endpoint.url=<XXXXXXXXX>/oauth/token > > > > > sasl.oauthbearer.audience=https://myprovider.com > > > > > allow.everyone.if.no.acl.found=true > > > > > ** > > > > > *Jaas Config* > > > > > KafkaServer { > > > > > > > org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule > > > > required > > > > > clientId="XXXXXXXXXXXXXXXXXX" > > > > > clientSecret="XXXXXXXXXXXXXXXXXX" > > > > > audience="https://myprovider.com"; > > > > > serviceName="kafka" > > > > > scope="kafka.read kafka.write"; > > > > > }; > > > > > > > > > > *Error* > > > > > [2025-03-19 16:05:43,465] INFO [Controller id=0, targetBrokerId=0] > > > > Failed authentication with localhost/127.0.0.1 (channelId=0) > > > > ({"status":"invalid_token"}) (org.apache.kafka.common.network.Selector) > > > > > > > > > > image.png > > > > > > > > > > image.png > > > > > > > > > > Thanks & Regards > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > ReplyForward > > > > > > > > > > > > > > > Add reaction > > > > > > > > > >
