apache · Caideyipi · Jul 10, 2026
diff --git a/...test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBLegacyPipeReceiverSecurityIT.java b/...test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBLegacyPipeReceiverSecurityIT.java
@@ -22,7 +22,10 @@
 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,22 +38,34 @@
 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;
 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 void testRejectPathTraversalFileNameInLegacyTransportFile() throws Except
     }
   }
 
+  @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/.../java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/IoTDBLegacyPipeReceiverAgent.java b/.../java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/IoTDBLegacyPipeReceiverAgent.java
@@ -22,21 +22,14 @@
 
 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.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 TSStatus handshake(
       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 @@ private void createConnection(final SyncIdentityInfo identityInfo) {
     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 @@ private boolean registerDatabase(
    * @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 TSStatus transportPipeData(ByteBuffer buff) throws TException {
         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 @@ private SyncIdentityInfo getCurrentSyncIdentityInfo() {
     }
   }
 
+  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/...rc/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/loader/DeletionLoader.java b/...rc/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.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 DeletionLoader(Deletion deletion) {
   }
 
   @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/...anode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/loader/ILoader.java b/...anode/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/.../src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/loader/TsFileLoader.java b/.../src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/loader/TsFileLoader.java
@@ -21,7 +21,6 @@
 
 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.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,22 +50,23 @@ public TsFileLoader(File tsFile, String database) {
   }
 
   @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 =
           Coordinator.getInstance()
               .executeForTreeModel(
                   statement,
                   queryId,
-                  new SessionInfo(0, AuthorityChecker.SUPER_USER, ZoneId.systemDefault(), ""),
+                  sessionInfo,
                   "",
                   PARTITION_FETCHER,
                   SCHEMA_FETCHER,

diff --git a/...datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java b/...datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java
@@ -2776,7 +2776,10 @@ public TSStatus sendPipeData(final ByteBuffer buff) throws TException {
       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();
     }