CVE-2023-46120 (Java, published 2023-10-25T18:17Z)
Description: 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.
Severity: nan (CVSS v3 base score: 7.5)
Weakness: CWE-400 \u2014 Uncontrolled Resource Consumption
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.
Repository: https://github.com/rabbitmq/rabbitmq-java-client
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 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)


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");
        }

    }