Kafkajs: Support for OAUTHBEARER SASL mechanism

[uwburn]

I have stumbled upon https://github.com/strimzi/strimzi-kafka-oauth that provides an implementation (both broker and Java client sides) for OAUTHBEARER mechanism.

As it looks like an interesting way to define programmatically authentication and authorization in Kafka, i'm guessing if there's any plan or interest in integrating some kind of support for OAUTHBEARER, so that it can be leveraged to send OAuth tokens for authentication.

PS
I might be willing to help with a PR, even if at the moment i didn't find a matching between the actual implementation of the existing mechanisms and the protocol spec at https://kafka.apache.org/protocol. Did you had to reverse engineer the Java client for it?

[tulios]

The protocol just tells you how to format the bytes; it's sadly hard to understand how everything is supposed to work. This feature is definitely wanted, and I think it is based on the token mechanism. Kafka introduced a change some time ago where the mechanisms can re-auth, so you can refresh the tokens.

I looked into this some time ago, I can try to find the KIP that describes the mechanism and it would be awesome to have someone driving the change.

[uwburn]

It's https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=75968876 the KIP you're referring to?

By looking at the Java library i linked it looked like a token renewal was needed, but looking at the proposal it seems the token is evaluated at connection time, and it stays valid as long as the connection is alive. Eventually, it needs to be refreshed in case of reconnections.

As the KIP notes, it might be a problem in case the token is used also for authorization, but the proposal seems to consider this an edge case and unlikely to be problematic.

I will try expiring the token with the Java impl. and see what happens, just to get the idea.

[tulios]

Yes @uwburn , but I remember another one handling re-auth. Nice, we can try to enrich this issue with information.

[uwburn]

Found it: https://cwiki.apache.org/confluence/display/KAFKA/KIP-368%3A+Allow+SASL+Connections+to+Periodically+Re-Authenticate

I have also found the following, giving some explanations at least about the initial authentication:

• Issue explainin part of the OAUTH SASL payload: https://issues.apache.org/jira/browse/KAFKA-7182
• RFC 5801 - GS header: https://tools.ietf.org/html/rfc5801
• A pull request about OAUTH SASL payload: https://github.com/apache/kafka/pull/5391
• An old version of a class with "easy" impl. of OAUTH SASL payload: https://github.com/rondagostino/kafka/blob/aad4465064afa655563a25cb3e8de1af9fcb5aa0/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/OAuthBearerSaslClient.java
• Actual classes for OAUTH SASL payload:
  https://github.com/apache/kafka/blob/b8651d4e82d45463d2c71798bd5852f8a605b440/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/OAuthBearerSaslClient.java
  https://github.com/apache/kafka/blob/7cef37cf55353f542db3562157547d26c992e782/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/OAuthBearerClientInitialResponse.java

With thoose and by looking at traffic with Wireshark i understood most of the initial auth sequence, i plan to have a prototype for it working by this weekend.

After that i will look into the reauth part.

[klalafaryan]

Hey @uwburn, are you still working on this PR? Do you have some expected delivery date ? Thanks :)

[uwburn]

Hello @klalafaryan, i've subbitted the PR, it's complete, but kafkajs team seems to have an hard time lately in dealing with incoming PRs.

As a workaround i'm using uwburn/kafkajs#oauthbearer-support in my dependencies, i've been using this in a test env for months now and it's working fine.

[Nevon]

I will be able to take a second look at it shortly, so if you go ahead and resolve the conflict I'll just need to get back into the context of the change and then if all's well we can get it merged.

[uwburn]

I've rebased the branch, fixed the quoting style in OAUTHBEARER compose file and moved it to Kafka 2.4. Except for a flaky test that seems unrelated COMPOSE_FILE=docker-compose.2_4_oauthbearer.yml OAUTHBEARER_ENABLED=1 yarn run test is passing.

[Nevon]

Fixed by #680

[j-a-h-i-r]

Hi, @uwburn. Thank you for adding support for OAUTHBEARER in KafkaJS. I am using it to do some testing and it is working fine. One option I'm interested to use is to run the oauthBearerProvider method on/before the token expiration. It seems by default, this method is run every 1 minute. How should I approach if I want to use the expiration time from the token itself?

Thanks for the great library @tulios!

[uwburn]

Hello @j-a-h-i-r, this is more or less the approach i'm using:

const DEFAULT_REAUTH_THRESHOLD = 10000;
const OAUTH_TOKEN_RENEWAL_THRESHOLD = 60000;

function oauthBearerProvider() {
  let tokenPromise;

  async function getToken() {
    // Get token from OAuth endpoint
  }

  async function refreshToken() {
    let response = await getToken();

    const refreshDelay = (response.expires_in * 1000) - DEFAULT_REAUTH_THRESHOLD - OAUTH_TOKEN_RENEWAL_THRESHOLD;
    setTimeout(() => {
      tokenPromise = refreshToken();
    }, refreshDelay);

    return response.access_token;
  }

  tokenPromise = refreshToken();

  return async function() {
    return {
      value: await tokenPromise
    };
  };
}

new Kafka({
  ...,
  sasl: {
    mechanism: "OAUTHBEARER",
    oauthBearerProvider()
  }
})

So when setting up the Kafka.js client the function gets the token from the endpoint and caches the token promise. Then it schedules the refresh of the token taking care of the expires_in parameter and applying an adequate offset.

Hope it helps.

[j-a-h-i-r]

That's a really elegant solution. Thank you @uwburn for sharing it. Much appreciated!