This is an automated email from the ASF dual-hosted git repository. rmaucher pushed a commit to branch 9.0.x in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 2379926ec498a200f093ff09cb246fe44069e0b2 Author: opencode <[email protected]> AuthorDate: Fri Oct 9 13:46:45 2026 +0200 Validate fragment metadata in FragmentationInterceptor reassembly The fragment count, fragment number and buffer sizes used for message reassembly are received as part of the message data and were used without validation. An oversized count made the fragment collection attempt a multi-GB array allocation. OutOfMemoryError is an Error and not caught by the per-message exception handler of the NIO receiver task, so the receive worker thread died and was never returned to the task pool. A negative count or an out-of-range fragment number produced array index exceptions, and the combined fragment size was accumulated in an int which could overflow. Validate the fragment count against a sane upper bound of 100,000 fragments, require the fragment number to be within the fragment set, reject fragments too short to carry their metadata before any buffer trimming, and accumulate the assembled size in a long. Invalid fragments are now logged and discarded. Malformed first fragments no longer poison later reassembly for the same message identifier. --- .../interceptors/FragmentationInterceptor.java | 60 ++++- .../group/interceptors/LocalStrings.properties | 5 + .../interceptors/TestFragmentationInterceptor.java | 261 +++++++++++++++++++++ 3 files changed, 317 insertions(+), 9 deletions(-) diff --git a/java/org/apache/catalina/tribes/group/interceptors/FragmentationInterceptor.java b/java/org/apache/catalina/tribes/group/interceptors/FragmentationInterceptor.java index eea432a979..ec550c411e 100644 --- a/java/org/apache/catalina/tribes/group/interceptors/FragmentationInterceptor.java +++ b/java/org/apache/catalina/tribes/group/interceptors/FragmentationInterceptor.java @@ -55,6 +55,14 @@ public class FragmentationInterceptor extends ChannelInterceptorBase implements */ protected static final StringManager sm = StringManager.getManager(FragmentationInterceptor.class); + /** + * The maximum number of fragments that will be accepted for a single message. The fragment count is received as + * part of the message data, so it is validated against this limit before it is used to allocate the fragment + * storage. The limit is far above any realistic message: with the default maxSize it allows messages of close to + * ten GB. + */ + public static final int MAX_FRAGMENTS = 100000; + /** * Map of fragment keys to their fragment collections for reassembly. */ @@ -130,14 +138,23 @@ public class FragmentationInterceptor extends ChannelInterceptorBase implements */ public void defrag(ChannelMessage msg) { FragKey key = new FragKey(msg.getUniqueId()); - FragCollection coll = getFragCollection(key, msg); - coll.addMessage((ChannelMessage) msg.deepclone()); + ChannelMessage complete = null; + try { + FragCollection coll = getFragCollection(key, msg); + coll.addMessage((ChannelMessage) msg.deepclone()); - if (coll.complete()) { - removeFragCollection(key); - ChannelMessage complete = coll.assemble(); - super.messageReceived(complete); + if (coll.complete()) { + removeFragCollection(key); + complete = coll.assemble(); + } + } catch (IllegalArgumentException | IllegalStateException e) { + // Fragment metadata received as part of the message data failed validation + log.warn(sm.getString("fragmentationInterceptor.fragment.discarded"), e); + return; + } + if (complete != null) { + super.messageReceived(complete); } } @@ -235,7 +252,16 @@ public class FragmentationInterceptor extends ChannelInterceptorBase implements */ public FragCollection(ChannelMessage msg) { // get the total messages - int count = XByteBuffer.toInt(msg.getMessage().getBytesDirect(), msg.getMessage().getLength() - 4); + int length = msg.getMessage().getLength(); + if (length < 4) { + throw new IllegalArgumentException( + sm.getString("fragmentationInterceptor.fragment.too-short", Integer.toString(length))); + } + int count = XByteBuffer.toInt(msg.getMessage().getBytesDirect(), length - 4); + if (count < 1 || count > MAX_FRAGMENTS) { + throw new IllegalArgumentException( + sm.getString("fragmentationInterceptor.invalid.frag.count", Integer.toString(count))); + } frags = new XByteBuffer[count]; this.msg = msg; } @@ -246,12 +272,24 @@ public class FragmentationInterceptor extends ChannelInterceptorBase implements * @param msg The fragment message to add */ public void addMessage(ChannelMessage msg) { + // At least the fragment number and the fragment count must be present. + // Check before trimming, trimming more bytes than are available throws + // an ArrayIndexOutOfBoundsException. + int length = msg.getMessage().getLength(); + if (length < 8) { + throw new IllegalArgumentException( + sm.getString("fragmentationInterceptor.fragment.too-short", Integer.toString(length))); + } // remove the total messages msg.getMessage().trim(4); // get the msg nr int nr = XByteBuffer.toInt(msg.getMessage().getBytesDirect(), msg.getMessage().getLength() - 4); // remove the msg nr msg.getMessage().trim(4); + if (nr < 0 || nr >= frags.length) { + throw new IllegalArgumentException(sm.getString("fragmentationInterceptor.invalid.frag.nr", + Integer.toString(nr), Integer.toString(frags.length))); + } frags[nr] = msg.getMessage(); } @@ -280,11 +318,15 @@ public class FragmentationInterceptor extends ChannelInterceptorBase implements if (!complete()) { throw new IllegalStateException(sm.getString("fragmentationInterceptor.fragments.missing")); } - int buffersize = 0; + long buffersize = 0; for (XByteBuffer frag : frags) { buffersize += frag.getLength(); } - XByteBuffer buf = new XByteBuffer(buffersize, false); + if (buffersize > Integer.MAX_VALUE) { + throw new IllegalStateException( + sm.getString("fragmentationInterceptor.fragments.too-large", Long.toString(buffersize))); + } + XByteBuffer buf = new XByteBuffer((int) buffersize, false); msg.setMessage(buf); for (XByteBuffer frag : frags) { msg.getMessage().append(frag.getBytesDirect(), 0, frag.getLength()); diff --git a/java/org/apache/catalina/tribes/group/interceptors/LocalStrings.properties b/java/org/apache/catalina/tribes/group/interceptors/LocalStrings.properties index e669178656..49a2a52fa6 100644 --- a/java/org/apache/catalina/tribes/group/interceptors/LocalStrings.properties +++ b/java/org/apache/catalina/tribes/group/interceptors/LocalStrings.properties @@ -32,8 +32,13 @@ encryptInterceptor.replayWindowMessageCount.tooSmall=Replay window message count encryptInterceptor.replayWindowTime.tooSmall=Replay window time must be at least 1 millisecond encryptInterceptor.tcpFailureDetector.ordering=EncryptInterceptor must be upstream of TcpFailureDetector. Please re-order EncryptInterceptor to be listed before TcpFailureDetector in your channel interceptor pipeline. +fragmentationInterceptor.fragment.discarded=Received fragment failed validation. The fragment is discarded. +fragmentationInterceptor.fragment.too-short=Received fragment with too little data, length [{0}]. fragmentationInterceptor.fragments.missing=Fragments are missing. +fragmentationInterceptor.fragments.too-large=The combined length [{0}] of the received fragments exceeds the maximum supported message size. fragmentationInterceptor.heartbeat.failed=Unable to perform heartbeat clean up in the frag interceptor +fragmentationInterceptor.invalid.frag.count=Received invalid fragment count [{0}]. The fragment is discarded. +fragmentationInterceptor.invalid.frag.nr=Received fragment number [{0}] for a message with [{1}] fragments. The fragment is discarded. gzipInterceptor.compress.failed=Unable to compress byte contents gzipInterceptor.decompress.failed=Unable to decompress byte contents diff --git a/test/org/apache/catalina/tribes/group/interceptors/TestFragmentationInterceptor.java b/test/org/apache/catalina/tribes/group/interceptors/TestFragmentationInterceptor.java new file mode 100644 index 0000000000..b9f70730f8 --- /dev/null +++ b/test/org/apache/catalina/tribes/group/interceptors/TestFragmentationInterceptor.java @@ -0,0 +1,261 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.catalina.tribes.group.interceptors; + +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import org.junit.Assert; +import org.junit.Test; + +import org.apache.catalina.tribes.ChannelException; +import org.apache.catalina.tribes.ChannelMessage; +import org.apache.catalina.tribes.Member; +import org.apache.catalina.tribes.group.ChannelInterceptorBase; +import org.apache.catalina.tribes.group.GroupChannel; +import org.apache.catalina.tribes.group.InterceptorPayload; +import org.apache.catalina.tribes.io.ChannelData; +import org.apache.catalina.tribes.io.XByteBuffer; +import org.apache.catalina.tribes.membership.MemberImpl; +import org.apache.catalina.tribes.util.UUIDGenerator; + +public class TestFragmentationInterceptor { + + private static class CapturingSender extends ChannelInterceptorBase { + + private final List<ChannelMessage> captured = new ArrayList<>(); + + @Override + public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) + throws ChannelException { + int len = msg.getMessage().getLength(); + byte[] copy = Arrays.copyOfRange(msg.getMessage().getBytesDirect(), 0, len); + ChannelData clone = new ChannelData(false); + clone.setUniqueId(Arrays.copyOf(msg.getUniqueId(), msg.getUniqueId().length)); + clone.setAddress(msg.getAddress()); + clone.setOptions(msg.getOptions()); + clone.setMessage(new XByteBuffer(copy, false)); + captured.add(clone); + } + } + + private static class DeliveringCollector extends ChannelInterceptorBase { + + private volatile ChannelMessage delivered; + + @Override + public void messageReceived(ChannelMessage msg) { + delivered = msg; + } + } + + private static ChannelData createMessage(byte[] payload, Member source) { + return createMessage(payload, source, UUIDGenerator.randomUUID(false)); + } + + private static ChannelData createMessage(byte[] payload, Member source, byte[] uniqueId) { + ChannelData data = new ChannelData(false); + data.setUniqueId(uniqueId); + data.setAddress(source); + data.setMessage(new XByteBuffer(payload, false)); + return data; + } + + private static byte[] bytesOf(ChannelMessage msg) { + int len = msg.getMessage().getLength(); + return Arrays.copyOfRange(msg.getMessage().getBytesDirect(), 0, len); + } + + @Test + public void testFragmentRoundTrip() throws Exception { + Member destination = new MemberImpl("localhost", 4000, -1); + byte[] payload = "The quick brown fox jumps over the lazy dog.".getBytes(StandardCharsets.UTF_8); + + FragmentationInterceptor sender = new FragmentationInterceptor(); + sender.setMaxSize(8); + CapturingSender capture = new CapturingSender(); + sender.setChannel(new GroupChannel()); + sender.setNext(capture); + sender.sendMessage(new Member[] { destination }, createMessage(payload, destination), null); + Assert.assertEquals(6, capture.captured.size()); + + FragmentationInterceptor receiver = new FragmentationInterceptor(); + DeliveringCollector delivery = new DeliveringCollector(); + receiver.setChannel(new GroupChannel()); + receiver.setPrevious(delivery); + for (ChannelMessage frag : capture.captured) { + receiver.messageReceived(frag); + } + + Assert.assertNotNull("Assembled message was not delivered", delivery.delivered); + Assert.assertArrayEquals("Assembled message does not match the original payload", + payload, bytesOf(delivery.delivered)); + } + + @Test + public void testHugeFragmentCountIsDiscarded() throws Exception { + Member destination = new MemberImpl("localhost", 4000, -1); + FragmentationInterceptor receiver = new FragmentationInterceptor(); + DeliveringCollector delivery = new DeliveringCollector(); + receiver.setChannel(new GroupChannel()); + receiver.setPrevious(delivery); + + XByteBuffer buffer = new XByteBuffer(16, false); + buffer.append("data".getBytes(StandardCharsets.UTF_8), 0, 4); + buffer.append(Integer.MAX_VALUE); + buffer.append(true); + ChannelData frag = createMessage(new byte[0], destination); + frag.setMessage(buffer); + + receiver.messageReceived(frag); + + Assert.assertNull("Message with a huge fragment count must not be delivered", delivery.delivered); + } + + @Test + public void testNegativeFragmentCountIsDiscarded() throws Exception { + Member destination = new MemberImpl("localhost", 4000, -1); + FragmentationInterceptor receiver = new FragmentationInterceptor(); + DeliveringCollector delivery = new DeliveringCollector(); + receiver.setChannel(new GroupChannel()); + receiver.setPrevious(delivery); + + XByteBuffer buffer = new XByteBuffer(16, false); + buffer.append("data".getBytes(StandardCharsets.UTF_8), 0, 4); + buffer.append(-5); + buffer.append(true); + ChannelData frag = createMessage(new byte[0], destination); + frag.setMessage(buffer); + + receiver.messageReceived(frag); + + Assert.assertNull("Message with a negative fragment count must not be delivered", delivery.delivered); + } + + @Test + public void testOutOfRangeFragmentNumberIsDiscarded() throws Exception { + Member destination = new MemberImpl("localhost", 4000, -1); + FragmentationInterceptor receiver = new FragmentationInterceptor(); + DeliveringCollector delivery = new DeliveringCollector(); + receiver.setChannel(new GroupChannel()); + receiver.setPrevious(delivery); + + XByteBuffer buffer = new XByteBuffer(16, false); + buffer.append("data".getBytes(StandardCharsets.UTF_8), 0, 4); + buffer.append(5); + buffer.append(1); + buffer.append(true); + ChannelData frag = createMessage(new byte[0], destination); + frag.setMessage(buffer); + + receiver.messageReceived(frag); + + Assert.assertNull("Fragment with an out of range number must not be delivered", delivery.delivered); + } + + @Test + public void testTooShortFragmentIsDiscarded() throws Exception { + Member destination = new MemberImpl("localhost", 4000, -1); + FragmentationInterceptor receiver = new FragmentationInterceptor(); + DeliveringCollector delivery = new DeliveringCollector(); + receiver.setChannel(new GroupChannel()); + receiver.setPrevious(delivery); + + XByteBuffer buffer = new XByteBuffer(16, false); + buffer.append("d".getBytes(StandardCharsets.UTF_8), 0, 1); + buffer.append(true); + ChannelData frag = createMessage(new byte[0], destination); + frag.setMessage(buffer); + + receiver.messageReceived(frag); + + Assert.assertNull("Fragment that is too short must not be delivered", delivery.delivered); + } + + @Test + public void testShortNonFirstFragmentIsDiscarded() throws Exception { + Member destination = new MemberImpl("localhost", 4000, -1); + FragmentationInterceptor receiver = new FragmentationInterceptor(); + DeliveringCollector delivery = new DeliveringCollector(); + receiver.setChannel(new GroupChannel()); + receiver.setPrevious(delivery); + + byte[] uniqueId = UUIDGenerator.randomUUID(false); + + XByteBuffer first = new XByteBuffer(16, false); + first.append("data".getBytes(StandardCharsets.UTF_8), 0, 4); + first.append(0); + first.append(2); + first.append(true); + ChannelData firstFrag = createMessage(new byte[0], destination, uniqueId); + firstFrag.setMessage(first); + receiver.messageReceived(firstFrag); + + XByteBuffer shortFrag = new XByteBuffer(16, false); + shortFrag.append("x".getBytes(StandardCharsets.UTF_8), 0, 1); + shortFrag.append(true); + ChannelData shortData = createMessage(new byte[0], destination, uniqueId); + shortData.setMessage(shortFrag); + receiver.messageReceived(shortData); + + Assert.assertNull("Incomplete message must not be delivered", delivery.delivered); + } + + @Test + public void testReassemblyAfterMalformedFirstFragment() throws Exception { + Member destination = new MemberImpl("localhost", 4000, -1); + FragmentationInterceptor receiver = new FragmentationInterceptor(); + DeliveringCollector delivery = new DeliveringCollector(); + receiver.setChannel(new GroupChannel()); + receiver.setPrevious(delivery); + + byte[] uniqueId = UUIDGenerator.randomUUID(false); + + XByteBuffer malformed = new XByteBuffer(16, false); + malformed.append("data".getBytes(StandardCharsets.UTF_8), 0, 4); + malformed.append(Integer.MAX_VALUE); + malformed.append(true); + ChannelData malformedFrag = createMessage(new byte[0], destination, uniqueId); + malformedFrag.setMessage(malformed); + receiver.messageReceived(malformedFrag); + + XByteBuffer first = new XByteBuffer(16, false); + first.append("ab".getBytes(StandardCharsets.UTF_8), 0, 2); + first.append(0); + first.append(2); + first.append(true); + ChannelData firstFrag = createMessage(new byte[0], destination, uniqueId); + firstFrag.setMessage(first); + receiver.messageReceived(firstFrag); + + XByteBuffer second = new XByteBuffer(16, false); + second.append("cd".getBytes(StandardCharsets.UTF_8), 0, 2); + second.append(1); + second.append(2); + second.append(true); + ChannelData secondFrag = createMessage(new byte[0], destination, uniqueId); + secondFrag.setMessage(second); + receiver.messageReceived(secondFrag); + + Assert.assertNotNull("Message must still be reassembled after a malformed fragment", + delivery.delivered); + Assert.assertArrayEquals("Assembled message does not match the expected content", + "abcd".getBytes(StandardCharsets.UTF_8), bytesOf(delivery.delivered)); + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
