I'm working with SASL_SSL in a Kafka Cluster and I am trying to configure some clients to get a token from an OAuth server. I've managed to setup my configs, however I hit a wall when faced with 2 options for my callback handlers. You see, there's 2 very similar properties and I can't seem to find anything verbose regarding them:
sasl.login.callback.handler.class
and
sasl.client.callback.handler.class
These properties require that the class implements AuthenticateCallbackHandler
From Kafka Docs I can see this regarding the login.callback.handler
The fully qualified name of a SASL login callback handler class that implements the AuthenticateCallbackHandler
interface. For brokers, login callback handler config must be prefixed with listener prefix and SASL
mechanism name in lower-case. For example, listener.name.sasl_ssl.scram-sha-256.sasl.login.callback.handler.class=com.example.CustomScramLoginCallbackHandler
which isn't very helpful. The client.callback.handler seems to have the same docs:
The fully qualified name of a SASL login callback handler class that implements the AuthenticateCallbackHandler interface.
Am I correct to assume that the login.callback.handler is a broker-side prop and the client.callback.handler is a client-side prop? Which one should I use to configure my clients(consumer/producer) to request a token from the OAuth Server? It seems to work with both properties
You should configure like this:
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
sasl.login.callback.handler.class is used to specify which loginmodule the client requests tokens based on, and needs to be configured in both the broker and client. （I don't know why the broker also needs to be configured, but when I test it, the broker must also be configured, otherwise an error will be reported.）
sasl.server.callback.handler.class is a class used to specify the server-side validation of tokens sent by the client, which needs to be configured on the broker side.
The whole related config is like this if you use AAD Client Credential Flow:
Broker config:
listeners=SASL://x.x.x.x:9770
advertised.listeners=SASL://x.x.x.x:9770
listener.security.protocol.map=SASL:SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=OAUTHBEARER
sasl.enabled.mechanisms=OAUTHBEARER
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
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:BrokerObjectId
Client config:
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';