From 8b3a7200a857a09377078fc2fceadbb9afb21c78 Mon Sep 17 00:00:00 2001 From: Lucas Bradstreet Date: Thu, 20 Feb 2020 11:04:19 -0800 Subject: [PATCH 1/3] SaslClientAuthenticator incorrectly negotiates supported SaslHandshakeRequest versions and uses the max versions supported by the broker whether or not the client supports it. This PR rolls back the SaslHandshake[Request,Response] bump, fixes the version negotiation, and adds a test to prevent anyone from accidentally bumping the version without a workaround (e.g. a new ApiKey). --- .../SaslClientAuthenticator.java | 25 ++++--- .../common/message/SaslHandshakeRequest.json | 6 +- .../common/message/SaslHandshakeResponse.json | 6 +- .../authenticator/SaslAuthenticatorTest.java | 74 ++++++++++++++++++- 4 files changed, 94 insertions(+), 17 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientAuthenticator.java b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientAuthenticator.java index 6784a0149c4c8..1c4de70de091b 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientAuthenticator.java +++ b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientAuthenticator.java @@ -133,6 +133,8 @@ public enum SaslState { private RequestHeader currentRequestHeader; // Version of SaslAuthenticate request/responses private short saslAuthenticateVersion; + // Version of SaslHandshake request/responses + private short saslHandshakeVersion; public SaslClientAuthenticator(Map configs, AuthenticateCallbackHandler callbackHandler, @@ -213,13 +215,13 @@ public void authenticate() throws IOException { if (apiVersionsResponse == null) break; else { - saslAuthenticateVersion(apiVersionsResponse); + saslApiVersions(apiVersionsResponse); reauthInfo.apiVersionsResponseReceivedFromBroker = apiVersionsResponse; setSaslState(SaslState.SEND_HANDSHAKE_REQUEST); // Fall through to send handshake request with the latest supported version } case SEND_HANDSHAKE_REQUEST: - sendHandshakeRequest(reauthInfo.apiVersionsResponseReceivedFromBroker); + sendHandshakeRequest(saslHandshakeVersion); setSaslState(SaslState.RECEIVE_HANDSHAKE_RESPONSE); break; case RECEIVE_HANDSHAKE_RESPONSE: @@ -236,11 +238,11 @@ public void authenticate() throws IOException { setSaslState(SaslState.INTERMEDIATE); break; case REAUTH_PROCESS_ORIG_APIVERSIONS_RESPONSE: - saslAuthenticateVersion(reauthInfo.apiVersionsResponseFromOriginalAuthentication); + saslApiVersions(reauthInfo.apiVersionsResponseFromOriginalAuthentication); setSaslState(SaslState.REAUTH_SEND_HANDSHAKE_REQUEST); // Will set immediately // Fall through to send handshake request with the latest supported version case REAUTH_SEND_HANDSHAKE_REQUEST: - sendHandshakeRequest(reauthInfo.apiVersionsResponseFromOriginalAuthentication); + sendHandshakeRequest(saslHandshakeVersion); setSaslState(SaslState.REAUTH_RECEIVE_HANDSHAKE_OR_OTHER_RESPONSE); break; case REAUTH_RECEIVE_HANDSHAKE_OR_OTHER_RESPONSE: @@ -285,9 +287,8 @@ public void authenticate() throws IOException { } } - private void sendHandshakeRequest(ApiVersionsResponse apiVersionsResponse) throws IOException { - SaslHandshakeRequest handshakeRequest = createSaslHandshakeRequest( - apiVersionsResponse.apiVersion(ApiKeys.SASL_HANDSHAKE.id).maxVersion()); + private void sendHandshakeRequest(short version) throws IOException { + SaslHandshakeRequest handshakeRequest = createSaslHandshakeRequest(version); send(handshakeRequest.toSend(node, nextRequestHeader(ApiKeys.SASL_HANDSHAKE, handshakeRequest.version()))); } @@ -345,11 +346,17 @@ protected SaslHandshakeRequest createSaslHandshakeRequest(short version) { } // Visible to override for testing - protected void saslAuthenticateVersion(ApiVersionsResponse apiVersionsResponse) { + protected void saslApiVersions(ApiVersionsResponse apiVersionsResponse) { ApiVersionsResponseKey authenticateVersion = apiVersionsResponse.apiVersion(ApiKeys.SASL_AUTHENTICATE.id); - if (authenticateVersion != null) + if (authenticateVersion != null) { this.saslAuthenticateVersion = (short) Math.min(authenticateVersion.maxVersion(), ApiKeys.SASL_AUTHENTICATE.latestVersion()); + } + ApiVersionsResponseKey handshakeVersion = apiVersionsResponse.apiVersion(ApiKeys.SASL_HANDSHAKE.id); + if (handshakeVersion != null) { + this.saslHandshakeVersion = (short) Math.min(handshakeVersion.maxVersion(), + ApiKeys.SASL_HANDSHAKE.latestVersion()); + } } private void setSaslState(SaslState saslState) { diff --git a/clients/src/main/resources/common/message/SaslHandshakeRequest.json b/clients/src/main/resources/common/message/SaslHandshakeRequest.json index 6cb9024737e14..bf94c4fdaaad1 100644 --- a/clients/src/main/resources/common/message/SaslHandshakeRequest.json +++ b/clients/src/main/resources/common/message/SaslHandshakeRequest.json @@ -18,9 +18,9 @@ "type": "request", "name": "SaslHandshakeRequest", // Version 1 supports SASL_AUTHENTICATE. - // Version 2 adds flexible version support - "validVersions": "0-2", - "flexibleVersions": "2+", + // NOTE: Version cannot be easily bumped due to incorrect + // client negotiation. See https://issues.apache.org/jira/browse/KAFKA-9577 + "validVersions": "0-1", "fields": [ { "name": "Mechanism", "type": "string", "versions": "0+", "about": "The SASL mechanism chosen by the client." } diff --git a/clients/src/main/resources/common/message/SaslHandshakeResponse.json b/clients/src/main/resources/common/message/SaslHandshakeResponse.json index 0ef871503447d..567aa0fca70ac 100644 --- a/clients/src/main/resources/common/message/SaslHandshakeResponse.json +++ b/clients/src/main/resources/common/message/SaslHandshakeResponse.json @@ -18,9 +18,9 @@ "type": "response", "name": "SaslHandshakeResponse", // Version 1 is the same as version 0. - // Version 2 adds flexible version support - "validVersions": "0-2", - "flexibleVersions": "2+", + // NOTE: Version cannot be easily bumped due to incorrect + // client negotiation. See https://issues.apache.org/jira/browse/KAFKA-9577 + "validVersions": "0-1", "fields": [ { "name": "ErrorCode", "type": "int16", "versions": "0+", "about": "The error code, or 0 if there was no error." }, diff --git a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java index 82da0c32446a8..c7614471428da 100644 --- a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java +++ b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java @@ -677,7 +677,7 @@ public void testUnauthenticatedApiVersionsRequestOverSslHandshakeVersion1() thro * {@link SaslServerAuthenticator} of the server prior to client authentication. */ @Test - public void testApiVersionsRequestWithUnsupportedVersion() throws Exception { + public void testApiVersionsRequestWithServerUnsupportedVersion() throws Exception { short handshakeVersion = ApiKeys.SASL_HANDSHAKE.latestVersion(); SecurityProtocol securityProtocol = SecurityProtocol.SASL_PLAINTEXT; configureMechanisms("PLAIN", Arrays.asList("PLAIN")); @@ -714,6 +714,25 @@ public void testApiVersionsRequestWithUnsupportedVersion() throws Exception { authenticateUsingSaslPlainAndCheckConnection(node, handshakeVersion > 0); } + /** + * Tests correct negotiation of handshake and authenticate api versions by having the server + * return a higher version than supported on the client. + * Note, that due to KAFKA-9577 this will require a workaround to effectively bump + * SASL_HANDSHAKE in the future. + */ + @Test + public void testSaslUnsupportedClientVersions() throws Exception { + configureMechanisms("SCRAM-SHA-512", Arrays.asList("SCRAM-SHA-512")); + + server = startServerApiVersionsUnsupportedByClient(SecurityProtocol.SASL_SSL, "SCRAM-SHA-512"); + updateScramCredentialCache(TestJaasConfig.USERNAME, TestJaasConfig.PASSWORD); + + String node = "0"; + + createClientConnection(SecurityProtocol.SASL_SSL, "SCRAM-SHA-512", node, true); + NetworkTestUtils.checkClientConnection(selector, "0", 100, 10); + } + /** * Tests that invalid ApiVersionRequest is handled by the server correctly and * returns an INVALID_REQUEST error. @@ -748,6 +767,16 @@ public void testInvalidApiVersionsRequest() throws Exception { authenticateUsingSaslPlainAndCheckConnection(node, handshakeVersion > 0); } + + @Test + public void testForBrokenSaslHandshakeVersionBump() { + assertEquals("It is not possible to easily bump SASL_HANDSHAKE schema without " + + "due to improper version negotiation for clients < 2.5. " + + "Please see https://issues.apache.org/jira/browse/KAFKA-9577", + ApiKeys.SASL_HANDSHAKE.latestVersion(), + 1); + } + /** * Tests that valid ApiVersionRequest is handled by the server correctly and * returns an NONE error. @@ -1730,6 +1759,47 @@ private void createClientConnection(SecurityProtocol securityProtocol, String sa createClientConnectionWithoutSaslAuthenticateHeader(securityProtocol, saslMechanism, node); } + private NioEchoServer startServerApiVersionsUnsupportedByClient(final SecurityProtocol securityProtocol, String saslMechanism) throws Exception { + final ListenerName listenerName = ListenerName.forSecurityProtocol(securityProtocol); + final Map configs = Collections.emptyMap(); + final JaasContext jaasContext = JaasContext.loadServerContext(listenerName, saslMechanism, configs); + final Map jaasContexts = Collections.singletonMap(saslMechanism, jaasContext); + + boolean isScram = ScramMechanism.isScram(saslMechanism); + if (isScram) + ScramCredentialUtils.createCache(credentialCache, Arrays.asList(saslMechanism)); + SaslChannelBuilder serverChannelBuilder = new SaslChannelBuilder(Mode.SERVER, jaasContexts, + securityProtocol, listenerName, false, saslMechanism, true, + credentialCache, null, time, new LogContext()) { + + @Override + protected SaslServerAuthenticator buildServerAuthenticator(Map configs, + Map callbackHandlers, + String id, + TransportLayer transportLayer, + Map subjects, + Map connectionsMaxReauthMsByMechanism, + ChannelMetadataRegistry metadataRegistry) { + return new SaslServerAuthenticator(configs, callbackHandlers, id, subjects, null, listenerName, + securityProtocol, transportLayer, connectionsMaxReauthMsByMechanism, metadataRegistry, time) { + + @Override + protected ApiVersionsResponse apiVersionsResponse() { + ApiVersionsResponseKeyCollection versionCollection = new ApiVersionsResponseKeyCollection(2); + versionCollection.add(new ApiVersionsResponseKey().setApiKey(ApiKeys.SASL_HANDSHAKE.id).setMinVersion((short) 0).setMaxVersion((short) 100)); + versionCollection.add(new ApiVersionsResponseKey().setApiKey(ApiKeys.SASL_AUTHENTICATE.id).setMinVersion((short) 0).setMaxVersion((short) 100)); + return new ApiVersionsResponse(new ApiVersionsResponseData().setApiKeys(versionCollection)); + } + }; + } + }; + serverChannelBuilder.configure(saslServerConfigs); + server = new NioEchoServer(listenerName, securityProtocol, new TestSecurityConfig(saslServerConfigs), + "localhost", serverChannelBuilder, credentialCache, time); + server.start(); + return server; + } + private NioEchoServer startServerWithoutSaslAuthenticateHeader(final SecurityProtocol securityProtocol, String saslMechanism) throws Exception { final ListenerName listenerName = ListenerName.forSecurityProtocol(securityProtocol); @@ -1821,7 +1891,7 @@ protected SaslHandshakeRequest createSaslHandshakeRequest(short version) { return buildSaslHandshakeRequest(saslMechanism, (short) 0); } @Override - protected void saslAuthenticateVersion(ApiVersionsResponse apiVersionsResponse) { + protected void saslApiVersions(ApiVersionsResponse apiVersionsResponse) { // Don't set version so that headers are disabled } }; From de7274f5673650db5aba20ce600ebb10e82cb04a Mon Sep 17 00:00:00 2001 From: Lucas Bradstreet Date: Thu, 20 Feb 2020 12:56:30 -0800 Subject: [PATCH 2/3] saslApiVersions -> setSaslAuthenticateAndHandshakeVersions. Improve test failure message. --- .../security/authenticator/SaslClientAuthenticator.java | 6 +++--- .../security/authenticator/SaslAuthenticatorTest.java | 8 ++++---- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientAuthenticator.java b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientAuthenticator.java index 1c4de70de091b..e972da15f32c0 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientAuthenticator.java +++ b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientAuthenticator.java @@ -215,7 +215,7 @@ public void authenticate() throws IOException { if (apiVersionsResponse == null) break; else { - saslApiVersions(apiVersionsResponse); + setSaslAuthenticateAndHandshakeVersions(apiVersionsResponse); reauthInfo.apiVersionsResponseReceivedFromBroker = apiVersionsResponse; setSaslState(SaslState.SEND_HANDSHAKE_REQUEST); // Fall through to send handshake request with the latest supported version @@ -238,7 +238,7 @@ public void authenticate() throws IOException { setSaslState(SaslState.INTERMEDIATE); break; case REAUTH_PROCESS_ORIG_APIVERSIONS_RESPONSE: - saslApiVersions(reauthInfo.apiVersionsResponseFromOriginalAuthentication); + setSaslAuthenticateAndHandshakeVersions(reauthInfo.apiVersionsResponseFromOriginalAuthentication); setSaslState(SaslState.REAUTH_SEND_HANDSHAKE_REQUEST); // Will set immediately // Fall through to send handshake request with the latest supported version case REAUTH_SEND_HANDSHAKE_REQUEST: @@ -346,7 +346,7 @@ protected SaslHandshakeRequest createSaslHandshakeRequest(short version) { } // Visible to override for testing - protected void saslApiVersions(ApiVersionsResponse apiVersionsResponse) { + protected void setSaslAuthenticateAndHandshakeVersions(ApiVersionsResponse apiVersionsResponse) { ApiVersionsResponseKey authenticateVersion = apiVersionsResponse.apiVersion(ApiKeys.SASL_AUTHENTICATE.id); if (authenticateVersion != null) { this.saslAuthenticateVersion = (short) Math.min(authenticateVersion.maxVersion(), diff --git a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java index c7614471428da..6c7d7c83606a8 100644 --- a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java +++ b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java @@ -770,9 +770,9 @@ public void testInvalidApiVersionsRequest() throws Exception { @Test public void testForBrokenSaslHandshakeVersionBump() { - assertEquals("It is not possible to easily bump SASL_HANDSHAKE schema without " - + "due to improper version negotiation for clients < 2.5. " - + "Please see https://issues.apache.org/jira/browse/KAFKA-9577", + assertEquals("It is not possible to easily bump SASL_HANDSHAKE schema" + + " due to improper version negotiation in clients < 2.5." + + " Please see https://issues.apache.org/jira/browse/KAFKA-9577", ApiKeys.SASL_HANDSHAKE.latestVersion(), 1); } @@ -1891,7 +1891,7 @@ protected SaslHandshakeRequest createSaslHandshakeRequest(short version) { return buildSaslHandshakeRequest(saslMechanism, (short) 0); } @Override - protected void saslApiVersions(ApiVersionsResponse apiVersionsResponse) { + protected void setSaslAuthenticateAndHandshakeVersions(ApiVersionsResponse apiVersionsResponse) { // Don't set version so that headers are disabled } }; From fb7ce3787a101b9243988c484c86238a8a61f97d Mon Sep 17 00:00:00 2001 From: Lucas Bradstreet Date: Fri, 21 Feb 2020 13:11:56 -0800 Subject: [PATCH 3/3] Better document bad negotiation --- .../main/resources/common/message/SaslHandshakeRequest.json | 3 ++- .../main/resources/common/message/SaslHandshakeResponse.json | 3 ++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/clients/src/main/resources/common/message/SaslHandshakeRequest.json b/clients/src/main/resources/common/message/SaslHandshakeRequest.json index bf94c4fdaaad1..f384f414c5c04 100644 --- a/clients/src/main/resources/common/message/SaslHandshakeRequest.json +++ b/clients/src/main/resources/common/message/SaslHandshakeRequest.json @@ -19,7 +19,8 @@ "name": "SaslHandshakeRequest", // Version 1 supports SASL_AUTHENTICATE. // NOTE: Version cannot be easily bumped due to incorrect - // client negotiation. See https://issues.apache.org/jira/browse/KAFKA-9577 + // client negotiation for clients <= 2.4. + // See https://issues.apache.org/jira/browse/KAFKA-9577 "validVersions": "0-1", "fields": [ { "name": "Mechanism", "type": "string", "versions": "0+", diff --git a/clients/src/main/resources/common/message/SaslHandshakeResponse.json b/clients/src/main/resources/common/message/SaslHandshakeResponse.json index 567aa0fca70ac..039841f4aff1b 100644 --- a/clients/src/main/resources/common/message/SaslHandshakeResponse.json +++ b/clients/src/main/resources/common/message/SaslHandshakeResponse.json @@ -19,7 +19,8 @@ "name": "SaslHandshakeResponse", // Version 1 is the same as version 0. // NOTE: Version cannot be easily bumped due to incorrect - // client negotiation. See https://issues.apache.org/jira/browse/KAFKA-9577 + // client negotiation for clients <= 2.4. + // See https://issues.apache.org/jira/browse/KAFKA-9577 "validVersions": "0-1", "fields": [ { "name": "ErrorCode", "type": "int16", "versions": "0+",