diff --git a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpFrameParser.java b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpFrameParser.java index 06bc97b219d..69f4918397d 100644 --- a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpFrameParser.java +++ b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpFrameParser.java @@ -83,10 +83,7 @@ private void validateFrameSize(int frameSize) throws IOException { if (wireFormat != null) { maxFrameSize = wireFormat.getMaxFrameSize(); } - - if (frameSize > maxFrameSize) { - throw IOExceptionSupport.createFrameSizeException(frameSize, maxFrameSize); - } + AmqpWireFormat.validateFrameSize(frameSize, maxFrameSize); } public void setWireFormat(AmqpWireFormat wireFormat) { diff --git a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpWireFormat.java b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpWireFormat.java index 033eae7c0bc..a3b72f1e0b0 100644 --- a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpWireFormat.java +++ b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpWireFormat.java @@ -30,6 +30,7 @@ import org.apache.activemq.util.ByteArrayInputStream; import org.apache.activemq.util.ByteArrayOutputStream; import org.apache.activemq.util.ByteSequence; +import org.apache.activemq.util.IOExceptionSupport; import org.apache.activemq.wireformat.WireFormat; import org.fusesource.hawtbuf.Buffer; import org.slf4j.Logger; @@ -109,11 +110,7 @@ public Object unmarshal(DataInput dataIn) throws IOException { return new AmqpHeader(magic, false); } else { int size = dataIn.readInt(); - if (size > maxFrameSize) { - throw new AmqpProtocolException("Frame size exceeded max frame length."); - } else if (size <= 0) { - throw new AmqpProtocolException("Frame size value was invalid: " + size); - } + validateFrameSize(size, maxFrameSize); Buffer frame = new Buffer(size); frame.bigEndianEditor().writeInt(size); frame.readFrom(dataIn); @@ -261,4 +258,14 @@ public int getIdleTimeout() { public void setIdleTimeout(int idelTimeout) { this.idelTimeout = idelTimeout; } + + static void validateFrameSize(int frameSize, long maxFrameSize) throws IOException { + if (frameSize < 0) { + throw new AmqpProtocolException("Frame size of " + frameSize + " exceeds the maximum frame configured or supported frame size limit"); + } else if (Integer.toUnsignedLong(frameSize) > maxFrameSize) { + throw IOExceptionSupport.createFrameSizeException(frameSize, maxFrameSize); + } else if (Integer.compareUnsigned(frameSize, 8) < 0) { + throw new AmqpProtocolException("Frame size of " + frameSize + " is smaller than the minimally viable frame size value"); + } + } } diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/protocol/AmqpFrameParserTest.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/protocol/AmqpFrameParserTest.java index 5f78aaaee3d..f3cfc6874de 100644 --- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/protocol/AmqpFrameParserTest.java +++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/protocol/AmqpFrameParserTest.java @@ -18,9 +18,11 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import java.io.IOException; import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.List; @@ -57,6 +59,7 @@ public void onFrame(Object frame) { frames.add(frame); } }); + amqpWireFormat.setMaxFrameSize(AmqpWireFormat.DEFAULT_MAX_FRAME_SIZE); codec.setWireFormat(amqpWireFormat); } @@ -345,6 +348,49 @@ public void testReadPartialPayload() throws Exception { assertEquals(2, frames.size()); } + @Test + public void testNegativeFrameSize() throws Exception { + AmqpHeader inputHeader = new AmqpHeader(); + + DataByteArrayOutputStream output = new DataByteArrayOutputStream(); + output.write(inputHeader.getBuffer()); + // the value is unsigned so negative means too large + output.writeInt(-100); + output.close(); + + IOException e = assertThrows(IOException.class, () -> codec.parse(output.toBuffer().toByteBuffer())); + assertTrue(e.getMessage().contains("exceeds the maximum frame configured or supported frame size limit")); + } + + @Test + public void testFrameSizeTooSmall() throws Exception { + AmqpHeader inputHeader = new AmqpHeader(); + + DataByteArrayOutputStream output = new DataByteArrayOutputStream(); + output.write(inputHeader.getBuffer()); + // less than 8 is too small + output.writeInt(3); + output.close(); + + IOException e = assertThrows(IOException.class, () -> codec.parse(output.toBuffer().toByteBuffer())); + assertTrue(e.getMessage().contains("is smaller than the minimally viable frame size value")); + } + + @Test + public void testFrameSizeTooLarge() throws Exception { + amqpWireFormat.setMaxFrameSize(100); + AmqpHeader inputHeader = new AmqpHeader(); + + DataByteArrayOutputStream output = new DataByteArrayOutputStream(); + output.write(inputHeader.getBuffer()); + // less than 300 is larger than 100 maxFrameSize + output.writeInt(300); + output.close(); + + IOException e = assertThrows(IOException.class, () -> codec.parse(output.toBuffer().toByteBuffer())); + assertTrue(e.getMessage().contains("larger than max allowed")); + } + private void assertHeadersEqual(AmqpHeader expected, AmqpHeader actual) { assertTrue(expected.getBuffer().equals(actual.getBuffer())); } diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/protocol/AmqpWireFormatTest.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/protocol/AmqpWireFormatTest.java index 7e320d24d3d..1cc1530aafb 100644 --- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/protocol/AmqpWireFormatTest.java +++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/protocol/AmqpWireFormatTest.java @@ -17,19 +17,29 @@ package org.apache.activemq.transport.amqp.protocol; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; +import java.io.IOException; +import java.nio.ByteBuffer; import java.util.concurrent.atomic.AtomicBoolean; import org.apache.activemq.transport.amqp.AmqpHeader; import org.apache.activemq.transport.amqp.AmqpWireFormat; import org.apache.activemq.transport.amqp.AmqpWireFormat.ResetListener; +import org.apache.activemq.util.ByteSequence; +import org.junit.Before; import org.junit.Test; public class AmqpWireFormatTest { private final AmqpWireFormat wireFormat = new AmqpWireFormat(); + @Before + public void setUp() throws Exception { + wireFormat.setMaxFrameSize(AmqpWireFormat.DEFAULT_MAX_FRAME_SIZE); + } + @Test public void testWhenSaslNotAllowedNonSaslHeaderIsInvliad() { wireFormat.setAllowNonSaslConnections(false); @@ -78,4 +88,39 @@ public void onProtocolReset() { wireFormat.resetMagicRead(); assertTrue(reset.get()); } + + + @Test + public void testNegativeFrameSize() throws Exception { + AmqpHeader inputHeader = new AmqpHeader(); + wireFormat.unmarshal(new ByteSequence(inputHeader.getBuffer().toByteArray())); + + ByteSequence bs = new ByteSequence(ByteBuffer.allocate(Integer.BYTES).putInt(-100).array()); + IOException e = assertThrows(IOException.class, () -> wireFormat.unmarshal(bs)); + assertTrue(e.getMessage().contains("exceeds the maximum frame configured or supported frame size limit")); + } + + @Test + public void testFrameSizeTooSmall() throws Exception { + AmqpHeader inputHeader = new AmqpHeader(); + wireFormat.unmarshal(new ByteSequence(inputHeader.getBuffer().toByteArray())); + + // less than 8 + ByteSequence bs = new ByteSequence(ByteBuffer.allocate(Integer.BYTES).putInt(3).array()); + IOException e = assertThrows(IOException.class, () -> wireFormat.unmarshal(bs)); + assertTrue(e.getMessage().contains("is smaller than the minimally viable frame size value")); + } + + @Test + public void testFrameSizeTooLarge() throws Exception { + wireFormat.setMaxFrameSize(100); + AmqpHeader inputHeader = new AmqpHeader(); + wireFormat.unmarshal(new ByteSequence(inputHeader.getBuffer().toByteArray())); + + // size 300 is larger than maxFrameSize + ByteSequence bs = new ByteSequence(ByteBuffer.allocate(Integer.BYTES).putInt(300).array()); + IOException e = assertThrows(IOException.class, () -> wireFormat.unmarshal(bs)); + assertTrue(e.getMessage().contains("larger than max allowed")); + } + } diff --git a/activemq-client/src/main/java/org/apache/activemq/openwire/OpenWireFormat.java b/activemq-client/src/main/java/org/apache/activemq/openwire/OpenWireFormat.java index a927ed8f4c6..e315320caef 100644 --- a/activemq-client/src/main/java/org/apache/activemq/openwire/OpenWireFormat.java +++ b/activemq-client/src/main/java/org/apache/activemq/openwire/OpenWireFormat.java @@ -211,6 +211,7 @@ public synchronized Object unmarshal(ByteSequence sequence) throws IOException { if (maxFrameSizeEnabled && size > maxFrameSize) { throw IOExceptionSupport.createFrameSizeException(size, maxFrameSize); } + // This will also verify size is not negative context.setFrameSize(size); } return doUnmarshal(bytesIn); @@ -286,11 +287,12 @@ public Object unmarshal(DataInput dis) throws IOException { final var context = new MarshallingContext(); marshallingContext.set(context); - if (!sizePrefixDisabled) { + if (!sizePrefixDisabled) { int size = dis.readInt(); if (maxFrameSizeEnabled && size > maxFrameSize) { throw IOExceptionSupport.createFrameSizeException(size, maxFrameSize); } + // This will also verify size is not negative context.setFrameSize(size); } return doUnmarshal(dis); diff --git a/activemq-client/src/main/java/org/apache/activemq/transport/nio/NIOSSLTransport.java b/activemq-client/src/main/java/org/apache/activemq/transport/nio/NIOSSLTransport.java index 606503288cf..384f3f2af06 100644 --- a/activemq-client/src/main/java/org/apache/activemq/transport/nio/NIOSSLTransport.java +++ b/activemq-client/src/main/java/org/apache/activemq/transport/nio/NIOSSLTransport.java @@ -374,6 +374,12 @@ protected void processCommand(ByteBuffer plain) throws Exception { } } + // This isn't strictly necessary as ByteBuffer.allocate() would also throw the same + // IllegalArgumentException but this provides a better error. + if (nextFrameSize < 0) { + throw new IllegalArgumentException("Frame size value " + nextFrameSize + " may not be negative."); + } + // now we got the data, lets reallocate and store the size for the marshaler. // if there's more data in plain, then the next call will start processing it. currentBuffer = ByteBuffer.allocate(nextFrameSize + 4); diff --git a/activemq-client/src/main/java/org/apache/activemq/transport/nio/NIOTransport.java b/activemq-client/src/main/java/org/apache/activemq/transport/nio/NIOTransport.java index 10aae3dd4bd..d0126ff8d48 100644 --- a/activemq-client/src/main/java/org/apache/activemq/transport/nio/NIOTransport.java +++ b/activemq-client/src/main/java/org/apache/activemq/transport/nio/NIOTransport.java @@ -148,6 +148,12 @@ protected void serviceRead() { } } + // This isn't strictly necessary as ByteBuffer.allocateDirect() would also throw the same + // IllegalArgumentException, but this provides a better error. OpenWire + if (nextFrameSize < 0) { + throw new IllegalArgumentException("Frame size value " + nextFrameSize + " may not be negative."); + } + if (nextFrameSize > inputBuffer.capacity()) { currentBuffer = ByteBuffer.allocateDirect(nextFrameSize); currentBuffer.putInt(nextFrameSize);