KAFKA-3652: Return error response for unsupported version of ApiVersionsRequest #1310
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -51,13 +51,15 @@ | |
| import org.apache.kafka.common.errors.AuthenticationException; | ||
| import org.apache.kafka.common.errors.IllegalSaslStateException; | ||
| import org.apache.kafka.common.errors.UnsupportedSaslMechanismException; | ||
| import org.apache.kafka.common.errors.UnsupportedVersionException; | ||
| import org.apache.kafka.common.network.Authenticator; | ||
| import org.apache.kafka.common.network.Mode; | ||
| import org.apache.kafka.common.network.NetworkSend; | ||
| import org.apache.kafka.common.network.NetworkReceive; | ||
| import org.apache.kafka.common.network.TransportLayer; | ||
| import org.apache.kafka.common.protocol.ApiKeys; | ||
| import org.apache.kafka.common.protocol.Errors; | ||
| import org.apache.kafka.common.protocol.Protocol; | ||
| import org.apache.kafka.common.protocol.types.SchemaException; | ||
| import org.apache.kafka.common.requests.AbstractRequest; | ||
| import org.apache.kafka.common.requests.AbstractRequestResponse; | ||
|
|
@@ -75,7 +77,7 @@ public class SaslServerAuthenticator implements Authenticator { | |
| private static final Logger LOG = LoggerFactory.getLogger(SaslServerAuthenticator.class); | ||
|
|
||
| public enum SaslState { | ||
| GSSAPI_OR_HANDSHAKE_REQUEST, HANDSHAKE_REQUEST, AUTHENTICATE, COMPLETE, FAILED | ||
| } | ||
|
|
||
| private final String node; | ||
|
|
@@ -85,7 +87,7 @@ public enum SaslState { | |
| private final String host; | ||
|
|
||
| // Current SASL state | ||
| private SaslState saslState = SaslState.GSSAPI_OR_HANDSHAKE_REQUEST; | ||
| // Next SASL state to be set when outgoing writes associated with the current SASL state complete | ||
| private SaslState pendingSaslState = null; | ||
| private SaslServer saslServer; | ||
|
|
@@ -215,6 +217,9 @@ public void authenticate() throws IOException { | |
| try { | ||
| switch (saslState) { | ||
| case HANDSHAKE_REQUEST: | ||
| handleKafkaRequest(clientToken); | ||
| break; | ||
| case GSSAPI_OR_HANDSHAKE_REQUEST: | ||
|
Contributor
There was a problem hiding this comment. Could we move this case to the first one after switch since it's the initial state?
Member
There was a problem hiding this comment. @junrao We are relying on fall-through for the second case, so it's not straightforward to make that change. |
||
| if (handleKafkaRequest(clientToken)) | ||
| break; | ||
| // For default GSSAPI, fall through to authenticate using the client token as the first GSSAPI packet. | ||
|
|
@@ -288,39 +293,53 @@ private boolean handleKafkaRequest(byte[] requestBytes) throws IOException, Auth | |
| try { | ||
| ByteBuffer requestBuffer = ByteBuffer.wrap(requestBytes); | ||
| RequestHeader requestHeader = RequestHeader.parse(requestBuffer); | ||
| ApiKeys apiKey = ApiKeys.forId(requestHeader.apiKey()); | ||
| // A valid Kafka request header was received. SASL authentication tokens are now expected only | ||
| // following a SaslHandshakeRequest since this is not a GSSAPI client token from a Kafka 0.9.0.x client. | ||
| setSaslState(SaslState.HANDSHAKE_REQUEST); | ||
| isKafkaRequest = true; | ||
|
|
||
| if (!Protocol.apiVersionSupported(requestHeader.apiKey(), requestHeader.apiVersion())) { | ||
|
Contributor
There was a problem hiding this comment. requestHeader.apiKey() can just be apiKey.
Member
There was a problem hiding this comment.
|
||
| if (apiKey == ApiKeys.API_VERSIONS) | ||
| sendKafkaResponse(requestHeader, ApiVersionsResponse.fromError(Errors.UNSUPPORTED_VERSION)); | ||
| else | ||
| throw new UnsupportedVersionException("Version " + requestHeader.apiVersion() + " is not supported for apiKey " + apiKey); | ||
| } else { | ||
| AbstractRequest request = AbstractRequest.getRequest(requestHeader.apiKey(), requestHeader.apiVersion(), requestBuffer); | ||
|
|
||
| LOG.debug("Handle Kafka request {}", apiKey); | ||
| switch (apiKey) { | ||
| case API_VERSIONS: | ||
| handleApiVersionsRequest(requestHeader, (ApiVersionsRequest) request); | ||
| break; | ||
| case SASL_HANDSHAKE: | ||
| clientMechanism = handleHandshakeRequest(requestHeader, (SaslHandshakeRequest) request); | ||
| break; | ||
| default: | ||
| throw new IllegalSaslStateException("Unexpected Kafka request of type " + apiKey + " during SASL handshake."); | ||
| } | ||
| } | ||
| } catch (SchemaException | IllegalArgumentException e) { | ||
| if (saslState == SaslState.GSSAPI_OR_HANDSHAKE_REQUEST) { | ||
| // SchemaException is thrown if the request is not in Kafka format. IIlegalArgumentException is thrown | ||
| // if the API key is invalid. For compatibility with 0.9.0.x where the first packet is a GSSAPI token | ||
| // starting with 0x60, revert to GSSAPI for both these exceptions. | ||
| if (LOG.isDebugEnabled()) { | ||
| StringBuilder tokenBuilder = new StringBuilder(); | ||
| for (byte b : requestBytes) { | ||
| tokenBuilder.append(String.format("%02x", b)); | ||
| if (tokenBuilder.length() >= 20) | ||
| break; | ||
| } | ||
| LOG.debug("Received client packet of length {} starting with bytes 0x{}, process as GSSAPI packet", requestBytes.length, tokenBuilder); | ||
| } | ||
| if (enabledMechanisms.contains(SaslConfigs.GSSAPI_MECHANISM)) { | ||
| LOG.debug("First client packet is not a SASL mechanism request, using default mechanism GSSAPI"); | ||
| clientMechanism = SaslConfigs.GSSAPI_MECHANISM; | ||
| } else | ||
| throw new UnsupportedSaslMechanismException("Exception handling first SASL packet from client, GSSAPI is not supported by server", e); | ||
| } else | ||
| throw e; | ||
| } | ||
| if (clientMechanism != null) { | ||
| createSaslServer(clientMechanism); | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -28,8 +28,8 @@ import kafka.metrics.KafkaMetricsGroup | |
| import kafka.utils.{Logging, SystemTime} | ||
| import org.apache.kafka.common.TopicPartition | ||
| import org.apache.kafka.common.network.Send | ||
| import org.apache.kafka.common.protocol.{ApiKeys, SecurityProtocol, Protocol} | ||
| import org.apache.kafka.common.requests.{RequestSend, ProduceRequest, AbstractRequest, RequestHeader, ApiVersionsRequest} | ||
| import org.apache.kafka.common.security.auth.KafkaPrincipal | ||
| import org.apache.log4j.Logger | ||
|
|
||
|
|
@@ -84,8 +84,13 @@ object RequestChannel extends Logging { | |
| null | ||
| val body: AbstractRequest = | ||
| if (requestObj == null) | ||
| try { | ||
| // For unsupported version of ApiVersionsRequest, create a dummy request to enable an error response to be returned later | ||
| if (header.apiKey == ApiKeys.API_VERSIONS.id && !Protocol.apiVersionSupported(header.apiKey, header.apiVersion)) | ||
| new ApiVersionsRequest | ||
|
Member
There was a problem hiding this comment. It may be better to just respond with the error from here instead of creating this dummy request. But let's see what @junrao says.
Contributor
There was a problem hiding this comment. Yes, the logic here is a bit awkward, but probably the simplest. Sending a response directly from SocketServer seems to require more work.
Contributor
Author
|
||
| else | ||
| AbstractRequest.getRequest(header.apiKey, header.apiVersion, buffer) | ||
| } catch { | ||
| case ex: Throwable => | ||
| throw new InvalidRequestException(s"Error getting request for apiKey: ${header.apiKey} and apiVersion: ${header.apiVersion}", ex) | ||
| } | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Hmm, I am not quite sure what the new state GSSAPI_OR_HANDSHAKE_REQUEST is for. It's making the same call as the HANDSHAKE_REQUEST and there is no code to change saslState to GSSAPI_OR_HANDSHAKE_REQUEST.
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@junrao Oops, sorry, missed out the change of initiate state. In HANDSHAKE_REQUEST state, any invalid request is treated as a GSSAPI token. This made sense when only a SaslHandshakeRequest was expected prior to SASL tokens. Since SaslHandshakeRequest can now be preceded by any number of ApiVersionsRequests, I think it makes sense to revert to 0.9.0.x GSSAPI authentication only for the first token. GSSAPI_OR_HANDSHAKE_REQUEST is the initiate state which checks that. Have updated the PR.