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 51e1e211535e8..8952c1ec4dd87 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 @@ 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.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.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/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 b9e26ee0fa57d..655bd0a1e19cc 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 @@ 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> connectionIdToStartIndexRecord = new ConcurrentHashMap<>(); - private final Map 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/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 019dfb7718b21..1c8771e218c56 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.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/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 22f9f4d44b277..b4aba1ddaae32 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 8459e22f1eed3..34028029de3ec 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 @@ 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,14 +50,15 @@ 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 = @@ -67,7 +66,7 @@ public void load() { .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 370e3ead82bd1..1b2f5f400b21f 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 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(); }