Merge pull request #1960 from rabbitmq/mergify/bp/v5.x/pr-1959

Implement a TODO in CommandAssembler (backport #1959)

Author: acogoluegnes

src/main/java/com/rabbitmq/client/impl/AMQConnection.java

@@ -252,7 +252,7 @@ public AMQConnection(ConnectionParams params, FrameHandler frameHandler,
 
         this.credentialsRefreshService = params.getCredentialsRefreshService();
 
-
+        this.maxInboundMessageBodySize = params.getMaxInboundMessageBodySize();
         this._channel0 = createChannel0();
 
         this._channelManager = null;
@@ -267,7 +267,6 @@ public AMQConnection(ConnectionParams params, FrameHandler frameHandler,
         this.errorOnWriteListener = params.getErrorOnWriteListener() != null ? params.getErrorOnWriteListener() :
             (connection, exception) -> { throw exception; }; // we just propagate the exception for non-recoverable connections
         this.workPoolTimeout = params.getWorkPoolTimeout();
-        this.maxInboundMessageBodySize = params.getMaxInboundMessageBodySize();
     }
 
     AMQChannel createChannel0() {

src/main/java/com/rabbitmq/client/impl/CommandAssembler.java

@@ -21,6 +21,7 @@
 import java.util.List;
 
 import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.MalformedFrameException;
 import com.rabbitmq.client.UnexpectedFrameError;
 import static java.lang.String.format;
 
@@ -142,14 +143,16 @@ private void consumeHeaderFrame(Frame f) throws IOException {
         }
     }
 
-    private void consumeBodyFrame(Frame f) {
+    private void consumeBodyFrame(Frame f) throws MalformedFrameException {
         if (f.getType() == AMQP.FRAME_BODY) {
             byte[] fragment = f.getPayload();
+            if (fragment.length > this.remainingBodyBytes) {
+                throw new MalformedFrameException(format(
+                    "Body frame payload (%d bytes) exceeds remaining expected body size (%d bytes)",
+                    fragment.length, this.remainingBodyBytes));
+            }
             this.remainingBodyBytes -= fragment.length;
             updateContentBodyState();
-            if (this.remainingBodyBytes < 0) {
-                throw new UnsupportedOperationException("%%%%%% FIXME unimplemented");
-            }
             appendBodyFragment(fragment);
         } else {
             throw new UnexpectedFrameError(f, AMQP.FRAME_BODY);

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

@@ -17,11 +17,15 @@
 package com.rabbitmq.client.test;
 
 import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.MalformedFrameException;
 import com.rabbitmq.client.UnexpectedFrameError;
 import com.rabbitmq.client.WriteListener;
 import com.rabbitmq.client.impl.AMQConnection;
+import com.rabbitmq.client.impl.AMQImpl;
 import com.rabbitmq.client.impl.AMQImpl.Basic.Publish;
+import com.rabbitmq.client.impl.DefaultExceptionHandler;
 import com.rabbitmq.client.impl.Frame;
 import com.rabbitmq.client.impl.FrameHandler;
 import org.junit.jupiter.api.AfterEach;
@@ -30,13 +34,14 @@
 
 import java.io.IOException;
 import java.net.InetAddress;
-import java.net.SocketException;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicReference;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.*;
 
 public class BrokenFramesTest {
@@ -98,6 +103,50 @@ public class BrokenFramesTest {
         fail("No UnexpectedFrameError thrown");
     }
 
+    @Test
+    public void bodyOverrunTriggersConnectionClosure() throws Exception {
+        List<Frame> frames = new ArrayList<>();
+        int channelNumber = 0;
+        factory.setMaxInboundMessageBodySize(1024);
+        AtomicReference<Throwable> exception = new AtomicReference<>();
+        factory.setExceptionHandler(
+            new DefaultExceptionHandler() {
+                @Override
+                public void handleUnexpectedConnectionDriverException(
+                    Connection conn, Throwable ex) {
+                    exception.set(ex);
+                }
+            });
+
+        // 1. Setup a valid method frame (Basic.Deliver expects content)
+        AMQImpl.Basic.Deliver method = new AMQImpl.Basic.Deliver("ctag", 1L, false, "", "rk");
+        frames.add(method.toFrame(channelNumber));
+
+        // 2. Declares exactly 1 byte of body size
+        AMQP.BasicProperties props = new AMQP.BasicProperties();
+        frames.add(props.toFrame(channelNumber, 1L));
+
+        // 3. Payload is 2 bytes (0x41, 0x42), exceeding the declared size
+        frames.add(new Frame(AMQP.FRAME_BODY, channelNumber, new byte[] { 0x41, 0x42 }));
+
+        myFrameHandler.setFrames(frames.iterator());
+
+        AMQConnection connection = null;
+        try {
+            connection = new AMQConnection(factory.params(Executors.newFixedThreadPool(1)), myFrameHandler);
+            connection.start();
+            fail("Expected an exception to be thrown due to body overrun, but none was.");
+        } catch (IOException e) {
+            assertThat(e).hasRootCauseInstanceOf(MalformedFrameException.class);
+            assertThat(exception.get()).isNotNull().isInstanceOf(MalformedFrameException.class);
+        } finally {
+            assertTrue(myFrameHandler.closeCalled, "The underlying FrameHandler (socket) should have been closed.");
+            if (connection != null) {
+                assertFalse(connection.isOpen(), "The AMQConnection should be marked as closed.");
+            }
+        }
+    }
+
     private UnexpectedFrameError findUnexpectedFrameError(Exception e) {
         Throwable t = e;
         while ((t = t.getCause()) != null) {
@@ -106,42 +155,42 @@ private UnexpectedFrameError findUnexpectedFrameError(Exception e) {
                 return (UnexpectedFrameError) t;
             }
         }
-
         return null;
     }
 
     private static class MyFrameHandler implements FrameHandler {
         private Iterator<Frame> frames;
+        public boolean closeCalled = false;
 
         public void setFrames(Iterator<Frame> frames) {
             this.frames = frames;
         }
 
-        public Frame readFrame() throws IOException {
+        public Frame readFrame() {
             return frames.next();
         }
 
-        public void sendHeader() throws IOException {
+        public void sendHeader() {
         }
 
         @Override
         public void initialize(AMQConnection connection) {
             connection.startMainLoop();
         }
 
-        public void setTimeout(int timeoutMs) throws SocketException {
+        public void setTimeout(int timeoutMs) {
             // no need to implement this: don't bother changing the timeout
         }
 
-        public void writeFrame(Frame frame) throws IOException {
+        public void writeFrame(Frame frame) {
             // no need to implement this: don't bother writing the frame
         }
 
         public void close() {
-            // nothing to do
+            this.closeCalled = true;
         }
 
-        public int getTimeout() throws SocketException {
+        public int getTimeout() {
             return 0;
         }
 
@@ -153,7 +202,7 @@ public int getPort() {
             return -1;
         }
 
-        public void flush(WriteListener listener) throws IOException {
+        public void flush(WriteListener listener) {
             // no need to implement this: don't bother writing the frame
         }