Validate transport handshake from known version

With parallel releases on multiple branches it's possible that an older
branch sees a transport version update that is not known to a
numerically newer but chronologically older version. In that case the
two nodes cannot intercommunicate, so with this commit we reject such
connection attempts at the version negotiation stage.

Backport of elastic#121747 to 8.x

Author: DaveCTurner

server/src/main/java/org/elasticsearch/TransportVersion.java

@@ -98,6 +98,14 @@ public static TransportVersion current() {
         return CurrentHolder.CURRENT;
     }
 
+    /**
+     * @return whether this is a known {@link TransportVersion}, i.e. one declared in {@link TransportVersions}. Other versions may exist
+     *         in the wild (they're sent over the wire by numeric ID) but we don't know how to communicate using such versions.
+     */
+    public boolean isKnown() {
+        return VersionsHolder.ALL_VERSIONS_MAP.containsKey(id);
+    }
+
     public static TransportVersion fromString(String str) {
         return TransportVersion.fromId(Integer.parseInt(str));
     }

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

@@ -11,15 +11,17 @@
 
 import org.elasticsearch.Build;
 import org.elasticsearch.TransportVersion;
-import org.elasticsearch.TransportVersions;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.metrics.CounterMetric;
+import org.elasticsearch.core.Strings;
 import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.logging.LogManager;
+import org.elasticsearch.logging.Logger;
 import org.elasticsearch.threadpool.ThreadPool;
 
 import java.io.EOFException;
