Resolve CVE-2026-49432, etc (5.15.x)

activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/protocol/AmqpSender.java

@@ -22,6 +22,7 @@
 import java.util.LinkedList;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import org.apache.activemq.ActiveMQMessageFormatException;
 import org.apache.activemq.broker.region.AbstractSubscription;
 import org.apache.activemq.command.ActiveMQDestination;
 import org.apache.activemq.command.ActiveMQMessage;
@@ -42,6 +43,7 @@
 import org.apache.activemq.transport.amqp.message.AutoOutboundTransformer;
 import org.apache.activemq.transport.amqp.message.EncodedMessage;
 import org.apache.activemq.transport.amqp.message.OutboundTransformer;
+import org.apache.activemq.util.ExceptionUtils;
 import org.apache.qpid.proton.amqp.messaging.Accepted;
 import org.apache.qpid.proton.amqp.messaging.Modified;
 import org.apache.qpid.proton.amqp.messaging.Outcome;
@@ -492,7 +494,16 @@ public void pumpOutbound() throws Exception {
                     }
                 }
             } catch (Exception e) {
-                LOG.warn("Error detected while flushing outbound messages: {}", e.getMessage());
+                // Check if there is a format error trying to convert the message. This error means the
+                // message can't be converted (corruption, etc). This will wrap and throw the message
+                // so it can be handled by the transport
+                ActiveMQMessageFormatException formatError = ExceptionUtils.createMessageFormatException(e);
+                if (formatError != null) {
+                    LOG.warn("Message conversion error while flushing outbound messages: {}", e.getMessage(), e);
+                    throw formatError;
+                } else {
+                    LOG.warn("Error detected while flushing outbound messages: {}", e.getMessage());
+                }
             }
         }
     }

activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSInteroperabilityTest.java

@@ -18,10 +18,12 @@
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.junit.Assume.assumeFalse;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -30,20 +32,25 @@
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
 
-import javax.jms.BytesMessage;
-import javax.jms.Connection;
-import javax.jms.Destination;
-import javax.jms.MapMessage;
-import javax.jms.Message;
-import javax.jms.MessageConsumer;
-import javax.jms.MessageProducer;
-import javax.jms.ObjectMessage;
-import javax.jms.Session;
-import javax.jms.TextMessage;
+import javax.jms.*;
 
 import org.apache.activemq.ActiveMQConnection;
+import org.apache.activemq.broker.Broker;
+import org.apache.activemq.broker.BrokerFilter;
+import org.apache.activemq.broker.BrokerPlugin;
+import org.apache.activemq.broker.BrokerPluginSupport;
+import org.apache.activemq.broker.ConnectionContext;
+import org.apache.activemq.broker.region.MessageReference;
+import org.apache.activemq.broker.region.Subscription;
+import org.apache.activemq.command.ActiveMQBytesMessage;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQMessage;
+import org.apache.activemq.util.ByteSequenceData;
+import org.apache.activemq.util.Wait;
 import org.apache.qpid.proton.amqp.Binary;
+import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
@@ -59,6 +66,7 @@ public class JMSInteroperabilityTest extends JMSClientTestSupport {
 
     protected static final Logger LOG = LoggerFactory.getLogger(JMSInteroperabilityTest.class);
 
+    private final AtomicBoolean sentToDlq = new AtomicBoolean(false);
     private final String transformer;
 
     @Parameters(name="Transformer->{0}")
@@ -70,6 +78,13 @@ public static Collection<Object[]> data() {
             });
     }
 
+    @Before
+    @Override
+    public void setUp() throws Exception {
+        super.setUp();
+        sentToDlq.set(false);
+    }
+
     public JMSInteroperabilityTest(String transformer) {
         this.transformer = transformer;
     }
@@ -84,7 +99,26 @@ protected String getAmqpTransformer() {
         return transformer;
     }
 
