From a98d7156b9c01e9ad7e4c5765b6a4af951d02acf Mon Sep 17 00:00:00 2001 From: Rajini Sivaram Date: Wed, 2 Dec 2020 19:59:29 +0000 Subject: [PATCH] KAFKA-10798; Ensure response is delayed for failed SASL authentication with connection close delay --- .../SaslServerAuthenticator.java | 2 +- .../kafka/common/network/NioEchoServer.java | 2 +- .../SaslAuthenticatorFailureDelayTest.java | 37 ++++++++++++++++++- 3 files changed, 37 insertions(+), 4 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java index 924ae4431d9ad..befbe4ec0a360 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java +++ b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java @@ -470,7 +470,7 @@ private void handleSaslToken(byte[] clientToken) throws IOException { String errorMessage = "Authentication failed during " + reauthInfo.authenticationOrReauthenticationText() + " due to invalid credentials with SASL mechanism " + saslMechanism; - sendKafkaResponse(requestContext, new SaslAuthenticateResponse( + buildResponseOnAuthenticateFailure(requestContext, new SaslAuthenticateResponse( new SaslAuthenticateResponseData() .setErrorCode(Errors.SASL_AUTHENTICATION_FAILED.code()) .setErrorMessage(errorMessage))); diff --git a/clients/src/test/java/org/apache/kafka/common/network/NioEchoServer.java b/clients/src/test/java/org/apache/kafka/common/network/NioEchoServer.java index b66d66baba628..1d6894b4dacd0 100644 --- a/clients/src/test/java/org/apache/kafka/common/network/NioEchoServer.java +++ b/clients/src/test/java/org/apache/kafka/common/network/NioEchoServer.java @@ -91,7 +91,7 @@ public NioEchoServer(ListenerName listenerName, SecurityProtocol securityProtoco public NioEchoServer(ListenerName listenerName, SecurityProtocol securityProtocol, AbstractConfig config, String serverHost, ChannelBuilder channelBuilder, CredentialCache credentialCache, int failedAuthenticationDelayMs, Time time) throws Exception { - this(listenerName, securityProtocol, config, serverHost, channelBuilder, credentialCache, 100, time, + this(listenerName, securityProtocol, config, serverHost, channelBuilder, credentialCache, failedAuthenticationDelayMs, time, new DelegationTokenCache(ScramMechanism.mechanismNames())); } diff --git a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorFailureDelayTest.java b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorFailureDelayTest.java index 599345a110ac3..19003ed56da90 100644 --- a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorFailureDelayTest.java +++ b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorFailureDelayTest.java @@ -45,6 +45,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.List; import java.util.Map; @@ -55,7 +56,7 @@ public class SaslAuthenticatorFailureDelayTest { private static final int BUFFER_SIZE = 4 * 1024; - private final MockTime time = new MockTime(10); + private final MockTime time = new MockTime(1); private NioEchoServer server; private Selector selector; private ChannelBuilder channelBuilder; @@ -118,6 +119,38 @@ public void testInvalidPasswordSaslPlain() throws Exception { server.verifyAuthenticationMetrics(0, 1); } + /** + * Tests that SASL/SCRAM clients with invalid password fail authentication with + * connection close delay if configured. + */ + @Test + public void testInvalidPasswordSaslScram() throws Exception { + String node = "0"; + SecurityProtocol securityProtocol = SecurityProtocol.SASL_SSL; + TestJaasConfig jaasConfig = configureMechanisms("SCRAM-SHA-256", Collections.singletonList("SCRAM-SHA-256")); + jaasConfig.setClientOptions("SCRAM-SHA-256", TestJaasConfig.USERNAME, "invalidpassword"); + + server = createEchoServer(securityProtocol); + createAndCheckClientAuthenticationFailure(securityProtocol, node, "SCRAM-SHA-256", null); + server.verifyAuthenticationMetrics(0, 1); + } + + /** + * Tests that clients with disabled SASL mechanism fail authentication with + * connection close delay if configured. + */ + @Test + public void testDisabledSaslMechanism() throws Exception { + String node = "0"; + SecurityProtocol securityProtocol = SecurityProtocol.SASL_SSL; + TestJaasConfig jaasConfig = configureMechanisms("SCRAM-SHA-256", Collections.singletonList("SCRAM-SHA-256")); + jaasConfig.setClientOptions("PLAIN", TestJaasConfig.USERNAME, "invalidpassword"); + + server = createEchoServer(securityProtocol); + createAndCheckClientAuthenticationFailure(securityProtocol, node, "SCRAM-SHA-256", null); + server.verifyAuthenticationMetrics(0, 1); + } + /** * Tests client connection close before response for authentication failure is sent. */ @@ -215,7 +248,7 @@ private void createAndCheckClientAuthenticationFailure(SecurityProtocol security Exception exception = finalState.exception(); assertTrue("Invalid exception class " + exception.getClass(), exception instanceof SaslAuthenticationException); if (expectedErrorMessage == null) - expectedErrorMessage = "Authentication failed due to invalid credentials with SASL mechanism " + mechanism; + expectedErrorMessage = "Authentication failed during authentication due to invalid credentials with SASL mechanism " + mechanism; assertEquals(expectedErrorMessage, exception.getMessage()); }