Validate build hash in handshake

server/src/main/java/org/elasticsearch/transport/TransportService.java

@@ -22,6 +22,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
+import org.elasticsearch.Build;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionListenerResponseHandler;
@@ -37,6 +38,7 @@
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.lease.Releasable;
+import org.elasticsearch.common.logging.DeprecationLogger;
 import org.elasticsearch.common.logging.Loggers;
 import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.common.settings.ClusterSettings;
@@ -73,10 +75,14 @@
 import java.util.function.Predicate;
 import java.util.function.Supplier;
 
-public class TransportService extends AbstractLifecycleComponent implements ReportingService<TransportInfo>, TransportMessageListener,
-    TransportConnectionListener {
+public class TransportService extends AbstractLifecycleComponent
+        implements ReportingService<TransportInfo>, TransportMessageListener, TransportConnectionListener {
+
     private static final Logger logger = LogManager.getLogger(TransportService.class);
 
+    private static final String PERMIT_HANDSHAKES_FROM_INCOMPATIBLE_BUILDS_KEY = "es.unsafely_permit_handshake_from_incompatible_builds";
+    private static final boolean PERMIT_HANDSHAKES_FROM_INCOMPATIBLE_BUILDS = getPermitHandshakesFromIncompatibleBuilds();
+
     public static final String DIRECT_RESPONSE_PROFILE = ".direct";
     public static final String HANDSHAKE_ACTION_NAME = "internal:transport/handshake";
 
@@ -115,6 +121,7 @@ protected boolean removeEldestEntry(Map.Entry eldest) {
     private final RemoteClusterService remoteClusterService;
 
     private final boolean validateConnections;
+    private final boolean requireCompatibleBuild;
 
     /** if set will call requests sent to this id to shortcut and executed locally */
     volatile DiscoveryNode localNode = null;
@@ -160,9 +167,15 @@ public TransportService(Settings settings, Transport transport, ThreadPool threa
     public TransportService(Settings settings, Transport transport, ThreadPool threadPool, TransportInterceptor transportInterceptor,
                             Function<BoundTransportAddress, DiscoveryNode> localNodeFactory, @Nullable ClusterSettings clusterSettings,
                             Set<String> taskHeaders, ConnectionManager connectionManager) {
+
+        final boolean isTransportClient = TransportClient.CLIENT_TYPE.equals(settings.get(Client.CLIENT_TYPE_SETTING_S.getKey()));
+
+        // If we are a transport client then we skip the check that the remote node has a compatible build hash
+        this.requireCompatibleBuild = isTransportClient == false;
+
         // The only time we do not want to validate node connections is when this is a transport client using the simple node sampler
-        this.validateConnections = TransportClient.CLIENT_TYPE.equals(settings.get(Client.CLIENT_TYPE_SETTING_S.getKey())) == false ||
-            TransportClient.CLIENT_TRANSPORT_SNIFF.get(settings);
+        this.validateConnections = isTransportClient == false || TransportClient.CLIENT_TRANSPORT_SNIFF.get(settings);
+
         this.transport = transport;
         transport.setSlowLogThreshold(TransportSettings.SLOW_OPERATION_THRESHOLD_SETTING.get(settings));
         this.threadPool = threadPool;
@@ -192,7 +205,14 @@ public TransportService(Settings settings, Transport transport, ThreadPool threa
             false, false,
             HandshakeRequest::new,
             (request, channel, task) -> channel.sendResponse(
-                new HandshakeResponse(localNode, clusterName, localNode.getVersion())));
+                new HandshakeResponse(localNode.getVersion(), Build.CURRENT.hash(), localNode, clusterName)));
+
+        if (PERMIT_HANDSHAKES_FROM_INCOMPATIBLE_BUILDS) {
+            logger.warn("transport handshakes from incompatible builds are unsafely permitted on this node; remove system property [" +
+                    PERMIT_HANDSHAKES_FROM_INCOMPATIBLE_BUILDS_KEY + "] to resolve this warning");
+            DeprecationLogger.getLogger(TransportService.class).deprecate("permit_handshake_from_incompatible_builds",
+                "system property [" + PERMIT_HANDSHAKES_FROM_INCOMPATIBLE_BUILDS_KEY + "] is deprecated and should be removed");
+        }
     }
 
     public RemoteClusterService getRemoteClusterService() {
@@ -482,7 +502,7 @@ public void onFailure(Exception e) {
                         listener.onFailure(e);
                     }
                 }
-                , HandshakeResponse::new, ThreadPool.Names.GENERIC
+                , in -> new HandshakeResponse(in, requireCompatibleBuild), ThreadPool.Names.GENERIC
             ));
     }
 
