authenticate on every request to broker when needed

[flucivja]

Fixes #1240

[Nevon]

Is this issue really limited to heartbeats? From reading the issue description, it sounds like the problem should occur on any request if the broker needs to reauthenticate.

If that's the case, then we could do the same thing in sendRequest, except when the request is apiVersions, since that happens before authentication.

[flucivja]

Is this issue really limited to heartbeats? From reading the issue description, it sounds like the problem should occur on any request if the broker needs to reauthenticate.

If that's the case, then we could do the same thing in sendRequest, except when the request is apiVersions, since that happens before authentication.

yes, it can happend at any point. I will add it into [sendRequest].
btw. how to check if request is not apiVersions? By simply comparing apiName with string? Or better to get version protocol by requests.ApiVersions.protocol and compare names?

[Nevon]

btw. how to check if request is not apiVersions? By simply comparing apiName with string? Or better to get version protocol by requests.ApiVersions.protocol and compare names?

Not sure it matters much, but you could grab the api key of ApiVersions from src/protocol/requests/apiKeys.js and compare it with request.apiKey. The apiName shouldn't change, but apiKey definitely won't.

[flucivja]

@Nevon everything should be ok now. Can you please check? :) Thanks.

[Nevon]

I've been trying to figure out how to test this, and I wrote a test that looks something like this:

test('Reauthenticates if needed before making a request', async () => {
  const saslEntry = saslEntries.pop()
  const connection = createConnection(saslEntry.opts())
  broker = new Broker({
    connection,
    logger: newLogger(),
  })
  await broker.connect()
  const spy = jest.spyOn(connection, 'send')

  broker.authenticatedAt = null
  await broker.listGroups()

  // Assert that sasl authenticate requests happen
  console.log(spy.mock.calls)
})

But I ran into the problem that the server rejects the second authentication request with an ILLEGAL_SASL_STATE error. This got me thinking and reading the KIP about reauthentication.

In the current implementation, if you have multiple requests going on when you realize that you have to reauthenticate, each one is going to try to reauthenticate, which is gonna cause a problem. This is relatively easily solved by having the isAuthenticated method wait for the authLock to be released, so that you wait to check if you need to authenticate while there's an authentication going on.

async isAuthenticated() {
  try {
    await this.authLock.acquire()
    return this.authenticatedAt != null && !this[PRIVATE.SHOULD_REAUTHENTICATE]()
  } finally {
    await this.authLock.release()
  }
}

Unfortunately it makes the method async, which ripples through and affects a lot of things. A pain in the ass to fix, but doable.

But it doesn't explain why I fail to authenticate. In my test I'm not making concurrent requests that trigger multiple authentications. I'm just authenticating once, then making it appear that I need to reauthenticate, and then I trigger a request that will cause me to reauthenticate.

[flucivja]

What do you think about putting isAuthenticated check into [PRIVATE.AUTHENTICATE]() after auth lock? And if it will be authenticated, then it will skip auth and release lock.

[t-d-d]

You might want to look at the sharedPromise in utils. Heartbeat() is wrapped with this so that when multiple are issues concurrently, only 1 actual request is sent.

…

On Tue, 30 Nov 2021, 17:21 Tommy Brunn, ***@***.***> wrote: I've been trying to figure out how to test this, and I wrote a test that looks something like this: test('Reauthenticates if needed before making a request', async () => { const saslEntry = saslEntries.pop() const connection = createConnection(saslEntry.opts()) broker = new Broker({ connection, logger: newLogger(), }) await broker.connect() const spy = jest.spyOn(connection, 'send').mockImplementation() broker.authenticatedAt = null await broker.listGroups() // Assert that sasl authenticate requests happen console.log(spy.mock.calls)}) But I ran into the problem that the server rejects the second authentication request with an ILLEGAL_SASL_STATE error. This got me thinking and reading the KIP about reauthentication <https://cwiki.apache.org/confluence/display/KAFKA/KIP-368%3A+Allow+SASL+Connections+to+Periodically+Re-Authenticate> . In the current implementation, if you have multiple requests going on when you realize that you have to reauthenticate, each one is going to try to reauthenticate, which is gonna cause a problem. This is relatively easily solved by having the isAuthenticated method wait for the authLock to be released, so that you wait to check if you need to authenticate while there's an authentication going on. Unfortunately it makes the method async, which ripples through and affects a lot of things. A pain in the ass to fix, but doable. — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub <#1241 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/ABDLW5TG5ZGLFA6B4MGVTFTUOSJQLANCNFSM5I6ZIHQQ> .

[flucivja]

sharedPromise looks handy for such case. But I am thinking if this is really an issue. I tried that test without my changes and it is failing too. @Nevon is that test passing for you without my changes (without check for re-authentication)?

[Nevon]

@Nevon is that test passing for you without my changes (without check for re-authentication)?

No, but I think it's an issue with the test. Maybe you can't authenticate again right after authenticating with sasl/scram like I'm doing. I'm guessing if you swap it out for oauthbearer with a short reauthentication timeout maybe it'll work. I just didn't have time to test it out.

But I am thinking if this is really an issue.

I would be very surprised if it isn't. Tossing broker.listGroups or something into a Promise.all should trigger it immediately.

[flucivja]

Ok then, are you ok to handle it with sharedPromise? It will look like this in the broker constructor:

this[PRIVATE.AUTHENTICATE] = sharedPromiseTo(async () => {
  ... authentication logic
})

[Nevon]

Sounds like a good solution! Let's give it a try. I would also really like to test these scenarios. We do have a test suite that uses oauthbearer, which should certainly allow for reauthentication. See the test:group:oauthbearer yarn script, which runs a subset of test files. There's a function called describeIfOauthbearerEnabled in testHelpers that will run tests nested in it only if oauthbearer is enabled.

[flucivja]

Hi @Nevon, I implemented tests and sorry it took longer time. I had issues with kafka SASL OAUTHBEARER authentication. Within test, the kafka returned me all the time that provided token created in test utils is invalid and I didn't know why. After debugging and looking to how other kafka clients are implemented, I found out that expiration time was missing in the token.

Btw. you were right with multiple parallel requests causing error: when I removed fix with sharedPromise my tests which runs multiple requests in parallel requiring authentication started failing with invalid SASL state.

Can you please check PR again when you will have time? If all will be good, can you merge it and release?

Thank you very much.

[flucivja]

@Nevon just a gentle reminder :)

[Nevon]

Haven't forgotten about you. I left the notification sitting on my phone so it wouldn't get lost. 😅 The change looks good overall - I'd just like to get your thoughts on whether we can get rid of those sleeps in the test (or at least reduce them).

[Nevon]

We don't need this lock anymore now right? If there's more callers, the first one will create a promise and start the authentication, the second one will just receive the promise from the first one so it'll just await the promise settlement from the first one. Since the promise creation itself is synchronous there shouldn't be any race condition there, I think.

[flucivja]

yes you are right. I will remove auth lock.

[flucivja]

@Nevon done, now my test is faster by 30 seconds (14s vs 44s).

[Nevon]

Thanks for your patience and for taking the time to contribute!

[j-a-h-i-r]

Sorry for reviving a merged thread but I'm curious about one thing @Nevon. Does this fix #1039? As far as I can remember, that issue occurred due to re-authentication. I saw that you mentioned testing SASL OauthBearer flows. So I was wondering if the changes potentially fix that issue too.