From d431bd5af60cb30ab5c9f92074436a352362cf0d Mon Sep 17 00:00:00 2001 From: HTHou Date: Mon, 14 Sep 2026 11:09:07 +0800 Subject: [PATCH 1/3] Harden Thrift container deserialization limits --- .../java/org/apache/iotdb/rpc/RpcUtils.java | 3 + .../conf/edge/iotdb-system.properties | 1 + .../resources/conf/iotdb-system.properties | 1 + .../conf/iotdb-system.properties.template | 6 ++ .../iotdb/commons/conf/CommonConfig.java | 14 ++++ .../iotdb/commons/conf/CommonDescriptor.java | 8 +++ .../service/AbstractThriftServiceThread.java | 12 ++-- .../AbstractThriftServiceThreadTest.java | 66 +++++++++++++++++++ 8 files changed, 106 insertions(+), 5 deletions(-) create mode 100644 iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/service/AbstractThriftServiceThreadTest.java diff --git a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/RpcUtils.java b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/RpcUtils.java index 9c88bf3f8b7d..5cec704d69d5 100644 --- a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/RpcUtils.java +++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/RpcUtils.java @@ -60,6 +60,9 @@ public class RpcUtils { */ public static final int THRIFT_FRAME_MAX_SIZE = 536870912; + /** Default maximum number of elements allowed in a Thrift container. */ + public static final int THRIFT_DEFAULT_CONTAINER_LENGTH_LIMIT = 100_000; + /** * if resizeIfNecessary is called continuously with a small size for more than * MAX_BUFFER_OVERSIZE_TIME times, we will shrink the buffer to reclaim space. diff --git a/iotdb-core/node-commons/src/assembly/resources/conf/edge/iotdb-system.properties b/iotdb-core/node-commons/src/assembly/resources/conf/edge/iotdb-system.properties index 82679f99e6a3..197aeb1e76c2 100644 --- a/iotdb-core/node-commons/src/assembly/resources/conf/edge/iotdb-system.properties +++ b/iotdb-core/node-commons/src/assembly/resources/conf/edge/iotdb-system.properties @@ -46,6 +46,7 @@ dn_internal_port=10730 dn_mpp_data_exchange_port=10740 dn_schema_region_consensus_port=10750 dn_data_region_consensus_port=10760 +thrift_container_length_limit=100000 #################### ### Replication configuration diff --git a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties index c8badb4101fa..0f948b819735 100644 --- a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties +++ b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties @@ -46,6 +46,7 @@ dn_internal_port=10730 dn_mpp_data_exchange_port=10740 dn_schema_region_consensus_port=10750 dn_data_region_consensus_port=10760 +thrift_container_length_limit=100000 #################### ### Replication configuration diff --git a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template index 5079abe5100a..c0cfb94f48ed 100644 --- a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template +++ b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template @@ -563,6 +563,12 @@ dn_rpc_max_concurrent_client_num=1000 # Datatype: int dn_thrift_max_frame_size=0 +# Maximum number of elements allowed in a Thrift container during RPC decoding. +# A non-positive value uses the default value of 100000. +# effectiveMode: restart +# Datatype: int +thrift_container_length_limit=100000 + # thrift init buffer size # effectiveMode: restart # Datatype: int diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java index 6b26b9d544dd..f90e748ec2f8 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java @@ -163,6 +163,9 @@ public class CommonConfig { /** Whether to use thrift compression. */ private boolean isRpcThriftCompressionEnabled = false; + /** Maximum number of elements allowed in a Thrift container. */ + private int thriftContainerLengthLimit = RpcUtils.THRIFT_DEFAULT_CONTAINER_LENGTH_LIMIT; + private int maxClientNumForEachNode = DefaultProperty.MAX_CLIENT_NUM_FOR_EACH_NODE; private int maxIdleClientNumForEachNode = DefaultProperty.MAX_IDLE_CLIENT_NUM_FOR_EACH_NODE; @@ -757,6 +760,17 @@ public void setRpcThriftCompressionEnabled(boolean rpcThriftCompressionEnabled) isRpcThriftCompressionEnabled = rpcThriftCompressionEnabled; } + public int getThriftContainerLengthLimit() { + return thriftContainerLengthLimit; + } + + public void setThriftContainerLengthLimit(int thriftContainerLengthLimit) { + this.thriftContainerLengthLimit = + thriftContainerLengthLimit > 0 + ? thriftContainerLengthLimit + : RpcUtils.THRIFT_DEFAULT_CONTAINER_LENGTH_LIMIT; + } + public int getMaxClientNumForEachNode() { return maxClientNumForEachNode; } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java index cd187a98bede..10798ff56e81 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java @@ -134,6 +134,14 @@ public void loadCommonProps(TrimProperties properties) throws IOException { String.valueOf(config.isRpcThriftCompressionEnabled())) .trim())); + config.setThriftContainerLengthLimit( + Integer.parseInt( + properties + .getProperty( + "thrift_container_length_limit", + String.valueOf(config.getThriftContainerLengthLimit())) + .trim())); + config.setCnConnectionTimeoutInMS( Integer.parseInt( properties diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/AbstractThriftServiceThread.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/AbstractThriftServiceThread.java index eb82bcd65488..188e43d11789 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/AbstractThriftServiceThread.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/AbstractThriftServiceThread.java @@ -21,6 +21,7 @@ import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory; import org.apache.iotdb.commons.concurrent.threadpool.WrappedThreadPoolExecutor; +import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.conf.IoTDBConstant; import org.apache.iotdb.commons.exception.runtime.RPCServiceException; import org.apache.iotdb.commons.i18n.ServiceMessages; @@ -66,9 +67,8 @@ public abstract class AbstractThriftServiceThread extends Thread { private TTransportFactory transportFactory; - // currently, we can reuse the ProtocolFactory instance. - private static TCompactProtocol.Factory compactProtocolFactory = new TCompactProtocol.Factory(); - private static TBinaryProtocol.Factory binaryProtocolFactory = new TBinaryProtocol.Factory(); + // Thrift uses -1 as the sentinel for an unlimited string length. + private static final long THRIFT_NO_STRING_LENGTH_LIMIT = -1; private void initProtocolFactory(boolean compress) { protocolFactory = getProtocolFactory(compress); @@ -79,10 +79,12 @@ public TTransportFactory getTTransportFactory() { } public static TProtocolFactory getProtocolFactory(boolean compress) { + int containerLengthLimit = + CommonDescriptor.getInstance().getConfig().getThriftContainerLengthLimit(); if (compress) { - return compactProtocolFactory; + return new TCompactProtocol.Factory(THRIFT_NO_STRING_LENGTH_LIMIT, containerLengthLimit); } else { - return binaryProtocolFactory; + return new TBinaryProtocol.Factory(THRIFT_NO_STRING_LENGTH_LIMIT, containerLengthLimit); } } diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/service/AbstractThriftServiceThreadTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/service/AbstractThriftServiceThreadTest.java new file mode 100644 index 000000000000..a6413c5f19bb --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/service/AbstractThriftServiceThreadTest.java @@ -0,0 +1,66 @@ +/* + * 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.iotdb.commons.service; + +import org.apache.iotdb.commons.conf.CommonConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; + +import org.apache.thrift.protocol.TProtocolException; +import org.apache.thrift.protocol.TType; +import org.apache.thrift.transport.TMemoryInputTransport; +import org.junit.Assert; +import org.junit.Test; + +public class AbstractThriftServiceThreadTest { + + @Test + public void testProtocolFactoriesLimitContainerLength() { + CommonConfig config = CommonDescriptor.getInstance().getConfig(); + int originalLimit = config.getThriftContainerLengthLimit(); + try { + config.setThriftContainerLengthLimit(1); + + Assert.assertThrows( + TProtocolException.class, + () -> + AbstractThriftServiceThread.getProtocolFactory(false) + .getProtocol(new TMemoryInputTransport(binaryListHeader(2))) + .readListBegin()); + Assert.assertThrows( + TProtocolException.class, + () -> + AbstractThriftServiceThread.getProtocolFactory(true) + .getProtocol(new TMemoryInputTransport(compactListHeader(2))) + .readListBegin()); + } finally { + config.setThriftContainerLengthLimit(originalLimit); + } + } + + private static byte[] binaryListHeader(int size) { + return new byte[] { + TType.BYTE, (byte) (size >>> 24), (byte) (size >>> 16), (byte) (size >>> 8), (byte) size, 0, 0 + }; + } + + private static byte[] compactListHeader(int size) { + return new byte[] {(byte) 0xF3, (byte) size, 0, 0}; + } +} From 3ed705ad5772c3ebb047314e17a1b93ccebde016 Mon Sep 17 00:00:00 2001 From: HTHou Date: Mon, 14 Sep 2026 11:26:36 +0800 Subject: [PATCH 2/3] Add generated Thrift request regression test --- .../AbstractThriftServiceThreadTest.java | 46 +++++++++++++++++++ 1 file changed, 46 insertions(+) diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/service/AbstractThriftServiceThreadTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/service/AbstractThriftServiceThreadTest.java index a6413c5f19bb..d1beb0e34fb8 100644 --- a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/service/AbstractThriftServiceThreadTest.java +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/service/AbstractThriftServiceThreadTest.java @@ -21,13 +21,23 @@ import org.apache.iotdb.commons.conf.CommonConfig; import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.service.rpc.thrift.TSInsertRecordReq; +import org.apache.thrift.TException; +import org.apache.thrift.protocol.TBinaryProtocol; +import org.apache.thrift.protocol.TCompactProtocol; +import org.apache.thrift.protocol.TProtocol; import org.apache.thrift.protocol.TProtocolException; import org.apache.thrift.protocol.TType; +import org.apache.thrift.transport.TMemoryBuffer; import org.apache.thrift.transport.TMemoryInputTransport; import org.junit.Assert; import org.junit.Test; +import java.nio.ByteBuffer; +import java.util.Arrays; +import java.util.Collections; + public class AbstractThriftServiceThreadTest { @Test @@ -54,6 +64,42 @@ public void testProtocolFactoriesLimitContainerLength() { } } + @Test + public void testGeneratedRequestRejectsOversizedByteList() throws TException { + CommonConfig config = CommonDescriptor.getInstance().getConfig(); + int originalLimit = config.getThriftContainerLengthLimit(); + try { + config.setThriftContainerLengthLimit(1); + + Assert.assertThrows( + TProtocolException.class, + () -> + new TSInsertRecordReq() + .read( + AbstractThriftServiceThread.getProtocolFactory(false) + .getProtocol(new TMemoryInputTransport(serializeRequest(false))))); + Assert.assertThrows( + TProtocolException.class, + () -> + new TSInsertRecordReq() + .read( + AbstractThriftServiceThread.getProtocolFactory(true) + .getProtocol(new TMemoryInputTransport(serializeRequest(true))))); + } finally { + config.setThriftContainerLengthLimit(originalLimit); + } + } + + private static byte[] serializeRequest(boolean compact) throws TException { + TSInsertRecordReq request = + new TSInsertRecordReq(1, "root.sg.d", Collections.emptyList(), ByteBuffer.allocate(0), 1) + .setColumnCategoryies(Arrays.asList((byte) 0, (byte) 1)); + TMemoryBuffer buffer = new TMemoryBuffer(128); + TProtocol protocol = compact ? new TCompactProtocol(buffer) : new TBinaryProtocol(buffer); + request.write(protocol); + return Arrays.copyOf(buffer.getArray(), buffer.length()); + } + private static byte[] binaryListHeader(int size) { return new byte[] { TType.BYTE, (byte) (size >>> 24), (byte) (size >>> 16), (byte) (size >>> 8), (byte) size, 0, 0 From 93008a370c44d90740b90a526dfa4747e49b2bed Mon Sep 17 00:00:00 2001 From: HTHou Date: Mon, 14 Sep 2026 12:40:49 +0800 Subject: [PATCH 3/3] Validate Thrift reads against remaining frame data --- .../apache/iotdb/rpc/i18n/RpcMessages.java | 3 + .../apache/iotdb/rpc/i18n/RpcMessages.java | 3 + .../java/org/apache/iotdb/rpc/RpcUtils.java | 3 - .../iotdb/rpc/TElasticFramedTransport.java | 42 ++- .../rpc/TElasticFramedTransportReadTest.java | 266 ++++++++++++++++++ .../conf/edge/iotdb-system.properties | 1 - .../resources/conf/iotdb-system.properties | 1 - .../conf/iotdb-system.properties.template | 6 - .../iotdb/commons/conf/CommonConfig.java | 14 - .../iotdb/commons/conf/CommonDescriptor.java | 8 - .../service/AbstractThriftServiceThread.java | 12 +- .../AbstractThriftServiceThreadTest.java | 112 -------- 12 files changed, 307 insertions(+), 164 deletions(-) create mode 100644 iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/TElasticFramedTransportReadTest.java delete mode 100644 iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/service/AbstractThriftServiceThreadTest.java diff --git a/iotdb-client/service-rpc/src/main/i18n/en/org/apache/iotdb/rpc/i18n/RpcMessages.java b/iotdb-client/service-rpc/src/main/i18n/en/org/apache/iotdb/rpc/i18n/RpcMessages.java index b9d76aa4fd95..33778bf8fb23 100644 --- a/iotdb-client/service-rpc/src/main/i18n/en/org/apache/iotdb/rpc/i18n/RpcMessages.java +++ b/iotdb-client/service-rpc/src/main/i18n/en/org/apache/iotdb/rpc/i18n/RpcMessages.java @@ -35,6 +35,9 @@ public final class RpcMessages { "Frame size (%d) larger than protect max size (%d)%s!"; public static final String FRAME_ERROR_STRING_LENGTH_EXCEEDED = "String length (%d) larger than protect max size (%d)%s!"; + public static final String + EXCEPTION_REQUIRED_READ_SIZE_ARG_EXCEEDS_REMAINING_FRAME_SIZE_ARG_ARG_9C0541EE = + "Required read size (%d) exceeds remaining frame size (%d)%s!"; // TElasticFramedTransport - SSL public static final String NON_SSL_TO_SSL_PORT = diff --git a/iotdb-client/service-rpc/src/main/i18n/zh/org/apache/iotdb/rpc/i18n/RpcMessages.java b/iotdb-client/service-rpc/src/main/i18n/zh/org/apache/iotdb/rpc/i18n/RpcMessages.java index 5ddfe9fc0b27..81b519fd4779 100644 --- a/iotdb-client/service-rpc/src/main/i18n/zh/org/apache/iotdb/rpc/i18n/RpcMessages.java +++ b/iotdb-client/service-rpc/src/main/i18n/zh/org/apache/iotdb/rpc/i18n/RpcMessages.java @@ -32,6 +32,9 @@ public final class RpcMessages { "帧大小 (%d) 超过保护最大值 (%d)%s！"; public static final String FRAME_ERROR_STRING_LENGTH_EXCEEDED = "字符串长度 (%d) 超过保护最大值 (%d)%s！"; + public static final String + EXCEPTION_REQUIRED_READ_SIZE_ARG_EXCEEDS_REMAINING_FRAME_SIZE_ARG_ARG_9C0541EE = + "请求读取的大小 (%d) 超过当前帧剩余大小 (%d)%s！"; // TElasticFramedTransport - SSL public static final String NON_SSL_TO_SSL_PORT = diff --git a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/RpcUtils.java b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/RpcUtils.java index 5cec704d69d5..9c88bf3f8b7d 100644 --- a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/RpcUtils.java +++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/RpcUtils.java @@ -60,9 +60,6 @@ public class RpcUtils { */ public static final int THRIFT_FRAME_MAX_SIZE = 536870912; - /** Default maximum number of elements allowed in a Thrift container. */ - public static final int THRIFT_DEFAULT_CONTAINER_LENGTH_LIMIT = 100_000; - /** * if resizeIfNecessary is called continuously with a small size for more than * MAX_BUFFER_OVERSIZE_TIME times, we will shrink the buffer to reclaim space. diff --git a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TElasticFramedTransport.java b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TElasticFramedTransport.java index 31f20f5c381e..d0c6a3d24cad 100644 --- a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TElasticFramedTransport.java +++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TElasticFramedTransport.java @@ -245,7 +245,9 @@ private enum FrameError { TLS_REQUEST(RpcMessages.FRAME_ERROR_TLS_REQUEST), NEGATIVE_FRAME_SIZE(RpcMessages.FRAME_ERROR_NEGATIVE_FRAME_SIZE), FRAME_SIZE_EXCEEDED(RpcMessages.FRAME_ERROR_FRAME_SIZE_EXCEEDED), - STRING_LENGTH_EXCEEDED(RpcMessages.FRAME_ERROR_STRING_LENGTH_EXCEEDED); + STRING_LENGTH_EXCEEDED(RpcMessages.FRAME_ERROR_STRING_LENGTH_EXCEEDED), + INSUFFICIENT_FRAME_DATA( + RpcMessages.EXCEPTION_REQUIRED_READ_SIZE_ARG_EXCEEDS_REMAINING_FRAME_SIZE_ARG_ARG_9C0541EE); private final String messageFormat; @@ -255,7 +257,9 @@ private enum FrameError { void throwException(long size, String remoteInfo, int maxSize) throws TTransportException { String message = - (this == FRAME_SIZE_EXCEEDED || this == STRING_LENGTH_EXCEEDED) + (this == FRAME_SIZE_EXCEEDED + || this == STRING_LENGTH_EXCEEDED + || this == INSUFFICIENT_FRAME_DATA) ? String.format(messageFormat, size, maxSize, remoteInfo) : String.format(messageFormat, size, remoteInfo); throw new TTransportException(TTransportException.CORRUPTED_DATA, message); @@ -308,18 +312,32 @@ public void updateKnownMessageSize(long size) throws TTransportException { @Override public void checkReadBytesAvailable(long numBytes) throws TTransportException { + // RPC messages are flushed as complete frames. Container checks pass their minimum encoded + // size here, before generated code allocates the container. Compare it with actual buffered + // data, not just the configured frame cap. Use readBuffer directly because copyBinary makes + // this transport's getBytesRemainingInBuffer() return -1. + int remaining = readBuffer.getBytesRemainingInBuffer(); + FrameError error; + int limit; if (numBytes >= thriftMaxFrameSize) { - SocketAddress remoteAddress = null; - if (underlying instanceof TSocket) { - remoteAddress = ((TSocket) underlying).getSocket().getRemoteSocketAddress(); - } - String remoteInfo = - (remoteAddress == null) - ? RpcMessages.EMPTY_MESSAGE - : RpcMessages.REMOTE_ADDRESS_PREFIX + remoteAddress; - close(); - FrameError.STRING_LENGTH_EXCEEDED.throwException(numBytes, remoteInfo, thriftMaxFrameSize); + error = FrameError.STRING_LENGTH_EXCEEDED; + limit = thriftMaxFrameSize; + } else if (numBytes > remaining) { + error = FrameError.INSUFFICIENT_FRAME_DATA; + limit = remaining; + } else { + return; + } + SocketAddress remoteAddress = null; + if (underlying instanceof TSocket) { + remoteAddress = ((TSocket) underlying).getSocket().getRemoteSocketAddress(); } + String remoteInfo = + (remoteAddress == null) + ? RpcMessages.EMPTY_MESSAGE + : RpcMessages.REMOTE_ADDRESS_PREFIX + remoteAddress; + close(); + error.throwException(numBytes, remoteInfo, limit); } @Override diff --git a/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/TElasticFramedTransportReadTest.java b/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/TElasticFramedTransportReadTest.java new file mode 100644 index 000000000000..93d12fd32b27 --- /dev/null +++ b/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/TElasticFramedTransportReadTest.java @@ -0,0 +1,266 @@ +/* + * 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.iotdb.rpc; + +import org.apache.iotdb.rpc.i18n.RpcMessages; +import org.apache.iotdb.service.rpc.thrift.IClientRPCService; +import org.apache.iotdb.service.rpc.thrift.TSInsertRecordReq; + +import org.apache.thrift.TException; +import org.apache.thrift.protocol.TBinaryProtocol; +import org.apache.thrift.protocol.TCompactProtocol; +import org.apache.thrift.protocol.TField; +import org.apache.thrift.protocol.TList; +import org.apache.thrift.protocol.TMap; +import org.apache.thrift.protocol.TMessage; +import org.apache.thrift.protocol.TMessageType; +import org.apache.thrift.protocol.TProtocol; +import org.apache.thrift.protocol.TSet; +import org.apache.thrift.protocol.TStruct; +import org.apache.thrift.protocol.TType; +import org.apache.thrift.transport.TMemoryBuffer; +import org.apache.thrift.transport.TMemoryInputTransport; +import org.apache.thrift.transport.TTransport; +import org.apache.thrift.transport.TTransportException; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; + +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThrows; + +@RunWith(Parameterized.class) +public class TElasticFramedTransportReadTest { + private static final int MAX_FRAME_SIZE = 1024 * 1024; + private static final int ELEMENT_COUNT = 200_001; + + private final boolean compact; + private final boolean copyBinary; + private final boolean snappy; + + @Parameterized.Parameters(name = "compact={0}, copyBinary={1}, snappy={2}") + public static Collection parameters() { + Collection parameters = new ArrayList<>(); + for (boolean compact : new boolean[] {false, true}) { + for (boolean copyBinary : new boolean[] {false, true}) { + for (boolean snappy : new boolean[] {false, true}) { + parameters.add(new Object[] {compact, copyBinary, snappy}); + } + } + } + return parameters; + } + + public TElasticFramedTransportReadTest(boolean compact, boolean copyBinary, boolean snappy) { + this.compact = compact; + this.copyBinary = copyBinary; + this.snappy = snappy; + } + + @Test + public void testTruncatedRequestRejectedBeforeContainerAllocation() throws Exception { + byte[] frame = + serialize( + p -> { + p.writeMessageBegin(new TMessage("insertRecord", TMessageType.CALL, 1)); + p.writeStructBegin(new TStruct("insertRecord_args")); + p.writeFieldBegin(new TField("req", TType.STRUCT, (short) 1)); + p.writeStructBegin(new TStruct("TSInsertRecordReq")); + p.writeFieldBegin(new TField("columnCategoryies", TType.LIST, (short) 8)); + p.writeListBegin(new TList(TType.BYTE, ELEMENT_COUNT)); + // Flush a complete frame that ends at the container header, without any elements. + }); + try (TElasticFramedTransport transport = transport(new TMemoryInputTransport(frame))) { + TProtocol protocol = protocol(transport); + assertEquals("insertRecord", protocol.readMessageBegin().name); + IClientRPCService.insertRecord_args args = new IClientRPCService.insertRecord_args(); + TTransportException exception = + assertThrows(TTransportException.class, () -> args.read(protocol)); + assertNotNull(args.req); + assertNull(args.req.columnCategoryies); + assertInsufficientFrame(exception, ELEMENT_COUNT, 0); + } + } + + @Test + public void testLargeCompleteRequestsInSuccessiveFrames() throws Exception { + TSInsertRecordReq large = + new TSInsertRecordReq(1, "root.sg.d", Collections.emptyList(), ByteBuffer.allocate(0), 1) + .setColumnCategoryies(Collections.nCopies(ELEMENT_COUNT, (byte) 0)); + TSInsertRecordReq empty = + new TSInsertRecordReq(1, "root.sg.d", Collections.emptyList(), ByteBuffer.allocate(0), 2) + .setColumnCategoryies(Collections.emptyList()); + TMemoryBuffer wire = new TMemoryBuffer(128); + try (TElasticFramedTransport output = transport(wire)) { + IClientRPCService.Client client = new IClientRPCService.Client(protocol(output)); + client.send_insertRecord(large); + client.send_insertRecord(empty); + } + try (TElasticFramedTransport input = + transport(new TMemoryInputTransport(Arrays.copyOf(wire.getArray(), wire.length())))) { + TProtocol protocol = protocol(input); + for (TSInsertRecordReq expected : new TSInsertRecordReq[] {large, empty}) { + assertEquals("insertRecord", protocol.readMessageBegin().name); + IClientRPCService.insertRecord_args args = new IClientRPCService.insertRecord_args(); + args.read(protocol); + protocol.readMessageEnd(); + assertEquals(expected, args.req); + } + } + } + + @Test + public void testTruncatedContainerHeaders() throws Exception { + for (byte elementType : new byte[] {TType.BYTE, TType.STRUCT, TType.LIST, TType.MAP}) { + for (byte containerType : new byte[] {TType.LIST, TType.SET, TType.MAP}) { + byte[] frame = + serialize( + p -> { + if (containerType == TType.LIST) { + p.writeListBegin(new TList(elementType, ELEMENT_COUNT)); + } else if (containerType == TType.SET) { + p.writeSetBegin(new TSet(elementType, ELEMENT_COUNT)); + } else { + p.writeMapBegin(new TMap(TType.BYTE, elementType, ELEMENT_COUNT)); + } + }); + try (TElasticFramedTransport input = transport(new TMemoryInputTransport(frame))) { + TProtocol p = protocol(input); + long minimumBytes = + (long) ELEMENT_COUNT + * (p.getMinSerializedSize(elementType) + (containerType == TType.MAP ? 1 : 0)); + TTransportException exception = + assertThrows( + TTransportException.class, + () -> { + if (containerType == TType.LIST) { + p.readListBegin(); + } else if (containerType == TType.SET) { + p.readSetBegin(); + } else { + p.readMapBegin(); + } + }); + assertInsufficientFrame(exception, minimumBytes, 0); + } + } + } + } + + @Test + public void testCompleteBinaryAtFrameBoundary() throws Exception { + ByteBuffer binary = ByteBuffer.wrap(new byte[ELEMENT_COUNT]); + byte[] frame = + serialize( + p -> { + p.writeString(""); + p.writeBinary(binary); + }); + try (TElasticFramedTransport input = transport(new TMemoryInputTransport(frame))) { + TProtocol protocol = protocol(input); + assertEquals("", protocol.readString()); + assertEquals(binary, protocol.readBinary()); + input.checkReadBytesAvailable(0); + } + } + + @Test + public void testTruncatedStringAndBinary() throws Exception { + TMemoryBuffer buffer = new TMemoryBuffer(128); + protocol(buffer).writeBinary(ByteBuffer.wrap(new byte[ELEMENT_COUNT])); + int headerLength = buffer.length() - ELEMENT_COUNT; + byte[] frame = serialize(p -> p.getTransport().write(buffer.getArray(), 0, headerLength)); + for (boolean binary : new boolean[] {false, true}) { + try (TElasticFramedTransport input = transport(new TMemoryInputTransport(frame))) { + TProtocol p = protocol(input); + TTransportException exception = + assertThrows( + TTransportException.class, + () -> { + if (binary) { + p.readBinary(); + } else { + p.readString(); + } + }); + assertInsufficientFrame(exception, ELEMENT_COUNT, 0); + } + } + } + + @Test + public void testExistingMaximumReadSizeProtection() throws Exception { + try (TElasticFramedTransport input = transport(new TMemoryInputTransport(new byte[0]))) { + TTransportException exception = + assertThrows( + TTransportException.class, () -> input.checkReadBytesAvailable(MAX_FRAME_SIZE)); + assertEquals(TTransportException.CORRUPTED_DATA, exception.getType()); + assertEquals( + String.format( + RpcMessages.FRAME_ERROR_STRING_LENGTH_EXCEEDED, MAX_FRAME_SIZE, MAX_FRAME_SIZE, ""), + exception.getMessage()); + } + } + + private void assertInsufficientFrame( + TTransportException exception, long required, int remaining) { + assertEquals(TTransportException.CORRUPTED_DATA, exception.getType()); + assertEquals( + String.format( + RpcMessages + .EXCEPTION_REQUIRED_READ_SIZE_ARG_EXCEEDS_REMAINING_FRAME_SIZE_ARG_ARG_9C0541EE, + required, + remaining, + ""), + exception.getMessage()); + } + + private TElasticFramedTransport transport(TTransport underlying) throws TTransportException { + return snappy + ? new TSnappyElasticFramedTransport(underlying, 128, MAX_FRAME_SIZE, copyBinary) + : new TElasticFramedTransport(underlying, 128, MAX_FRAME_SIZE, copyBinary); + } + + private TProtocol protocol(TTransport transport) { + return compact ? new TCompactProtocol(transport) : new TBinaryProtocol(transport); + } + + private byte[] serialize(ProtocolWriter writer) throws TException { + TMemoryBuffer wire = new TMemoryBuffer(128); + try (TElasticFramedTransport output = transport(wire)) { + writer.write(protocol(output)); + output.flush(); + } + return Arrays.copyOf(wire.getArray(), wire.length()); + } + + @FunctionalInterface + private interface ProtocolWriter { + void write(TProtocol protocol) throws TException; + } +} diff --git a/iotdb-core/node-commons/src/assembly/resources/conf/edge/iotdb-system.properties b/iotdb-core/node-commons/src/assembly/resources/conf/edge/iotdb-system.properties index 197aeb1e76c2..82679f99e6a3 100644 --- a/iotdb-core/node-commons/src/assembly/resources/conf/edge/iotdb-system.properties +++ b/iotdb-core/node-commons/src/assembly/resources/conf/edge/iotdb-system.properties @@ -46,7 +46,6 @@ dn_internal_port=10730 dn_mpp_data_exchange_port=10740 dn_schema_region_consensus_port=10750 dn_data_region_consensus_port=10760 -thrift_container_length_limit=100000 #################### ### Replication configuration diff --git a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties index 0f948b819735..c8badb4101fa 100644 --- a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties +++ b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties @@ -46,7 +46,6 @@ dn_internal_port=10730 dn_mpp_data_exchange_port=10740 dn_schema_region_consensus_port=10750 dn_data_region_consensus_port=10760 -thrift_container_length_limit=100000 #################### ### Replication configuration diff --git a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template index c0cfb94f48ed..5079abe5100a 100644 --- a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template +++ b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template @@ -563,12 +563,6 @@ dn_rpc_max_concurrent_client_num=1000 # Datatype: int dn_thrift_max_frame_size=0 -# Maximum number of elements allowed in a Thrift container during RPC decoding. -# A non-positive value uses the default value of 100000. -# effectiveMode: restart -# Datatype: int -thrift_container_length_limit=100000 - # thrift init buffer size # effectiveMode: restart # Datatype: int diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java index f90e748ec2f8..6b26b9d544dd 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java @@ -163,9 +163,6 @@ public class CommonConfig { /** Whether to use thrift compression. */ private boolean isRpcThriftCompressionEnabled = false; - /** Maximum number of elements allowed in a Thrift container. */ - private int thriftContainerLengthLimit = RpcUtils.THRIFT_DEFAULT_CONTAINER_LENGTH_LIMIT; - private int maxClientNumForEachNode = DefaultProperty.MAX_CLIENT_NUM_FOR_EACH_NODE; private int maxIdleClientNumForEachNode = DefaultProperty.MAX_IDLE_CLIENT_NUM_FOR_EACH_NODE; @@ -760,17 +757,6 @@ public void setRpcThriftCompressionEnabled(boolean rpcThriftCompressionEnabled) isRpcThriftCompressionEnabled = rpcThriftCompressionEnabled; } - public int getThriftContainerLengthLimit() { - return thriftContainerLengthLimit; - } - - public void setThriftContainerLengthLimit(int thriftContainerLengthLimit) { - this.thriftContainerLengthLimit = - thriftContainerLengthLimit > 0 - ? thriftContainerLengthLimit - : RpcUtils.THRIFT_DEFAULT_CONTAINER_LENGTH_LIMIT; - } - public int getMaxClientNumForEachNode() { return maxClientNumForEachNode; } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java index 10798ff56e81..cd187a98bede 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java @@ -134,14 +134,6 @@ public void loadCommonProps(TrimProperties properties) throws IOException { String.valueOf(config.isRpcThriftCompressionEnabled())) .trim())); - config.setThriftContainerLengthLimit( - Integer.parseInt( - properties - .getProperty( - "thrift_container_length_limit", - String.valueOf(config.getThriftContainerLengthLimit())) - .trim())); - config.setCnConnectionTimeoutInMS( Integer.parseInt( properties diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/AbstractThriftServiceThread.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/AbstractThriftServiceThread.java index 188e43d11789..eb82bcd65488 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/AbstractThriftServiceThread.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/AbstractThriftServiceThread.java @@ -21,7 +21,6 @@ import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory; import org.apache.iotdb.commons.concurrent.threadpool.WrappedThreadPoolExecutor; -import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.conf.IoTDBConstant; import org.apache.iotdb.commons.exception.runtime.RPCServiceException; import org.apache.iotdb.commons.i18n.ServiceMessages; @@ -67,8 +66,9 @@ public abstract class AbstractThriftServiceThread extends Thread { private TTransportFactory transportFactory; - // Thrift uses -1 as the sentinel for an unlimited string length. - private static final long THRIFT_NO_STRING_LENGTH_LIMIT = -1; + // currently, we can reuse the ProtocolFactory instance. + private static TCompactProtocol.Factory compactProtocolFactory = new TCompactProtocol.Factory(); + private static TBinaryProtocol.Factory binaryProtocolFactory = new TBinaryProtocol.Factory(); private void initProtocolFactory(boolean compress) { protocolFactory = getProtocolFactory(compress); @@ -79,12 +79,10 @@ public TTransportFactory getTTransportFactory() { } public static TProtocolFactory getProtocolFactory(boolean compress) { - int containerLengthLimit = - CommonDescriptor.getInstance().getConfig().getThriftContainerLengthLimit(); if (compress) { - return new TCompactProtocol.Factory(THRIFT_NO_STRING_LENGTH_LIMIT, containerLengthLimit); + return compactProtocolFactory; } else { - return new TBinaryProtocol.Factory(THRIFT_NO_STRING_LENGTH_LIMIT, containerLengthLimit); + return binaryProtocolFactory; } } diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/service/AbstractThriftServiceThreadTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/service/AbstractThriftServiceThreadTest.java deleted file mode 100644 index d1beb0e34fb8..000000000000 --- a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/service/AbstractThriftServiceThreadTest.java +++ /dev/null @@ -1,112 +0,0 @@ -/* - * 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.iotdb.commons.service; - -import org.apache.iotdb.commons.conf.CommonConfig; -import org.apache.iotdb.commons.conf.CommonDescriptor; -import org.apache.iotdb.service.rpc.thrift.TSInsertRecordReq; - -import org.apache.thrift.TException; -import org.apache.thrift.protocol.TBinaryProtocol; -import org.apache.thrift.protocol.TCompactProtocol; -import org.apache.thrift.protocol.TProtocol; -import org.apache.thrift.protocol.TProtocolException; -import org.apache.thrift.protocol.TType; -import org.apache.thrift.transport.TMemoryBuffer; -import org.apache.thrift.transport.TMemoryInputTransport; -import org.junit.Assert; -import org.junit.Test; - -import java.nio.ByteBuffer; -import java.util.Arrays; -import java.util.Collections; - -public class AbstractThriftServiceThreadTest { - - @Test - public void testProtocolFactoriesLimitContainerLength() { - CommonConfig config = CommonDescriptor.getInstance().getConfig(); - int originalLimit = config.getThriftContainerLengthLimit(); - try { - config.setThriftContainerLengthLimit(1); - - Assert.assertThrows( - TProtocolException.class, - () -> - AbstractThriftServiceThread.getProtocolFactory(false) - .getProtocol(new TMemoryInputTransport(binaryListHeader(2))) - .readListBegin()); - Assert.assertThrows( - TProtocolException.class, - () -> - AbstractThriftServiceThread.getProtocolFactory(true) - .getProtocol(new TMemoryInputTransport(compactListHeader(2))) - .readListBegin()); - } finally { - config.setThriftContainerLengthLimit(originalLimit); - } - } - - @Test - public void testGeneratedRequestRejectsOversizedByteList() throws TException { - CommonConfig config = CommonDescriptor.getInstance().getConfig(); - int originalLimit = config.getThriftContainerLengthLimit(); - try { - config.setThriftContainerLengthLimit(1); - - Assert.assertThrows( - TProtocolException.class, - () -> - new TSInsertRecordReq() - .read( - AbstractThriftServiceThread.getProtocolFactory(false) - .getProtocol(new TMemoryInputTransport(serializeRequest(false))))); - Assert.assertThrows( - TProtocolException.class, - () -> - new TSInsertRecordReq() - .read( - AbstractThriftServiceThread.getProtocolFactory(true) - .getProtocol(new TMemoryInputTransport(serializeRequest(true))))); - } finally { - config.setThriftContainerLengthLimit(originalLimit); - } - } - - private static byte[] serializeRequest(boolean compact) throws TException { - TSInsertRecordReq request = - new TSInsertRecordReq(1, "root.sg.d", Collections.emptyList(), ByteBuffer.allocate(0), 1) - .setColumnCategoryies(Arrays.asList((byte) 0, (byte) 1)); - TMemoryBuffer buffer = new TMemoryBuffer(128); - TProtocol protocol = compact ? new TCompactProtocol(buffer) : new TBinaryProtocol(buffer); - request.write(protocol); - return Arrays.copyOf(buffer.getArray(), buffer.length()); - } - - private static byte[] binaryListHeader(int size) { - return new byte[] { - TType.BYTE, (byte) (size >>> 24), (byte) (size >>> 16), (byte) (size >>> 8), (byte) size, 0, 0 - }; - } - - private static byte[] compactListHeader(int size) { - return new byte[] {(byte) 0xF3, (byte) size, 0, 0}; - } -}