Stricter validation of pre-negotiated values in the frame reader

(cherry picked from commit 08790f0)

Author: michaelklishin

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

@@ -433,6 +433,14 @@ public void start()
                                    connTune.getFrameMax());
             this._frameMax = frameMax;
 
+            // Bound inbound frames to the negotiated frame_max. EMPTY_FRAME_SIZE is
+            // the per-frame overhead; +1 because the reader rejects payloads >= the limit.
+            if (frameMax > 0) {
+                _frameHandler.setMaxInboundFramePayloadSize(
+                    Math.min(this.maxInboundMessageBodySize,
+                             frameMax - AMQCommand.EMPTY_FRAME_SIZE + 1));
+            }
+
             int negotiatedHeartbeat =
                 negotiatedMaxValue(this.requestedHeartbeat,
                                    connTune.getHeartbeat());

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

@@ -59,6 +59,11 @@ default void finishConnectionNegotiation() {
 
     }
 
+    /** Cap inbound frame payloads, applied once frame_max is negotiated. */
+    default void setMaxInboundFramePayloadSize(int maxPayloadSize) {
+
+    }
+
     /**
      * Read a {@link Frame} from the underlying data connection.
      * @return an incoming Frame, or null if there is none

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

@@ -351,6 +351,11 @@ public void finishConnectionNegotiation() {
       maybeRemoveHandler(HANDLER_PROTOCOL_VERSION_MISMATCH);
     }
 
+    @Override
+    public void setMaxInboundFramePayloadSize(int maxPayloadSize) {
+      this.handler.maxPayloadSize = maxPayloadSize;
+    }
+
     @Override
     public Frame readFrame() {
       throw new UnsupportedOperationException();
@@ -506,7 +511,7 @@ InetSocketAddress maybeInetSocketAddress(SocketAddress socketAddress) {
 
   private static class AmqpHandler extends ChannelInboundHandlerAdapter {
 
-    private final int maxPayloadSize;
+    private volatile int maxPayloadSize;
     private final Runnable closeSequence;
     private final Predicate<ShutdownSignalException> willRecover;
     private volatile AMQConnection connection;

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

@@ -57,7 +57,7 @@ public class SocketFrameHandler implements FrameHandler {
     private final DataOutputStream _outputStream;
     private final Lock _outputStreamLock = new ReentrantLock();
 
-    private final int maxInboundMessageBodySize;
+    private volatile int maxInboundMessageBodySize;
 
     /** Time to linger before closing the socket forcefully. */
     public static final int SOCKET_CLOSING_TIMEOUT = 1;
@@ -193,6 +193,11 @@ public void initialize(AMQConnection connection) {
         connection.startMainLoop();
     }
 
+    @Override
+    public void setMaxInboundFramePayloadSize(int maxPayloadSize) {
+        this.maxInboundMessageBodySize = maxPayloadSize;
+    }
+
     @Override
     public Frame readFrame() throws IOException {
         _inputStreamLock.lock();

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

@@ -0,0 +1,76 @@
+// Copyright (c) 2026 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+//
+// This software, the RabbitMQ Java client library, is triple-licensed under the
+// Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2
+// ("GPL") and the Apache License version 2 ("ASL"). For the MPL, please see
+// LICENSE-MPL-RabbitMQ. For the GPL, please see LICENSE-GPL2.  For the ASL,
+// please see LICENSE-APACHE2.
+//
+// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+// either express or implied. See the LICENSE file for specific language governing
+// rights and limitations of this software.
+//
+// If you have any questions regarding licensing, please contact us at
+// info@rabbitmq.com.
+
+package com.rabbitmq.client.test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.MalformedFrameException;
+import com.rabbitmq.client.impl.AMQCommand;
+import com.rabbitmq.client.impl.Frame;
+import com.rabbitmq.client.impl.SocketFrameHandler;
+import java.io.DataOutputStream;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Verifies that once the negotiated frame_max is applied to the inbound reader, a frame whose total
+ * size exceeds frame_max is rejected, while a frame of exactly frame_max is still accepted.
+ */
+public class NegotiatedFrameMaxInboundTest {
+
+  private static final int FRAME_MAX = 4096;
+
+  // Same limit AMQConnection applies once frame_max is negotiated.
+  private static final int INBOUND_PAYLOAD_LIMIT = FRAME_MAX - AMQCommand.EMPTY_FRAME_SIZE + 1;
+
+  @Test
+  void frameOfExactlyFrameMaxIsAccepted() throws Exception {
+    int payloadSize = FRAME_MAX - AMQCommand.EMPTY_FRAME_SIZE;
+    Frame frame = readSingleFrame(payloadSize);
+    assertThat(frame.getPayload()).hasSize(payloadSize);
+    assertThat(frame.size()).isEqualTo(FRAME_MAX);
+  }
+
+  @Test
+  void frameLargerThanFrameMaxIsRejected() {
+    int payloadSize = FRAME_MAX - AMQCommand.EMPTY_FRAME_SIZE + 1; // total frame size = frame_max + 1
+    assertThatThrownBy(() -> readSingleFrame(payloadSize))
+        .isInstanceOf(MalformedFrameException.class);
+  }
+
+  private static Frame readSingleFrame(int payloadSize) throws Exception {
+    InetAddress loopback = InetAddress.getLoopbackAddress();
+    try (ServerSocket server = new ServerSocket(0, 1, loopback);
+        Socket client = new Socket(loopback, server.getLocalPort());
+        Socket peer = server.accept()) {
+      DataOutputStream out = new DataOutputStream(peer.getOutputStream());
+      out.writeByte(AMQP.FRAME_METHOD);
+      out.writeShort(0);
+      out.writeInt(payloadSize);
+      out.write(new byte[payloadSize]);
+      out.writeByte(AMQP.FRAME_END);
+      out.flush();
+
+      SocketFrameHandler handler = new SocketFrameHandler(client);
+      handler.setMaxInboundFramePayloadSize(INBOUND_PAYLOAD_LIMIT);
+      return handler.readFrame();
+    }
+  }
+}