@@ -157,6 +159,8 @@ final class TransportHandshaker {
      * [3] Parent task ID should be empty; see org.elasticsearch.tasks.TaskId.writeTo for its structure.
      */
 
+    private static final Logger logger = LogManager.getLogger(TransportHandshaker.class);
+
     static final TransportVersion V7_HANDSHAKE_VERSION = TransportVersion.fromId(6_08_00_99);
     static final TransportVersion V8_HANDSHAKE_VERSION = TransportVersion.fromId(7_17_00_99);
     static final TransportVersion V9_HANDSHAKE_VERSION = TransportVersion.fromId(8_800_00_0);
@@ -195,7 +199,7 @@ void sendHandshake(
         ActionListener<TransportVersion> listener
     ) {
         numHandshakes.inc();
-        final HandshakeResponseHandler handler = new HandshakeResponseHandler(requestId, listener);
+        final HandshakeResponseHandler handler = new HandshakeResponseHandler(requestId, channel, listener);
         pendingHandshakes.put(requestId, handler);
         channel.addCloseListener(
             ActionListener.running(() -> handler.handleLocalException(new TransportException("handshake failed because connection reset")))
@@ -221,9 +225,9 @@ void sendHandshake(
     }
 
     void handleHandshake(TransportChannel channel, long requestId, StreamInput stream) throws IOException {
+        final HandshakeRequest handshakeRequest;
         try {
-            // Must read the handshake request to exhaust the stream
-            new HandshakeRequest(stream);
+            handshakeRequest = new HandshakeRequest(stream);
         } catch (Exception e) {
             assert ignoreDeserializationErrors : e;
             throw e;
@@ -242,9 +246,44 @@ void handleHandshake(TransportChannel channel, long requestId, StreamInput strea
             assert ignoreDeserializationErrors : exception;
             throw exception;
         }
+        ensureCompatibleVersion(version, handshakeRequest.transportVersion, handshakeRequest.releaseVersion, channel);
         channel.sendResponse(new HandshakeResponse(this.version, Build.current().version()));
     }
 
+    static void ensureCompatibleVersion(
+        TransportVersion localTransportVersion,
+        TransportVersion remoteTransportVersion,
+        String releaseVersion,
+        Object channel
+    ) {
+        if (TransportVersion.isCompatible(remoteTransportVersion)) {
+            if (remoteTransportVersion.onOrAfter(localTransportVersion)) {
+                // Remote is newer than us, so we will be using our transport protocol and it's up to the other end to decide whether it
+                // knows how to do that.
+                return;
+            }
+            if (remoteTransportVersion.isKnown()) {
+                // Remote is older than us, so we will be using its transport protocol, which we can only do if and only if its protocol
+                // version is known to us.
+                return;
+            }
+        }
+
+        final var message = Strings.format(
+            """
+                Rejecting unreadable transport handshake from remote node with version [%s/%s] received on [%s] since this node has \
+                version [%s/%s] which has an incompatible wire format.""",
+            releaseVersion,
+            remoteTransportVersion,
+            channel,
+            Build.current().version(),
+            localTransportVersion
+        );
+        logger.warn(message);
+        throw new IllegalStateException(message);
+
+    }
+
     TransportResponseHandler<HandshakeResponse> removeHandlerForHandshake(long requestId) {
         return pendingHandshakes.remove(requestId);
     }
@@ -260,11 +299,13 @@ long getNumHandshakes() {
     private class HandshakeResponseHandler implements TransportResponseHandler<HandshakeResponse> {
 
         private final long requestId;
+        private final TcpChannel channel;
         private final ActionListener<TransportVersion> listener;
         private final AtomicBoolean isDone = new AtomicBoolean(false);
 
-        private HandshakeResponseHandler(long requestId, ActionListener<TransportVersion> listener) {
+        private HandshakeResponseHandler(long requestId, TcpChannel channel, ActionListener<TransportVersion> listener) {
             this.requestId = requestId;
+            this.channel = channel;
             this.listener = listener;
         }
 
@@ -281,20 +322,13 @@ public Executor executor() {
         @Override
         public void handleResponse(HandshakeResponse response) {
             if (isDone.compareAndSet(false, true)) {
-                TransportVersion responseVersion = response.transportVersion;
-                if (TransportVersion.isCompatible(responseVersion) == false) {
-                    listener.onFailure(
-                        new IllegalStateException(
-                            "Received message from unsupported version: ["
-                                + responseVersion
-                                + "] minimal compatible version is: ["
-                                + TransportVersions.MINIMUM_COMPATIBLE
-                                + "]"
-                        )
-                    );
-                } else {
-                    listener.onResponse(TransportVersion.min(TransportHandshaker.this.version, response.getTransportVersion()));
-                }
+                ActionListener.completeWith(listener, () -> {
+                    ensureCompatibleVersion(version, response.getTransportVersion(), response.getReleaseVersion(), channel);
+                    final var resultVersion = TransportVersion.min(TransportHandshaker.this.version, response.getTransportVersion());
+                    assert TransportVersion.current().before(version) // simulating a newer-version transport service for test purposes
+                        || resultVersion.isKnown() : "negotiated unknown version " + resultVersion;
+                    return resultVersion;
+                });
             }
         }
 

server/src/test/java/org/elasticsearch/transport/InboundHandlerTests.java

@@ -290,7 +290,9 @@ public void testLogsSlowInboundProcessing() throws Exception {
             );
             BytesStreamOutput byteData = new BytesStreamOutput();
             TaskId.EMPTY_TASK_ID.writeTo(byteData);
+            // simulate bytes of a transport handshake: vInt transport version then release version string
             TransportVersion.writeVersion(remoteVersion, byteData);
+            byteData.writeString(randomIdentifier());
             final InboundMessage requestMessage = new InboundMessage(
                 requestHeader,
                 ReleasableBytesReference.wrap(byteData.bytes()),

server/src/test/java/org/elasticsearch/transport/TransportHandshakerTests.java

@@ -8,8 +8,12 @@
  */
 package org.elasticsearch.transport;
 
+import org.apache.logging.log4j.Level;
+import org.elasticsearch.Build;
 import org.elasticsearch.TransportVersion;
+import org.elasticsearch.TransportVersions;
 import org.elasticsearch.Version;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodeUtils;
@@ -18,13 +22,17 @@
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.tasks.TaskId;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.MockLog;
 import org.elasticsearch.test.TransportVersionUtils;
+import org.elasticsearch.test.junit.annotations.TestLogging;
 import org.elasticsearch.threadpool.TestThreadPool;
 
 import java.io.IOException;
 import java.util.Collections;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 
+import static org.hamcrest.Matchers.allOf;
 import static org.hamcrest.Matchers.containsString;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
@@ -91,6 +99,40 @@ public void testHandshakeRequestAndResponse() throws IOException {
         assertEquals(TransportVersion.current(), versionFuture.actionGet());
     }
 
+    @TestLogging(reason = "testing WARN logging", value = "org.elasticsearch.transport.TransportHandshaker:WARN")
+    public void testIncompatibleHandshakeRequest() throws IOException {
+        TransportHandshaker.HandshakeRequest handshakeRequest = new TransportHandshaker.HandshakeRequest(
+            getRandomIncompatibleTransportVersion(),
+            randomIdentifier()
+        );
+        BytesStreamOutput bytesStreamOutput = new BytesStreamOutput();
+        bytesStreamOutput.setTransportVersion(HANDSHAKE_REQUEST_VERSION);
+        handshakeRequest.writeTo(bytesStreamOutput);
+        StreamInput input = bytesStreamOutput.bytes().streamInput();
+        input.setTransportVersion(HANDSHAKE_REQUEST_VERSION);
+        final TestTransportChannel channel = new TestTransportChannel(ActionListener.running(() -> fail("should not complete")));
+
+        MockLog.assertThatLogger(
+            () -> assertThat(
+                expectThrows(IllegalStateException.class, () -> handshaker.handleHandshake(channel, randomNonNegativeLong(), input))
+                    .getMessage(),
+                allOf(
+                    containsString("Rejecting unreadable transport handshake"),
+                    containsString("[" + handshakeRequest.releaseVersion + "/" + handshakeRequest.transportVersion + "]"),
+                    containsString("[" + Build.current().version() + "/" + TransportVersion.current() + "]"),
+                    containsString("which has an incompatible wire format")
+                )
+            ),
+            TransportHandshaker.class,
+            new MockLog.SeenEventExpectation(
+                "warning",
+                TransportHandshaker.class.getCanonicalName(),
+                Level.WARN,
+                "Rejecting unreadable transport handshake * incompatible wire format."
+            )
+        );
+    }
+
     public void testHandshakeResponseFromOlderNode() throws Exception {
         final PlainActionFuture<TransportVersion> versionFuture = new PlainActionFuture<>();
         final long reqId = randomNonNegativeLong();
@@ -106,6 +148,54 @@ public void testHandshakeResponseFromOlderNode() throws Exception {
         assertEquals(remoteVersion, versionFuture.result());
     }
 
+    @TestLogging(reason = "testing WARN logging", value = "org.elasticsearch.transport.TransportHandshaker:WARN")
+    public void testHandshakeResponseFromOlderNodeWithPatchedProtocol() {
+        final PlainActionFuture<TransportVersion> versionFuture = new PlainActionFuture<>();
+        final long reqId = randomNonNegativeLong();
+        handshaker.sendHandshake(reqId, node, channel, SAFE_AWAIT_TIMEOUT, versionFuture);
+        TransportResponseHandler<TransportHandshaker.HandshakeResponse> handler = handshaker.removeHandlerForHandshake(reqId);
+
+        assertFalse(versionFuture.isDone());
+
+        final var handshakeResponse = new TransportHandshaker.HandshakeResponse(
+            getRandomIncompatibleTransportVersion(),
+            randomIdentifier()
+        );
+
+        MockLog.assertThatLogger(
+            () -> handler.handleResponse(handshakeResponse),
+            TransportHandshaker.class,
+            new MockLog.SeenEventExpectation(
+                "warning",
+                TransportHandshaker.class.getCanonicalName(),
+                Level.WARN,
+                "Rejecting unreadable transport handshake * incompatible wire format."
+            )
+        );
+
+        assertTrue(versionFuture.isDone());
+        assertThat(
+            expectThrows(ExecutionException.class, IllegalStateException.class, versionFuture::result).getMessage(),
+            allOf(
+                containsString("Rejecting unreadable transport handshake"),
+                containsString("[" + handshakeResponse.getReleaseVersion() + "/" + handshakeResponse.getTransportVersion() + "]"),
+                containsString("[" + Build.current().version() + "/" + TransportVersion.current() + "]"),
+                containsString("which has an incompatible wire format")
+            )
+        );
+    }
+
+    private static TransportVersion getRandomIncompatibleTransportVersion() {
+        return randomBoolean()
+            // either older than MINIMUM_COMPATIBLE
+            ? new TransportVersion(between(1, TransportVersions.MINIMUM_COMPATIBLE.id() - 1))
+            // or between MINIMUM_COMPATIBLE and current but not known
+            : randomValueOtherThanMany(
+                TransportVersion::isKnown,
+                () -> new TransportVersion(between(TransportVersions.MINIMUM_COMPATIBLE.id(), TransportVersion.current().id()))
+            );
+    }
+
     public void testHandshakeResponseFromNewerNode() throws Exception {
         final PlainActionFuture<TransportVersion> versionFuture = new PlainActionFuture<>();
         final long reqId = randomNonNegativeLong();