Skip to content

[KAFKA-13848] Clients remain connected after SASL re-authentication f… #12179

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
showuon merged 2 commits into from
Jun 10, 2022
Merged

[KAFKA-13848] Clients remain connected after SASL re-authentication f… #12179

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
bfbd750
[KAFKA-13848] Clients remain connected after SASL re-authentication f…
May 19, 2022
c3ef32e
[KAFKA-13848] Clients remain connected after SASL re-authentication f…
Jun 9, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1243,7 +1243,7 @@ project(':clients') {

testImplementation libs.bcpkix
testImplementation libs.junitJupiter
testImplementation libs.mockitoCore
testImplementation libs.mockitoInline

@showuon showuon Jun 2, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could you explain why we need MockitoInline here? Any method only exists in mockitoInline?

@acsaki acsaki Jun 3, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I needed this to mock out ChannelBuilders#createPrincipalBuilder and Sasl#createSaslServer. It's a petty there are these static calls, hard to test them.

@ijuma ijuma Jun 10, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ChannelBuilders is our code, we should avoid mocking static methods.

@showuon showuon Jun 13, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@acsaki , as @ijuma suggested, would you file another small PR to update the tests to avoid mocking the static methods? For the ChannelBuilders#createPrincipalBuilder, it looks like we actually don't need it for these tests, right? (the tests still pass after I removed them)

@stanislavkozlovski stanislavkozlovski Aug 4, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

bump, I second we should aim to remove this


testRuntimeOnly libs.slf4jlog4j
testRuntimeOnly libs.jacksonDatabind
Expand Down
32 changes: 14 additions & 18 deletions ...src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -673,30 +673,26 @@ private long calcCompletionTimesAndReturnSessionLifetimeMs() {
Long credentialExpirationMs = (Long) saslServer
.getNegotiatedProperty(SaslInternalConfigs.CREDENTIAL_LIFETIME_MS_SASL_NEGOTIATED_PROPERTY_KEY);
Long connectionsMaxReauthMs = connectionsMaxReauthMsByMechanism.get(saslMechanism);
if (credentialExpirationMs != null || connectionsMaxReauthMs != null) {
boolean maxReauthSet = connectionsMaxReauthMs != null && connectionsMaxReauthMs > 0;

@showuon showuon Jun 10, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

boolean maxReauthSet = connectionsMaxReauthMs != null && connectionsMaxReauthMs > 0;

Make sense to me.

@acsaki acsaki Jun 10, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thank you! Some tests are failing on my machine also but pass when I run them in the IDE. :( Might be some flakiness but I'm looking into it right now.


if (credentialExpirationMs != null || maxReauthSet) {
if (credentialExpirationMs == null)
retvalSessionLifetimeMs = zeroIfNegative(connectionsMaxReauthMs);
else if (connectionsMaxReauthMs == null)
else if (!maxReauthSet)
retvalSessionLifetimeMs = zeroIfNegative(credentialExpirationMs - authenticationEndMs);
else
retvalSessionLifetimeMs = zeroIfNegative(
Math.min(credentialExpirationMs - authenticationEndMs, connectionsMaxReauthMs));
if (retvalSessionLifetimeMs > 0L)
sessionExpirationTimeNanos = authenticationEndNanos + 1000 * 1000 * retvalSessionLifetimeMs;
retvalSessionLifetimeMs = zeroIfNegative(Math.min(credentialExpirationMs - authenticationEndMs, connectionsMaxReauthMs));

sessionExpirationTimeNanos = authenticationEndNanos + 1000 * 1000 * retvalSessionLifetimeMs;

@tombentley tombentley May 25, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Probably worth a comment explaining that sessionExpirationTimeNanos should always be set to a non-null value.

@showuon showuon May 25, 2022 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

After some investigation, I found the null value is indicating the re-auth is disabled:
https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/network/KafkaChannel.java#L535-L543

So, maybe here, we should make sure we need re-auth (that is, we did have expiration time, if it's 0, which means it's expired, not re-auth disabled). ex:

if (retvalSessionLifetimeMs > 0L || credentialExpirationMs != null || connectionsMaxReauthMs != null)
                    sessionExpirationTimeNanos = authenticationEndNanos + 1000 * 1000 * retvalSessionLifetimeMs;

So, if the retvalSessionLifetimeMs==0, and we don't have credential expiration ms and connection max reauth ms, we should keep it as null
Does that make sense?

@SamBarker SamBarker May 26, 2022

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Adding || credentialExpirationMs != null || connectionsMaxReauthMs != null to the if condition adds no value as we already know one of them has to be true to reach the if guard in general.

I think the correct fix is to delay the initialisation of retvalSessionLifetimeMs so that we can use a >= test in the expiry case and 0 in the non expiry cases while still allowing sessionExpirationTimeNanos being null to signal that re-auth is disabled.

}

if (credentialExpirationMs != null) {
if (sessionExpirationTimeNanos != null)
LOG.debug(
"Authentication complete; session max lifetime from broker config={} ms, credential expiration={} ({} ms); session expiration = {} ({} ms), sending {} ms to client",
connectionsMaxReauthMs, new Date(credentialExpirationMs),
credentialExpirationMs - authenticationEndMs,
new Date(authenticationEndMs + retvalSessionLifetimeMs), retvalSessionLifetimeMs,
retvalSessionLifetimeMs);
else
LOG.debug(
"Authentication complete; session max lifetime from broker config={} ms, credential expiration={} ({} ms); no session expiration, sending 0 ms to client",
connectionsMaxReauthMs, new Date(credentialExpirationMs),
credentialExpirationMs - authenticationEndMs);
LOG.debug(
"Authentication complete; session max lifetime from broker config={} ms, credential expiration={} ({} ms); session expiration = {} ({} ms), sending {} ms to client",
connectionsMaxReauthMs, new Date(credentialExpirationMs),
credentialExpirationMs - authenticationEndMs,
new Date(authenticationEndMs + retvalSessionLifetimeMs), retvalSessionLifetimeMs,
retvalSessionLifetimeMs);
} else {
if (sessionExpirationTimeNanos != null)
LOG.debug(
Expand Down
8 changes: 5 additions & 3 deletions clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -781,14 +781,13 @@ public void testConnectDisconnectDuringInSinglePoll() throws Exception {
when(kafkaChannel.selectionKey()).thenReturn(selectionKey);
when(selectionKey.channel()).thenReturn(SocketChannel.open());
when(selectionKey.readyOps()).thenReturn(SelectionKey.OP_CONNECT);
when(selectionKey.attachment()).thenReturn(kafkaChannel);

selectionKey.attach(kafkaChannel);
Set<SelectionKey> selectionKeys = Utils.mkSet(selectionKey);
selector.pollSelectionKeys(selectionKeys, false, System.nanoTime());

assertFalse(selector.connected().contains(kafkaChannel.id()));
assertTrue(selector.disconnected().containsKey(kafkaChannel.id()));
assertNull(selectionKey.attachment());
Comment on lines +784 to -791

@showuon showuon Jun 2, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why we did this change? Did we change anything affect this test?

@acsaki acsaki Jun 3, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This test failed after my change to use mockitoInline to mock statics. The selectionKey.attach line called the mock's method directly (pointless?) while it looked like the actual intention was selectionKey.attachment() to return the kafkaChannel. That's how it worked actually. Calling the mock method directly in the test and later asserting on attachment being null seemed confusing to me and the assert actually fails too. (with #attachment returning the kafkaChannel). Maybe this was sort of overbearing, is there a simpler way to fix the test?

@showuon showuon Jun 10, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the explanation. I agree with the change. Make sense.


verify(kafkaChannel, atLeastOnce()).ready();
verify(kafkaChannel).disconnect();
Expand Down Expand Up @@ -976,8 +975,11 @@ public void testChannelCloseWhileProcessingReceives() throws Exception {
SelectionKey selectionKey = mock(SelectionKey.class);
when(channel.selectionKey()).thenReturn(selectionKey);
when(selectionKey.isValid()).thenReturn(true);
when(selectionKey.isReadable()).thenReturn(true);
when(selectionKey.readyOps()).thenReturn(SelectionKey.OP_READ);
selectionKey.attach(channel);
when(selectionKey.attachment())
.thenReturn(channel)
.thenReturn(null);
Comment on lines +978 to +982

@showuon showuon Jun 2, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same here, why this change?

@acsaki acsaki Jun 3, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same as above.

selectionKeys.add(selectionKey);

NetworkReceive receive = mock(NetworkReceive.class);
Expand Down
Loading