Handle protocol version mismatch in frame builder

References #319

Author: acogoluegnes

src/main/java/com/rabbitmq/client/impl/nio/FrameBuilder.java

@@ -19,6 +19,7 @@
 import com.rabbitmq.client.MalformedFrameException;
 import com.rabbitmq.client.impl.Frame;
 
+import java.io.DataInputStream;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.ReadableByteChannel;
@@ -32,6 +33,7 @@
  * start where it left off when the {@link NioLoop} comes back to
  * this connection.
  * This class is not thread safe.
+ *
  * @since 4.4.0
  */
 public class FrameBuilder {
@@ -41,17 +43,14 @@ public class FrameBuilder {
     protected final ReadableByteChannel channel;
 
     protected final ByteBuffer applicationBuffer;
-
-    private int frameType;
-    private int frameChannel;
-    private byte [] framePayload;
-
-    private int bytesRead = 0;
-
     // to store the bytes of the outstanding data
     // 3 byte-long because the longest we read is an unsigned int
     // (not need to store the latest byte)
-    private final int [] frameBuffer = new int[3];
+    private final int[] frameBuffer = new int[3];
+    private int frameType;
+    private int frameChannel;
+    private byte[] framePayload;
+    private int bytesRead = 0;
 
     public FrameBuilder(ReadableByteChannel channel, ByteBuffer buffer) {
         this.channel = channel;
@@ -63,15 +62,19 @@ public FrameBuilder(ReadableByteChannel channel, ByteBuffer buffer) {
      * This method returns null f a frame could not have been fully built from
      * the network. The client must then retry later (typically
      * when the channel notifies it has something to read).
+     *
      * @return a complete frame or null if a frame couldn't have been fully built
      * @throws IOException
+     * @see Frame#readFrom(DataInputStream)
      */
     public Frame readFrame() throws IOException {
-        while(somethingToRead()) {
+        while (somethingToRead()) {
             if (bytesRead == 0) {
                 // type
-                // FIXME check first byte isn't 'A' and thus a header indicating protocol version mismatch
                 frameType = readFromBuffer();
+                if (frameType == 'A') {
+                    handleProtocolVersionMismatch();
+                }
             } else if (bytesRead == 1) {
                 // channel 1/2
                 frameBuffer[0] = readFromBuffer();
@@ -108,24 +111,17 @@ public Frame readFrame() throws IOException {
         return null;
     }
 
-    private int read() throws IOException {
-        return NioHelper.read(channel, applicationBuffer);
-    }
-
-    private int readFromBuffer() {
-        return applicationBuffer.get() & 0xff;
-    }
-
     /**
      * Tells whether there's something to read in the application buffer or not.
      * Tries to read from the network if necessary.
+     *
      * @return true if there's something to read in the application buffer
      * @throws IOException
      */
     protected boolean somethingToRead() throws IOException {
-        if(!applicationBuffer.hasRemaining()) {
+        if (!applicationBuffer.hasRemaining()) {
             applicationBuffer.clear();
-            int read = read();
+            int read = NioHelper.read(channel, applicationBuffer);
             applicationBuffer.flip();
             if (read > 0) {
                 return true;
@@ -136,4 +132,72 @@ protected boolean somethingToRead() throws IOException {
             return true;
         }
     }
+
+    private int readFromBuffer() {
+        return applicationBuffer.get() & 0xff;
+    }
+
+    /**
+     * Handle a protocol version mismatch.
+     * @return
+     * @throws IOException
+     * @see Frame#protocolVersionMismatch(DataInputStream)
+     */
+    private void handleProtocolVersionMismatch() throws IOException {
+        // Probably an AMQP.... header indicating a version mismatch
+        // Otherwise meaningless, so try to read the version,
+        // and throw an exception, whether we read the version
+        // okay or not.
+        // Try to read everything from the network, this header
+        // is small and should never require several network reads.
+        byte[] expectedBytes = new byte[] { 'M', 'Q', 'P' };
+        int expectedBytesCount = 0;
+        while (somethingToRead() && expectedBytesCount < 3) {
+            // We expect the letters M, Q, P in that order: generate an informative error if they're not found
+            int nextByte = readFromBuffer();
+            if (nextByte != expectedBytes[expectedBytesCount]) {
+                throw new MalformedFrameException("Invalid AMQP protocol header from server: expected character " +
+                    expectedBytes[expectedBytesCount] + ", got " + nextByte);
+            }
+            expectedBytesCount++;
+        }
+
+        if (expectedBytesCount != 3) {
+            throw new MalformedFrameException("Invalid AMQP protocol header from server: read only "
+                + (expectedBytesCount + 1) + " byte(s) instead of 4");
+        }
+
+        int[] signature = new int[4];
+
+        for (int i = 0; i < 4; i++) {
+            if (somethingToRead()) {
+                signature[i] = readFromBuffer();
+            } else {
+                throw new MalformedFrameException("Invalid AMQP protocol header from server");
+            }
+        }
+
+        MalformedFrameException x;
+
+        if (signature[0] == 1 &&
+            signature[1] == 1 &&
+            signature[2] == 8 &&
+            signature[3] == 0) {
+            x = new MalformedFrameException("AMQP protocol version mismatch; we are version " +
+                AMQP.PROTOCOL.MAJOR + "-" + AMQP.PROTOCOL.MINOR + "-" + AMQP.PROTOCOL.REVISION +
+                ", server is 0-8");
+        } else {
+            String sig = "";
+            for (int i = 0; i < 4; i++) {
+                if (i != 0)
+                    sig += ",";
+                sig += signature[i];
+            }
+
+            x = new MalformedFrameException("AMQP protocol version mismatch; we are version " +
+                AMQP.PROTOCOL.MAJOR + "-" + AMQP.PROTOCOL.MINOR + "-" + AMQP.PROTOCOL.REVISION +
+                ", server sent signature " + sig);
+        }
+        throw x;
+    }
 }

src/main/java/com/rabbitmq/client/impl/nio/HeaderWriteRequest.java

@@ -25,6 +25,10 @@
  */
 public class HeaderWriteRequest implements WriteRequest {
 
+    public static final WriteRequest SINGLETON = new HeaderWriteRequest();
+
+    private HeaderWriteRequest() { }
+
     @Override
     public void handle(DataOutputStream outputStream) throws IOException {
         outputStream.write("AMQP".getBytes("US-ASCII"));

src/main/java/com/rabbitmq/client/impl/nio/SocketChannelFrameHandlerState.java

@@ -120,7 +120,7 @@ public Queue<WriteRequest> getWriteQueue() {
     }
 
     public void sendHeader() throws IOException {
-        sendWriteRequest(new HeaderWriteRequest());
+        sendWriteRequest(HeaderWriteRequest.SINGLETON);
     }
 
     public void write(Frame frame) throws IOException {
@@ -192,12 +192,8 @@ boolean continueReading() throws IOException {
             if (!plainIn.hasRemaining() && !cipherIn.hasRemaining()) {
                 // need to try to read something
                 cipherIn.clear();
-
-                // FIXME this logic may be simplified:
-                // flipping cipherIn and return cipherIn.hasRemaining should be enough
-
                 int bytesRead = NioHelper.read(channel, cipherIn);
-                if (bytesRead <= 0) {
+                if (bytesRead == 0) {
                     return false;
                 } else {
                     cipherIn.flip();

src/test/java/com/rabbitmq/client/test/FrameBuilderTest.java

@@ -16,6 +16,7 @@
 package com.rabbitmq.client.test;
 
 import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.MalformedFrameException;
 import com.rabbitmq.client.impl.Frame;
 import com.rabbitmq.client.impl.nio.FrameBuilder;
 import org.junit.Test;
@@ -31,6 +32,7 @@
 import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.nullValue;
 import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
 
 /**
  *
@@ -47,7 +49,7 @@ public class FrameBuilderTest {
 
     @Test
     public void buildFrameInOneGo() throws IOException {
-        buffer = ByteBuffer.wrap(new byte[]{1, 0, 0, 0, 0, 0, 3, 1, 2, 3, end()});
+        buffer = ByteBuffer.wrap(new byte[] { 1, 0, 0, 0, 0, 0, 3, 1, 2, 3, end() });
         builder = new FrameBuilder(channel, buffer);
         Frame frame = builder.readFrame();
         assertThat(frame, notNullValue());
@@ -58,10 +60,10 @@ public void buildFrameInOneGo() throws IOException {
 
     @Test
     public void buildFramesInOneGo() throws IOException {
-        byte[] frameContent = new byte[]{1, 0, 0, 0, 0, 0, 3, 1, 2, 3, end()};
+        byte[] frameContent = new byte[] { 1, 0, 0, 0, 0, 0, 3, 1, 2, 3, end() };
         int nbFrames = 13;
         byte[] frames = new byte[frameContent.length * nbFrames];
-        for(int i = 0; i < nbFrames; i++) {
+        for (int i = 0; i < nbFrames; i++) {
             for (int j = 0; j < frameContent.length; j++) {
                 frames[i * frameContent.length + j] = frameContent[j];
             }
@@ -82,7 +84,7 @@ public void buildFramesInOneGo() throws IOException {
 
     @Test
     public void buildFrameInSeveralCalls() throws IOException {
-        buffer = ByteBuffer.wrap(new byte[]{1, 0, 0, 0, 0, 0, 3, 1, 2});
+        buffer = ByteBuffer.wrap(new byte[] { 1, 0, 0, 0, 0, 0, 3, 1, 2 });
         builder = new FrameBuilder(channel, buffer);
         Frame frame = builder.readFrame();
         assertThat(frame, nullValue());
@@ -98,12 +100,43 @@ public void buildFrameInSeveralCalls() throws IOException {
         assertThat(frame.getPayload().length, is(3));
     }
 
+    @Test
+    public void protocolMismatchHeader() throws IOException {
+        ByteBuffer[] buffers = new ByteBuffer[] {
+            ByteBuffer.wrap(new byte[] { 'A' }),
+            ByteBuffer.wrap(new byte[] { 'A', 'M', 'Q' }),
+            ByteBuffer.wrap(new byte[] { 'A', 'N', 'Q', 'P' }),
+            ByteBuffer.wrap(new byte[] { 'A', 'M', 'Q', 'P' }),
+            ByteBuffer.wrap(new byte[] { 'A', 'M', 'Q', 'P', 1, 1, 8 }),
+            ByteBuffer.wrap(new byte[] { 'A', 'M', 'Q', 'P', 1, 1, 8, 0 }),
+            ByteBuffer.wrap(new byte[] { 'A', 'M', 'Q', 'P', 1, 1, 9, 1 })
+        };
+        String[] messages = new String[] {
+            "Invalid AMQP protocol header from server: read only 1 byte(s) instead of 4",
+            "Invalid AMQP protocol header from server: read only 3 byte(s) instead of 4",
+            "Invalid AMQP protocol header from server: expected character 77, got 78",
+            "Invalid AMQP protocol header from server",
+            "Invalid AMQP protocol header from server",
+            "AMQP protocol version mismatch; we are version 0-9-1, server is 0-8",
+            "AMQP protocol version mismatch; we are version 0-9-1, server sent signature 1,1,9,1"
+        };
+
+        for (int i = 0; i < buffers.length; i++) {
+            builder = new FrameBuilder(channel, buffers[i]);
+            try {
+                builder.readFrame();
+                fail("protocol header not correct, exception should have been thrown");
+            } catch (MalformedFrameException e) {
+                assertThat(e.getMessage(), is(messages[i]));
+            }
+        }
+    }
+
     byte b(int b) {
         return (byte) b;
     }
 
     byte end() {
         return (byte) AMQP.FRAME_END;
     }
-
 }