Merge pull request #2064 from rabbitmq/mk-item-697fx

Be more defensive when parsing shortstr values in tables

Author: michaelklishin

src/main/java/com/rabbitmq/client/RpcServer.java

@@ -17,6 +17,8 @@
 package com.rabbitmq.client;
 
 import com.rabbitmq.utility.Utility;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.util.concurrent.BlockingQueue;
@@ -27,6 +29,9 @@
  * The class is agnostic about the format of RPC arguments / return values.
 */
 public class RpcServer {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(RpcServer.class);
+
     /** Channel we are communicating on */
     private final Channel _channel;
     /** Queue to receive requests from */
@@ -119,8 +124,15 @@ public ShutdownSignalException mainloop()
                     _mainloopRunning = false;
                     continue;
                 }
-                processRequest(request);
-                _channel.basicAck(request.getEnvelope().getDeliveryTag(), false);
+                try {
+                    processRequest(request);
+                    _channel.basicAck(request.getEnvelope().getDeliveryTag(), false);
+                } catch (ShutdownSignalException sse) {
+                    throw sse;
+                } catch (RuntimeException e) {
+                    LOGGER.warn("Discarding request that could not be processed", e);
+                    _channel.basicReject(request.getEnvelope().getDeliveryTag(), false);
+                }
             }
             return null;
         } catch (ShutdownSignalException sse) {

src/main/java/com/rabbitmq/client/impl/ValueReader.java

@@ -52,6 +52,9 @@ private static long unsignedExtend(int value)
         return extended & INT_MASK;
     }
 
+    /** Maximum length, in bytes, of a shortstr on the wire. */
+    private static final int MAX_SHORTSTR_LENGTH = 255;
+
     /** The stream we are reading from. */
     private final DataInputStream in;
 
@@ -71,7 +74,25 @@ private static String readShortstr(DataInputStream in)
     {
         byte [] b = new byte[in.readUnsignedByte()];
         in.readFully(b);
-        return new String(b, StandardCharsets.UTF_8);
+        return truncateToMaxUtf8Length(new String(b, StandardCharsets.UTF_8), MAX_SHORTSTR_LENGTH);
+    }
+
+    private static String truncateToMaxUtf8Length(String s, int maxBytes) {
+        if (s.length() <= maxBytes / 3 || s.indexOf('\uFFFD') < 0) {
+            return s;
+        }
+        int bytes = 0;
+        int i = 0;
+        while (i < s.length()) {
+            int codePoint = s.codePointAt(i);
+            int width = codePoint < 0x80 ? 1 : codePoint < 0x800 ? 2 : codePoint < 0x10000 ? 3 : 4;
+            if (bytes + width > maxBytes) {
+                return s.substring(0, i);
+            }
+            bytes += width;
+            i += Character.charCount(codePoint);
+        }
+        return s;
     }
 
     /** Public API - reads a short string. */

src/test/java/com/rabbitmq/client/test/ClientTestSuite.java

