diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileAuthIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileAuthIT.java new file mode 100644 index 0000000000000..d842941e0cc9c --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileAuthIT.java @@ -0,0 +1,285 @@ +/* + * 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.db.it; + +import org.apache.iotdb.commons.auth.entity.PrivilegeType; +import org.apache.iotdb.it.env.EnvFactory; +import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper; +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.it.utils.TsFileGenerator; +import org.apache.iotdb.itbase.category.ClusterIT; +import org.apache.iotdb.itbase.category.LocalStandaloneIT; +import org.apache.iotdb.jdbc.IoTDBSQLException; + +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.external.commons.io.FileUtils; +import org.apache.tsfile.file.metadata.enums.TSEncoding; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.junit.After; +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.File; +import java.nio.file.Files; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.Collections; +import java.util.concurrent.TimeUnit; + +import static org.apache.iotdb.db.it.utils.TestUtils.assertNonQueryTestFail; +import static org.apache.iotdb.db.it.utils.TestUtils.createUser; +import static org.apache.iotdb.db.it.utils.TestUtils.executeNonQuery; +import static org.apache.iotdb.db.it.utils.TestUtils.grantUserSeriesPrivilege; + +@RunWith(IoTDBTestRunner.class) +@Category({LocalStandaloneIT.class, ClusterIT.class}) +public class IoTDBLoadTsFileAuthIT { + private static final long PARTITION_INTERVAL = 10 * 1000L; + private static final String DATABASE = "root.load_auth"; + private static final String DEVICE = DATABASE + ".d1"; + private static final IMeasurementSchema MEASUREMENT = + new MeasurementSchema("s1", TSDataType.INT32, TSEncoding.RLE); + private static final String NO_WRITE_USER = "load_no_write_user"; + private static final String WRITE_USER = "load_write_user"; + private static final String OTHER_PATH_WRITE_USER = "load_other_path_write_user"; + private static final String ASYNC_NO_WRITE_USER = "async_load_no_write_user"; + private static final String ASYNC_WRITE_USER = "async_load_write_user"; + private static final String PASSWORD = "test123123456"; + private static final long UNALLOCATABLE_TABLET_CONVERSION_BATCH_MEMORY_SIZE_IN_BYTES = + Long.MAX_VALUE / 4; + + private static File tmpDir; + + @BeforeClass + public static void setUp() throws Exception { + tmpDir = new File(Files.createTempDirectory("load-auth").toUri()); + EnvFactory.getEnv().getConfig().getCommonConfig().setTimePartitionInterval(PARTITION_INTERVAL); + EnvFactory.getEnv().getConfig().getCommonConfig().setEnforceStrongPassword(false); + EnvFactory.getEnv().getConfig().getCommonConfig().setAutoCreateSchemaEnabled(false); + EnvFactory.getEnv() + .getConfig() + .getDataNodeConfig() + .setMaxAllocateMemoryRatioForLoad(1.0) + .setLoadTsFileAnalyzeSchemaMemorySizeInBytes(10 * 1024L) + .setLoadTsFileTabletConversionBatchMemorySizeInBytes( + UNALLOCATABLE_TABLET_CONVERSION_BATCH_MEMORY_SIZE_IN_BYTES) + .setLoadActiveListeningCheckIntervalSeconds(1); + + EnvFactory.getEnv().initClusterEnvironment(); + } + + @AfterClass + public static void tearDown() throws Exception { + deleteDatabase(); + EnvFactory.getEnv().cleanClusterEnvironment(); + FileUtils.deleteDirectory(tmpDir); + } + + @After + public void cleanData() throws Exception { + deleteDatabase(); + } + + @Test + public void testLoadWithoutSchemaCheckStillChecksWriteDataPermission() throws Exception { + final File tsFile = new File(tmpDir, "1-0-0-0.tsfile"); + prepareSchemaAndTsFile(tsFile); + createUser(NO_WRITE_USER, PASSWORD); + + assertNonQueryTestFail( + String.format("load \"%s\" with ('database-level'='2', 'verify'='false')", tsFile), + "No permissions for this operation, please add privilege WRITE_DATA", + NO_WRITE_USER, + PASSWORD); + } + + @Test + public void testLoadWithoutSchemaCheckAllowsUserWithWriteDataPermission() throws Exception { + final File tsFile = new File(tmpDir, "2-0-0-0.tsfile"); + prepareSchemaAndTsFile(tsFile); + createUser(WRITE_USER, PASSWORD); + grantUserSeriesPrivilege(WRITE_USER, PrivilegeType.WRITE_DATA, DATABASE + ".**"); + + executeNonQuery( + String.format("load \"%s\" with ('database-level'='2', 'verify'='false')", tsFile), + WRITE_USER, + PASSWORD); + + try (final Connection connection = EnvFactory.getEnv().getConnection(); + final Statement statement = connection.createStatement(); + final ResultSet resultSet = statement.executeQuery("select count(s1) from " + DEVICE)) { + Assert.assertTrue(resultSet.next()); + Assert.assertEquals(10, resultSet.getLong(1)); + } + } + + @Test + public void testLoadWithoutSchemaCheckRejectsUserWithOtherPathWriteDataPermission() + throws Exception { + final File tsFile = new File(tmpDir, "3-0-0-0.tsfile"); + prepareSchemaAndTsFile(tsFile); + createUser(OTHER_PATH_WRITE_USER, PASSWORD); + grantUserSeriesPrivilege(OTHER_PATH_WRITE_USER, PrivilegeType.WRITE_DATA, "root.other.**"); + + assertNonQueryTestFail( + String.format("load \"%s\" with ('database-level'='2', 'verify'='false')", tsFile), + "No permissions for this operation, please add privilege WRITE_DATA", + OTHER_PATH_WRITE_USER, + PASSWORD); + } + + @Test + public void testAsyncLoadShouldCheckWriteDataPermissionWithStoredUser() throws Exception { + final File noWriteTsFile = new File(tmpDir, "4-0-0-0.tsfile"); + final File writeTsFile = new File(tmpDir, "5-0-0-0.tsfile"); + prepareSchemaAndTsFile(noWriteTsFile); + generateTsFile(writeTsFile); + createUser(ASYNC_NO_WRITE_USER, PASSWORD); + createUser(ASYNC_WRITE_USER, PASSWORD); + grantUserSeriesPrivilege(ASYNC_WRITE_USER, PrivilegeType.WRITE_DATA, DATABASE + ".**"); + + executeNonQuery( + String.format( + "load \"%s\" with ('database-level'='2', 'async'='true', 'on-success'='none', " + + "'verify'='false')", + noWriteTsFile.getAbsolutePath()), + ASYNC_NO_WRITE_USER, + PASSWORD); + executeNonQuery( + String.format( + "load \"%s\" with ('database-level'='2', 'async'='true', 'on-success'='none', " + + "'verify'='false')", + writeTsFile.getAbsolutePath()), + ASYNC_WRITE_USER, + PASSWORD); + + waitUntilAllActiveLoadPendingDirsAreEmpty(TimeUnit.SECONDS.toMillis(60)); + assertCountEventually(10, TimeUnit.SECONDS.toMillis(60)); + } + + private static void prepareSchemaAndTsFile(final File tsFile) throws Exception { + prepareSchema(MEASUREMENT.getType()); + generateTsFile(tsFile); + } + + private static void prepareSchema(final TSDataType dataType) throws Exception { + try (final Connection connection = EnvFactory.getEnv().getConnection(); + final Statement statement = connection.createStatement()) { + statement.execute("create database " + DATABASE); + statement.execute( + String.format( + "create timeseries %s.%s %s", DEVICE, MEASUREMENT.getMeasurementName(), dataType)); + } + } + + private static void generateTsFile(final File tsFile) throws Exception { + try (final TsFileGenerator generator = new TsFileGenerator(tsFile)) { + generator.registerTimeseries(DEVICE, Collections.singletonList(MEASUREMENT)); + generator.generateData(DEVICE, 10, PARTITION_INTERVAL / 10, false); + } + } + + private static void deleteDatabase() throws Exception { + try (final Connection connection = EnvFactory.getEnv().getConnection(); + final Statement statement = connection.createStatement()) { + statement.execute("delete database " + DATABASE); + } catch (final IoTDBSQLException ignored) { + } + } + + private File getActiveLoadPendingDir(final DataNodeWrapper dataNodeWrapper) { + return new File( + dataNodeWrapper.getNodePath() + + File.separator + + "ext" + + File.separator + + "load" + + File.separator + + "pending"); + } + + private void waitUntilAllActiveLoadPendingDirsAreEmpty(final long timeoutMs) + throws InterruptedException { + final long deadline = System.currentTimeMillis() + timeoutMs; + while (System.currentTimeMillis() < deadline) { + boolean hasTsFile = false; + for (final DataNodeWrapper dataNodeWrapper : EnvFactory.getEnv().getDataNodeWrapperList()) { + if (containsTsFile(getActiveLoadPendingDir(dataNodeWrapper))) { + hasTsFile = true; + break; + } + } + if (!hasTsFile) { + return; + } + Thread.sleep(500L); + } + Assert.fail("Timed out waiting for active load pending dirs to become empty"); + } + + private void assertCountEventually(final long expected, final long timeoutMs) throws Exception { + final long deadline = System.currentTimeMillis() + timeoutMs; + AssertionError lastError = null; + while (System.currentTimeMillis() < deadline) { + try (final Connection connection = EnvFactory.getEnv().getConnection(); + final Statement statement = connection.createStatement(); + final ResultSet resultSet = + statement.executeQuery( + "select count(" + MEASUREMENT.getMeasurementName() + ") from " + DEVICE)) { + Assert.assertTrue(resultSet.next()); + Assert.assertEquals(expected, resultSet.getLong(1)); + return; + } catch (final AssertionError e) { + lastError = e; + } + Thread.sleep(500L); + } + if (lastError != null) { + throw lastError; + } + Assert.fail("Timed out waiting for count " + expected); + } + + private boolean containsTsFile(final File root) { + if (root == null || !root.exists()) { + return false; + } + if (root.isFile()) { + return root.getName().endsWith(".tsfile"); + } + + final File[] children = root.listFiles(); + if (children == null) { + return false; + } + for (final File child : children) { + if (containsTsFile(child)) { + return true; + } + } + return false; + } +} diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java index f8b6e50ed5372..512879bfd3cb0 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java @@ -538,11 +538,14 @@ private StorageEngineMessages() {} public static final String FAILED_COUNT_ACTIVE_DIRS_FILE_NUMBER = "Failed to count active listening dirs file number."; public static final String ACTIVE_LOAD_METRIC_COLLECTOR_REGISTERED = "Active load metric collector periodical jobs registered"; public static final String DATABASE_NAME_MUST_NOT_BE_EMPTY = "Database name must not be empty."; + public static final String USER_NAME_MUST_NOT_BE_EMPTY = "User name must not be empty."; public static final String ERROR_EXECUTING_ACTIVE_LOAD_JOB = "Error occurred when executing active load periodical job."; public static final String ACTIVE_LOAD_EXECUTOR_STARTED = "Active load periodical jobs executor is started successfully."; public static final String ACTIVE_LOAD_EXECUTOR_STOPPED = "Active load periodical jobs executor is stopped successfully."; public static final String ACTIVE_LOAD_TEMPORARILY_UNAVAILABLE = "Rejecting auto load tsfile {} (isGeneratedByPipe = {}) due to temporary unavailability, will retry later. Status: {}"; + public static final String USER_IN_ACTIVE_LOAD_PATH_DOES_NOT_EXIST = + "The user in the active load path does not exist"; public static final String ERROR_MOVING_FILE_TO_FAIL_DIR = "Error occurred during moving file {} to fail directory."; public static final String FAILED_COUNT_FILES_IN_FAIL_DIR = "Failed to count failed files in fail directory."; diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java index bffda3d52879e..96d09c1c14140 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java @@ -538,11 +538,14 @@ private StorageEngineMessages() {} public static final String FAILED_COUNT_ACTIVE_DIRS_FILE_NUMBER = "统计 Active 监听目录文件数量失败。"; public static final String ACTIVE_LOAD_METRIC_COLLECTOR_REGISTERED = "Active 加载指标收集定期任务已注册"; public static final String DATABASE_NAME_MUST_NOT_BE_EMPTY = "数据库名称不能为空。"; + public static final String USER_NAME_MUST_NOT_BE_EMPTY = "用户名不能为空。"; public static final String ERROR_EXECUTING_ACTIVE_LOAD_JOB = "执行 Active 加载定期任务时发生错误。"; public static final String ACTIVE_LOAD_EXECUTOR_STARTED = "Active 加载定期任务执行器已成功启动。"; public static final String ACTIVE_LOAD_EXECUTOR_STOPPED = "Active 加载定期任务执行器已成功停止。"; public static final String ACTIVE_LOAD_TEMPORARILY_UNAVAILABLE = "拒绝自动加载 TsFile {} (isGeneratedByPipe = {})，原因是系统暂时不可用，将稍后重试。状态: {}"; + public static final String USER_IN_ACTIVE_LOAD_PATH_DOES_NOT_EXIST = + "Active 加载路径中的用户不存在"; public static final String ERROR_MOVING_FILE_TO_FAIL_DIR = "将文件 {} 移动到失败目录时发生错误。"; public static final String FAILED_COUNT_FILES_IN_FAIL_DIR = "统计失败目录中的失败文件数量失败。"; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java index 46c00a095c841..2697eb0bf7e31 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java @@ -653,7 +653,8 @@ static Map buildLoadTsFileAttributesForAsync( shouldConvertDataTypeOnTypeMismatch, validateTsFile || shouldConvertDataTypeOnTypeMismatch, null, - shouldMarkAsPipeRequest); + shouldMarkAsPipeRequest, + AuthorityChecker.SUPER_USER); } private TSStatus loadTsFileSync(final String dataBaseName, final String fileAbsolutePath) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java index 89b2f8f2c5238..f459c899f0432 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java @@ -295,7 +295,8 @@ private boolean doAsyncLoad(final IAnalysis analysis) { isConvertOnTypeMismatch, isVerifySchema, tabletConversionThresholdBytes, - isGeneratedByPipe); + isGeneratedByPipe, + Objects.nonNull(context) ? context.getUsername() : null); if (LoadUtil.loadTsFileAsyncToActiveDir(tsFiles, activeLoadAttributes, isDeleteAfterLoad)) { analysis.setFinishQueryAfterAnalyze(true); @@ -528,6 +529,8 @@ private void doAnalyzeSingleTreeFile( if (isAutoCreateSchemaOrVerifySchemaEnabled) { getOrCreateTreeSchemaVerifier().autoCreateAndVerify(reader, device2TimeseriesMetadata); + } else { + getOrCreateTreeSchemaVerifier().checkWritePermission(device2TimeseriesMetadata); } // TODO: how to get the correct write point count when diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/TreeSchemaAutoCreatorAndVerifier.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/TreeSchemaAutoCreatorAndVerifier.java index a5e8f2d1c3c9e..6d1704c0cfd0a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/TreeSchemaAutoCreatorAndVerifier.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/TreeSchemaAutoCreatorAndVerifier.java @@ -154,20 +154,7 @@ public void autoCreateAndVerify( // not a timeseries, skip } else { // check WRITE_DATA permission of timeseries - long startTime = System.nanoTime(); - try { - UserEntity userEntity = loadTsFileAnalyzer.context.getSession().getUserEntity(); - TSStatus status = - AuthorityChecker.getAccessControl() - .checkFullPathWriteDataPermission( - userEntity, device, timeseriesMetadata.getMeasurementId()); - if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { - throw new AuthException( - TSStatusCode.representOf(status.getCode()), status.getMessage()); - } - } finally { - PerformanceOverviewMetrics.getInstance().recordAuthCost(System.nanoTime() - startTime); - } + checkWritePermission(device, timeseriesMetadata.getMeasurementId()); final Pair compressionEncodingPair = reader.readTimeseriesCompressionTypeAndEncoding(timeseriesMetadata); schemaCache.addTimeSeries( @@ -191,6 +178,63 @@ public void autoCreateAndVerify( } } + public void checkWritePermission( + Map> device2TimeseriesMetadataList) throws AuthException { + for (final Map.Entry> entry : + device2TimeseriesMetadataList.entrySet()) { + final IDeviceID device = entry.getKey(); + + try { + if (schemaCache.isDeviceDeletedByMods(device)) { + continue; + } + } catch (IllegalPathException e) { + LOGGER.warn( + DataNodeQueryMessages + .FAILED_TO_CHECK_IF_DEVICE_ARG_IS_DELETED_BY_MODS_WILL_SEE_IT_AS_NOT_DELETED, + device, + e); + } + + for (final TimeseriesMetadata timeseriesMetadata : entry.getValue()) { + try { + if (schemaCache.isTimeSeriesDeletedByMods(device, timeseriesMetadata)) { + continue; + } + } catch (IllegalPathException e) { + if (!timeseriesMetadata.getMeasurementId().isEmpty()) { + LOGGER.warn( + DataNodeQueryMessages + .FAILED_TO_CHECK_IF_DEVICE_ARG_TIMESERIES_ARG_IS_DELETED_BY_MODS_WILL_SEE_IT_AS_NOT, + device, + timeseriesMetadata.getMeasurementId(), + e); + } + } + + if (!TSDataType.VECTOR.equals(timeseriesMetadata.getTsDataType())) { + checkWritePermission(device, timeseriesMetadata.getMeasurementId()); + } + } + } + } + + private void checkWritePermission(final IDeviceID device, final String measurementId) + throws AuthException { + final long startTime = System.nanoTime(); + try { + UserEntity userEntity = loadTsFileAnalyzer.context.getSession().getUserEntity(); + TSStatus status = + AuthorityChecker.getAccessControl() + .checkFullPathWriteDataPermission(userEntity, device, measurementId); + if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + throw new AuthException(TSStatusCode.representOf(status.getCode()), status.getMessage()); + } + } finally { + PerformanceOverviewMetrics.getInstance().recordAuthCost(System.nanoTime() - startTime); + } + } + /** * This can only be invoked after all timeseries in the current tsfile have been processed. * Otherwise, the isAligned status may be wrong. diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPathHelper.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPathHelper.java index 7b971c15a10b1..991d368d39530 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPathHelper.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPathHelper.java @@ -25,6 +25,8 @@ import org.apache.iotdb.db.queryengine.plan.statement.crud.LoadTsFileStatement; import org.apache.iotdb.db.storageengine.load.config.LoadTsFileConfigurator; +import com.google.common.io.BaseEncoding; + import java.io.File; import java.io.UnsupportedEncodingException; import java.net.URLDecoder; @@ -43,10 +45,15 @@ public final class ActiveLoadPathHelper { private static final String SEGMENT_SEPARATOR = "-"; + public static final String USER_KEY = "user"; + // Keep a version in the user path segment so future encryption algorithms can be added safely. + private static final String USER_VALUE_MASK_PREFIX = "v1-"; + private static final BaseEncoding USER_VALUE_ENCODING = BaseEncoding.base32().omitPadding(); private static final List KEY_ORDER = Collections.unmodifiableList( Arrays.asList( + USER_KEY, LoadTsFileConfigurator.DATABASE_NAME_KEY, LoadTsFileConfigurator.DATABASE_LEVEL_KEY, LoadTsFileConfigurator.CONVERT_ON_TYPE_MISMATCH_KEY, @@ -65,8 +72,13 @@ public static Map buildAttributes( final Boolean convertOnTypeMismatch, final Boolean verify, final Long tabletConversionThresholdBytes, - final Boolean pipeGenerated) { + final Boolean pipeGenerated, + final String userName) { final Map attributes = new LinkedHashMap<>(); + if (Objects.nonNull(userName) && !userName.isEmpty()) { + attributes.put(USER_KEY, userName); + } + if (Objects.nonNull(databaseName) && !databaseName.isEmpty()) { attributes.put(LoadTsFileConfigurator.DATABASE_NAME_KEY, databaseName); } @@ -208,7 +220,15 @@ public static boolean containsDatabaseName(final Map attributes) } private static String formatSegment(final String key, final String value) { - return key + SEGMENT_SEPARATOR + encodeValue(value); + return key + SEGMENT_SEPARATOR + encodeValue(maskValueIfNecessary(key, value)); + } + + private static String maskValueIfNecessary(final String key, final String value) { + if (!USER_KEY.equals(key)) { + return value; + } + return USER_VALUE_MASK_PREFIX + + USER_VALUE_ENCODING.encode(value.getBytes(StandardCharsets.UTF_8)); } private static String encodeValue(final String value) { @@ -227,7 +247,11 @@ private static Optional extractAndValidateAttributeValue( } final String encodedValue = dirName.substring(prefixLength); - final String decodedValue = decodeValue(encodedValue); + final String rawDecodedValue = decodeValue(encodedValue); + if (USER_KEY.equals(key) && !rawDecodedValue.startsWith(USER_VALUE_MASK_PREFIX)) { + return Optional.empty(); + } + final String decodedValue = unmaskValueIfNecessary(key, rawDecodedValue); try { validateAttributeValue(key, decodedValue); return Optional.of(decodedValue); @@ -255,6 +279,11 @@ private static void validateAttributeValue(final String key, final String value) case LoadTsFileConfigurator.VERIFY_KEY: LoadTsFileConfigurator.validateVerifyParam(value); break; + case USER_KEY: + if (value == null || value.isEmpty()) { + throw new SemanticException(StorageEngineMessages.USER_NAME_MUST_NOT_BE_EMPTY); + } + break; default: LoadTsFileConfigurator.validateParameters(key, value); } @@ -279,4 +308,19 @@ private static String decodeValue(final String value) { return value; } } + + private static String unmaskValueIfNecessary(final String key, final String value) { + if (!USER_KEY.equals(key) || !value.startsWith(USER_VALUE_MASK_PREFIX)) { + return value; + } + return decodeUserName(value.substring(USER_VALUE_MASK_PREFIX.length()), value); + } + + private static String decodeUserName(final String encodedUserName, final String fallback) { + try { + return new String(USER_VALUE_ENCODING.decode(encodedUserName), StandardCharsets.UTF_8); + } catch (final IllegalArgumentException e) { + return fallback; + } + } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoader.java index 5e6cade3087c4..ff5222e575440 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoader.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoader.java @@ -261,6 +261,15 @@ private TSStatus loadTsFile( : new File(entry.getPendingDir()); final Map attributes = ActiveLoadPathHelper.parseAttributes(tsFile, pendingDir); ActiveLoadPathHelper.applyAttributesToStatement(attributes, statement, isVerify); + final String userName = + attributes.getOrDefault(ActiveLoadPathHelper.USER_KEY, AuthorityChecker.SUPER_USER); + final Optional userId = AuthorityChecker.getUserId(userName); + if (!userId.isPresent()) { + return new TSStatus(TSStatusCode.USER_NOT_EXIST.getStatusCode()) + .setMessage(StorageEngineMessages.USER_IN_ACTIVE_LOAD_PATH_DOES_NOT_EXIST); + } + session.setUserId(userId.get()); + session.setUsername(userName); final File parentFile; if (statement.getDatabase() == null && entry.isTableModel()) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/util/LoadUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/util/LoadUtil.java index cafb1b0e70022..88c71fb1cef28 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/util/LoadUtil.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/util/LoadUtil.java @@ -23,8 +23,11 @@ import org.apache.iotdb.commons.disk.strategy.DirectoryStrategyType; import org.apache.iotdb.commons.exception.DiskSpaceInsufficientException; import org.apache.iotdb.commons.utils.RetryUtils; +import org.apache.iotdb.db.auth.AuthorityChecker; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.i18n.StorageEngineMessages; +import org.apache.iotdb.db.protocol.session.IClientSession; +import org.apache.iotdb.db.protocol.session.SessionManager; import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile; import org.apache.iotdb.db.storageengine.dataregion.modification.v1.ModificationFileV1; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; @@ -37,7 +40,7 @@ import java.io.File; import java.io.IOException; import java.util.Arrays; -import java.util.Collections; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Objects; @@ -124,8 +127,7 @@ private static boolean loadTsFilesToActiveDir( LOGGER.warn(StorageEngineMessages.LOAD_ACTIVE_LISTENING_DIR_NOT_SET); return false; } - final Map attributes = - Objects.nonNull(loadAttributes) ? loadAttributes : Collections.emptyMap(); + final Map attributes = appendCurrentUserIfAbsent(loadAttributes); final File targetDir = ActiveLoadPathHelper.resolveTargetDir(targetFilePath, attributes); loadTsFileAsyncToTargetDir( @@ -138,6 +140,23 @@ private static boolean loadTsFilesToActiveDir( return true; } + private static Map appendCurrentUserIfAbsent( + final Map loadAttributes) { + final Map attributes = + Objects.nonNull(loadAttributes) + ? new LinkedHashMap<>(loadAttributes) + : new LinkedHashMap<>(); + if (!attributes.containsKey(ActiveLoadPathHelper.USER_KEY)) { + final IClientSession session = SessionManager.getInstance().getCurrSession(); + attributes.put( + ActiveLoadPathHelper.USER_KEY, + session == null || session.getUsername() == null + ? AuthorityChecker.SUPER_USER + : session.getUsername()); + } + return attributes; + } + public static boolean loadFilesToActiveDir( final Map loadAttributes, final List files, @@ -161,8 +180,7 @@ public static boolean loadFilesToActiveDir( LOGGER.warn(StorageEngineMessages.LOAD_ACTIVE_LISTENING_DIR_NOT_SET); return false; } - final Map attributes = - Objects.nonNull(loadAttributes) ? loadAttributes : Collections.emptyMap(); + final Map attributes = appendCurrentUserIfAbsent(loadAttributes); final File targetDir = ActiveLoadPathHelper.resolveTargetDir(targetFilePath, attributes); for (final String file : files) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPathHelperTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPathHelperTest.java new file mode 100644 index 0000000000000..77936a653c7b5 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPathHelperTest.java @@ -0,0 +1,146 @@ +/* + * 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.db.storageengine.load.active; + +import org.apache.iotdb.db.queryengine.plan.statement.crud.LoadTsFileStatement; +import org.apache.iotdb.db.storageengine.load.config.LoadTsFileConfigurator; + +import org.junit.Assert; +import org.junit.Test; + +import java.io.File; +import java.nio.file.Files; +import java.util.Map; + +public class ActiveLoadPathHelperTest { + + @Test + public void testUserAttributeShouldBeMaskedInPathAndDecodedWhenParsing() throws Exception { + final String userName = "active_load_user"; + final File pendingDir = Files.createTempDirectory("active-load-path").toFile(); + try { + final File targetDir = + ActiveLoadPathHelper.resolveTargetDir( + pendingDir, + ActiveLoadPathHelper.buildAttributes(null, null, null, null, null, null, userName)); + final File tsFile = new File(targetDir, "1-0-0-0.tsfile"); + + Assert.assertTrue(targetDir.getAbsolutePath().contains("user-v1-")); + Assert.assertFalse(targetDir.getAbsolutePath().contains(userName)); + Assert.assertFalse(targetDir.getAbsolutePath().contains("b64%3A")); + + final Map attributes = + ActiveLoadPathHelper.parseAttributes(tsFile, pendingDir); + Assert.assertEquals(userName, attributes.get(ActiveLoadPathHelper.USER_KEY)); + } finally { + deleteRecursively(pendingDir); + } + } + + @Test + public void testRawUserAttributeShouldBeIgnored() throws Exception { + final File pendingDir = Files.createTempDirectory("active-load-path").toFile(); + try { + final File tsFile = new File(new File(pendingDir, "user-active_load_user"), "1-0-0-0.tsfile"); + + final Map attributes = + ActiveLoadPathHelper.parseAttributes(tsFile, pendingDir); + Assert.assertFalse(attributes.containsKey(ActiveLoadPathHelper.USER_KEY)); + } finally { + deleteRecursively(pendingDir); + } + } + + @Test + public void testNonV1UserAttributeShouldBeIgnored() throws Exception { + final File pendingDir = Files.createTempDirectory("active-load-path").toFile(); + try { + final File tsFile = + new File(new File(pendingDir, "user-v2-active_load_user"), "1-0-0-0.tsfile"); + + final Map attributes = + ActiveLoadPathHelper.parseAttributes(tsFile, pendingDir); + Assert.assertFalse(attributes.containsKey(ActiveLoadPathHelper.USER_KEY)); + } finally { + deleteRecursively(pendingDir); + } + } + + @Test + public void testUnknownRawAttributeDirectoryShouldBeIgnoredForDowngradeCompatibility() + throws Exception { + final File pendingDir = Files.createTempDirectory("active-load-path").toFile(); + try { + final File tsFile = + new File(new File(pendingDir, "future-load-param-future-value"), "1-0-0-0.tsfile"); + createFile(tsFile); + + final Map attributes = + ActiveLoadPathHelper.parseAttributes(tsFile, pendingDir); + Assert.assertFalse(attributes.containsKey("future-load-param")); + + final LoadTsFileStatement statement = + LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()); + ActiveLoadPathHelper.applyAttributesToStatement(attributes, statement, true); + Assert.assertTrue(statement.isVerifySchema()); + } finally { + deleteRecursively(pendingDir); + } + } + + @Test + public void testKnownPrefixWithInvalidFutureLikeValueShouldBeIgnoredForDowngradeCompatibility() + throws Exception { + final File pendingDir = Files.createTempDirectory("active-load-path").toFile(); + try { + final File tsFile = new File(new File(pendingDir, "verify-future-value"), "1-0-0-0.tsfile"); + createFile(tsFile); + + final Map attributes = + ActiveLoadPathHelper.parseAttributes(tsFile, pendingDir); + Assert.assertFalse(attributes.containsKey(LoadTsFileConfigurator.VERIFY_KEY)); + + final LoadTsFileStatement statement = + LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()); + ActiveLoadPathHelper.applyAttributesToStatement(attributes, statement, true); + Assert.assertTrue(statement.isVerifySchema()); + } finally { + deleteRecursively(pendingDir); + } + } + + private static void deleteRecursively(final File file) { + if (file == null || !file.exists()) { + return; + } + final File[] children = file.listFiles(); + if (children != null) { + for (final File child : children) { + deleteRecursively(child); + } + } + Assert.assertTrue(file.delete()); + } + + private static void createFile(final File file) throws Exception { + Assert.assertTrue(file.getParentFile().mkdirs()); + Assert.assertTrue(file.createNewFile()); + } +}