IGNITE-27554 Add check for serMode

Author: chesnokoff

modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryIoSession.java

@@ -168,7 +168,16 @@ <T> T readMessage() throws IgniteCheckedException, IOException {
 
         try {
             if (MESSAGE_SERIALIZATION != serMode) {
-                detectSslAlert(serMode, in);
+                byte[] hdr = new byte[4];
+                hdr[0] = serMode;
+
+                int read = in.readNBytes(hdr, 1, 3);
+
+                if (read < 3)
+                    throw new EOFException();
+
+                detectSslAlert(hdr);
+                detectJavaObjectStreamHeader(hdr);
 
                 // IOException type is important for ServerImpl. It may search the cause (X.hasCause).
                 // The connection error processing behavior depends on it.
@@ -307,17 +316,23 @@ private void serializeMessage(Message m, OutputStream out) throws IOException {
      * See handling {@code StreamCorruptedException} in {@link #readMessage()}.
      * Keeps logic similar to {@link java.io.ObjectInputStream#readStreamHeader}.
      */
-    private void detectSslAlert(byte firstByte, InputStream in) throws IOException {
-        byte[] hdr = new byte[4];
-        hdr[0] = firstByte;
-        int read = in.readNBytes(hdr, 1, 3);
-
-        if (read < 3)
-            throw new EOFException();
-
+    private void detectSslAlert(byte[] hdr) throws IOException {
         String hex = String.format("%02x%02x%02x%02x", hdr[0], hdr[1], hdr[2], hdr[3]);
 
         if (hex.matches("15....00"))
             throw new StreamCorruptedException("invalid stream header: " + hex);
     }
+
+    /**
+     * Detects Java Object Serialization stream header (AC ED 00 05).
+     * This indicates that the remote node sends discovery messages without leading serMode byte.
+     */
+    private void detectJavaObjectStreamHeader(byte[] hdr) throws IOException {
+        if ((hdr[0] == (byte)0xAC) && (hdr[1] == (byte)0xED) && (hdr[2] == (byte)0x00) && (hdr[3] == (byte)0x05))
+            throw new IOException(
+                "Incompatible discovery protocol: received Java ObjectStream header (AC ED 00 05) where serMode byte is expected. "
+                    + "Remote node likely uses a discovery message format without serMode. "
+                    + "All nodes must run compatible Ignite versions."
+            );
+    }
 }

modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryIoSessionDifferentSerializationTest.java

@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.spi.discovery.tcp;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.Socket;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.util.typedef.X;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryHandshakeRequest;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.junit.Test;
+
+/** Tests Java serialization header detection in discovery messages. */
+public class TcpDiscoveryIoSessionDifferentSerializationTest extends GridCommonAbstractTest {
+    /** {@inheritDoc} */
+    @Override protected void afterTest() throws Exception {
+        stopAllGrids();
+    }
+
+    /** */
+    @Test
+    public void testDetectJavaObjectStreamHeader() throws Exception {
+        IgniteEx grid = startGrid(0);
+
+        TcpDiscoverySpi spi = (TcpDiscoverySpi)grid.configuration().getDiscoverySpi();
+
+        TcpDiscoveryHandshakeRequest req = new TcpDiscoveryHandshakeRequest();
+
+        byte[] bytes = U.marshal(spi.marshaller(), req);
+
+        TcpDiscoveryIoSession ses = new TcpDiscoveryIoSession(
+            new TestSocket(new ByteArrayInputStream(bytes), new ByteArrayOutputStream()),
+            spi
+        );
+
+        Throwable e = GridTestUtils.assertThrows(log, () -> ses.readMessage(), IgniteCheckedException.class, null);
+
+        assertTrue(X.hasCause(e, "Incompatible discovery protocol: received Java ObjectStream header", IOException.class));
+    }
+
+    /** */
+    private static class TestSocket extends Socket {
+        /** */
+        private final InputStream in;
+
+        /** */
+        private final OutputStream out;
+
+        /**
+         * @param in Input stream.
+         * @param out Output stream.
+         */
+        private TestSocket(InputStream in, OutputStream out) {
+            this.in = in;
+            this.out = out;
+        }
+
+        /** {@inheritDoc} */
+        @Override public InputStream getInputStream() {
+            return in;
+        }
+
+        /** {@inheritDoc} */
+        @Override public OutputStream getOutputStream() {
+            return out;
+        }
+    }
+}

modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSpiDiscoverySelfTestSuite.java

@@ -47,6 +47,7 @@
 import org.apache.ignite.spi.discovery.tcp.TcpDiscoveryCoordinatorFailureTest;
 import org.apache.ignite.spi.discovery.tcp.TcpDiscoveryDeadNodeAddressResolvingTest;
 import org.apache.ignite.spi.discovery.tcp.TcpDiscoveryFailedJoinTest;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoveryIoSessionDifferentSerializationTest;
 import org.apache.ignite.spi.discovery.tcp.TcpDiscoveryIpFinderCleanerTest;
 import org.apache.ignite.spi.discovery.tcp.TcpDiscoveryIpFinderFailureTest;
 import org.apache.ignite.spi.discovery.tcp.TcpDiscoveryMetricsWarnLogTest;
@@ -157,6 +158,7 @@
     IgniteClientReconnectEventHandlingTest.class,
 
     TcpDiscoveryFailedJoinTest.class,
+    TcpDiscoveryIoSessionDifferentSerializationTest.class,
 
     // SSL.
     TcpDiscoverySslSelfTest.class,