@@ -24,6 +24,7 @@
 @Suite
 @SelectClasses({
     TableTest.class,
+    ShortstrRoundTripTest.class,
     LongStringTest.class,
     BlockingCellTest.class,
     TruncatedInputStreamTest.class,

src/test/java/com/rabbitmq/client/test/RpcTest.java

@@ -330,6 +330,100 @@ public void handleRecoveryStarted(Recoverable recoverable) {
         client.close();
     }
 
+    @Test
+    public void serverKeepsRunningWhenHandlerFails() throws Exception {
+        rpcServer = new FailingRpcServer(serverChannel, queue);
+        Thread serverThread = new Thread(() -> {
+            try {
+                rpcServer.mainloop();
+            } catch (Exception e) {
+            }
+        });
+        serverThread.start();
+        RpcClient client = new RpcClient(new RpcClientParams()
+                .channel(clientChannel).exchange("").routingKey(queue).timeout(1000));
+
+        try {
+            client.doCall(null, "boom".getBytes());
+            fail("The handler failed, the call should have timed out");
+        } catch (TimeoutException e) {
+        }
+
+        RpcClient.Response response = client.doCall(null, "hello".getBytes());
+        assertEquals("*** hello ***", new String(response.getBody()));
+        assertTrue(serverThread.isAlive());
+
+        client.close();
+    }
+
+    @Test
+    public void requestThatCannotBeProcessedIsNotRequeued() throws Exception {
+        rpcServer = new FailingRpcServer(serverChannel, queue);
+        Thread serverThread = new Thread(() -> {
+            try {
+                rpcServer.mainloop();
+            } catch (Exception e) {
+            }
+        });
+        serverThread.start();
+        RpcClient client = new RpcClient(new RpcClientParams()
+                .channel(clientChannel).exchange("").routingKey(queue).timeout(1000));
+
+        try {
+            client.doCall(null, "boom".getBytes());
+            fail("The handler failed, the call should have timed out");
+        } catch (TimeoutException e) {
+        }
+
+        waitAtMost(Duration.ofSeconds(5), () -> clientChannel.messageCount(queue) == 0);
+        assertTrue(serverThread.isAlive());
+
+        client.close();
+    }
+
+    @Test
+    public void serverKeepsRunningWhenSeveralHandlerCallsFail() throws Exception {
+        rpcServer = new FailingRpcServer(serverChannel, queue);
+        Thread serverThread = new Thread(() -> {
+            try {
+                rpcServer.mainloop();
+            } catch (Exception e) {
+            }
+        });
+        serverThread.start();
+        RpcClient client = new RpcClient(new RpcClientParams()
+                .channel(clientChannel).exchange("").routingKey(queue).timeout(1000));
+
+        for (int i = 0; i < 5; i++) {
+            try {
+                client.doCall(null, "boom".getBytes());
+                fail("The handler failed, the call should have timed out");
+            } catch (TimeoutException e) {
+            }
+        }
+
+        RpcClient.Response response = client.doCall(null, "hello".getBytes());
+        assertEquals("*** hello ***", new String(response.getBody()));
+        assertTrue(serverThread.isAlive());
+
+        client.close();
+    }
+
+    private static class FailingRpcServer extends TestRpcServer {
+
+        public FailingRpcServer(Channel channel, String queueName) throws IOException {
+            super(channel, queueName);
+        }
+
+        @Override
+        public byte[] handleCall(Delivery request, AMQP.BasicProperties replyProperties) {
+            if ("boom".equals(new String(request.getBody()))) {
+                throw new IllegalArgumentException("cannot handle this request");
+            }
+            return super.handleCall(request, replyProperties);
+        }
+    }
+
     private static class TestRpcServer extends RpcServer {
 
         public TestRpcServer(Channel channel, String queueName) throws IOException {

src/test/java/com/rabbitmq/client/test/ShortstrRoundTripTest.java

@@ -0,0 +1,144 @@
+// Copyright (c) 2007-2026 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+//
+// 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 org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.impl.ValueReader;
+import com.rabbitmq.client.impl.ValueWriter;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import org.junit.jupiter.api.Test;
+
+public class ShortstrRoundTripTest {
+
+  private static String read(byte[] payload) throws IOException {
+    ByteArrayOutputStream frame = new ByteArrayOutputStream();
+    frame.write(payload.length);
+    frame.write(payload);
+    return new ValueReader(new DataInputStream(new ByteArrayInputStream(frame.toByteArray())))
+        .readShortstr();
+  }
+
+  private static byte[] write(String s) throws IOException {
+    ByteArrayOutputStream out = new ByteArrayOutputStream();
+    new ValueWriter(new DataOutputStream(out)).writeShortstr(s);
+    return out.toByteArray();
+  }
+
+  @Test
+  public void valueReadFromWireCanBeWrittenBack() throws IOException {
+    byte[] payload = new byte[255];
+    Arrays.fill(payload, (byte) 0xFF);
+    String decoded = read(payload);
+    assertThat(decoded.getBytes(StandardCharsets.UTF_8).length).isLessThanOrEqualTo(255);
+    assertThatCode(() -> write(decoded)).doesNotThrowAnyException();
+  }
+
+  @Test
+  public void wellFormedValuesArePreserved() throws IOException {
+    for (String value : new String[] {"", "hello", "santé", "你好", "😀"}) {
+      assertThat(read(value.getBytes(StandardCharsets.UTF_8))).isEqualTo(value);
+    }
+    StringBuilder sb = new StringBuilder();
+    for (int i = 0; i < 85; i++) {
+      sb.append("你");
+    }
+    String maxLength = sb.toString();
+    assertThat(maxLength.getBytes(StandardCharsets.UTF_8).length).isEqualTo(255);
+    assertThat(read(maxLength.getBytes(StandardCharsets.UTF_8))).isEqualTo(maxLength);
+  }
+
+  @Test
+  public void partiallyMalformedValueStaysWithinLimit() throws IOException {
+    byte[] payload = new byte[255];
+    Arrays.fill(payload, (byte) 'a');
+    for (int i = 100; i < 255; i++) {
+      payload[i] = (byte) 0xFF;
+    }
+    String decoded = read(payload);
+    assertThat(decoded.getBytes(StandardCharsets.UTF_8).length).isLessThanOrEqualTo(255);
+    assertThatCode(() -> write(decoded)).doesNotThrowAnyException();
+    assertThat(decoded).startsWith("aaaa");
+  }
+
+  @Test
+  public void messagePropertiesReadFromWireCanBeWrittenBack() throws IOException {
+    byte[] malformed = new byte[255];
+    Arrays.fill(malformed, (byte) 0xFF);
+
+    ByteArrayOutputStream header = new ByteArrayOutputStream();
+    DataOutputStream out = new DataOutputStream(header);
+    out.writeShort(0);
+    out.writeLong(6);
+    out.writeShort((1 << 10) | (1 << 9) | (1 << 7) | (1 << 5) | (1 << 4));
+    out.writeByte(malformed.length);
+    out.write(malformed);
+    out.writeByte(malformed.length);
+    out.write(malformed);
+    out.writeByte(malformed.length);
+    out.write(malformed);
+    out.writeByte(malformed.length);
+    out.write(malformed);
+    out.writeByte(malformed.length);
+    out.write(malformed);
+    out.flush();
+
+    AMQP.BasicProperties properties =
+        new AMQP.BasicProperties(
+            new DataInputStream(new ByteArrayInputStream(header.toByteArray())));
+
+    assertThat(properties.getCorrelationId()).isNotNull();
+    assertThat(properties.getReplyTo()).isNotNull();
+    assertThat(properties.getMessageId()).isNotNull();
+    assertThat(properties.getType()).isNotNull();
+    assertThat(properties.getUserId()).isNotNull();
+
+    AMQP.BasicProperties echoed =
+        new AMQP.BasicProperties.Builder()
+            .correlationId(properties.getCorrelationId())
+            .replyTo(properties.getReplyTo())
+            .messageId(properties.getMessageId())
+            .type(properties.getType())
+            .userId(properties.getUserId())
+            .build();
+
+    assertThatCode(
+            () -> {
+              ByteArrayOutputStream sink = new ByteArrayOutputStream();
+              echoed.writePropertiesTo(
+                  new com.rabbitmq.client.impl.ContentHeaderPropertyWriter(
+                      new DataOutputStream(sink)));
+            })
+        .doesNotThrowAnyException();
+  }
+
+  @Test
+  public void oversizedValuesAreStillRejected() {
+    StringBuilder sb = new StringBuilder();
+    for (int i = 0; i < 256; i++) {
+      sb.append('a');
+    }
+    assertThatCode(() -> write(sb.toString())).isInstanceOf(IllegalArgumentException.class);
+  }
+}