CVE-2023-46120: Uncontrolled Resource Consumption (CWE-400)
value: The RabbitMQ Java client library allows Java and JVM-based applications to connect to and interact with RabbitMQ nodes. `maxBodyLebgth` was not used when receiving Message objects.  Attackers could send a very large Message causing a memory overflow and triggering an OOM Error. Users of RabbitMQ may suffer from  DoS attacks from RabbitMQ Java client which will ultimately exhaust the memory of the consumer. This vulnerability was patched in version 5.18.0.
CWE description: The product does not properly control the allocation and maintenance of a limited resource, thereby enabling an actor to influence the amount of resources consumed, eventually leading to the exhaustion of available resources.
CVSS v3 base score: 7.5
Published: 2023-10-25T18:17Z
Repository: https://github.com/rabbitmq/rabbitmq-java-client
Language: Java
Files changed: src/main/java/com/rabbitmq/client/ConnectionFactory.java, src/main/java/com/rabbitmq/client/impl/AMQChannel.java, src/main/java/com/rabbitmq/client/impl/AMQCommand.java, src/main/java/com/rabbitmq/client/impl/AMQConnection.java, src/main/java/com/rabbitmq/client/impl/AbstractFrameHandlerFactory.java, src/main/java/com/rabbitmq/client/impl/CommandAssembler.java, src/main/java/com/rabbitmq/client/impl/ConnectionParams.java, src/main/java/com/rabbitmq/client/impl/Frame.java, src/main/java/com/rabbitmq/client/impl/SocketFrameHandler.java, src/main/java/com/rabbitmq/client/impl/SocketFrameHandlerFactory.java, src/main/java/com/rabbitmq/client/impl/nio/FrameBuilder.java, src/main/java/com/rabbitmq/client/impl/nio/SocketChannelFrameHandlerFactory.java, src/main/java/com/rabbitmq/client/impl/nio/SocketChannelFrameHandlerState.java, src/main/java/com/rabbitmq/client/impl/nio/SslEngineFrameBuilder.java, src/test/java/com/rabbitmq/client/test/FrameBuilderTest.java, src/test/java/com/rabbitmq/client/test/MaxInboundMessageSizeTest.java, src/test/java/com/rabbitmq/client/test/TestUtils.java
Commit date: 2023-06-15 17:31:57 +0200
Commit message: Add max inbound message size to ConnectionFactory

To avoid OOM with a very large message.
The default value is 64 MiB.

Fixes #1062

(cherry picked from commit 9ed45fde52224ec74fc523321efdf9a157d5cfca)