-    //----- Tests for property handling between protocols --------------------//
+    @Override
+    protected void addAdditionalPlugins(List<BrokerPlugin> plugins) throws Exception {
+        super.addAdditionalPlugins(plugins);
+        plugins.add(new BrokerPluginSupport() {
+            @Override
+            public Broker installPlugin(Broker broker) {
+                return new BrokerFilter(broker) {
+                    @Override
+                    public boolean sendToDeadLetterQueue(ConnectionContext context,
+                                                         MessageReference messageReference, Subscription subscription,
+                                                         Throwable poisonCause) {
+                        sentToDlq.set(true);
+                        return super.sendToDeadLetterQueue(context, messageReference,
+                                subscription, poisonCause);
+                    }
+                };
+            }
+        });
+    }
+//----- Tests for property handling between protocols --------------------//
 
     @SuppressWarnings("unchecked")
     @Test(timeout = 60000)
@@ -483,8 +517,127 @@ public void testOpenWireToQpidObjectMessageWithOpenWireCompression() throws Exce
         assertNotNull(payload);
         assertTrue(payload instanceof UUID);
 
-        amqp.close();
-        openwire.close();
+    }
+
+    // The following tests for corruption will corrupt the headers or body
+    // to test that the AMQP protocol correctly passes the error during
+    // dispatch to allow the Transport Connection to properly handle
+    // with a poison ack so the message will be removed from the subscription.
+    // No selectors are set so these messages are only going to error
+    // during the protocol conversion.
+
+    @Test
+    public void testCorruptMessageErrorHeaders() throws Exception {
+        testCorruptMessageError(session -> {
+            ActiveMQBytesMessage message = (ActiveMQBytesMessage) session.createBytesMessage();
+            message.setStringProperty("testestt", "Testestt");
+            message.setStringProperty("prop2", "Testestt");
+            try {
+                message.beforeMarshall(null);
+                ByteSequenceData.writeIntBig(message.getMarshalledProperties(), 1024 * 1024);
+            } catch (IOException e) {
+                throw new RuntimeException(e);
+            }
+            return message;
+        }, false);
+    }
+
+    @Test
+    public void testCorruptMessageErrorMap() throws Exception {
+        testCorruptMessageError(session -> {
+            MapMessage message = session.createMapMessage();
+            message.setString("id", UUID.randomUUID().toString());
+            return message;
+        }, false);
+    }
+
+    // Check durable sub as well which is also a prefetch subscription so a
+    // poison ack will be handled the same way and DLQ
+    @Test
+    public void testCorruptMessageErrorMapDurableSub() throws Exception {
+        testCorruptMessageError(session -> {
+            MapMessage message = session.createMapMessage();
+            message.setString("id", UUID.randomUUID().toString());
+            return message;
+        }, true);
+    }
+
+    @Test
+    public void testCorruptMessageErrorText() throws Exception {
+        testCorruptMessageError(session -> {
+            TextMessage message = session.createTextMessage();
+            message.setText(UUID.randomUUID().toString());
+            return message;
+        }, false);
+    }
+
+    @Test
+    public void testCorruptMessageErrorTextDurableSub() throws Exception {
+        testCorruptMessageError(session -> {
+            TextMessage message = session.createTextMessage();
+            message.setText(UUID.randomUUID().toString());
+            return message;
+        }, true);
+    }
+
+
+    @Test
+    public void testCorruptMessageErrorStream() throws Exception {
+        testCorruptMessageError(session -> {
+            StreamMessage message = session.createStreamMessage();
+            message.writeBytes(UUID.randomUUID().toString().getBytes());
+            return message;
+        }, false);
+    }
+
+    private void testCorruptMessageError(MessageCreator messageCreator, boolean topic) throws Exception {
+        // Raw Transformer doesn't expand the body
+        assumeFalse(!transformer.equals("jms"));
+
+        Connection openwire = createJMSConnection();
+        Connection amqp = createConnection();
+        try {
+            openwire.start();
+            amqp.start();
+            Session openwireSession = openwire.createSession(false, Session.AUTO_ACKNOWLEDGE);
+            Session amqpSession = amqp.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+            ActiveMQDestination dest = topic ?
+                    (ActiveMQDestination)  openwireSession.createTopic(getDestinationName()) :
+                    (ActiveMQDestination)  openwireSession.createQueue(getDestinationName());
+            MessageProducer openwireProducer = openwireSession.createProducer(dest);
+            MessageConsumer amqpConsumer = topic ?
+                    amqpSession.createDurableSubscriber((Topic) dest, "sub") :
+                    amqpSession.createConsumer(dest);
+
+            // Create and send the Message
+            ActiveMQMessage outgoing = (ActiveMQMessage) messageCreator.create(openwireSession);
+            outgoing.storeContentAndClear();
+
+            // corrupt the buffer
+            // might be null if we are testing headers only
+            if (outgoing.getContent() != null && outgoing.getContent().length > 0) {
+                ByteSequenceData.writeIntBig(outgoing.getContent(), 1000);
+            }
+
+            openwireProducer.send(outgoing);
+
+            // Now try to consume the Message, should not be received
+            Message received = amqpConsumer.receive(2000);
+            assertNull(received);
+
+            // verify message is gone off the dest and went to the DLQ
+            assertTrue(Wait.waitFor(() -> brokerService.getDestination(dest)
+                    .getDestinationStatistics().getMessages().getCount() == 0, 500, 10));
+            assertTrue(sentToDlq.get());
+        } finally {
+            amqp.close();
+            openwire.close();
+        }
+    }
+
+    private interface MessageCreator {
+        Message create(Session session) throws JMSException;
     }
 
     @SuppressWarnings("unchecked")

activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java

@@ -99,6 +99,8 @@
 import org.apache.activemq.transport.TransmitCallback;
 import org.apache.activemq.transport.Transport;
 import org.apache.activemq.transport.TransportDisposedIOException;
+import org.apache.activemq.util.ExceptionUtils;
+import org.apache.activemq.ActiveMQMessageFormatException;
 import org.apache.activemq.util.IntrospectionSupport;
 import org.apache.activemq.util.MarshallingSupport;
 import org.apache.activemq.util.NetworkBridgeUtils;
@@ -995,6 +997,18 @@ protected void processDispatch(Command command) throws IOException {
                 if (sub != null) {
                     sub.onFailure();
                 }
+                // Check if this is a type of message format error which indicates the
+                // message was corrupt and there was some problem unmarshaling. For these
+                // errors we can handle by acking with a poison ack (which will send to the DLQ
+                // if durable/queue sub) and remove them from the consumer so the consumer can
+                // continue. We do not want to throw the exception as that would close the connection.
+                ActiveMQMessageFormatException marshallingError = ExceptionUtils.createMessageFormatException(e);
+                if (marshallingError != null) {
+                    handleMessageFormatError(marshallingError, messageDispatch);
+                    // must set to null so when we return the finally block is skipped
+                    messageDispatch = null;
+                    return;
+                }
                 messageDispatch = null;
                 throw e;
             } else {
@@ -1014,6 +1028,34 @@ protected void processDispatch(Command command) throws IOException {
         }
     }
 
