GG-23022: Fixed exception on message parsing failure

This closes apache#455

Author: isapego

modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientListenerBufferedParser.java

@@ -25,6 +25,8 @@
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 
+import static org.apache.ignite.internal.processors.odbc.ClientListenerNioListener.CONN_CTX_HANDSHAKE_TIMEOUT_TASK;
+
 /**
  * This class implements stream parser based on {@link ClientListenerNioServerBuffer}.
  * <p>
@@ -40,6 +42,9 @@ public class ClientListenerBufferedParser implements GridNioParser {
     /** Buffer metadata key. */
     private static final int BUF_META_KEY = GridNioSessionMetaKey.nextUniqueKey();
 
+    /** Length limit of handshake message - 128 MB.*/
+    private static final int HANDSHAKE_MSG_LEN_LIMIT = 1024 * 1024 * 128;
+
     /** {@inheritDoc} */
     @Override public byte[] decode(GridNioSession ses, ByteBuffer buf) throws IOException, IgniteCheckedException {
         ClientListenerNioServerBuffer nioBuf = ses.meta(BUF_META_KEY);
@@ -54,7 +59,18 @@ public class ClientListenerBufferedParser implements GridNioParser {
             assert old == null;
         }
 
-        return nioBuf.read(buf);
+        int msgLenLimit = 0;
+
+        // It means, we are still in handshake process
+        if (ses.meta(CONN_CTX_HANDSHAKE_TIMEOUT_TASK) != null)
+            msgLenLimit = HANDSHAKE_MSG_LEN_LIMIT;
+
+        try {
+            return nioBuf.read(buf, msgLenLimit);
+        }
+        catch (IgniteCheckedException e) {
+            throw new IgniteCheckedException("Failed to parse message from remote " + ses.remoteAddress(), e);
+        }
     }
 
     /** {@inheritDoc} */

modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientListenerNioServerBuffer.java

@@ -66,7 +66,7 @@ public byte[] data() {
      * @return Message bytes or {@code null} if message is not fully read yet.
      * @throws IgniteCheckedException If failed to parse message.
      */
-    @Nullable public byte[] read(ByteBuffer buf) throws IgniteCheckedException {
+    @Nullable public byte[] read(ByteBuffer buf, int msgLenLimit) throws IgniteCheckedException {
         if (cnt < 0) {
             for (; cnt < 0 && buf.hasRemaining(); cnt++)
                 msgSize |= (buf.get() & 0xFF) << (8*(4 + cnt));
@@ -75,7 +75,7 @@ public byte[] data() {
                 return null;
 
             // If count is 0 then message size should be inited.
-            if (msgSize <= 0)
+            if (msgSize <= 0 || (msgLenLimit > 0 && msgSize > msgLenLimit))
                 throw new IgniteCheckedException("Invalid message size: " + msgSize);
 
             data = new byte[msgSize];

modules/core/src/test/java/org/apache/ignite/client/ConnectionTest.java

@@ -60,6 +60,15 @@ public void testValidInvalidNodeAddressesMix() throws Exception {
         testConnection("127.0.0.1:47500", "127.0.0.1:10801", Config.SERVER);
     }
 
+    /** */
+    @Test(expected = org.apache.ignite.client.ClientConnectionException.class)
+    public void testInvalidBigHandshakeMessage() throws Exception {
+        char[] data = new char[1024 * 1024 * 128];
+        String userName = new String(data);
+
+        testConnectionWithUsername(userName, Config.SERVER);
+    }
+
     /**
      * @param addrs Addresses to connect.
      */
@@ -69,4 +78,15 @@ private void testConnection(String... addrs) throws Exception {
                      .setAddresses(addrs))) {
         }
     }
+    /**
+     * @param userName User name.
+     * @param addrs Addresses to connect.
+     */
+    private void testConnectionWithUsername(String userName, String... addrs) throws Exception {
+        try (LocalIgniteCluster cluster = LocalIgniteCluster.start(1);
+             IgniteClient client = Ignition.startClient(new ClientConfiguration()
+                 .setAddresses(addrs)
+                 .setUserName(userName))) {
+        }
+    }
 }