Diff: diff --git a/src/main/java/com/rabbitmq/client/ConnectionFactory.java b/src/main/java/com/rabbitmq/client/ConnectionFactory.java
index 3bff5471..92566cb2 100644
--- a/src/main/java/com/rabbitmq/client/ConnectionFactory.java
+++ b/src/main/java/com/rabbitmq/client/ConnectionFactory.java
@@ -1,6 +1,6 @@
-// Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // 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,
@@ -203,10 +203,17 @@ public class ConnectionFactory implements Cloneable {
      */
     private TrafficListener trafficListener = TrafficListener.NO_OP;
 
     private CredentialsRefreshService credentialsRefreshService;
 
+    /**
+     * Maximum body size of inbound (received) messages in bytes.
+     *
+     * <p>Default value is 67,108,864 (64 MiB).
+     */
+    private int maxInboundMessageBodySize = 1_048_576 * 64;
+
     /** @return the default host to use for connections */
     public String getHost() {
         return host;
     }
 
@@ -968,15 +975,19 @@ public class ConnectionFactory implements Cloneable {
         if(nio) {
             if(this.frameHandlerFactory == null) {
                 if(this.nioParams.getNioExecutor() == null && this.nioParams.getThreadFactory() == null) {
                     this.nioParams.setThreadFactory(getThreadFactory());
                 }
-                this.frameHandlerFactory = new SocketChannelFrameHandlerFactory(connectionTimeout, nioParams, isSSL(), sslContextFactory);
+                this.frameHandlerFactory = new SocketChannelFrameHandlerFactory(
+                    connectionTimeout, nioParams, isSSL(), sslContextFactory,
+                    this.maxInboundMessageBodySize);
             }
             return this.frameHandlerFactory;
         } else {
-            return new SocketFrameHandlerFactory(connectionTimeout, socketFactory, socketConf, isSSL(), this.shutdownExecutor, sslContextFactory);
+            return new SocketFrameHandlerFactory(connectionTimeout, socketFactory,
+                socketConf, isSSL(), this.shutdownExecutor, sslContextFactory,
+                this.maxInboundMessageBodySize);
         }
 
     }
 
     /**
@@ -1271,10 +1282,11 @@ public class ConnectionFactory implements Cloneable {
         result.setConnectionRecoveryTriggeringCondition(connectionRecoveryTriggeringCondition);
         result.setTopologyRecoveryRetryHandler(topologyRecoveryRetryHandler);
         result.setRecoveredQueueNameSupplier(recoveredQueueNameSupplier);
         result.setTrafficListener(trafficListener);
         result.setCredentialsRefreshService(credentialsRefreshService);
+        result.setMaxInboundMessageBodySize(maxInboundMessageBodySize);
         return result;
     }
 
     protected AMQConnection createConnection(ConnectionParams params, FrameHandler frameHandler, MetricsCollector metricsCollector) {
         return new AMQConnection(params, frameHandler, metricsCollector);
@@ -1554,10 +1566,25 @@ public class ConnectionFactory implements Cloneable {
      */
     public int getChannelRpcTimeout() {
         return channelRpcTimeout;
     }
 
+    /**
+     * Maximum body size of inbound (received) messages in bytes.
+     *
+     * <p>Default value is 67,108,864 (64 MiB).
+     *
+     * @param maxInboundMessageBodySize the maximum size of inbound messages
+     */
+    public void setMaxInboundMessageBodySize(int maxInboundMessageBodySize) {
+        if (maxInboundMessageBodySize <= 0) {
+            throw new IllegalArgumentException("Max inbound message body size must be greater than 0: "
+                + maxInboundMessageBodySize);
+        }
+        this.maxInboundMessageBodySize = maxInboundMessageBodySize;
+    }
+
     /**
      * The factory to create SSL contexts.
      * This provides more flexibility to create {@link SSLContext}s
      * for different connections than sharing the {@link SSLContext}
      * with all the connections produced by the connection factory
diff --git a/src/main/java/com/rabbitmq/client/impl/AMQChannel.java b/src/main/java/com/rabbitmq/client/impl/AMQChannel.java
index fe64ab37..f863edfd 100644
--- a/src/main/java/com/rabbitmq/client/impl/AMQChannel.java
+++ b/src/main/java/com/rabbitmq/client/impl/AMQChannel.java
@@ -1,6 +1,6 @@
-// Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // 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,
@@ -60,11 +60,11 @@ public abstract class AMQChannel extends ShutdownNotifierComponent {
 
     /** This channel's channel number. */
     private final int _channelNumber;
 
     /** Command being assembled */
-    private AMQCommand _command = new AMQCommand();
+    private AMQCommand _command;
 
     /** The current outstanding RPC request, if any. (Could become a queue in future.) */
     private RpcWrapper _activeRpc = null;
 
     /** Whether transmission of content-bearing methods should be blocked */
@@ -74,10 +74,11 @@ public abstract class AMQChannel extends ShutdownNotifierComponent {
     protected final int _rpcTimeout;
 
     private final boolean _checkRpcResponseType;
 
     private final TrafficListener _trafficListener;
+    private final int maxInboundMessageBodySize;
 
     /**
      * Construct a channel on the given connection, with the given channel number.
      * @param connection the underlying connection for this channel
      * @param channelNumber the allocated reference number for this channel
@@ -89,10 +90,12 @@ public abstract class AMQChannel extends ShutdownNotifierComponent {
             throw new IllegalArgumentException("Continuation timeout on RPC calls cannot be less than 0");
         }
         this._rpcTimeout = connection.getChannelRpcTimeout();
         this._checkRpcResponseType = connection.willCheckRpcResponseType();
         this._trafficListener = connection.getTrafficListener();
+        this.maxInboundMessageBodySize = connection.getMaxInboundMessageBodySize();
+        this._command = new AMQCommand(this.maxInboundMessageBodySize);
     }
 
     /**
      * Public API - Retrieves this channel's channel number.
      * @return the channel number
@@ -108,11 +111,11 @@ public abstract class AMQChannel extends ShutdownNotifierComponent {
      * @throws IOException if an error is encountered
      */
     public void handleFrame(Frame frame) throws IOException {
         AMQCommand command = _command;
         if (command.handleFrame(frame)) { // a complete command has rolled off the assembly line
-            _command = new AMQCommand(); // prepare for the next one
+            _command = new AMQCommand(this.maxInboundMessageBodySize); // prepare for the next one
             handleCompleteInboundCommand(command);
         }
     }
 
     /**
diff --git a/src/main/java/com/rabbitmq/client/impl/AMQCommand.java b/src/main/java/com/rabbitmq/client/impl/AMQCommand.java
index b8cb3c97..a761e5cc 100644
--- a/src/main/java/com/rabbitmq/client/impl/AMQCommand.java
+++ b/src/main/java/com/rabbitmq/client/impl/AMQCommand.java
@@ -1,6 +1,6 @@
-// Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // 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,
@@ -42,31 +42,47 @@ public class AMQCommand implements Command {
     public static final int EMPTY_FRAME_SIZE = 8;
 
     /** The assembler for this command - synchronised on - contains all the state */
     private final CommandAssembler assembler;
 
+    AMQCommand(int maxBodyLength) {
+        this(null, null, null, maxBodyLength);
+    }
+
     /** Construct a command ready to fill in by reading frames */
     public AMQCommand() {
-        this(null, null, null);
+        this(null, null, null, Integer.MAX_VALUE);
     }
 
     /**
      * Construct a command with just a method, and without header or body.
      * @param method the wrapped method
      */
     public AMQCommand(com.rabbitmq.client.Method method) {
-        this(method, null, null);
+        this(method, null, null, Integer.MAX_VALUE);
     }
 
     /**
      * Construct a command with a specified method, header and body.
      * @param method the wrapped method
      * @param contentHeader the wrapped content header
      * @param body the message body data
      */
     public AMQCommand(com.rabbitmq.client.Method method, AMQContentHeader contentHeader, byte[] body) {
-        this.assembler = new CommandAssembler((Method) method, contentHeader, body);
+        this.assembler = new CommandAssembler((Method) method, contentHeader, body, Integer.MAX_VALUE);
+    }
+
+    /**
+     * Construct a command with a specified method, header and body.
+     * @param method the wrapped method
+     * @param contentHeader the wrapped content header
+     * @param body the message body data
+     * @param maxBodyLength the maximum size for an inbound message body
+     */
+    public AMQCommand(com.rabbitmq.client.Method method, AMQContentHeader contentHeader, byte[] body,
+                      int maxBodyLength) {
+        this.assembler = new CommandAssembler((Method) method, contentHeader, body, maxBodyLength);
     }
 
     /** Public API - {@inheritDoc} */
     @Override
     public Method getMethod() {
diff --git a/src/main/java/com/rabbitmq/client/impl/AMQConnection.java b/src/main/java/com/rabbitmq/client/impl/AMQConnection.java
index dc077080..fd1d4531 100644
--- a/src/main/java/com/rabbitmq/client/impl/AMQConnection.java
+++ b/src/main/java/com/rabbitmq/client/impl/AMQConnection.java
@@ -155,10 +155,11 @@ public class AMQConnection extends ShutdownNotifierComponent implements Connecti
     private volatile int _heartbeat = 0;
     /** Object that manages a set of channels */
     private volatile ChannelManager _channelManager;
     /** Saved server properties field from connection.start */
     private volatile Map<String, Object> _serverProperties;
+    private final int maxInboundMessageBodySize;
 
     /**
      * Protected API - respond, in the driver thread, to a ShutdownSignal.
      * @param channel the channel to disconnect
      */
@@ -242,10 +243,11 @@ public class AMQConnection extends ShutdownNotifierComponent implements Connecti
 
         this.trafficListener = params.getTrafficListener() == null ? TrafficListener.NO_OP : params.getTrafficListener();
 
         this.credentialsRefreshService = params.getCredentialsRefreshService();
 
+
         this._channel0 = createChannel0();
 
         this._channelManager = null;
 
         this._brokerInitiatedShutdown = false;
@@ -255,10 +257,11 @@ public class AMQConnection extends ShutdownNotifierComponent implements Connecti
         this.metricsCollector = metricsCollector;
 
         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() {
         return new AMQChannel(this, 0) {
             @Override public boolean processAsync(Command c) throws IOException {
@@ -1200,6 +1203,10 @@ public class AMQConnection extends ShutdownNotifierComponent implements Connecti
     }
 
     public TrafficListener getTrafficListener() {
         return trafficListener;
     }
+
+    int getMaxInboundMessageBodySize() {
+        return maxInboundMessageBodySize;
+    }
 }
diff --git a/src/main/java/com/rabbitmq/client/impl/AbstractFrameHandlerFactory.java b/src/main/java/com/rabbitmq/client/impl/AbstractFrameHandlerFactory.java
index 32eb4671..61f4c513 100644
--- a/src/main/java/com/rabbitmq/client/impl/AbstractFrameHandlerFactory.java
+++ b/src/main/java/com/rabbitmq/client/impl/AbstractFrameHandlerFactory.java
@@ -1,5 +1,20 @@
+// Copyright (c) 2016-2023 VMware, Inc. or its affiliates.  All rights reserved.
+//
+// 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.impl;
 
 import com.rabbitmq.client.SocketConfigurator;
 
 /**
@@ -8,12 +23,15 @@ import com.rabbitmq.client.SocketConfigurator;
 public abstract class AbstractFrameHandlerFactory implements FrameHandlerFactory {
 
     protected final int connectionTimeout;
     protected final SocketConfigurator configurator;
     protected final boolean ssl;
+    protected final int maxInboundMessageBodySize;
 
-    protected AbstractFrameHandlerFactory(int connectionTimeout, SocketConfigurator configurator, boolean ssl) {
+    protected AbstractFrameHandlerFactory(int connectionTimeout, SocketConfigurator configurator,
+                                          boolean ssl, int maxInboundMessageBodySize) {
         this.connectionTimeout = connectionTimeout;
         this.configurator = configurator;
         this.ssl = ssl;
+        this.maxInboundMessageBodySize = maxInboundMessageBodySize;
     }
 }
diff --git a/src/main/java/com/rabbitmq/client/impl/CommandAssembler.java b/src/main/java/com/rabbitmq/client/impl/CommandAssembler.java
index f1929486..bc548efe 100644
--- a/src/main/java/com/rabbitmq/client/impl/CommandAssembler.java
+++ b/src/main/java/com/rabbitmq/client/impl/CommandAssembler.java
@@ -1,6 +1,6 @@
-// Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // 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,
@@ -19,10 +19,11 @@ import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.UnexpectedFrameError;
+import static java.lang.String.format;
 
 /**
  * Class responsible for piecing together a command from a series of {@link Frame}s.
  * <p/><b>Concurrency</b><br/>
  * This class is thread-safe, since all methods are synchronised. Callers should not
@@ -50,16 +51,20 @@ final class CommandAssembler {
     private int bodyLength;
 
     /** No bytes of content body not yet accumulated */
     private long remainingBodyBytes;
 
-    public CommandAssembler(Method method, AMQContentHeader contentHeader, byte[] body) {
+    private final int maxBodyLength;
+
+    public CommandAssembler(Method method, AMQContentHeader contentHeader, byte[] body,
+                            int maxBodyLength) {
         this.method = method;
         this.contentHeader = contentHeader;
-        this.bodyN = new ArrayList<byte[]>(2);
+        this.bodyN = new ArrayList<>(2);
         this.bodyLength = 0;
         this.remainingBodyBytes = 0;
+        this.maxBodyLength = maxBodyLength;
         appendBodyFragment(body);
         if (method == null) {
             this.state = CAState.EXPECTING_METHOD;
         } else if (contentHeader == null) {
             this.state = method.hasContent() ? CAState.EXPECTING_CONTENT_HEADER : CAState.COMPLETE;
@@ -97,11 +102,18 @@ final class CommandAssembler {
     }
 
     private void consumeHeaderFrame(Frame f) throws IOException {
         if (f.type == AMQP.FRAME_HEADER) {
             this.contentHeader = AMQImpl.readContentHeaderFrom(f.getInputStream());
-            this.remainingBodyBytes = this.contentHeader.getBodySize();
+            long bodySize = this.contentHeader.getBodySize();
+            if (bodySize >= this.maxBodyLength) {
+                throw new IllegalStateException(format(
+                    "Message body is too large (%d), maximum size is %d",
+                    bodySize, this.maxBodyLength
+                ));
+            }
+            this.remainingBodyBytes = bodySize;
             updateContentBodyState();
         } else {
             throw new UnexpectedFrameError(f, AMQP.FRAME_HEADER);
         }
     }
diff --git a/src/main/java/com/rabbitmq/client/impl/ConnectionParams.java b/src/main/java/com/rabbitmq/client/impl/ConnectionParams.java
index 24d20040..9fe6e855 100644
--- a/src/main/java/com/rabbitmq/client/impl/ConnectionParams.java
+++ b/src/main/java/com/rabbitmq/client/impl/ConnectionParams.java
@@ -1,6 +1,6 @@
-// Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // 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,
@@ -62,10 +62,12 @@ public class ConnectionParams {
 
     private TrafficListener trafficListener;
 
     private CredentialsRefreshService credentialsRefreshService;
 
+    private int maxInboundMessageBodySize;
+
     public ConnectionParams() {}
 
     public CredentialsProvider getCredentialsProvider() {
         return credentialsProvider;
     }
@@ -295,6 +297,14 @@ public class ConnectionParams {
     }
 
     public CredentialsRefreshService getCredentialsRefreshService() {
         return credentialsRefreshService;
     }
+
+    public int getMaxInboundMessageBodySize() {
+        return maxInboundMessageBodySize;
+    }
+
+    public void setMaxInboundMessageBodySize(int maxInboundMessageBodySize) {
+        this.maxInboundMessageBodySize = maxInboundMessageBodySize;
+    }
 }
diff --git a/src/main/java/com/rabbitmq/client/impl/Frame.java b/src/main/java/com/rabbitmq/client/impl/Frame.java
index ce7531cc..3608b37f 100644
--- a/src/main/java/com/rabbitmq/client/impl/Frame.java
+++ b/src/main/java/com/rabbitmq/client/impl/Frame.java
@@ -1,6 +1,6 @@
-// Copyright (c) 2007-2022 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // 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,
@@ -23,10 +23,11 @@ import java.io.*;
 import java.math.BigDecimal;
 import java.net.SocketTimeoutException;
 import java.util.Date;
 import java.util.List;
 import java.util.Map;
+import static java.lang.String.format;
 
 /**
  * Represents an AMQP wire-protocol frame, with frame type, channel number, and payload bytes.
  * TODO: make state private
  */
@@ -80,11 +81,11 @@ public class Frame {
      * Protected API - Factory method to instantiate a Frame by reading an
      * AMQP-wire-protocol frame from the given input stream.
      *
      * @return a new Frame if we read a frame successfully, otherwise null
      */
-    public static Frame readFrom(DataInputStream is) throws IOException {
+    public static Frame readFrom(DataInputStream is, int maxPayloadSize) throws IOException {
         int type;
         int channel;
 
         try {
             type = is.readUnsignedByte();
@@ -106,10 +107,16 @@ public class Frame {
             protocolVersionMismatch(is);
         }
 
         channel = is.readUnsignedShort();
         int payloadSize = is.readInt();
+        if (payloadSize >= maxPayloadSize) {
+            throw new IllegalStateException(format(
+                "Frame body is too large (%d), maximum size is %d",
+                payloadSize, maxPayloadSize
+            ));
+        }
         byte[] payload = new byte[payloadSize];
         is.readFully(payload);
 
         int frameEndMarker = is.readUnsignedByte();
         if (frameEndMarker != AMQP.FRAME_END) {
diff --git a/src/main/java/com/rabbitmq/client/impl/SocketFrameHandler.java b/src/main/java/com/rabbitmq/client/impl/SocketFrameHandler.java
index 6efd6f83..5048100a 100644
--- a/src/main/java/com/rabbitmq/client/impl/SocketFrameHandler.java
+++ b/src/main/java/com/rabbitmq/client/impl/SocketFrameHandler.java
@@ -1,6 +1,6 @@
-// Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // 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,
@@ -50,26 +50,30 @@ public class SocketFrameHandler implements FrameHandler {
     private final DataInputStream _inputStream;
 
     /** Socket's outputstream - data to the broker - synchronized on */
     private final DataOutputStream _outputStream;
 
+    private final int maxInboundMessageBodySize;
+
     /** Time to linger before closing the socket forcefully. */
     public static final int SOCKET_CLOSING_TIMEOUT = 1;
 
     /**
      * @param socket the socket to use
      */
     public SocketFrameHandler(Socket socket) throws IOException {
-        this(socket, null);
+        this(socket, null, Integer.MAX_VALUE);
     }
 
     /**
      * @param socket the socket to use
      */
-    public SocketFrameHandler(Socket socket, ExecutorService shutdownExecutor) throws IOException {
+    public SocketFrameHandler(Socket socket, ExecutorService shutdownExecutor,
+                              int maxInboundMessageBodySize) throws IOException {
         _socket = socket;
         _shutdownExecutor = shutdownExecutor;
+        this.maxInboundMessageBodySize = maxInboundMessageBodySize;
 
         _inputStream = new DataInputStream(new BufferedInputStream(socket.getInputStream()));
         _outputStream = new DataOutputStream(new BufferedOutputStream(socket.getOutputStream()));
     }
 
@@ -179,11 +183,11 @@ public class SocketFrameHandler implements FrameHandler {
     }
 
     @Override
     public Frame readFrame() throws IOException {
         synchronized (_inputStream) {
-            return Frame.readFrom(_inputStream);
+            return Frame.readFrom(_inputStream, this.maxInboundMessageBodySize);
         }
     }
 
     @Override
     public void writeFrame(Frame frame) throws IOException {
diff --git a/src/main/java/com/rabbitmq/client/impl/SocketFrameHandlerFactory.java b/src/main/java/com/rabbitmq/client/impl/SocketFrameHandlerFactory.java
index d3fdfa9e..40818c46 100644
--- a/src/main/java/com/rabbitmq/client/impl/SocketFrameHandlerFactory.java
+++ b/src/main/java/com/rabbitmq/client/impl/SocketFrameHandlerFactory.java
@@ -1,6 +1,6 @@
-// Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // 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,
@@ -36,16 +36,18 @@ public class SocketFrameHandlerFactory extends AbstractFrameHandlerFactory {
         this(connectionTimeout, socketFactory, configurator, ssl, null);
     }
 
     public SocketFrameHandlerFactory(int connectionTimeout, SocketFactory socketFactory, SocketConfigurator configurator,
                                      boolean ssl, ExecutorService shutdownExecutor) {
-        this(connectionTimeout, socketFactory, configurator, ssl, shutdownExecutor, null);
+        this(connectionTimeout, socketFactory, configurator, ssl, shutdownExecutor, null,
+             Integer.MAX_VALUE);
     }
 
     public SocketFrameHandlerFactory(int connectionTimeout, SocketFactory socketFactory, SocketConfigurator configurator,
-                                     boolean ssl, ExecutorService shutdownExecutor, SslContextFactory sslContextFactory) {
-        super(connectionTimeout, configurator, ssl);
+                                     boolean ssl, ExecutorService shutdownExecutor, SslContextFactory sslContextFactory,
+                                     int maxInboundMessageBodySize) {
+        super(connectionTimeout, configurator, ssl, maxInboundMessageBodySize);
         this.socketFactory = socketFactory;
         this.shutdownExecutor = shutdownExecutor;
         this.sslContextFactory = sslContextFactory;
     }
 
@@ -77,11 +79,11 @@ public class SocketFrameHandlerFactory extends AbstractFrameHandlerFactory {
         }
     }
 
     public FrameHandler create(Socket sock) throws IOException
     {
-        return new SocketFrameHandler(sock, this.shutdownExecutor);
+        return new SocketFrameHandler(sock, this.shutdownExecutor, this.maxInboundMessageBodySize);
     }
 
     private static void quietTrySocketClose(Socket socket) {
         if (socket != null)
             try { socket.close(); } catch (Exception _e) {/*ignore exceptions*/}
diff --git a/src/main/java/com/rabbitmq/client/impl/nio/FrameBuilder.java b/src/main/java/com/rabbitmq/client/impl/nio/FrameBuilder.java
index 813e328b..a14bbaf1 100644
--- a/src/main/java/com/rabbitmq/client/impl/nio/FrameBuilder.java
+++ b/src/main/java/com/rabbitmq/client/impl/nio/FrameBuilder.java
@@ -1,6 +1,6 @@
-// Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // 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,
@@ -21,10 +21,11 @@ import com.rabbitmq.client.impl.Frame;
 
 import java.io.DataInputStream;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.ReadableByteChannel;
+import static java.lang.String.format;
 
 /**
  * Class to create AMQP frames from a {@link ReadableByteChannel}.
  * Supports partial frames: a frame can be read in several attempts
  * from the {@link NioLoop}. This can happen when the channel won't
@@ -41,33 +42,35 @@ public class FrameBuilder {
     private static final int PAYLOAD_OFFSET = 1 /* type */ + 2 /* channel */ + 4 /* payload size */;
 
     protected final ReadableByteChannel channel;
 
     protected final ByteBuffer applicationBuffer;
+    private final int maxPayloadSize;
     // 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 int frameType;
     private int frameChannel;
     private byte[] framePayload;
     private int bytesRead = 0;
 
-    public FrameBuilder(ReadableByteChannel channel, ByteBuffer buffer) {
+    public FrameBuilder(ReadableByteChannel channel, ByteBuffer buffer, int maxPayloadSize) {
         this.channel = channel;
         this.applicationBuffer = buffer;
+        this.maxPayloadSize = maxPayloadSize;
     }
 
     /**
      * Read a frame from the network.
      * 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)
+     * @see Frame#readFrom(DataInputStream, int)
      */
     public Frame readFrame() throws IOException {
         while (somethingToRead()) {
             if (bytesRead == 0) {
                 // type
@@ -91,10 +94,16 @@ public class FrameBuilder {
                 // payload size 3/4
                 frameBuffer[2] = readFromBuffer();
             } else if (bytesRead == 6) {
                 // payload size 4/4
                 int framePayloadSize = (frameBuffer[0] << 24) + (frameBuffer[1] << 16) + (frameBuffer[2] << 8) + readFromBuffer();
+                if (framePayloadSize >= maxPayloadSize) {
+                    throw new IllegalStateException(format(
+                        "Frame body is too large (%d), maximum size is %d",
+                        framePayloadSize, maxPayloadSize
+                    ));
+                }
                 framePayload = new byte[framePayloadSize];
             } else if (bytesRead >= PAYLOAD_OFFSET && bytesRead < framePayload.length + PAYLOAD_OFFSET) {
                 framePayload[bytesRead - PAYLOAD_OFFSET] = (byte) readFromBuffer();
             } else if (bytesRead == framePayload.length + PAYLOAD_OFFSET) {
                 int frameEndMarker = readFromBuffer();
diff --git a/src/main/java/com/rabbitmq/client/impl/nio/SocketChannelFrameHandlerFactory.java b/src/main/java/com/rabbitmq/client/impl/nio/SocketChannelFrameHandlerFactory.java
index 2cab2066..ed0a4f20 100644
--- a/src/main/java/com/rabbitmq/client/impl/nio/SocketChannelFrameHandlerFactory.java
+++ b/src/main/java/com/rabbitmq/client/impl/nio/SocketChannelFrameHandlerFactory.java
@@ -1,6 +1,6 @@
-// Copyright (c) 2007-2022 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // 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,
@@ -56,12 +56,14 @@ public class SocketChannelFrameHandlerFactory extends AbstractFrameHandlerFactor
 
     private final AtomicLong globalConnectionCount = new AtomicLong();
 
     private final List<NioLoopContext> nioLoopContexts;
 
-    public SocketChannelFrameHandlerFactory(int connectionTimeout, NioParams nioParams, boolean ssl, SslContextFactory sslContextFactory) {
-        super(connectionTimeout, null, ssl);
+    public SocketChannelFrameHandlerFactory(int connectionTimeout, NioParams nioParams, boolean ssl,
+                                            SslContextFactory sslContextFactory,
+                                            int maxInboundMessageBodySize) {
+        super(connectionTimeout, null, ssl, maxInboundMessageBodySize);
         this.nioParams = new NioParams(nioParams);
         this.sslContextFactory = sslContextFactory;
         this.nioLoopContexts = new ArrayList<>(this.nioParams.getNbIoThreads());
         for (int i = 0; i < this.nioParams.getNbIoThreads(); i++) {
             this.nioLoopContexts.add(new NioLoopContext(this, this.nioParams));
@@ -132,11 +134,12 @@ public class SocketChannelFrameHandlerFactory extends AbstractFrameHandlerFactor
                 nioLoopContext.initStateIfNecessary();
                 SocketChannelFrameHandlerState state = new SocketChannelFrameHandlerState(
                     channel,
                     nioLoopContext,
                     nioParams,
-                    sslEngine
+                    sslEngine,
+                    this.maxInboundMessageBodySize
                 );
                 state.startReading();
                 SocketChannelFrameHandler frameHandler = new SocketChannelFrameHandler(state);
                 return frameHandler;
             } finally {
diff --git a/src/main/java/com/rabbitmq/client/impl/nio/SocketChannelFrameHandlerState.java b/src/main/java/com/rabbitmq/client/impl/nio/SocketChannelFrameHandlerState.java
index 4f1e4dc8..b1a391fd 100644
--- a/src/main/java/com/rabbitmq/client/impl/nio/SocketChannelFrameHandlerState.java
+++ b/src/main/java/com/rabbitmq/client/impl/nio/SocketChannelFrameHandlerState.java
@@ -1,6 +1,6 @@
-// Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // 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,
@@ -69,11 +69,13 @@ public class SocketChannelFrameHandlerState {
 
     final DataOutputStream outputStream;
 
     final FrameBuilder frameBuilder;
 
-    public SocketChannelFrameHandlerState(SocketChannel channel, NioLoopContext nioLoopsState, NioParams nioParams, SSLEngine sslEngine) {
+    public SocketChannelFrameHandlerState(SocketChannel channel, NioLoopContext nioLoopsState,
+                                          NioParams nioParams, SSLEngine sslEngine,
+                                          int maxFramePayloadSize) {
         this.channel = channel;
         this.readSelectorState = nioLoopsState.readSelectorState;
         this.writeSelectorState = nioLoopsState.writeSelectorState;
 
         NioContext nioContext = new NioContext(nioParams, sslEngine);
@@ -92,11 +94,11 @@ public class SocketChannelFrameHandlerState {
 
             this.outputStream = new DataOutputStream(
                 new ByteBufferOutputStream(channel, plainOut)
             );
 
-            this.frameBuilder = new FrameBuilder(channel, plainIn);
+            this.frameBuilder = new FrameBuilder(channel, plainIn, maxFramePayloadSize);
 
         } else {
             this.ssl = true;
             this.plainOut = nioParams.getByteBufferFactory().createWriteBuffer(nioContext);
             this.cipherOut = nioParams.getByteBufferFactory().createEncryptedWriteBuffer(nioContext);
@@ -104,11 +106,12 @@ public class SocketChannelFrameHandlerState {
             this.cipherIn = nioParams.getByteBufferFactory().createEncryptedReadBuffer(nioContext);
 
             this.outputStream = new DataOutputStream(
                 new SslEngineByteBufferOutputStream(sslEngine, plainOut, cipherOut, channel)
             );
-            this.frameBuilder = new SslEngineFrameBuilder(sslEngine, plainIn, cipherIn, channel);
+            this.frameBuilder = new SslEngineFrameBuilder(sslEngine, plainIn,
+                cipherIn, channel, maxFramePayloadSize);
         }
 
     }
 
     public SocketChannel getChannel() {
diff --git a/src/main/java/com/rabbitmq/client/impl/nio/SslEngineFrameBuilder.java b/src/main/java/com/rabbitmq/client/impl/nio/SslEngineFrameBuilder.java
index 8b6ecaf8..48072988 100644
--- a/src/main/java/com/rabbitmq/client/impl/nio/SslEngineFrameBuilder.java
+++ b/src/main/java/com/rabbitmq/client/impl/nio/SslEngineFrameBuilder.java
@@ -1,6 +1,6 @@
-// Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // 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,
@@ -33,12 +33,14 @@ public class SslEngineFrameBuilder extends FrameBuilder {
 
     private final ByteBuffer cipherBuffer;
 
     private boolean isUnderflowHandlingEnabled = false;
 
-    public SslEngineFrameBuilder(SSLEngine sslEngine, ByteBuffer plainIn, ByteBuffer cipherIn, ReadableByteChannel channel) {
-        super(channel, plainIn);
+    public SslEngineFrameBuilder(SSLEngine sslEngine, ByteBuffer plainIn,
+                                 ByteBuffer cipherIn, ReadableByteChannel channel,
+                                 int maxPayloadSize) {
+        super(channel, plainIn, maxPayloadSize);
         this.sslEngine = sslEngine;
         this.cipherBuffer = cipherIn;
     }
 
     @Override
diff --git a/src/test/java/com/rabbitmq/client/test/FrameBuilderTest.java b/src/test/java/com/rabbitmq/client/test/FrameBuilderTest.java
index 3e8b462f..5c227ae1 100644
--- a/src/test/java/com/rabbitmq/client/test/FrameBuilderTest.java
+++ b/src/test/java/com/rabbitmq/client/test/FrameBuilderTest.java
@@ -57,11 +57,11 @@ 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() });
-        builder = new FrameBuilder(channel, buffer);
+        builder = new FrameBuilder(channel, buffer, Integer.MAX_VALUE);
         Frame frame = builder.readFrame();
         assertThat(frame).isNotNull();
         assertThat(frame.type).isEqualTo(1);
         assertThat(frame.channel).isEqualTo(0);
         assertThat(frame.getPayload()).hasSize(3);
@@ -76,11 +76,11 @@ public class FrameBuilderTest {
             for (int j = 0; j < frameContent.length; j++) {
                 frames[i * frameContent.length + j] = frameContent[j];
             }
         }
         buffer = ByteBuffer.wrap(frames);
-        builder = new FrameBuilder(channel, buffer);
+        builder = new FrameBuilder(channel, buffer, Integer.MAX_VALUE);
         int frameCount = 0;
         Frame frame;
         while ((frame = builder.readFrame()) != null) {
             assertThat(frame).isNotNull();
             assertThat(frame.type).isEqualTo(1);
@@ -92,11 +92,11 @@ public class FrameBuilderTest {
     }
 
     @Test
     public void buildFrameInSeveralCalls() throws IOException {
         buffer = ByteBuffer.wrap(new byte[] { 1, 0, 0, 0, 0, 0, 3, 1, 2 });
-        builder = new FrameBuilder(channel, buffer);
+        builder = new FrameBuilder(channel, buffer, Integer.MAX_VALUE);
         Frame frame = builder.readFrame();
         assertThat(frame).isNull();
 
         buffer.clear();
         buffer.put(b(3)).put(end());
@@ -129,11 +129,11 @@ public class FrameBuilderTest {
             "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]);
+            builder = new FrameBuilder(channel, buffers[i], Integer.MAX_VALUE);
             try {
                 builder.readFrame();
                 fail("protocol header not correct, exception should have been thrown");
             } catch (MalformedFrameException e) {
                 assertThat(e.getMessage()).isEqualTo(messages[i]);
diff --git a/src/test/java/com/rabbitmq/client/test/MaxInboundMessageSizeTest.java b/src/test/java/com/rabbitmq/client/test/MaxInboundMessageSizeTest.java
new file mode 100644
index 00000000..da534d9c
--- /dev/null
+++ b/src/test/java/com/rabbitmq/client/test/MaxInboundMessageSizeTest.java
@@ -0,0 +1,97 @@
+// Copyright (c) 2023 VMware, Inc. or its affiliates.  All rights reserved.
+//
+// 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 com.rabbitmq.client.test.TestUtils.LatchConditions.completed;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.rabbitmq.client.*;
+import java.io.IOException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+public class MaxInboundMessageSizeTest extends BrokerTestCase {
+
+  String q;
+
+  private static void safeClose(Connection c) {
+    try {
+      c.close();
+    } catch (Exception e) {
+      // OK
+    }
+  }
+
+  @Override
+  protected void createResources() throws IOException, TimeoutException {
+    q = generateQueueName();
+    declareTransientQueue(q);
+    super.createResources();
+  }
+
+  @CsvSource({
+    "20000,5000,true",
+    "20000,100000,true",
+    "20000,5000,false",
+    "20000,100000,false",
+  })
+  @ParameterizedTest
+  void maxInboundMessageSizeMustBeEnforced(int maxMessageSize, int frameMax, boolean basicGet)
+      throws Exception {
+    ConnectionFactory cf = newConnectionFactory();
+    cf.setMaxInboundMessageBodySize(maxMessageSize);
+    cf.setRequestedFrameMax(frameMax);
+    Connection c = cf.newConnection();
+    try {
+      Channel ch = c.createChannel();
+      ch.confirmSelect();
+      byte[] body = new byte[maxMessageSize * 2];
+      ch.basicPublish("", q, null, body);
+      ch.waitForConfirmsOrDie();
+      AtomicReference<Throwable> exception = new AtomicReference<>();
+      CountDownLatch errorLatch = new CountDownLatch(1);
+      ch.addShutdownListener(
+          cause -> {
+            exception.set(cause.getCause());
+            errorLatch.countDown();
+          });
+      if (basicGet) {
+        try {
+          ch.basicGet(q, true);
+        } catch (Exception e) {
+          // OK for basicGet
+        }
+      } else {
+        ch.basicConsume(q, new DefaultConsumer(ch));
+      }
+      assertThat(errorLatch).is(completed());
+      assertThat(exception.get())
+          .isInstanceOf(IllegalStateException.class)
+          .hasMessageContaining("Message body is too large");
+    } finally {
+      safeClose(c);
+    }
+  }
+
+  @Override
+  protected void releaseResources() throws IOException {
+    deleteQueue(q);
+    super.releaseResources();
+  }
+}
diff --git a/src/test/java/com/rabbitmq/client/test/TestUtils.java b/src/test/java/com/rabbitmq/client/test/TestUtils.java
index cbb2885f..ec495bcb 100644
--- a/src/test/java/com/rabbitmq/client/test/TestUtils.java
+++ b/src/test/java/com/rabbitmq/client/test/TestUtils.java
@@ -24,15 +24,15 @@ import java.lang.annotation.ElementType;
 import java.lang.annotation.Retention;
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 import java.util.function.Function;
 import org.assertj.core.api.Assertions;
+import org.assertj.core.api.Condition;
 import org.junit.jupiter.api.extension.ConditionEvaluationResult;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.api.extension.ExtensionContext;
 import org.junit.jupiter.api.extension.ExtensionContext.Namespace;
-import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.net.ServerSocket;
 import java.time.Duration;
@@ -45,12 +45,10 @@ import java.util.concurrent.TimeoutException;
 
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 public class TestUtils {
 
-    private static final Logger LOGGER = LoggerFactory.getLogger(TestUtils.class);
-
     public static final boolean USE_NIO = System.getProperty("use.nio") != null;
 
     public static ConnectionFactory connectionFactory() {
         ConnectionFactory connectionFactory = new ConnectionFactory();
         if (USE_NIO) {
@@ -301,10 +299,26 @@ public class TestUtils {
 
         R apply(T t) throws Exception;
 
     }
 
+    public static class LatchConditions {
+
+        static Condition<CountDownLatch> completed() {
+            return new Condition<>(
+                countDownLatch-> {
+                    try {
+                        return countDownLatch.await(10, TimeUnit.SECONDS);
+                    } catch (InterruptedException e) {
+                        throw new RuntimeException(e);
+                    }
+                },
+                "Latch did not complete in 10 seconds");
+        }
+
+    }
+
     public static boolean basicGetBasicConsume(Connection connection, String queue, final CountDownLatch latch, int msgSize)
         throws Exception {
         Channel channel = connection.createChannel();
         channel.queueDeclare(queue, false, true, false, null);
         channel.queuePurge(queue);

Vulnerable code: // Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
                this.frameHandlerFactory = new SocketChannelFrameHandlerFactory(connectionTimeout, nioParams, isSSL(), sslContextFactory);
            return new SocketFrameHandlerFactory(connectionTimeout, socketFactory, socketConf, isSSL(), this.shutdownExecutor, sslContextFactory);
// Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
    private AMQCommand _command = new AMQCommand();
            _command = new AMQCommand(); // prepare for the next one
// Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
        this(null, null, null);
        this(method, null, null);
        this.assembler = new CommandAssembler((Method) method, contentHeader, body);
    protected AbstractFrameHandlerFactory(int connectionTimeout, SocketConfigurator configurator, boolean ssl) {
// Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
    public CommandAssembler(Method method, AMQContentHeader contentHeader, byte[] body) {
        this.bodyN = new ArrayList<byte[]>(2);
            this.remainingBodyBytes = this.contentHeader.getBodySize();
// Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
// Copyright (c) 2007-2022 VMware, Inc. or its affiliates.  All rights reserved.
    public static Frame readFrom(DataInputStream is) throws IOException {
// Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
        this(socket, null);
    public SocketFrameHandler(Socket socket, ExecutorService shutdownExecutor) throws IOException {
            return Frame.readFrom(_inputStream);
// Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
        this(connectionTimeout, socketFactory, configurator, ssl, shutdownExecutor, null);
                                     boolean ssl, ExecutorService shutdownExecutor, SslContextFactory sslContextFactory) {
        super(connectionTimeout, configurator, ssl);
        return new SocketFrameHandler(sock, this.shutdownExecutor);
// Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
    public FrameBuilder(ReadableByteChannel channel, ByteBuffer buffer) {
     * @see Frame#readFrom(DataInputStream)
// Copyright (c) 2007-2022 VMware, Inc. or its affiliates.  All rights reserved.
    public SocketChannelFrameHandlerFactory(int connectionTimeout, NioParams nioParams, boolean ssl, SslContextFactory sslContextFactory) {
        super(connectionTimeout, null, ssl);
                    sslEngine
// Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
    public SocketChannelFrameHandlerState(SocketChannel channel, NioLoopContext nioLoopsState, NioParams nioParams, SSLEngine sslEngine) {
            this.frameBuilder = new FrameBuilder(channel, plainIn);
            this.frameBuilder = new SslEngineFrameBuilder(sslEngine, plainIn, cipherIn, channel);
// Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
    public SslEngineFrameBuilder(SSLEngine sslEngine, ByteBuffer plainIn, ByteBuffer cipherIn, ReadableByteChannel channel) {
        super(channel, plainIn);
        builder = new FrameBuilder(channel, buffer);
        builder = new FrameBuilder(channel, buffer);
        builder = new FrameBuilder(channel, buffer);
            builder = new FrameBuilder(channel, buffers[i]);
import org.slf4j.Logger;
    private static final Logger LOGGER = LoggerFactory.getLogger(TestUtils.class);

Fixed code: // Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
    /**
     * Maximum body size of inbound (received) messages in bytes.
     *
     * <p>Default value is 67,108,864 (64 MiB).
     */
    private int maxInboundMessageBodySize = 1_048_576 * 64;

                this.frameHandlerFactory = new SocketChannelFrameHandlerFactory(
                    connectionTimeout, nioParams, isSSL(), sslContextFactory,
                    this.maxInboundMessageBodySize);
            return new SocketFrameHandlerFactory(connectionTimeout, socketFactory,
                socketConf, isSSL(), this.shutdownExecutor, sslContextFactory,
                this.maxInboundMessageBodySize);
        result.setMaxInboundMessageBodySize(maxInboundMessageBodySize);
    /**
     * Maximum body size of inbound (received) messages in bytes.
     *
     * <p>Default value is 67,108,864 (64 MiB).
     *
     * @param maxInboundMessageBodySize the maximum size of inbound messages
     */
    public void setMaxInboundMessageBodySize(int maxInboundMessageBodySize) {
        if (maxInboundMessageBodySize <= 0) {
            throw new IllegalArgumentException("Max inbound message body size must be greater than 0: "
                + maxInboundMessageBodySize);
        }
        this.maxInboundMessageBodySize = maxInboundMessageBodySize;
    }

// Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
    private AMQCommand _command;
    private final int maxInboundMessageBodySize;
        this.maxInboundMessageBodySize = connection.getMaxInboundMessageBodySize();
        this._command = new AMQCommand(this.maxInboundMessageBodySize);
            _command = new AMQCommand(this.maxInboundMessageBodySize); // prepare for the next one
// Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
    AMQCommand(int maxBodyLength) {
        this(null, null, null, maxBodyLength);
    }

        this(null, null, null, Integer.MAX_VALUE);
        this(method, null, null, Integer.MAX_VALUE);
        this.assembler = new CommandAssembler((Method) method, contentHeader, body, Integer.MAX_VALUE);
    }

    /**
     * Construct a command with a specified method, header and body.
     * @param method the wrapped method
     * @param contentHeader the wrapped content header
     * @param body the message body data
     * @param maxBodyLength the maximum size for an inbound message body
     */
    public AMQCommand(com.rabbitmq.client.Method method, AMQContentHeader contentHeader, byte[] body,
                      int maxBodyLength) {
        this.assembler = new CommandAssembler((Method) method, contentHeader, body, maxBodyLength);
    private final int maxInboundMessageBodySize;

        this.maxInboundMessageBodySize = params.getMaxInboundMessageBodySize();

    int getMaxInboundMessageBodySize() {
        return maxInboundMessageBodySize;
    }
// Copyright (c) 2016-2023 VMware, Inc. or its affiliates.  All rights reserved.
//
// 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.

    protected final int maxInboundMessageBodySize;
    protected AbstractFrameHandlerFactory(int connectionTimeout, SocketConfigurator configurator,
                                          boolean ssl, int maxInboundMessageBodySize) {
        this.maxInboundMessageBodySize = maxInboundMessageBodySize;
// Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
import static java.lang.String.format;
    private final int maxBodyLength;

    public CommandAssembler(Method method, AMQContentHeader contentHeader, byte[] body,
                            int maxBodyLength) {
        this.bodyN = new ArrayList<>(2);
        this.maxBodyLength = maxBodyLength;
            long bodySize = this.contentHeader.getBodySize();
            if (bodySize >= this.maxBodyLength) {
                throw new IllegalStateException(format(
                    "Message body is too large (%d), maximum size is %d",
                    bodySize, this.maxBodyLength
                ));
            }
            this.remainingBodyBytes = bodySize;
// Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
    private int maxInboundMessageBodySize;


    public int getMaxInboundMessageBodySize() {
        return maxInboundMessageBodySize;
    }

    public void setMaxInboundMessageBodySize(int maxInboundMessageBodySize) {
        this.maxInboundMessageBodySize = maxInboundMessageBodySize;
    }
// Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
import static java.lang.String.format;
    public static Frame readFrom(DataInputStream is, int maxPayloadSize) throws IOException {
        if (payloadSize >= maxPayloadSize) {
            throw new IllegalStateException(format(
                "Frame body is too large (%d), maximum size is %d",
                payloadSize, maxPayloadSize
            ));
        }
// Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
    private final int maxInboundMessageBodySize;

        this(socket, null, Integer.MAX_VALUE);
    public SocketFrameHandler(Socket socket, ExecutorService shutdownExecutor,
                              int maxInboundMessageBodySize) throws IOException {
        this.maxInboundMessageBodySize = maxInboundMessageBodySize;
            return Frame.readFrom(_inputStream, this.maxInboundMessageBodySize);
// Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
        this(connectionTimeout, socketFactory, configurator, ssl, shutdownExecutor, null,
             Integer.MAX_VALUE);
                                     boolean ssl, ExecutorService shutdownExecutor, SslContextFactory sslContextFactory,
                                     int maxInboundMessageBodySize) {
        super(connectionTimeout, configurator, ssl, maxInboundMessageBodySize);
        return new SocketFrameHandler(sock, this.shutdownExecutor, this.maxInboundMessageBodySize);
// Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
import static java.lang.String.format;
    private final int maxPayloadSize;
    public FrameBuilder(ReadableByteChannel channel, ByteBuffer buffer, int maxPayloadSize) {
        this.maxPayloadSize = maxPayloadSize;
     * @see Frame#readFrom(DataInputStream, int)
                if (framePayloadSize >= maxPayloadSize) {
                    throw new IllegalStateException(format(
                        "Frame body is too large (%d), maximum size is %d",
                        framePayloadSize, maxPayloadSize
                    ));
                }
// Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
    public SocketChannelFrameHandlerFactory(int connectionTimeout, NioParams nioParams, boolean ssl,
                                            SslContextFactory sslContextFactory,
                                            int maxInboundMessageBodySize) {
        super(connectionTimeout, null, ssl, maxInboundMessageBodySize);
                    sslEngine,
                    this.maxInboundMessageBodySize
// Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
    public SocketChannelFrameHandlerState(SocketChannel channel, NioLoopContext nioLoopsState,
                                          NioParams nioParams, SSLEngine sslEngine,
                                          int maxFramePayloadSize) {
            this.frameBuilder = new FrameBuilder(channel, plainIn, maxFramePayloadSize);
            this.frameBuilder = new SslEngineFrameBuilder(sslEngine, plainIn,
                cipherIn, channel, maxFramePayloadSize);
// Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
    public SslEngineFrameBuilder(SSLEngine sslEngine, ByteBuffer plainIn,
                                 ByteBuffer cipherIn, ReadableByteChannel channel,
                                 int maxPayloadSize) {
        super(channel, plainIn, maxPayloadSize);
        builder = new FrameBuilder(channel, buffer, Integer.MAX_VALUE);
        builder = new FrameBuilder(channel, buffer, Integer.MAX_VALUE);
        builder = new FrameBuilder(channel, buffer, Integer.MAX_VALUE);
            builder = new FrameBuilder(channel, buffers[i], Integer.MAX_VALUE);
// Copyright (c) 2023 VMware, Inc. or its affiliates.  All rights reserved.
//
// 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 com.rabbitmq.client.test.TestUtils.LatchConditions.completed;
import static org.assertj.core.api.Assertions.assertThat;

import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;

public class MaxInboundMessageSizeTest extends BrokerTestCase {

  String q;

  private static void safeClose(Connection c) {
    try {
      c.close();
    } catch (Exception e) {
      // OK
    }
  }

  @Override
  protected void createResources() throws IOException, TimeoutException {
    q = generateQueueName();
    declareTransientQueue(q);
    super.createResources();
  }

  @CsvSource({
    "20000,5000,true",
    "20000,100000,true",
    "20000,5000,false",
    "20000,100000,false",
  })
  @ParameterizedTest
  void maxInboundMessageSizeMustBeEnforced(int maxMessageSize, int frameMax, boolean basicGet)
      throws Exception {
    ConnectionFactory cf = newConnectionFactory();
    cf.setMaxInboundMessageBodySize(maxMessageSize);
    cf.setRequestedFrameMax(frameMax);
    Connection c = cf.newConnection();
    try {
      Channel ch = c.createChannel();
      ch.confirmSelect();
      byte[] body = new byte[maxMessageSize * 2];
      ch.basicPublish("", q, null, body);
      ch.waitForConfirmsOrDie();
      AtomicReference<Throwable> exception = new AtomicReference<>();
      CountDownLatch errorLatch = new CountDownLatch(1);
      ch.addShutdownListener(
          cause -> {
            exception.set(cause.getCause());
            errorLatch.countDown();
          });
      if (basicGet) {
        try {
          ch.basicGet(q, true);
        } catch (Exception e) {
          // OK for basicGet
        }
      } else {
        ch.basicConsume(q, new DefaultConsumer(ch));
      }
      assertThat(errorLatch).is(completed());
      assertThat(exception.get())
          .isInstanceOf(IllegalStateException.class)
          .hasMessageContaining("Message body is too large");
    } finally {
      safeClose(c);
    }
  }

  @Override
  protected void releaseResources() throws IOException {
    deleteQueue(q);
    super.releaseResources();
  }
}
import org.assertj.core.api.Condition;
    public static class LatchConditions {

        static Condition<CountDownLatch> completed() {
            return new Condition<>(
                countDownLatch-> {
                    try {
                        return countDownLatch.await(10, TimeUnit.SECONDS);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                },
                "Latch did not complete in 10 seconds");
        }

    }