+    private void handleMessageFormatError(ActiveMQMessageFormatException e, MessageDispatch messageDispatch) {
+        if (TRANSPORTLOG.isDebugEnabled()) {
+            TRANSPORTLOG.debug("{} had an unexpected Message format error: {}", this, e.getMessage(), e);
+        } else if (TRANSPORTLOG.isWarnEnabled()) {
+            TRANSPORTLOG.warn("{} had an unexpected Message format  error: {}", this, e.getMessage());
+        }
+
+        ConsumerBrokerExchange consumerExchange = getConsumerBrokerExchange(messageDispatch.getConsumerId());
+        try {
+            // acknowledge with the consumer exchange for this dispatch
+            // This should exist because this error happened during dispatch, but if for some
+            // reason it is null it should get handled when delivery is attempted again
+            if (consumerExchange != null) {
+                MessageAck ack = new MessageAck();
+                // Acking with a poison ack will send to the DLQ
+                ack.setAckType(MessageAck.POSION_ACK_TYPE);
+                ack.setPoisonCause(e);
+                ack.setConsumerId(messageDispatch.getConsumerId());
+                ack.setDestination(messageDispatch.getDestination());
+                ack.setMessageID(messageDispatch.getMessage().getMessageId());
+                broker.acknowledge(consumerExchange, ack);
+            }
+        } catch (Exception ex) {
+            TRANSPORTLOG.warn("{} could not acknowledge and send message to the DLQ after"
+                    + " ActiveMQMessageFormatException: {}", this, e.getMessage());
+        }
+    }
+
     @Override
     public boolean iterate() {
         try {
@@ -1390,7 +1432,14 @@ private NetworkBridgeConfiguration getNetworkConfiguration(final BrokerInfo info
     }
 
     @Override
-    public Response processBrokerInfo(BrokerInfo info) {
+    public Response processBrokerInfo(BrokerInfo info) throws IOException {
+        // We only expect to get at most one broker info command per connection
+        // Log and throw an IOException to close the connection if we receive more
+        // one because this is a protocol violation
+        if (this.brokerInfo != null) {
+            LOG.warn("Unexpected extra broker info command received: {}", info);
+            throw new IOException("Unexpected extra broker info command received from: " + info.getBrokerId());
+        }
         if (info.isSlaveBroker()) {
             LOG.error(" Slave Brokers are no longer supported - slave trying to attach is: {}", info.getBrokerName());
         } else if (info.isNetworkConnection() && !info.isDuplexConnection()) {
@@ -1463,10 +1512,6 @@ public Response processBrokerInfo(BrokerInfo info) {
                 return null;
             }
         }
-        // We only expect to get one broker info command per connection
-        if (this.brokerInfo != null) {
-            LOG.warn("Unexpected extra broker info command received: {}", info);
-        }
         this.brokerInfo = info;
         networkConnection = true;
         List<TransportConnectionState> connectionStates = listConnectionStates();

activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnector.java

@@ -41,6 +41,7 @@
 import org.apache.activemq.transport.TransportServer;
 import org.apache.activemq.transport.discovery.DiscoveryAgent;
 import org.apache.activemq.transport.discovery.DiscoveryAgentFactory;
+import org.apache.activemq.util.ExceptionUtils;
 import org.apache.activemq.util.ServiceStopper;
 import org.apache.activemq.util.ServiceSupport;
 import org.slf4j.Logger;
@@ -242,10 +243,10 @@ public void onAcceptError(Exception error) {
 
             private void onAcceptError(Exception error, String remoteHost) {
                 if (brokerService != null && brokerService.isStopping()) {
-                    LOG.info("Could not accept connection during shutdown {} : {} ({})", (remoteHost == null ? "" : "from " + remoteHost), error.getLocalizedMessage(), getRootCause(error).getMessage());
+                    LOG.info("Could not accept connection during shutdown {} : {} ({})", (remoteHost == null ? "" : "from " + remoteHost), error.getLocalizedMessage(),  ExceptionUtils.getRootCause(error).getMessage());
                 } else {
-                    LOG.warn("Could not accept connection {}: {} ({})", (remoteHost == null ? "" : "from " + remoteHost), error.getMessage(), getRootCause(error).getMessage());
-                    LOG.debug("Reason: " + error.getMessage(), error);
+                    LOG.warn("Could not accept connection {}: {} ({})", (remoteHost == null ? "" : "from " + remoteHost), error.getMessage(), ExceptionUtils.getRootCause(error).getMessage());
+                    LOG.debug("Reason: {}", error.getMessage(), error);
                 }
             }
         });
@@ -265,20 +266,6 @@ private void onAcceptError(Exception error, String remoteHost) {
         LOG.info("Connector {} started", getName());
     }
 
-    static Throwable getRootCause(final Throwable throwable) {
-        final List<Throwable> list = getThrowableList(throwable);
-        return list.isEmpty() ? null : list.get(list.size() - 1);
-    }
-
-    static List<Throwable> getThrowableList(Throwable throwable) {
-        final List<Throwable> list = new ArrayList<>();
-        while (throwable != null && !list.contains(throwable)) {
-            list.add(throwable);
-            throwable = throwable.getCause();
-        }
-        return list;
-    }
-
     public String getPublishableConnectString() throws Exception {
         String publishableConnectString = publishedAddressPolicy.getPublishableConnectString(this);
         LOG.debug("Publishing: {} for broker transport URI: {}", publishableConnectString, getConnectUri());