Test & document raw transport handshakes (#120785)

DaveCTurner · web-flow · commit 60f78e45a75c · 2025-01-25T03:18:56.000+11:00
Updates the `TransportHandshaker` code comments to reflect the new handshake format introduced in #120744, and adds some low-level tests to verify the bytes on the wire are as described in those docs.
diff --git a/server/src/main/java/org/elasticsearch/transport/TransportHandshaker.java b/server/src/main/java/org/elasticsearch/transport/TransportHandshaker.java
@@ -53,7 +53,7 @@ final class TransportHandshaker {
      * rely on them matching the real transport protocol (which itself matched the release version numbers), but these days that's no longer
      * true.
      *
-     * Here are some example messages, broken down to show their structure:
+     * Here are some example messages, broken down to show their structure. See TransportHandshakerRawMessageTests for supporting tests.
      *
      * ## v6080099 Request:
      *
@@ -87,7 +87,7 @@ final class TransportHandshaker {
      *    c3 f9 eb 03                   -- max acceptable protocol version (vInt: 00000011 11101011 11111001 11000011 == 8060099)
      *
      *
-     * ## v7170099 and v8800000 Requests:
+     * ## v7170099 Requests:
      *
      * 45 53                            -- 'ES' marker
      * 00 00 00 31                      -- total message length
@@ -106,7 +106,7 @@ final class TransportHandshaker {
      *    04                            -- payload length
      *       c3 f9 eb 03                -- max acceptable protocol version (vInt: 00000011 11101011 11111001 11000011 == 8060099)
      *
-     * ## v7170099 and v8800000 Responses:
+     * ## v7170099 Responses:
      *
      * 45 53                            -- 'ES' marker
      * 00 00 00 17                      -- total message length
@@ -118,6 +118,40 @@ final class TransportHandshaker {
      *       00                         -- no response headers [1]
      *    c3 f9 eb 03                   -- max acceptable protocol version (vInt: 00000011 11101011 11111001 11000011 == 8060099)
      *
+     * ## v8800000 Requests:
+     *
+     * 45 53                            -- 'ES' marker
+     * 00 00 00 36                      -- total message length
+     *    00 00 00 00 00 00 00 01       -- request ID
+     *    08                            -- status flags (0b1000 == handshake request)
+     *    00 86 47 00                   -- handshake protocol version (0x6d6833 == 7170099)
+     *    00 00 00 19                   -- length of variable portion of header
+     *       00                         -- no request headers [1]
+     *       00                         -- no response headers [1]
+     *       16                         -- action string size
+     *       69 6e 74 65 72 6e 61 6c    }
+     *       3a 74 63 70 2f 68 61 6e    }- ASCII representation of HANDSHAKE_ACTION_NAME
+     *       64 73 68 61 6b 65          }
+     *    00                            -- no parent task ID [3]
+     *    0a                            -- payload length
+     *       e8 8f 9b 04                -- requesting node transport version (vInt: 00000100 10011011 10001111 11101000 == 8833000)
+     *       05                         -- requesting node release version string length
+     *          39 2e 30 2e 30          -- requesting node release version string "9.0.0"
+     *
+     * ## v8800000 Responses:
+     *
+     * 45 53                            -- 'ES' marker
+     * 00 00 00 1d                      -- total message length
+     *    00 00 00 00 00 00 00 01       -- request ID (copied from request)
+     *    09                            -- status flags (0b1001 == handshake response)
+     *    00 86 47 00                   -- handshake protocol version (0x864700 == 8800000, copied from request)
+     *    00 00 00 02                   -- length of following variable portion of header
+     *       00                         -- no request headers [1]
+     *       00                         -- no response headers [1]
+     *    e8 8f 9b 04                   -- responding node transport version (vInt: 00000100 10011011 10001111 11101000 == 8833000)
+     *    05                            -- responding node release version string length
+     *       39 2e 30 2e 30             -- responding node release version string "9.0.0"
+     *
      * [1] Thread context headers should be empty; see org.elasticsearch.common.util.concurrent.ThreadContext.ThreadContextStruct.writeTo
      *     for their structure.
      * [2] A list of strings, which can safely be ignored
diff --git a/server/src/main/resources/org/elasticsearch/bootstrap/test-framework.policy b/server/src/main/resources/org/elasticsearch/bootstrap/test-framework.policy
@@ -127,6 +127,7 @@ grant codeBase "${codebase.netty-transport}" {
 
 grant {
   permission java.net.SocketPermission "127.0.0.1", "accept, connect, resolve";
+  permission java.net.SocketPermission "[0:0:0:0:0:0:0:1]", "accept, connect, resolve";
   permission java.nio.file.LinkPermission "symbolic";
   // needed for keystore tests
   permission java.lang.RuntimePermission "accessUserInformation";
diff --git a/server/src/test/java/org/elasticsearch/transport/TransportHandshakerRawMessageTests.java b/server/src/test/java/org/elasticsearch/transport/TransportHandshakerRawMessageTests.java
@@ -0,0 +1,251 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.transport;
+
+import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.Build;
+import org.elasticsearch.TransportVersion;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.cluster.node.DiscoveryNodeUtils;
+import org.elasticsearch.common.bytes.BytesArray;
+import org.elasticsearch.common.io.stream.BytesStreamOutput;
+import org.elasticsearch.common.io.stream.InputStreamStreamInput;
+import org.elasticsearch.common.io.stream.OutputStreamStreamOutput;
+import org.elasticsearch.common.transport.TransportAddress;
+import org.elasticsearch.core.UpdateForV10;
+import org.elasticsearch.core.UpdateForV9;
+import org.elasticsearch.test.ESSingleNodeTestCase;
+import org.elasticsearch.test.TransportVersionUtils;
+
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.security.AccessController;
+import java.security.PrivilegedExceptionAction;
+
+import static org.hamcrest.Matchers.allOf;
+import static org.hamcrest.Matchers.greaterThan;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.lessThan;
+
+public class TransportHandshakerRawMessageTests extends ESSingleNodeTestCase {
+
+    @UpdateForV9(owner = UpdateForV9.Owner.CORE_INFRA) // remove support for v7 handshakes in v9
+    public void testV7Handshake() throws Exception {
+        final BytesRef handshakeRequestBytes;
+        final var requestId = randomNonNegativeLong();
+        try (var outputStream = new BytesStreamOutput()) {
+            outputStream.setTransportVersion(TransportHandshaker.V7_HANDSHAKE_VERSION);
+            outputStream.writeLong(requestId);
+            outputStream.writeByte(TransportStatus.setRequest(TransportStatus.setHandshake((byte) 0)));
+            outputStream.writeInt(TransportHandshaker.V7_HANDSHAKE_VERSION.id());
+            outputStream.writeByte((byte) 0); // no request headers;
+            outputStream.writeByte((byte) 0); // no response headers;
+            outputStream.writeStringArray(new String[] { "x-pack" }); // one feature
+            outputStream.writeString("internal:tcp/handshake");
+            outputStream.writeByte((byte) 0); // no parent task ID;
+
+            final var requestNodeTransportVersionId = TransportVersionUtils.randomCompatibleVersion(random()).id();
+            assertThat(requestNodeTransportVersionId, allOf(greaterThanOrEqualTo(1 << 22), lessThan(1 << 28))); // 4-byte vInt
+            outputStream.writeByte((byte) 4); // payload length
+            outputStream.writeVInt(requestNodeTransportVersionId);
+
+            handshakeRequestBytes = outputStream.bytes().toBytesRef();
+        }
+
+        final BytesRef handshakeResponseBytes;
+        try (var socket = openTransportConnection()) {
+            var streamOutput = new OutputStreamStreamOutput(socket.getOutputStream());
+            streamOutput.write("ES".getBytes(StandardCharsets.US_ASCII));
+            streamOutput.writeInt(handshakeRequestBytes.length);
+            streamOutput.writeBytes(handshakeRequestBytes.bytes, handshakeRequestBytes.offset, handshakeRequestBytes.length);
+            streamOutput.flush();
+
+            var streamInput = new InputStreamStreamInput(socket.getInputStream());
+            assertEquals((byte) 'E', streamInput.readByte());
+            assertEquals((byte) 'S', streamInput.readByte());
+            var responseLength = streamInput.readInt();
+            handshakeResponseBytes = streamInput.readBytesRef(responseLength);
+        }
+
+        try (var inputStream = new BytesArray(handshakeResponseBytes).streamInput()) {
+            assertEquals(requestId, inputStream.readLong());
+            assertEquals(TransportStatus.setResponse(TransportStatus.setHandshake((byte) 0)), inputStream.readByte());
+            assertEquals(TransportHandshaker.V7_HANDSHAKE_VERSION.id(), inputStream.readInt());
+            assertEquals((byte) 0, inputStream.readByte()); // no request headers
+            assertEquals((byte) 0, inputStream.readByte()); // no response headers
+            inputStream.setTransportVersion(TransportHandshaker.V7_HANDSHAKE_VERSION);
+            assertEquals(TransportVersion.current().id(), inputStream.readVInt());
+            assertEquals(-1, inputStream.read());
+        }
+    }
+
+    @UpdateForV10(owner = UpdateForV10.Owner.CORE_INFRA) // remove support for v8 handshakes in v10
+    public void testV8Handshake() throws Exception {
+        final BytesRef handshakeRequestBytes;
+        final var requestId = randomNonNegativeLong();
+        try (var outputStream = new BytesStreamOutput()) {
+            outputStream.setTransportVersion(TransportHandshaker.V8_HANDSHAKE_VERSION);
+            outputStream.writeLong(requestId);
+            outputStream.writeByte(TransportStatus.setRequest(TransportStatus.setHandshake((byte) 0)));
+            outputStream.writeInt(TransportHandshaker.V8_HANDSHAKE_VERSION.id());
+            outputStream.writeInt(0x1a); // length of variable-length header, always 0x1a
+            outputStream.writeByte((byte) 0); // no request headers;
+            outputStream.writeByte((byte) 0); // no response headers;
+            outputStream.writeByte((byte) 0); // no features;
+            outputStream.writeString("internal:tcp/handshake");
+            outputStream.writeByte((byte) 0); // no parent task ID;
+
+            final var requestNodeTransportVersionId = TransportVersionUtils.randomCompatibleVersion(random()).id();
+            assertThat(requestNodeTransportVersionId, allOf(greaterThanOrEqualTo(1 << 22), lessThan(1 << 28))); // 4-byte vInt
+            outputStream.writeByte((byte) 4); // payload length
+            outputStream.writeVInt(requestNodeTransportVersionId);
+
+            handshakeRequestBytes = outputStream.bytes().toBytesRef();
+        }
+
+        final BytesRef handshakeResponseBytes;
+        try (var socket = openTransportConnection()) {
+            var streamOutput = new OutputStreamStreamOutput(socket.getOutputStream());
+            streamOutput.write("ES".getBytes(StandardCharsets.US_ASCII));
+            streamOutput.writeInt(handshakeRequestBytes.length);
+            streamOutput.writeBytes(handshakeRequestBytes.bytes, handshakeRequestBytes.offset, handshakeRequestBytes.length);
+            streamOutput.flush();
+
+            var streamInput = new InputStreamStreamInput(socket.getInputStream());
+            assertEquals((byte) 'E', streamInput.readByte());
+            assertEquals((byte) 'S', streamInput.readByte());
+            var responseLength = streamInput.readInt();
+            handshakeResponseBytes = streamInput.readBytesRef(responseLength);
+        }
+
+        try (var inputStream = new BytesArray(handshakeResponseBytes).streamInput()) {
+            assertEquals(requestId, inputStream.readLong());
+            assertEquals(TransportStatus.setResponse(TransportStatus.setHandshake((byte) 0)), inputStream.readByte());
+            assertEquals(TransportHandshaker.V8_HANDSHAKE_VERSION.id(), inputStream.readInt());
+            assertEquals(2, inputStream.readInt()); // length of variable-length header, always 0x02
+            assertEquals((byte) 0, inputStream.readByte()); // no request headers
+            assertEquals((byte) 0, inputStream.readByte()); // no response headers
+            inputStream.setTransportVersion(TransportHandshaker.V8_HANDSHAKE_VERSION);
+            assertEquals(TransportVersion.current().id(), inputStream.readVInt());
+            assertEquals(-1, inputStream.read());
+        }
+    }
+
+    @UpdateForV10(owner = UpdateForV10.Owner.CORE_INFRA) // remove support for v9 handshakes in v11
+    public void testV9Handshake() throws Exception {
+        final BytesRef handshakeRequestBytes;
+        final var requestId = randomNonNegativeLong();
+        try (var outputStream = new BytesStreamOutput()) {
+            outputStream.setTransportVersion(TransportHandshaker.V9_HANDSHAKE_VERSION);
+            outputStream.writeLong(requestId);
+            outputStream.writeByte(TransportStatus.setRequest(TransportStatus.setHandshake((byte) 0)));
+            outputStream.writeInt(TransportHandshaker.V9_HANDSHAKE_VERSION.id());
+            outputStream.writeInt(0x19); // length of variable-length header, always 0x19
+            outputStream.writeByte((byte) 0); // no request headers;
+            outputStream.writeByte((byte) 0); // no response headers;
+            outputStream.writeString("internal:tcp/handshake");
+            outputStream.writeByte((byte) 0); // no parent task ID;
+
+            final var requestNodeTransportVersionId = TransportVersionUtils.randomCompatibleVersion(random()).id();
+            assertThat(requestNodeTransportVersionId, allOf(greaterThanOrEqualTo(1 << 22), lessThan(1 << 28))); // 4-byte vInt
+            final var releaseVersionLength = between(0, 127 - 5); // so that its length, and the length of the payload, is a one-byte vInt
+            final var requestNodeReleaseVersion = randomAlphaOfLength(releaseVersionLength);
+            outputStream.writeByte((byte) (4 + 1 + releaseVersionLength)); // payload length
+            outputStream.writeVInt(requestNodeTransportVersionId);
+            outputStream.writeString(requestNodeReleaseVersion);
+
+            handshakeRequestBytes = outputStream.bytes().toBytesRef();
+        }
+
+        final BytesRef handshakeResponseBytes;
+        try (var socket = openTransportConnection()) {
+            var streamOutput = new OutputStreamStreamOutput(socket.getOutputStream());
+            streamOutput.write("ES".getBytes(StandardCharsets.US_ASCII));
+            streamOutput.writeInt(handshakeRequestBytes.length);
+            streamOutput.writeBytes(handshakeRequestBytes.bytes, handshakeRequestBytes.offset, handshakeRequestBytes.length);
+            streamOutput.flush();
+
+            var streamInput = new InputStreamStreamInput(socket.getInputStream());
+            assertEquals((byte) 'E', streamInput.readByte());
+            assertEquals((byte) 'S', streamInput.readByte());
+            var responseLength = streamInput.readInt();
+            handshakeResponseBytes = streamInput.readBytesRef(responseLength);
+        }
+
+        try (var inputStream = new BytesArray(handshakeResponseBytes).streamInput()) {
+            assertEquals(requestId, inputStream.readLong());
+            assertEquals(TransportStatus.setResponse(TransportStatus.setHandshake((byte) 0)), inputStream.readByte());
+            assertEquals(TransportHandshaker.V9_HANDSHAKE_VERSION.id(), inputStream.readInt());
+            assertEquals(2, inputStream.readInt()); // length of variable-length header, always 0x02
+            assertEquals((byte) 0, inputStream.readByte()); // no request headers
+            assertEquals((byte) 0, inputStream.readByte()); // no response headers
+            inputStream.setTransportVersion(TransportHandshaker.V9_HANDSHAKE_VERSION);
+            assertEquals(TransportVersion.current().id(), inputStream.readVInt());
+            assertEquals(Build.current().version(), inputStream.readString());
+            assertEquals(-1, inputStream.read());
+        }
+    }
+
+    public void testOutboundHandshake() throws Exception {
+        final BytesRef handshakeRequestBytes;
+
+        try (var serverSocket = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) {
+            getInstanceFromNode(TransportService.class).openConnection(
+                DiscoveryNodeUtils.builder(randomIdentifier())
+                    .address(new TransportAddress(serverSocket.getInetAddress(), serverSocket.getLocalPort()))
+                    .build(),
+                ConnectionProfile.buildSingleChannelProfile(TransportRequestOptions.Type.REG, null, null, null, null, null),
+                ActionListener.noop()
+            );
+
+            try (
+                var acceptedSocket = serverSocket.accept();
+                var streamInput = new InputStreamStreamInput(acceptedSocket.getInputStream())
+            ) {
+                assertEquals((byte) 'E', streamInput.readByte());
+                assertEquals((byte) 'S', streamInput.readByte());
+                var responseLength = streamInput.readInt();
+                handshakeRequestBytes = streamInput.readBytesRef(responseLength);
+            }
+        }
+
+        final BytesRef payloadBytes;
+
+        try (var inputStream = new BytesArray(handshakeRequestBytes).streamInput()) {
+            assertThat(inputStream.readLong(), greaterThan(0L));
+            assertEquals(TransportStatus.setRequest(TransportStatus.setHandshake((byte) 0)), inputStream.readByte());
+            assertEquals(TransportHandshaker.V8_HANDSHAKE_VERSION.id(), inputStream.readInt());
+            assertEquals(0x1a, inputStream.readInt()); // length of variable-length header, always 0x1a
+            assertEquals((byte) 0, inputStream.readByte()); // no request headers
+            assertEquals((byte) 0, inputStream.readByte()); // no response headers
+            assertEquals((byte) 0, inputStream.readByte()); // no features
+            assertEquals("internal:tcp/handshake", inputStream.readString());
+            assertEquals((byte) 0, inputStream.readByte()); // no parent task
+            inputStream.setTransportVersion(TransportHandshaker.V8_HANDSHAKE_VERSION);
+            payloadBytes = inputStream.readBytesRef();
+            assertEquals(-1, inputStream.read());
+        }
+
+        try (var inputStream = new BytesArray(payloadBytes).streamInput()) {
+            inputStream.setTransportVersion(TransportHandshaker.V8_HANDSHAKE_VERSION);
+            assertEquals(TransportVersion.current().id(), inputStream.readVInt());
+            assertEquals(-1, inputStream.read());
+        }
+    }
+
+    private Socket openTransportConnection() throws Exception {
+        final var transportAddress = randomFrom(getInstanceFromNode(TransportService.class).boundAddress().boundAddresses()).address();
+        return AccessController.doPrivileged(
+            (PrivilegedExceptionAction<Socket>) (() -> new Socket(transportAddress.getAddress(), transportAddress.getPort()))
+        );
+    }
+}
