This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 2c089d39326 Pipe: use session context in legacy receiver loaders
(#18111) (#18184)
2c089d39326 is described below
commit 2c089d393263b4605450be6d82547612e2bf24b3
Author: Caideyipi <[email protected]>
AuthorDate: Mon Jul 13 10:47:28 2026 +0800
Pipe: use session context in legacy receiver loaders (#18111) (#18184)
* Pipe: use session context in legacy receiver loaders
* Update DeletionLoader.java
* Fix legacy pipe receiver i18n message
---
.../single/IoTDBLegacyPipeReceiverSecurityIT.java | 109 ++++++++++++++++++++-
.../legacy/IoTDBLegacyPipeReceiverAgent.java | 70 +++----------
.../protocol/legacy/loader/DeletionLoader.java | 12 ++-
.../receiver/protocol/legacy/loader/ILoader.java | 3 +-
.../protocol/legacy/loader/TsFileLoader.java | 9 +-
.../protocol/thrift/impl/ClientRPCServiceImpl.java | 5 +-
6 files changed, 142 insertions(+), 66 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBLegacyPipeReceiverSecurityIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBLegacyPipeReceiverSecurityIT.java
index 51e1e211535..8952c1ec4dd 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBLegacyPipeReceiverSecurityIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBLegacyPipeReceiverSecurityIT.java
@@ -22,7 +22,10 @@ package org.apache.iotdb.pipe.it.single;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.client.property.ThriftClientProperty;
import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.commons.pipe.sink.client.IoTDBSyncClient;
+import org.apache.iotdb.db.pipe.sink.payload.legacy.PipeData;
+import org.apache.iotdb.db.storageengine.dataregion.modification.Deletion;
import org.apache.iotdb.it.env.EnvFactory;
import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
import org.apache.iotdb.it.framework.IoTDBTestRunner;
@@ -35,6 +38,7 @@ import org.apache.iotdb.service.rpc.thrift.TSProtocolVersion;
import org.apache.iotdb.service.rpc.thrift.TSyncIdentityInfo;
import org.apache.iotdb.service.rpc.thrift.TSyncTransportMetaInfo;
+import org.apache.tsfile.utils.ReadWriteIOUtils;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
@@ -42,15 +46,26 @@ import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
import java.io.File;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
import java.time.ZoneId;
@RunWith(IoTDBTestRunner.class)
@Category({LocalStandaloneIT.class})
public class IoTDBLegacyPipeReceiverSecurityIT {
+ private static final String LEGACY_PIPE_USER = "pipeHack";
+ private static final String LEGACY_PIPE_PASSWORD = "StrngPsWd@623451";
+ private static final String LEGACY_DATABASE = "root.legacy_poc";
+ private static final String LEGACY_TIMESERIES = LEGACY_DATABASE + ".d1.s1";
+
@BeforeClass
public static void setUp() {
EnvFactory.getEnv().initClusterEnvironment();
@@ -99,11 +114,101 @@ public class IoTDBLegacyPipeReceiverSecurityIT {
}
}
+ @Test
+ public void testLegacyPipeDataDeleteUsesAuthenticatedUserPermission() throws
Exception {
+ prepareLegacyPipePrivilegeEscalationData();
+ assertDirectDeleteDeniedForLegacyPipeUser();
+
+ final DataNodeWrapper dataNode = EnvFactory.getEnv().getDataNodeWrapper(0);
+ try (final IoTDBSyncClient client =
+ new IoTDBSyncClient(
+ new ThriftClientProperty.Builder().build(),
+ dataNode.getIp(),
+ dataNode.getPort(),
+ false,
+ null,
+ null)) {
+ final TSOpenSessionResp openSessionResp =
+ client.openSession(createOpenSessionReq(LEGACY_PIPE_USER,
LEGACY_PIPE_PASSWORD));
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
openSessionResp.getStatus().getCode());
+
+ try {
+ final TSStatus handshakeStatus =
+ client.handshake(
+ new TSyncIdentityInfo(
+ "legacyPipePrivilege", System.currentTimeMillis(),
"UNKNOWN", ""));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
handshakeStatus.getCode());
+
+ final TSStatus status =
+
client.sendPipeData(ByteBuffer.wrap(createDeletionPipeDataPayload()));
+
+ Assert.assertEquals(TSStatusCode.PIPESERVER_ERROR.getStatusCode(),
status.getCode());
+ } finally {
+ client.closeSession(new
TSCloseSessionReq(openSessionResp.getSessionId()));
+ }
+ }
+
+ assertLegacyPocRowCount(2);
+ }
+
+ private void prepareLegacyPipePrivilegeEscalationData() throws SQLException {
+ try (final Connection connection = EnvFactory.getEnv().getConnection();
+ final Statement statement = connection.createStatement()) {
+ statement.execute("CREATE DATABASE " + LEGACY_DATABASE);
+ statement.execute(
+ "CREATE TIMESERIES " + LEGACY_TIMESERIES + " WITH
DATATYPE=INT64,ENCODING=PLAIN");
+ statement.execute("INSERT INTO root.legacy_poc.d1(time,s1) VALUES
(1,1),(2,2)");
+ statement.execute("CREATE USER " + LEGACY_PIPE_USER + " '" +
LEGACY_PIPE_PASSWORD + "'");
+ statement.execute("GRANT USE_PIPE ON root.** TO USER " +
LEGACY_PIPE_USER);
+ }
+ }
+
+ private void assertDirectDeleteDeniedForLegacyPipeUser() throws SQLException
{
+ try (final Connection connection =
+ EnvFactory.getEnv().getConnection(LEGACY_PIPE_USER,
LEGACY_PIPE_PASSWORD);
+ final Statement statement = connection.createStatement()) {
+ final SQLException exception =
+ Assert.assertThrows(
+ SQLException.class,
+ () -> statement.execute("DELETE FROM " + LEGACY_TIMESERIES + "
WHERE time <= 1"));
+ Assert.assertTrue(exception.getMessage().contains("WRITE_DATA"));
+ }
+ }
+
+ private byte[] createDeletionPipeDataPayload() throws Exception {
+ final ByteArrayOutputStream byteStream = new ByteArrayOutputStream();
+ try (final DataOutputStream stream = new DataOutputStream(byteStream)) {
+ stream.writeByte(PipeData.PipeDataType.DELETION.getType());
+ stream.writeLong(1L);
+ ReadWriteIOUtils.write(LEGACY_DATABASE, stream);
+ new Deletion(new MeasurementPath(LEGACY_TIMESERIES), 0, Long.MIN_VALUE,
1)
+ .serializeWithoutFileOffset(stream);
+ }
+ return byteStream.toByteArray();
+ }
+
+ private void assertLegacyPocRowCount(final int expectedCount) throws
SQLException {
+ try (final Connection connection = EnvFactory.getEnv().getConnection();
+ final Statement statement = connection.createStatement();
+ final ResultSet resultSet = statement.executeQuery("SELECT s1 FROM
root.legacy_poc.d1")) {
+ int actualCount = 0;
+ while (resultSet.next()) {
+ ++actualCount;
+ }
+ Assert.assertEquals(expectedCount, actualCount);
+ }
+ }
+
private TSOpenSessionReq createOpenSessionReq() {
+ return createOpenSessionReq("root", "root");
+ }
+
+ private TSOpenSessionReq createOpenSessionReq(final String username, final
String password) {
final TSOpenSessionReq req = new TSOpenSessionReq();
req.setClient_protocol(TSProtocolVersion.IOTDB_SERVICE_PROTOCOL_V3);
- req.setUsername("root");
- req.setPassword("root");
+ req.setUsername(username);
+ req.setPassword(password);
req.setZoneId(ZoneId.systemDefault().toString());
req.putToConfiguration("version",
IoTDBConstant.ClientVersion.V_1_0.toString());
req.putToConfiguration("sql_dialect", "tree");
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/IoTDBLegacyPipeReceiverAgent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/IoTDBLegacyPipeReceiverAgent.java
index b9e26ee0fa5..655bd0a1e19 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/IoTDBLegacyPipeReceiverAgent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/IoTDBLegacyPipeReceiverAgent.java
@@ -22,21 +22,14 @@ package org.apache.iotdb.db.pipe.receiver.protocol.legacy;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.conf.CommonDescriptor;
-import org.apache.iotdb.commons.exception.IllegalPathException;
-import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.pipe.receiver.PipeReceiverFilePathUtils;
import org.apache.iotdb.commons.utils.FileUtils;
-import org.apache.iotdb.db.auth.AuthorityChecker;
-import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.pipe.sink.payload.legacy.PipeData;
import org.apache.iotdb.db.pipe.sink.payload.legacy.TsFilePipeData;
import org.apache.iotdb.db.protocol.session.SessionManager;
import org.apache.iotdb.db.queryengine.common.SessionInfo;
-import org.apache.iotdb.db.queryengine.plan.Coordinator;
import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
-import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
-import
org.apache.iotdb.db.queryengine.plan.statement.metadata.DatabaseSchemaStatement;
import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -53,7 +46,6 @@ import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.file.Paths;
-import java.time.ZoneId;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
@@ -76,8 +68,6 @@ public class IoTDBLegacyPipeReceiverAgent {
private final Map<Long, Map<String, Long>> connectionIdToStartIndexRecord =
new ConcurrentHashMap<>();
- private final Map<String, String> registeredDatabase = new
ConcurrentHashMap<>();
-
// The sync connectionId is unique in one IoTDB instance.
private final AtomicLong connectionIdGenerator = new AtomicLong();
@@ -118,12 +108,6 @@ public class IoTDBLegacyPipeReceiverAgent {
new File(getFileDataDir(identityInfo)).mkdirs();
}
createConnection(identityInfo);
- if (!StringUtils.isEmpty(identityInfo.getDatabase())
- && !registerDatabase(identityInfo.getDatabase(), partitionFetcher,
schemaFetcher)) {
- return RpcUtils.getStatus(
- TSStatusCode.PIPESERVER_ERROR,
- String.format("Auto register database %s error.",
identityInfo.getDatabase()));
- }
return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS, "");
}
@@ -138,42 +122,6 @@ public class IoTDBLegacyPipeReceiverAgent {
connectionIdToIdentityInfoMap.put(connectionId, identityInfo);
}
- private boolean registerDatabase(
- String database, IPartitionFetcher partitionFetcher, ISchemaFetcher
schemaFetcher) {
- if (registeredDatabase.containsKey(database)) {
- return true;
- }
- try {
- DatabaseSchemaStatement statement =
- new
DatabaseSchemaStatement(DatabaseSchemaStatement.DatabaseSchemaStatementType.CREATE);
- statement.setDatabasePath(new PartialPath(database));
- long queryId = SessionManager.getInstance().requestQueryId();
- ExecutionResult result =
- Coordinator.getInstance()
- .executeForTreeModel(
- statement,
- queryId,
- new SessionInfo(0, AuthorityChecker.SUPER_USER,
ZoneId.systemDefault(), ""),
- "",
- partitionFetcher,
- schemaFetcher,
-
IoTDBDescriptor.getInstance().getConfig().getQueryTimeoutThreshold(),
- false);
- if (result.status.code != TSStatusCode.SUCCESS_STATUS.getStatusCode()
- && result.status.code !=
TSStatusCode.DATABASE_ALREADY_EXISTS.getStatusCode()) {
- LOGGER.error(
- "Create Database error, statement: {}, result status : {}.",
statement, result.status);
- return false;
- }
- } catch (IllegalPathException e) {
- LOGGER.error("Parse database PartialPath {} error", database, e);
- return false;
- }
-
- registeredDatabase.put(database, "");
- return true;
- }
-
/**
* Receive {@link PipeData} and load it into IoTDB Engine.
*
@@ -182,7 +130,12 @@ public class IoTDBLegacyPipeReceiverAgent {
* @throws TException The connection between the sender and the receiver has
not been established
* by {@link IoTDBLegacyPipeReceiverAgent#handshake}
*/
- public TSStatus transportPipeData(ByteBuffer buff) throws TException {
+ public TSStatus transportPipeData(final ByteBuffer buff) throws TException {
+ return transportPipeData(buff, null);
+ }
+
+ public TSStatus transportPipeData(final ByteBuffer buff, final SessionInfo
sessionInfo)
+ throws TException {
// step1. check connection
SyncIdentityInfo identityInfo = getCurrentSyncIdentityInfo();
if (identityInfo == null) {
@@ -217,7 +170,7 @@ public class IoTDBLegacyPipeReceiverAgent {
pipeData.getPipeDataType(),
pipeData);
try {
- pipeData.createLoader().load();
+ pipeData.createLoader().load(sessionInfo == null ?
getCurrentSessionInfo() : sessionInfo);
LOGGER.info(
"Load pipeData with serialize number {} successfully.",
pipeData.getSerialNumber());
} catch (PipeException e) {
@@ -243,6 +196,15 @@ public class IoTDBLegacyPipeReceiverAgent {
}
}
+ private SessionInfo getCurrentSessionInfo() {
+ final SessionManager sessionManager = SessionManager.getInstance();
+ if (!sessionManager.checkLogin(sessionManager.getCurrSession())) {
+ throw new PipeException("Legacy pipe receiver requires a logged-in
session.");
+ }
+
+ return sessionManager.getSessionInfo(sessionManager.getCurrSession());
+ }
+
/**
* handle when successfully receive tsFilePipeData. Rename .patch file and
reset tsFilePipeData's
* path.
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/loader/DeletionLoader.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/loader/DeletionLoader.java
index 019dfb7718b..1c8771e218c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/loader/DeletionLoader.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/loader/DeletionLoader.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.pipe.receiver.protocol.legacy.loader;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.db.auth.AuthorityChecker;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -37,7 +38,6 @@ import org.apache.iotdb.rpc.TSStatusCode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.time.ZoneId;
import java.util.Collections;
/** This loader is used to load deletion plan. */
@@ -52,19 +52,25 @@ public class DeletionLoader implements ILoader {
}
@Override
- public void load() throws PipeException {
+ public void load(final SessionInfo sessionInfo) throws PipeException {
if (CommonDescriptor.getInstance().getConfig().isReadOnly()) {
throw new PipeException("storage engine readonly");
}
try {
Statement statement = generateStatement();
+ TSStatus authorityStatus =
+ AuthorityChecker.checkAuthority(statement,
sessionInfo.getUserName());
+ if (authorityStatus.getCode() !=
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new PipeException(authorityStatus.getMessage());
+ }
+
long queryId = SessionManager.getInstance().requestQueryId();
ExecutionResult result =
Coordinator.getInstance()
.executeForTreeModel(
statement,
queryId,
- new SessionInfo(0, AuthorityChecker.SUPER_USER,
ZoneId.systemDefault(), ""),
+ sessionInfo,
"",
PARTITION_FETCHER,
SCHEMA_FETCHER,
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/loader/ILoader.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/loader/ILoader.java
index 22f9f4d44b2..b4aba1ddaae 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/loader/ILoader.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/loader/ILoader.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.pipe.receiver.protocol.legacy.loader;
+import org.apache.iotdb.db.queryengine.common.SessionInfo;
import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
import
org.apache.iotdb.db.queryengine.plan.analyze.schema.ClusterSchemaFetcher;
@@ -34,5 +35,5 @@ public interface ILoader {
ISchemaFetcher SCHEMA_FETCHER = ClusterSchemaFetcher.getInstance();
- void load();
+ void load(SessionInfo sessionInfo);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/loader/TsFileLoader.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/loader/TsFileLoader.java
index 8459e22f1ee..34028029de3 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/loader/TsFileLoader.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/loader/TsFileLoader.java
@@ -21,7 +21,6 @@ package
org.apache.iotdb.db.pipe.receiver.protocol.legacy.loader;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.PartialPath;
-import org.apache.iotdb.db.auth.AuthorityChecker;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.load.LoadFileException;
import org.apache.iotdb.db.protocol.session.SessionManager;
@@ -36,7 +35,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
-import java.time.ZoneId;
/** This loader is used to load tsFiles. If .mods file exists, it will be
loaded as well. */
public class TsFileLoader implements ILoader {
@@ -52,14 +50,15 @@ public class TsFileLoader implements ILoader {
}
@Override
- public void load() {
+ public void load(final SessionInfo sessionInfo) {
try {
LoadTsFileStatement statement =
LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath());
statement.setDeleteAfterLoad(true);
statement.setConvertOnTypeMismatch(true);
statement.setDatabaseLevel(parseSgLevel());
statement.setVerifySchema(true);
- statement.setAutoCreateDatabase(false);
+ statement.setAutoCreateDatabase(
+
IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled());
long queryId = SessionManager.getInstance().requestQueryId();
ExecutionResult result =
@@ -67,7 +66,7 @@ public class TsFileLoader implements ILoader {
.executeForTreeModel(
statement,
queryId,
- new SessionInfo(0, AuthorityChecker.SUPER_USER,
ZoneId.systemDefault(), ""),
+ sessionInfo,
"",
PARTITION_FETCHER,
SCHEMA_FETCHER,
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java
index 370e3ead82b..1b2f5f400b2 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java
@@ -2776,7 +2776,10 @@ public class ClientRPCServiceImpl implements
IClientRPCServiceWithHandler {
if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
return status;
}
- return PipeDataNodeAgent.receiver().legacy().transportPipeData(buff);
+ return PipeDataNodeAgent.receiver()
+ .legacy()
+ .transportPipeData(
+ buff,
SESSION_MANAGER.getSessionInfo(SESSION_MANAGER.getCurrSession()));
} finally {
SESSION_MANAGER.updateIdleTime();
}
