Adding protection against malformed entries on mapPersister.

Say someone says it has 30K entries while the persister can only generate 8

I'm allowing 100 entries (which should be way beyond necessary to
support newer versions).

Author: clebertsuconic

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessageMetadataPersister.java

@@ -50,6 +50,17 @@ public static AMQPMessageMetadataPersister getInstance() {
    protected static final short KEY_EXTRA_PROPERTIES = 8;
    protected static final short KEY_IS_REENCODED = 9; // used on large messages only
 
+   // This persister cannot generate more elements than defined keys.
+   // Currently, KEY_IS_REENCODED is the last defined key (9), with KEY_MESSAGE_ID being the first (1).
+   // Reading beyond 9 elements indicates either invalid/malicious data or a message from a future version.
+   // We set a reasonable limit to support future versions while protecting the decoder from malformed data.
+   protected static final short MAX_ELEMENTS = 100;
+
+   @Override
+   protected int getMaxAllowedElements() {
+      return MAX_ELEMENTS;
+   }
+
    /** This is the minimum size this codec will generate, as these will always be persisted */
    private static final int BASIC_SIZE = headerSize() +
                                          payloadSizeLong() + // messageID

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/util/AbstractMapPersister.java

@@ -58,6 +58,8 @@ public static Datatypes fromId(byte id) {
       }
    }
 
+   protected abstract int getMaxAllowedElements();
+
    protected abstract void onMapReadInteger(short key, int value, T decodingObject);
    protected abstract void onMapReadByte(short key, byte value, T decodingObject);
    protected abstract void onMapReadBoolean(short key, boolean value, T decodingObject);
@@ -152,6 +154,10 @@ public void decode(ActiveMQBuffer buffer, T decodingObject) {
       checkReadableBytes(buffer, DataConstants.SIZE_SHORT, endPosition);
       int entries = buffer.readUnsignedShort();
 
+      if (entries > getMaxAllowedElements()) {
+         throw new IllegalStateException("Invalid entries size " + entries + " beyond max allowed elements of " + getMaxAllowedElements());
+      }
+
       for (int i = 0; i < entries; i++) {
          // This will check that each entry is valid. If entries is set to an invalid value through malicious data
          // this check here would interrupt any further parsing

artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/util/MapPersisterTest.java

@@ -66,6 +66,11 @@ private void calculatePayloadSize() {
          encodeSize += headerSize();
       }
 
+      @Override
+      protected int getMaxAllowedElements() {
+         return Short.MAX_VALUE;
+      }
+
       public int getEncodeSize() {
          if (encodeSize == 0) {
             calculatePayloadSize();