@@ -504,28 +524,89 @@ private HandshakeRequest() {
     }
 
     public static class HandshakeResponse extends TransportResponse {
+
+        private static final Version BUILD_HASH_HANDSHAKE_VERSION = Version.V_7_11_0;
+
+        private final Version version;
+
+        @Nullable // if version < BUILD_HASH_HANDSHAKE_VERSION
+        private final String buildHash;
+
         private final DiscoveryNode discoveryNode;
+
         private final ClusterName clusterName;
-        private final Version version;
 
-        public HandshakeResponse(DiscoveryNode discoveryNode, ClusterName clusterName, Version version) {
-            this.discoveryNode = discoveryNode;
-            this.version = version;
-            this.clusterName = clusterName;
+        public HandshakeResponse(Version version, String buildHash, DiscoveryNode discoveryNode, ClusterName clusterName) {
+            this.buildHash = Objects.requireNonNull(buildHash);
+            this.discoveryNode = Objects.requireNonNull(discoveryNode);
+            this.version = Objects.requireNonNull(version);
+            this.clusterName = Objects.requireNonNull(clusterName);
         }
 
-        public HandshakeResponse(StreamInput in) throws IOException {
+        public HandshakeResponse(StreamInput in, boolean requireCompatibleBuild) throws IOException {
             super(in);
-            discoveryNode = in.readOptionalWriteable(DiscoveryNode::new);
-            clusterName = new ClusterName(in);
-            version = Version.readVersion(in);
+            if (in.getVersion().onOrAfter(BUILD_HASH_HANDSHAKE_VERSION)) {
+                // the first two fields need only VInts and raw (ASCII) characters, so we cross our fingers and hope that they appear
+                // on the wire as we expect them to even if this turns out to be an incompatible build
+                version = Version.readVersion(in);
+                buildHash = in.readString();
+
+                try {
+                    // If the remote node is incompatible then make an effort to identify it anyway, so we can mention it in the exception
+                    // message, but recognise that this may fail
+                    discoveryNode = new DiscoveryNode(in);
+                } catch (Exception e) {
+                    if (isIncompatibleBuild(version, buildHash, requireCompatibleBuild)) {
+                        throw new IllegalArgumentException("unidentifiable remote node is build [" + buildHash +
+                                "] of version [" + version + "] but this node is build [" + Build.CURRENT.hash() +
+                                "] of version [" + Version.CURRENT + "] which has an incompatible wire format", e);
+                    } else {
+                        throw e;
+                    }
+                }
+
+                if (isIncompatibleBuild(version, buildHash, requireCompatibleBuild)) {
+                    if (PERMIT_HANDSHAKES_FROM_INCOMPATIBLE_BUILDS) {
+                        logger.warn("remote node [{}] is build [{}] of version [{}] but this node is build [{}] of version [{}] " +
+                                        "which may not be compatible; remove system property [{}] to resolve this warning",
+                                discoveryNode, buildHash, version, Build.CURRENT.hash(), Version.CURRENT,
+                                PERMIT_HANDSHAKES_FROM_INCOMPATIBLE_BUILDS_KEY);
+                    } else {
+                        throw new IllegalArgumentException("remote node [" + discoveryNode + "] is build [" + buildHash +
+                                "] of version [" + version + "] but this node is build [" + Build.CURRENT.hash() +
+                                "] of version [" + Version.CURRENT + "] which has an incompatible wire format");
+                    }
+                }
+
+                clusterName = new ClusterName(in);
+            } else {
+                discoveryNode = in.readOptionalWriteable(DiscoveryNode::new);
+                clusterName = new ClusterName(in);
+                version = Version.readVersion(in);
+                buildHash = null;
+            }
         }
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
-            out.writeOptionalWriteable(discoveryNode);
-            clusterName.writeTo(out);
-            Version.writeVersion(version, out);
+            if (out.getVersion().onOrAfter(BUILD_HASH_HANDSHAKE_VERSION)) {
+                Version.writeVersion(version, out);
+                out.writeString(buildHash);
+                discoveryNode.writeTo(out);
+                clusterName.writeTo(out);
+            } else {
+                out.writeOptionalWriteable(discoveryNode);
+                clusterName.writeTo(out);
+                Version.writeVersion(version, out);
+            }
+        }
+
+        public Version getVersion() {
+            return version;
+        }
+
+        public String getBuildHash() {
+            return buildHash;
         }
 
         public DiscoveryNode getDiscoveryNode() {
@@ -535,6 +616,10 @@ public DiscoveryNode getDiscoveryNode() {
         public ClusterName getClusterName() {
             return clusterName;
         }
+
+        private static boolean isIncompatibleBuild(Version version, String buildHash, boolean requireCompatibleBuild) {
+            return requireCompatibleBuild && version == Version.CURRENT && Build.CURRENT.hash().equals(buildHash) == false;
+        }
     }
 
     public void disconnectFromNode(DiscoveryNode node) {
@@ -1354,4 +1439,17 @@ public void onResponseReceived(long requestId, Transport.ResponseContext holder)
             }
         }
     }
+
+    private static boolean getPermitHandshakesFromIncompatibleBuilds() {
+        final String value = System.getProperty(PERMIT_HANDSHAKES_FROM_INCOMPATIBLE_BUILDS_KEY);
+        if (value == null) {
+            return false;
+        }
+        if (Boolean.parseBoolean(value)) {
+            return true;
+        }
+        throw new IllegalArgumentException("invalid value [" + value + "] for system property ["
+                + PERMIT_HANDSHAKES_FROM_INCOMPATIBLE_BUILDS_KEY + "]");
+    }
+
 }

