Kafka SASL: OAUTHBEARER and PLAIN simultaniously

Question

What I am trying to do is -

For Clients to Broker communication - use OAUTHBEARER authentication
For Broker to Broker communication - use PLAIN authentication

I Have following JAAS configuration:

{
  KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="inter"
    password="inter-secret"
    user_inter="inter-secret"
    user_admin="YvNzcbmqhA0DfxjP";

    org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;
  };

  Client {
    org.apache.zookeeper.server.auth.DigestLoginModule required
    username="zookeeper"
    password="zookeeper-secret";
  };
}

And I have following configs in server.properties:

sasl.enabled.mechanisms=PLAIN,OAUTHBEARER
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.server.callback.handler.class=br.com.jairsjunior.security.oauthbearer.OauthAuthenticateValidatorCallbackHandler

But if start the kafka service I am seeing the error like below:

used by: java.lang.IllegalArgumentException: Must supply exactly 1 non-null JAAS mechanism configuration (size was 2)
at org.apache.kafka.common.security.oauthbearer.internals.unsecured.OAuthBearerUnsecuredValidatorCallbackHandler.configure(OAuthBearerUnsecuredValidatorCallbackHandler.java:114)
at org.apache.kafka.common.network.SaslChannelBuilder.configure(SaslChannelBuilder.java:122)
... 17 more

which indicates kafka is not allowing to specify multiple JAAS mechanism configurations.

So how can I specify multiple JAAS configs, and setup authentication mechanisms like below:

CLient to Broker ----> OAUTHBEARER
Broker to Broker ----> PLAIN

Thanks!

Answer 1

You can configure two mechanisms like this (if your oauth use AAD Client Credential Flow), it works well in my broker:

Broker config:

listeners=SASL://0.0.0.0:9770,SASL_CONTROLLER://0.0.0.0:9773,SASL_INTERNAL://0.0.0.0:9774
advertised.listeners=SASL://localhost:9770,SASL_CONTROLLER://localhost:9773,SASL_INTERNAL://localhost:9774
listener.security.protocol.map=SASL:SASL_PLAINTEXT,SASL_CONTROLLER:SASL_PLAINTEXT,SASL_INTERNAL:SASL_PLAINTEXT

sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=OAUTHBEARER,PLAIN
control.plane.listener.name = SASL_CONTROLLER
inter.broker.listener.name = SASL_INTERNAL

listener.name.sasl.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="AdminName" password="AdminPassword" user_UserName="UserPassword";
listener.name.sasl.plain.sasl.server.callback.handler.class=com.XXXAuthenticateCallbackHandler
listener.name.sasl_controller.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="AdminName" password="AdminPassword" user_UserName="UserPassword";
listener.name.sasl_controller.plain.sasl.server.callback.handler.class=com.XXXAuthenticateCallbackHandler
listener.name.sasl_internal.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="AdminName" password="AdminPassword" user_UserName="UserPassword";
listener.name.sasl_internal.plain.sasl.server.callback.handler.class=com.XXXAuthenticateCallbackHandler

listener.name.sasl.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required clientId='BrokerApplicationId' clientSecret='BrokerApplicationSecret' scope='BrokerApplicationId/.default';
listener.name.sasl.oauthbearer.sasl.login.callback.handler.class=org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler
listener.name.sasl.oauthbearer.sasl.server.callback.handler.class=org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerValidatorCallbackHandler
listener.name.sasl_controller.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required clientId='BrokerApplicationId' clientSecret='BrokerApplicationSecret' scope='BrokerApplicationId/.default';
listener.name.sasl_controller.oauthbearer.sasl.login.callback.handler.class=org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler
listener.name.sasl_controller.oauthbearer.sasl.server.callback.handler.class=org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerValidatorCallbackHandler
listener.name.sasl_internal.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required clientId='BrokerApplicationId' clientSecret='BrokerApplicationSecret' scope='BrokerApplicationId/.default';
listener.name.sasl_internal.oauthbearer.sasl.login.callback.handler.class=org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler
listener.name.sasl_internal.oauthbearer.sasl.server.callback.handler.class=org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerValidatorCallbackHandler

sasl.oauthbearer.jwks.endpoint.url=https://login.microsoftonline.com/BrokerTenantId/discovery/v2.0/keys
sasl.oauthbearer.token.endpoint.url=https://login.microsoftonline.com/BrokerTenantId/oauth2/v2.0/token
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
allow.everyone.if.no.acl.found=true
super.users=User:AdminName

Client config:

Plain:

bootstrap.servers=xxxx:9770
compression.type=none
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="UserName" password="UserPassword";

Oauth:

bootstrap.servers=xxxx:9770
compression.type=none
sasl.oauthbearer.token.endpoint.url=https://login.microsoftonline.com/ClientTenantId/oauth2/v2.0/token
sasl.login.callback.handler.class=org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler
sasl.mechanism=OAUTHBEARER
security.protocol=SASL_PLAINTEXT
sasl.jaas.config= \
  org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
    clientId='ClientApplicationId' \
    scope='BrokerApplicationId/.default' \
    clientSecret='ClientApplicationSecret';

Answer 2

I am currently also working on the problem to use plain and oauthbearer simultaniously, which I have not solved yet but I solved your specific question in the following way. This is my Jaas Configuration:

internal.KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin-secret"
    user_admin="admin-secret"
    user_test="test";
};

external.KafkaServer {
    org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required
};

Client {
   org.apache.zookeeper.server.auth.DigestLoginModule required
   username="username"
   password="pw";
};

Then I set the setting in the server.properties the following way:

  inter.broker.listener.name: INTERNAL 
  sasl.mechanism.inter.broker.protocol: PLAIN
  listener.security.protocol.map: INTERNAL:SASL_PLAINTEXT,EXTERNAL:SASL_SSL
  listeners: "INTERNAL://0.0.0.0:9092,EXTERNAL://0.0.0.0:19092"
  sasl.enabled.mechanisms: PLAIN,OAUTHBEARER

  listener.name.external.oauthbearer.sasl.server.callback.handler.class: my.module.kafka.security.oauthbearer.OauthAuthenticateValidatorCallbackHandler
  listener.name.external.oauthbearer.sasl.login.callback.handler.class: my.module.kafka.security.oauthbearer.OauthAuthenticateLoginCallbackHandler

When you it this way you won't get your error. Sadly I get another error when the broker want to set up the external connection:

javax.security.auth.callback.UnsupportedCallbackException: Unrecognized SASL Login callback
    at org.apache.kafka.common.security.authenticator.AbstractLogin$DefaultLoginCallbackHandler.handle(AbstractLogin.java:105)
    at org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule.identifyToken(OAuthBearerLoginModule.java:316)
    at org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule.login(OAuthBearerLoginModule.java:301)
    ... 32 more

It seems like the kafka brokers are ignoring oauthbearer callbackhandler. This is a bit strange because external is working perfectly when I configure it as the only listener.

I hope it helps you with your problem!