server/src/test/java/org/elasticsearch/client/transport/FailAndRetryMockTransport.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.client.transport;
 
+import org.elasticsearch.Build;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.admin.cluster.node.liveness.LivenessResponse;
@@ -102,7 +103,11 @@ public void sendRequest(long requestId, String action, TransportRequest request,
                     } else if (TransportService.HANDSHAKE_ACTION_NAME.equals(action)) {
                         TransportResponseHandler transportResponseHandler = responseHandlers.onResponseReceived(requestId, listener);
                         Version version = node.getVersion();
-                        transportResponseHandler.handleResponse(new TransportService.HandshakeResponse(node, clusterName, version));
+                        transportResponseHandler.handleResponse(new TransportService.HandshakeResponse(
+                                version,
+                                Build.CURRENT.hash(),
+                                node,
+                                clusterName));
 
                     } else {
                         throw new UnsupportedOperationException("Mock transport does not understand action " + action);

server/src/test/java/org/elasticsearch/client/transport/TransportClientHeadersTests.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.client.transport;
 
+import org.elasticsearch.Build;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.admin.cluster.node.liveness.LivenessResponse;
@@ -180,7 +181,11 @@ public <T extends TransportResponse> void sendRequest(Transport.Connection conne
                         clusterStateLatch.countDown();
                     } else if (TransportService.HANDSHAKE_ACTION_NAME .equals(action)) {
                         ((TransportResponseHandler<TransportService.HandshakeResponse>) handler).handleResponse(
-                            new TransportService.HandshakeResponse(connection.getNode(), clusterName, connection.getNode().getVersion()));
+                                new TransportService.HandshakeResponse(
+                                        connection.getNode().getVersion(),
+                                        Build.CURRENT.hash(),
+                                        connection.getNode(),
+                                        clusterName));
                     } else {
                         handler.handleException(new TransportException("", new InternalException(action)));
                     }

server/src/test/java/org/elasticsearch/cluster/NodeConnectionsServiceTests.java

@@ -21,6 +21,7 @@
 
 import org.apache.logging.log4j.Level;
 import org.apache.logging.log4j.LogManager;
+import org.elasticsearch.Build;
 import org.elasticsearch.ElasticsearchTimeoutException;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
@@ -479,7 +480,7 @@ private TestTransportService(Transport transport, ThreadPool threadPool) {
         @Override
         public void handshake(Transport.Connection connection, TimeValue timeout, Predicate<ClusterName> clusterNamePredicate,
                               ActionListener<HandshakeResponse> listener) {
-            listener.onResponse(new HandshakeResponse(connection.getNode(), new ClusterName(""), Version.CURRENT));
+            listener.onResponse(new HandshakeResponse(Version.CURRENT, Build.CURRENT.hash(), connection.getNode(), new ClusterName("")));
         }
 
         @Override

server/src/test/java/org/elasticsearch/cluster/coordination/FollowersCheckerTests.java

@@ -18,6 +18,7 @@
  */
 package org.elasticsearch.cluster.coordination;
 
+import org.elasticsearch.Build;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.Version;
 import org.elasticsearch.cluster.ClusterName;
@@ -226,7 +227,11 @@ public void testFailsNodeThatDisconnects() {
             protected void onSendRequest(long requestId, String action, TransportRequest request, DiscoveryNode node) {
                 assertFalse(node.equals(localNode));
                 if (action.equals(HANDSHAKE_ACTION_NAME)) {
-                    handleResponse(requestId, new TransportService.HandshakeResponse(node, ClusterName.DEFAULT, Version.CURRENT));
+                    handleResponse(requestId, new TransportService.HandshakeResponse(
+                            Version.CURRENT,
+                            Build.CURRENT.hash(),
+                            node,
+                            ClusterName.DEFAULT));
                     return;
                 }
                 deterministicTaskQueue.scheduleNow(new Runnable() {

server/src/test/java/org/elasticsearch/cluster/coordination/LeaderCheckerTests.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.cluster.coordination;
 
+import org.elasticsearch.Build;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.Version;
 import org.elasticsearch.cluster.ClusterName;
@@ -221,7 +222,11 @@ public void testFollowerFailsImmediatelyOnDisconnection() {
             @Override
             protected void onSendRequest(long requestId, String action, TransportRequest request, DiscoveryNode node) {
                 if (action.equals(HANDSHAKE_ACTION_NAME)) {
-                    handleResponse(requestId, new TransportService.HandshakeResponse(node, ClusterName.DEFAULT, Version.CURRENT));
+                    handleResponse(requestId, new TransportService.HandshakeResponse(
+                            Version.CURRENT,
+                            Build.CURRENT.hash(),
+                            node,
+                            ClusterName.DEFAULT));
                     return;
                 }
                 assertThat(action, equalTo(LEADER_CHECK_ACTION_NAME));
@@ -330,7 +335,11 @@ public void testFollowerFailsImmediatelyOnHealthCheckFailure() {
             @Override
             protected void onSendRequest(long requestId, String action, TransportRequest request, DiscoveryNode node) {
                 if (action.equals(HANDSHAKE_ACTION_NAME)) {
-                    handleResponse(requestId, new TransportService.HandshakeResponse(node, ClusterName.DEFAULT, Version.CURRENT));
+                    handleResponse(requestId, new TransportService.HandshakeResponse(
+                            Version.CURRENT,
+                            Build.CURRENT.hash(),
+                            node,
+                            ClusterName.DEFAULT));
                     return;
                 }
                 assertThat(action, equalTo(LEADER_CHECK_ACTION_NAME));

server/src/test/java/org/elasticsearch/cluster/coordination/NodeJoinTests.java

@@ -19,6 +19,7 @@
 package org.elasticsearch.cluster.coordination;
 
 import org.apache.logging.log4j.message.ParameterizedMessage;
+import org.elasticsearch.Build;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.cluster.ClusterName;
@@ -162,8 +163,12 @@ private void setupMasterServiceAndCoordinator(long term, ClusterState initialSta
             @Override
             protected void onSendRequest(long requestId, String action, TransportRequest request, DiscoveryNode destination) {
                 if (action.equals(HANDSHAKE_ACTION_NAME)) {
-                    handleResponse(requestId, new TransportService.HandshakeResponse(destination, initialState.getClusterName(),
-                        destination.getVersion()));
+                    handleResponse(requestId, new TransportService.HandshakeResponse(
+                            destination.getVersion(),
+                            Build.CURRENT.hash(),
+                            destination,
+                            initialState.getClusterName()
+                    ));
                 } else if (action.equals(JoinHelper.VALIDATE_JOIN_ACTION_NAME)) {
                     handleResponse(requestId, new TransportResponse.Empty());
                 } else {

server/src/test/java/org/elasticsearch/discovery/HandshakingTransportAddressConnectorTests.java

@@ -23,6 +23,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.lucene.util.SetOnce;
+import org.elasticsearch.Build;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
@@ -98,7 +99,11 @@ protected void onSendRequest(long requestId, String action, TransportRequest req
                     if (fullConnectionFailure != null && node.getAddress().equals(remoteNode.getAddress())) {
                         handleError(requestId, fullConnectionFailure);
                     } else {
-                        handleResponse(requestId, new HandshakeResponse(remoteNode, new ClusterName(remoteClusterName), Version.CURRENT));
+                        handleResponse(requestId, new HandshakeResponse(
+                                Version.CURRENT,
+                                Build.CURRENT.hash(),
+                                remoteNode,
+                                new ClusterName(remoteClusterName)));
                     }
                 }
             }