This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 45c962bc4c2 Fix the identity used by the asynchronous LOAD of the pipe
receiver (#19029)
45c962bc4c2 is described below
commit 45c962bc4c23fca8a9fd098270768fe7241e66bc
Author: Zhenyu Luo <[email protected]>
AuthorDate: Fri Oct 30 12:17:45 2026 +0800
Fix the identity used by the asynchronous LOAD of the pipe receiver (#19029)
* Fix the identity used by the asynchronous LOAD of the pipe receiver
The load attributes of a TsFile received by the pipe receiver were built
from the
default user instead of from the sender of the transfer. The receiver now
passes
the authenticated sender to the active loader, and a seal request whose
user can
not be determined is rejected. The type mismatch converter inherits the
identity
of the load request. TsFiles placed directly in the active listening
directory are
still loaded with the default user.
Tests: IoTDBPipeReceiverAsyncLoadAuthIT and IoTDBDataNodeReceiverTest.
* Fix the user resolution in the LOAD pipelines
Several places substituted the default user when a load carried no identity.
LoadUtil, ActiveLoadTsFileLoader and the type mismatch converter now
require the
user of the load and fail otherwise, while a TsFile placed directly in the
active
listening directory still falls back to the default user. The worker
session of
the active loader is created after the per-file identity is known, and the
thrift
pipe receiver returns NOT_LOGIN instead of dereferencing a missing session.
Tests: LoadUtilTest and IoTDBPipeReceiverAsyncLoadAuthIT.
* Fix the identity resolution of the pipe receiver
The V2 handshake took the user id from the handshake parameters, and the
receiver
fell back to default credentials when the handshake carried none. The
receiver now
resolves the identity from the handshake user name with the local user
manager or
the permission manager, creates the audit entity only after a successful
login,
and clears it when the handshake fails or the connection exits. Two null
pointer
exceptions on the LOAD paths are fixed as well.
Tests: IoTDBFileReceiverTest and IoTDBDataNodeReceiverAuditEntityTest.
* Fix the identity binding of a pipe transfer connection
A connection that was already authenticated kept the identity of its first
transfer while the audit entity followed the new handshake. A handshake
that does
not match the user of the connection is now rejected with NO_PERMISSION,
and the
identity of a transfer is resolved from the verified user name, with the
address
of the connection preferred over the hostname of the handshake.
Tests: IoTDBFileReceiverTest.
* Harden the user resolution of the pipe receiver and the load pipeline
- Use "{}" placeholders in the new LOG_ templates and keep the en and zh
constants in parity.
- Fail closed when the session of a pipe connection is not logged in: an
unauthenticated session still carries the id of the super user.
- Resolve the user id of a load from its user name instead of trusting the
default id of the session, in the tree model schema verifier and in the
TsFile data type converter.
- Allow a local TsFile that is dropped into a sub directory of the active
listening directory to be loaded with the default user, while a file that
carries attributes but no user is still rejected.
* Make the load user identity requirement configurable
The pipe receiver and the LOAD pipeline now fail when the user that
executes a load can not be resolved. Keep the legacy default-user
fallback by default and gate the strict behavior behind two internal
switches instead of changing the existing load code:
- pipe_receiver_require_user_identity (default false): an asynchronous
LOAD of the pipe receiver without an authenticated sender is executed
as the default user instead of being rejected.
- load_require_user_identity (default false): an asynchronous LOAD
without a user is queued with the default user instead of letting the
strict LoadUtil check reject it.
The switches are not added to iotdb-system.properties.template and
therefore stay internal. ActiveLoadTsFileLoader,
LoadTsFileDataTypeConverter,
LoadUtil and TreeSchemaAutoCreatorAndVerifier keep the behavior of this
branch.
Tests: LoadUtilTest, ActiveLoadTsFileLoaderTest, IoTDBDataNodeReceiverTest
and LoadTsFileAnalyzerTest.
* Keep the load user identity switches minimal
Reduce the diff of the configurable load identity requirement:
- Move the LOAD switch back into LoadUtil so LoadTsFileAnalyzer keeps the
behavior of this branch.
- Read the two switches only when the properties are loaded; they are not
hot reloadable.
- Shorten the comments and javadocs of the new switches.
ActiveLoadTsFileLoader, LoadTsFileDataTypeConverter,
TreeSchemaAutoCreatorAndVerifier
and the asynchronous LOAD acknowledgement semantics are unchanged.
Tests: LoadUtilTest.
* Drop the redundant user resolution in the LOAD pipeline
The load path already carries a locally resolved user and LoadUtil always
queues a user when the strict switch is enabled, so the extra resolution
layers are redundant:
- TreeSchemaAutoCreatorAndVerifier uses the user of the load session again
instead of resolving the id per measurement.
- ActiveLoadTsFileLoader uses the user attribute with the default user as
fallback, instead of walking the listening directory and failing files.
- Remove the i18n constants that are no longer used.
Tests: LoadUtilTest, ActiveLoadTsFileLoaderTest, IoTDBDataNodeReceiverTest,
LoadTsFileAnalyzerTest.
* Keep the legacy user fallback of LoadUtil
LoadUtil only needs to use the user name of the session when it is present
and fall back to the default user otherwise, which is what the original
code does. Drop the strict switch and restore the original method.
Remove the now unused i18n constant.
Tests: LoadUtilTest, ActiveLoadTsFileLoaderTest, IoTDBDataNodeReceiverTest,
LoadTsFileAnalyzerTest.
* Require a logged in user when the load identity switches are enabled
- pipe_receiver_require_user_identity: the asynchronous LOAD of the pipe
receiver requires the authenticated sender to be the logged in user of
the connection.
- load_require_user_identity: a LOAD without a user requires a logged in
session that carries a user name instead of falling back to the default
user.
Both switches keep the legacy fallback when they are disabled.
Tests: LoadUtilTest, ActiveLoadTsFileLoaderTest, IoTDBDataNodeReceiverTest,
LoadTsFileAnalyzerTest.
* Align the legacy receiver runtime test and cover local LOAD options
- A V1 handshake carries no credentials, so the receiver runtime reports an
unknown user instead of the default user. Update the legacy receiver
expectation accordingly, keeping the removal of privileged defaults.
- Add a regression test for a locally dropped TsFile under an attribute
directory without a user; such a local input must still be loaded with the
default user.
* Require an authenticated sender for asynchronous LOAD by default
Enable pipe_receiver_require_user_identity by default: the receiver only
queues an asynchronous LOAD when the sender is authenticated and the
connection is logged in as that user. Set the switch to false to restore
the legacy default user fallback. The load_require_user_identity switch
keeps its default (false).
Tests: LoadUtilTest, ActiveLoadTsFileLoaderTest, IoTDBDataNodeReceiverTest,
LoadTsFileAnalyzerTest.
* Compare the number of points of the replicas instead of their TsFile sizes
The restart case waited until every data region replica on the restarted
DataNode had the same on-disk size as the other replicas of that region. A
replica that replayed the LOAD can keep the same points in fewer physical
TsFiles than the replicas that stayed up, so the sizes never converge and
the case fails with a size mismatch that says nothing about the data.
Hard-link the TsFiles of every replica of a region into
<temp dir>/<region>/<DataNode>/, read them back point by point, and wait
until all replicas of the region hold the same number of points. Duplicate
physical copies collapse in the point set, so the check no longer depends on
how the same data is spread over files. Every waiting attempt deletes its
snapshots, and tearDownClass removes what an interrupted attempt left
behind.
---
.../iotdb/db/it/IoTDBLoadTsFileClusterIT.java | 247 +++++++++---
.../single/IoTDBPipeReceiverAsyncLoadAuthIT.java | 413 +++++++++++++++++++++
.../iotdb/pipe/it/single/IoTDBShowReceiversIT.java | 4 +-
.../iotdb/confignode/i18n/ManagerMessages.java | 3 +
.../iotdb/confignode/i18n/ManagerMessages.java | 3 +
.../receiver/protocol/IoTDBConfigNodeReceiver.java | 29 ++
.../iotdb/db/i18n/StorageEngineMessages.java | 5 +
.../iotdb/db/i18n/StorageEngineMessages.java | 5 +
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 28 ++
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 11 +
.../protocol/thrift/IoTDBDataNodeReceiver.java | 154 +++++---
.../plan/analyze/load/LoadTsFileAnalyzer.java | 4 +-
.../converter/LoadTsFileDataTypeConverter.java | 89 +++--
.../iotdb/db/storageengine/load/util/LoadUtil.java | 25 +-
.../IoTDBDataNodeReceiverAuditEntityTest.java | 79 ++++
.../protocol/thrift/IoTDBDataNodeReceiverTest.java | 20 +-
.../db/storageengine/load/util/LoadUtilTest.java | 62 ++++
.../apache/iotdb/commons/i18n/PipeMessages.java | 2 +
.../apache/iotdb/commons/i18n/PipeMessages.java | 2 +
.../commons/pipe/receiver/IoTDBFileReceiver.java | 74 +++-
.../pipe/receiver/IoTDBFileReceiverTest.java | 147 +++++++-
21 files changed, 1263 insertions(+), 143 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileClusterIT.java
b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileClusterIT.java
index 0cb700627ef..6b51c1ae06d 100644
---
a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileClusterIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileClusterIT.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.db.it;
import org.apache.iotdb.commons.schema.column.ColumnHeaderConstant;
import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.env.cluster.ClusterConstant;
import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
import org.apache.iotdb.it.framework.IoTDBTestRunner;
import org.apache.iotdb.it.utils.TsFileGenerator;
@@ -30,6 +31,13 @@ import org.apache.iotdb.jdbc.Config;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.TsFileReader;
+import org.apache.tsfile.read.TsFileSequenceReader;
+import org.apache.tsfile.read.common.Field;
+import org.apache.tsfile.read.common.Path;
+import org.apache.tsfile.read.common.RowRecord;
+import org.apache.tsfile.read.expression.QueryExpression;
+import org.apache.tsfile.read.query.dataset.QueryDataSet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
import org.awaitility.Awaitility;
@@ -43,7 +51,9 @@ import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;
import java.io.File;
+import java.io.IOException;
import java.nio.file.Files;
+import java.nio.file.StandardCopyOption;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
@@ -52,9 +62,12 @@ import java.sql.Statement;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
@@ -80,6 +93,16 @@ public class IoTDBLoadTsFileClusterIT {
/** Bounds the load of a cluster that cannot reach consensus, so the test
cannot hang forever. */
private static final int DEGRADED_LOAD_TIMEOUT_SECONDS = 300;
+ /** The directories a DataRegion keeps its TsFiles in, below
`<datanode>/data`. */
+ private static final List<String> TSFILE_DIRECTORIES =
Arrays.asList("sequence", "unsequence");
+
+ private static final String TSFILE_SUFFIX = ".tsfile";
+
+ /**
+ * The prefix of the temporary directories that hold the hard-linked TsFiles
of the restart case.
+ */
+ private static final String TSFILE_SNAPSHOT_PREFIX = "load-cluster-tsfile";
+
private File tmpDir;
@BeforeClass
@@ -99,8 +122,9 @@ public class IoTDBLoadTsFileClusterIT {
.setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS)
.setSchemaReplicationFactor(REPLICATION_FACTOR)
.setDataReplicationFactor(REPLICATION_FACTOR);
- // The restart case compares the on-disk size of the replicas, so keep
compaction from changing
- // those sizes while the test waits for the restarted DataNode to catch up.
+ // Keep every kind of merge off for this class: the restart case
hard-links the replicas'
+ // TsFiles out of the DataNode directory and compares their points while
the restarted DataNode
+ // catches up, so compaction must not rewrite those files in the meantime.
EnvFactory.getEnv()
.getConfig()
.getCommonConfig()
@@ -112,6 +136,17 @@ public class IoTDBLoadTsFileClusterIT {
@AfterClass
public static void tearDownClass() {
+ // Every waiting attempt deletes its own snapshot directory; this also
removes what a JVM that
+ // died in the middle of an attempt left behind next to the DataNode
directories.
+ final File targetDir =
+ new File(System.getProperty(ClusterConstant.USER_DIR),
ClusterConstant.TARGET);
+ final File[] snapshotDirs =
+ targetDir.listFiles((dir, name) ->
name.startsWith(TSFILE_SNAPSHOT_PREFIX));
+ if (snapshotDirs != null) {
+ for (final File snapshotDir : snapshotDirs) {
+ deleteRecursively(snapshotDir);
+ }
+ }
EnvFactory.getEnv().cleanClusterEnvironment();
}
@@ -316,58 +351,70 @@ public class IoTDBLoadTsFileClusterIT {
}
/**
- * Waits until every data region replica on the given DataNode has the same
on-disk size as the
- * other replicas of that region. A node that did not replay the LOAD keeps
its pre-LOAD size.
+ * Waits until every data region replica on the given DataNode holds as many
points as the other
+ * replicas of that region.
+ *
+ * <p>A replica that did not replay the LOAD keeps its pre-LOAD data, but
the replicas that stayed
+ * up may keep the same points in more physical TsFiles than the restarted
one, so on-disk sizes
+ * are not comparable. Every replica's TsFiles are hard-linked into a
temporary directory and read
+ * back point by point instead.
*/
private static void awaitRestartedLocalReplicasHaveTheLoadedFile(
final DataNodeWrapper restartedDataNode) {
Awaitility.await()
.atMost(5, TimeUnit.MINUTES)
- .pollInterval(1, TimeUnit.SECONDS)
+ .pollInterval(5, TimeUnit.SECONDS)
.ignoreExceptions()
.untilAsserted(
() -> {
- final Map<Integer, Map<Integer, String>> regionSizes =
dataRegionSizes();
+ final Map<Integer, Set<Integer>> regionReplicas =
dataRegionReplicas();
Assert.assertFalse(
- "the loaded database must have data regions",
regionSizes.isEmpty());
+ "the loaded database must have data regions",
regionReplicas.isEmpty());
- int restartedDataNodeId;
+ final Map<Integer, DataNodeWrapper> dataNodesById = new
HashMap<>();
try (final Connection connection =
EnvFactory.getEnv().getConnection();
final Statement statement = connection.createStatement()) {
- restartedDataNodeId = dataNodeId(statement, restartedDataNode);
+ for (int dataNodeIndex = 0; dataNodeIndex < DATA_NODE_NUM;
dataNodeIndex++) {
+ final DataNodeWrapper dataNode =
+ EnvFactory.getEnv().getDataNodeWrapper(dataNodeIndex);
+ dataNodesById.put(dataNodeId(statement, dataNode), dataNode);
+ }
}
- for (final Map.Entry<Integer, Map<Integer, String>> region :
regionSizes.entrySet()) {
- final Map<Integer, String> replicaSizes = region.getValue();
- // A region without data on any replica has nothing to catch
up with.
- if
(replicaSizes.values().stream().noneMatch(IoTDBLoadTsFileClusterIT::hasData)) {
- continue;
+ // Create the snapshots beside the DataNode directories, so that
the hard links stay
+ // on the file system of the TsFiles they point at.
+ final File snapshotRoot =
+ Files.createTempDirectory(
+ new
File(restartedDataNode.getNodePath()).getParentFile().toPath(),
+ TSFILE_SNAPSHOT_PREFIX)
+ .toFile();
+ try {
+ for (final Map.Entry<Integer, Set<Integer>> region :
regionReplicas.entrySet()) {
+ final Map<Integer, Integer> pointCounts = new TreeMap<>();
+ for (final int dataNodeId : region.getValue()) {
+ final DataNodeWrapper dataNode =
dataNodesById.get(dataNodeId);
+ Assert.assertNotNull(
+ "region "
+ + region.getKey()
+ + " is hosted by an unknown DataNode "
+ + dataNodeId,
+ dataNode);
+ pointCounts.put(
+ dataNodeId,
+ readLocalReplicaPointCount(dataNode, region.getKey(),
snapshotRoot));
+ }
+ assertSamePointCounts(region.getKey(), pointCounts);
}
- final String restartedSize =
replicaSizes.get(restartedDataNodeId);
+ } finally {
Assert.assertTrue(
- "region " + region.getKey() + " has no data on the
restarted DataNode",
- hasData(restartedSize));
- replicaSizes.forEach(
- (dataNodeId, size) -> {
- if (dataNodeId != restartedDataNodeId) {
- Assert.assertEquals(
- "region " + region.getKey() + " differs on
DataNode " + dataNodeId,
- size,
- restartedSize);
- }
- });
+ "cannot delete the TsFile snapshots",
deleteRecursively(snapshotRoot));
}
});
}
- /** Whether a `show regions` size means the replica holds data. */
- private static boolean hasData(final String size) {
- return size != null && !"Unknown".equals(size) && !"0 B".equals(size);
- }
-
- /** The DataRegion rows of `show regions`, mapped to each replica's on-disk
size. */
- private static Map<Integer, Map<Integer, String>> dataRegionSizes() throws
Exception {
- final Map<Integer, Map<Integer, String>> regionSizes = new HashMap<>();
+ /** The DataRegion rows of `show regions`, mapped to the DataNodes that hold
a replica. */
+ private static Map<Integer, Set<Integer>> dataRegionReplicas() throws
SQLException {
+ final Map<Integer, Set<Integer>> regionReplicas = new HashMap<>();
try (final Connection connection = EnvFactory.getEnv().getConnection();
final Statement statement = connection.createStatement();
final ResultSet resultSet = statement.executeQuery("show regions")) {
@@ -376,15 +423,124 @@ public class IoTDBLoadTsFileClusterIT {
||
!DATABASE.equals(resultSet.getString(ColumnHeaderConstant.DATABASE))) {
continue;
}
- regionSizes
+ regionReplicas
.computeIfAbsent(
- resultSet.getInt(ColumnHeaderConstant.REGION_ID), id -> new
HashMap<>())
- .put(
- resultSet.getInt(ColumnHeaderConstant.DATA_NODE_ID),
- resultSet.getString(ColumnHeaderConstant.TSFILE_SIZE));
+ resultSet.getInt(ColumnHeaderConstant.REGION_ID), id -> new
HashSet<>())
+ .add(resultSet.getInt(ColumnHeaderConstant.DATA_NODE_ID));
+ }
+ }
+ return regionReplicas;
+ }
+
+ /**
+ * Counts the points of one local replica, after hard-linking the replica's
TsFiles into
+ * `<snapshotRoot>/<region>/<node>` so that a concurrent flush cannot change
them while they are
+ * read.
+ */
+ private static int readLocalReplicaPointCount(
+ final DataNodeWrapper dataNode, final int regionId, final File
snapshotRoot)
+ throws IOException {
+ final File replicaDir =
+ new File(new File(snapshotRoot, String.valueOf(regionId)),
dataNode.getId());
+ final File dataRoot = new File(dataNode.getDataNodeDir(), "data");
+ final List<File> tsFiles = new ArrayList<>();
+ for (final String directory : TSFILE_DIRECTORIES) {
+ final File regionDir =
+ new File(new File(dataRoot, directory), DATABASE + File.separator +
regionId);
+ tsFiles.addAll(snapshotTsFiles(regionDir, new File(replicaDir,
directory)));
+ }
+ final Set<String> points = new HashSet<>();
+ for (final File tsFile : tsFiles) {
+ readPoints(tsFile, points);
+ }
+ return points.size();
+ }
+
+ /** Hard-links the TsFiles below a replica directory into the snapshot
directory. */
+ private static List<File> snapshotTsFiles(final File regionDir, final File
snapshotDir)
+ throws IOException {
+ final List<File> tsFiles = new ArrayList<>();
+ collectTsFiles(regionDir, tsFiles);
+ if (tsFiles.isEmpty()) {
+ return tsFiles;
+ }
+ Files.createDirectories(snapshotDir.toPath());
+ final List<File> snapshots = new ArrayList<>();
+ for (final File tsFile : tsFiles) {
+ // The time partition directory keeps the names unique inside one
replica.
+ final File snapshot =
+ new File(snapshotDir, tsFile.getParentFile().getName() + "-" +
tsFile.getName());
+ try {
+ Files.createLink(snapshot.toPath(), tsFile.toPath());
+ } catch (final IOException | UnsupportedOperationException e) {
+ // A hard link needs the same file system; copy if the snapshots land
on another one.
+ Files.copy(tsFile.toPath(), snapshot.toPath(),
StandardCopyOption.REPLACE_EXISTING);
}
+ snapshots.add(snapshot);
}
- return regionSizes;
+ return snapshots;
+ }
+
+ private static void collectTsFiles(final File directory, final List<File>
tsFiles) {
+ final File[] children = directory.listFiles();
+ if (children == null) {
+ return;
+ }
+ for (final File child : children) {
+ if (child.isDirectory()) {
+ collectTsFiles(child, tsFiles);
+ } else if (child.getName().endsWith(TSFILE_SUFFIX)) {
+ tsFiles.add(child);
+ }
+ }
+ }
+
+ /** Adds every non-null point of the given TsFile to the point set. */
+ private static void readPoints(final File tsFile, final Set<String> points)
throws IOException {
+ try (final TsFileSequenceReader sequenceReader = new
TsFileSequenceReader(tsFile.getPath());
+ final TsFileReader reader = new TsFileReader(sequenceReader)) {
+ final List<Path> queryPaths = new ArrayList<>();
+ for (final String device : allDevices()) {
+ for (final String measurement : MEASUREMENTS) {
+ queryPaths.add(new Path(device, measurement, false));
+ }
+ }
+ final QueryDataSet dataSet =
reader.query(QueryExpression.create(queryPaths, null));
+ // The data set keeps only the series this TsFile actually holds, in the
query order.
+ final List<Path> storedPaths = dataSet.getPaths();
+ while (dataSet.hasNext()) {
+ final RowRecord row = dataSet.next();
+ final List<Field> fields = row.getFields();
+ for (int column = 0; column < storedPaths.size() && column <
fields.size(); column++) {
+ final Field field = fields.get(column);
+ if (field == null || field.getDataType() == null) {
+ continue;
+ }
+ points.add(
+ storedPaths.get(column).getFullPath()
+ + "@"
+ + row.getTimestamp()
+ + "="
+ + field.getObjectValue(field.getDataType()));
+ }
+ }
+ }
+ }
+
+ /**
+ * Asserts that every replica of the region holds the same number of points.
A replica that did
+ * not replay the LOAD keeps its pre-LOAD data and stays behind.
+ */
+ private static void assertSamePointCounts(
+ final int regionId, final Map<Integer, Integer> pointCounts) {
+ if (new HashSet<>(pointCounts.values()).size() <= 1) {
+ return;
+ }
+ Assert.fail(
+ "region "
+ + regionId
+ + " holds a different number of points on its replicas: "
+ + pointCounts);
}
/** The cluster id of the given DataNode, matched by its client RPC
endpoint. */
@@ -533,9 +689,7 @@ public class IoTDBLoadTsFileClusterIT {
private static long queryPointCount(final Statement statement) throws
Exception {
long actualPointCount = 0;
- final List<String> allDevices = new ArrayList<>(DEVICES);
- allDevices.addAll(ALIGNED_DEVICES);
- for (final String device : allDevices) {
+ for (final String device : allDevices()) {
for (final String measurement : MEASUREMENTS) {
try (final ResultSet resultSet =
statement.executeQuery("select count(" + measurement + ") from " +
device)) {
@@ -546,4 +700,11 @@ public class IoTDBLoadTsFileClusterIT {
}
return actualPointCount;
}
+
+ /** Every device the loaded files hold. */
+ private static List<String> allDevices() {
+ final List<String> devices = new ArrayList<>(DEVICES);
+ devices.addAll(ALIGNED_DEVICES);
+ return devices;
+ }
}
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipeReceiverAsyncLoadAuthIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipeReceiverAsyncLoadAuthIT.java
new file mode 100644
index 00000000000..ba2a15d34a8
--- /dev/null
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipeReceiverAsyncLoadAuthIT.java
@@ -0,0 +1,413 @@
+/*
+ * 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.pipe.it.single;
+
+import org.apache.iotdb.commons.auth.entity.PrivilegeType;
+import org.apache.iotdb.commons.client.property.ThriftClientProperty;
+import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant;
+import org.apache.iotdb.commons.pipe.sink.client.IoTDBSyncClient;
+import
org.apache.iotdb.commons.pipe.sink.payload.thrift.common.PipeTransferHandshakeConstant;
+import
org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferDataNodeHandshakeV2Req;
+import
org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFilePieceReq;
+import
org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFileSealWithModReq;
+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.LocalStandaloneIT;
+import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
+import org.apache.iotdb.service.rpc.thrift.TPipeTransferResp;
+import org.apache.iotdb.service.rpc.thrift.TSCloseSessionReq;
+import org.apache.iotdb.service.rpc.thrift.TSOpenSessionReq;
+import org.apache.iotdb.service.rpc.thrift.TSOpenSessionResp;
+import org.apache.iotdb.service.rpc.thrift.TSProtocolVersion;
+
+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.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.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.StandardCopyOption;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.time.ZoneId;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.iotdb.db.it.utils.TestUtils.createUser;
+import static org.apache.iotdb.db.it.utils.TestUtils.grantUserSeriesPrivilege;
+
+/**
+ * Verifies that the thrift pipe receiver loads asynchronously TsFiles with
the privileges of the
+ * authenticated sender. The asynchronous load is executed by the active
loader in another thread,
+ * so the receiver must hand the sender's identity over to it explicitly.
+ */
+@RunWith(IoTDBTestRunner.class)
+@Category({LocalStandaloneIT.class})
+public class IoTDBPipeReceiverAsyncLoadAuthIT {
+
+ private static final String DATABASE = "root.pipe_receiver_async_auth";
+ private static final String DEVICE = DATABASE + ".d1";
+ private static final String NO_WRITE_MEASUREMENT = "no_write";
+ private static final String WRITE_MEASUREMENT = "write";
+ private static final String LOCALLY_DROPPED_MEASUREMENT =
"dropped_by_operator";
+ private static final String LOCALLY_DROPPED_WITH_OPTIONS_MEASUREMENT =
"dropped_with_options";
+ private static final IMeasurementSchema NO_WRITE_SCHEMA =
+ new MeasurementSchema(NO_WRITE_MEASUREMENT, TSDataType.INT32,
TSEncoding.RLE);
+ private static final IMeasurementSchema WRITE_SCHEMA =
+ new MeasurementSchema(WRITE_MEASUREMENT, TSDataType.INT32,
TSEncoding.RLE);
+ private static final IMeasurementSchema LOCALLY_DROPPED_SCHEMA =
+ new MeasurementSchema(LOCALLY_DROPPED_MEASUREMENT, TSDataType.INT32,
TSEncoding.RLE);
+ private static final IMeasurementSchema LOCALLY_DROPPED_WITH_OPTIONS_SCHEMA =
+ new MeasurementSchema(
+ LOCALLY_DROPPED_WITH_OPTIONS_MEASUREMENT, TSDataType.INT32,
TSEncoding.RLE);
+
+ private static final String NO_WRITE_USER = "pipe_receiver_no_write_user";
+ private static final String WRITE_USER = "pipe_receiver_write_user";
+ private static final String PASSWORD = "StrngPsWd@623451";
+ private static final String SENDER_CLUSTER_ID =
"IT_PIPE_RECEIVER_SENDER_CLUSTER";
+ private static final long TIME_PARTITION_INTERVAL = 10 * 1000L;
+ private static final int POINT_COUNT = 10;
+
+ private static File tmpDir;
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ tmpDir = new
File(Files.createTempDirectory("pipe-receiver-async-auth").toUri());
+ // Match the configuration of the other pipe integration tests: the
receiver refuses handshakes
+ // when the DataNode cannot reserve pipe memory.
+ EnvFactory.getEnv()
+ .getConfig()
+ .getCommonConfig()
+ .setAutoCreateSchemaEnabled(true)
+ .setDatanodeMemoryProportion("3:3:1:1:1:0")
+ .setPipeMemoryManagementEnabled(false)
+ .setIsPipeEnableMemoryCheck(false)
+ .setPipeAutoSplitFullEnabled(false)
+ .setEnforceStrongPassword(false)
+ .setTimePartitionInterval(TIME_PARTITION_INTERVAL);
+ EnvFactory.getEnv()
+ .getConfig()
+ .getDataNodeConfig()
+ .setLoadActiveListeningCheckIntervalSeconds(1);
+ EnvFactory.getEnv().initClusterEnvironment();
+ prepareSchema();
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ EnvFactory.getEnv().cleanClusterEnvironment();
+ FileUtils.deleteDirectory(tmpDir);
+ }
+
+ @Test
+ public void testAsyncLoadStrategyDoesNotElevatePipeSender() throws Exception
{
+ createUser(NO_WRITE_USER, PASSWORD);
+ createUser(WRITE_USER, PASSWORD);
+ grantUserSeriesPrivilege(WRITE_USER, PrivilegeType.WRITE_DATA, DATABASE +
".**");
+
+ final File noWriteTsFile = new File(tmpDir, "1-0-0-0.tsfile");
+ final File writeTsFile = new File(tmpDir, "2-0-0-0.tsfile");
+ generateTsFile(noWriteTsFile, NO_WRITE_SCHEMA);
+ generateTsFile(writeTsFile, WRITE_SCHEMA);
+
+ // Control: the synchronous strategy loads with the privileges of the
sender, so the account
+ // without grants is rejected here.
+ assertSealRejected(noWriteTsFile, NO_WRITE_USER, false);
+
+ // The asynchronous strategy must not turn the same account into the super
user.
+ assertSealAccepted(noWriteTsFile, NO_WRITE_USER, true);
+
+ // Positive control: the same asynchronous flow succeeds for an account
holding WRITE_DATA.
+ assertSealAccepted(writeTsFile, WRITE_USER, true);
+
+ waitUntilActiveLoadIsIdle(TimeUnit.SECONDS.toMillis(60));
+ assertCount(NO_WRITE_MEASUREMENT, 0);
+ assertCount(WRITE_MEASUREMENT, POINT_COUNT);
+ }
+
+ @Test
+ public void testTsFileDroppedIntoActiveListeningDirIsStillLoaded() throws
Exception {
+ final File tsFile = new File(tmpDir, "3-0-0-0.tsfile");
+ generateTsFile(tsFile, LOCALLY_DROPPED_SCHEMA);
+
+ // An operator may place a TsFile directly into the active listening
directory. Such a file has
+ // no user attribute and must still be loaded, with the default user,
because writing it to the
+ // DataNode file system is already a privileged operation.
+ final DataNodeWrapper dataNode = EnvFactory.getEnv().getDataNodeWrapper(0);
+ final File pendingDir = getActiveLoadPendingDir(dataNode);
+ Files.createDirectories(pendingDir.toPath());
+ // The staging copy must not end with ".tsfile", otherwise the scanner
could pick it up while it
+ // is still being copied.
+ final File staging = new File(pendingDir, "staging-" + tsFile.getName() +
".tmp");
+ Files.copy(tsFile.toPath(), staging.toPath(),
StandardCopyOption.REPLACE_EXISTING);
+ Files.move(
+ staging.toPath(),
+ new File(pendingDir, tsFile.getName()).toPath(),
+ StandardCopyOption.ATOMIC_MOVE);
+
+ assertCountEventually(LOCALLY_DROPPED_MEASUREMENT, POINT_COUNT,
TimeUnit.SECONDS.toMillis(60));
+ }
+
+ @Test
+ public void
testTsFileDroppedIntoAttributeDirectoryWithoutUserIsStillLoaded() throws
Exception {
+ final File tsFile = new File(tmpDir, "4-0-0-0.tsfile");
+ generateTsFile(tsFile, LOCALLY_DROPPED_WITH_OPTIONS_SCHEMA);
+
+ // A locally dropped file may carry recognized LOAD options, e.g.
verify=false, without a user
+ // attribute. It is not a receiver handoff, so it must still be loaded
with the default user.
+ final DataNodeWrapper dataNode = EnvFactory.getEnv().getDataNodeWrapper(0);
+ final File attributeDir = new File(getActiveLoadPendingDir(dataNode),
"verify-false");
+ Files.createDirectories(attributeDir.toPath());
+ final File staging = new File(attributeDir, "staging-" + tsFile.getName()
+ ".tmp");
+ Files.copy(tsFile.toPath(), staging.toPath(),
StandardCopyOption.REPLACE_EXISTING);
+ Files.move(
+ staging.toPath(),
+ new File(attributeDir, tsFile.getName()).toPath(),
+ StandardCopyOption.ATOMIC_MOVE);
+
+ assertCountEventually(
+ LOCALLY_DROPPED_WITH_OPTIONS_MEASUREMENT, POINT_COUNT,
TimeUnit.SECONDS.toMillis(60));
+ }
+
+ private static void prepareSchema() 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, NO_WRITE_MEASUREMENT, NO_WRITE_SCHEMA.getType()));
+ statement.execute(
+ String.format(
+ "create timeseries %s.%s %s", DEVICE, WRITE_MEASUREMENT,
WRITE_SCHEMA.getType()));
+ statement.execute(
+ String.format(
+ "create timeseries %s.%s %s",
+ DEVICE, LOCALLY_DROPPED_MEASUREMENT,
LOCALLY_DROPPED_SCHEMA.getType()));
+ statement.execute(
+ String.format(
+ "create timeseries %s.%s %s",
+ DEVICE,
+ LOCALLY_DROPPED_WITH_OPTIONS_MEASUREMENT,
+ LOCALLY_DROPPED_WITH_OPTIONS_SCHEMA.getType()));
+ }
+ }
+
+ private static void generateTsFile(final File tsFile, final
IMeasurementSchema schema)
+ throws Exception {
+ try (final TsFileGenerator generator = new TsFileGenerator(tsFile)) {
+ generator.registerTimeseries(DEVICE, Collections.singletonList(schema));
+ generator.generateData(DEVICE, POINT_COUNT, TIME_PARTITION_INTERVAL /
10, false);
+ }
+ }
+
+ /**
+ * Sends a TsFile to the DataNode pipe receiver with the given load strategy
and returns the seal
+ * status. The handshake carries a cluster id different from the receiver's,
so the receiver
+ * treats the sender as a remote cluster.
+ */
+ private static TPipeTransferResp sendTsFile(
+ final File tsFile, final String user, final boolean asyncLoadStrategy)
throws Exception {
+ 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(buildOpenSessionReq(user));
+ Assert.assertEquals(
+ "Failed to open the session of " + user + ": " +
openSessionResp.getStatus(),
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ openSessionResp.getStatus().getCode());
+
+ try {
+ final TPipeTransferResp handshakeResp =
+ client.pipeTransfer(buildHandshakeReq(user, asyncLoadStrategy));
+ Assert.assertEquals(
+ "Handshake failed: " + handshakeResp.getStatus(),
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ handshakeResp.getStatus().getCode());
+
+ final String fileName = tsFile.getName();
+ final TPipeTransferResp pieceResp =
+ client.pipeTransfer(
+ PipeTransferTsFilePieceReq.toTPipeTransferReq(
+ fileName, 0L, Files.readAllBytes(tsFile.toPath())));
+ Assert.assertEquals(
+ "File piece rejected: " + pieceResp.getStatus(),
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ pieceResp.getStatus().getCode());
+
+ return client.pipeTransfer(
+ PipeTransferTsFileSealWithModReq.toTPipeTransferReq(
+ fileName, tsFile.length(), DATABASE));
+ } finally {
+ client.closeSession(new
TSCloseSessionReq(openSessionResp.getSessionId()));
+ }
+ }
+ }
+
+ private static TSOpenSessionReq buildOpenSessionReq(final String user) {
+ final TSOpenSessionReq req = new TSOpenSessionReq();
+ req.setClient_protocol(TSProtocolVersion.IOTDB_SERVICE_PROTOCOL_V3);
+ req.setUsername(user);
+ req.setPassword(PASSWORD);
+ req.setZoneId(ZoneId.systemDefault().toString());
+ req.putToConfiguration("version",
IoTDBConstant.ClientVersion.V_1_0.toString());
+ req.putToConfiguration("sql_dialect", "tree");
+ return req;
+ }
+
+ private static TPipeTransferReq buildHandshakeReq(
+ final String user, final boolean asyncLoadStrategy) throws IOException {
+ final Map<String, String> params = new HashMap<>();
+ params.put(PipeTransferHandshakeConstant.HANDSHAKE_KEY_CLUSTER_ID,
SENDER_CLUSTER_ID);
+ params.put(
+ PipeTransferHandshakeConstant.HANDSHAKE_KEY_TIME_PRECISION,
+ CommonDescriptor.getInstance().getConfig().getTimestampPrecision());
+ params.put(PipeTransferHandshakeConstant.HANDSHAKE_KEY_USERNAME, user);
+ params.put(PipeTransferHandshakeConstant.HANDSHAKE_KEY_PASSWORD, PASSWORD);
+ params.put(
+ PipeTransferHandshakeConstant.HANDSHAKE_KEY_LOAD_TSFILE_STRATEGY,
+ asyncLoadStrategy
+ ? PipeSinkConstant.CONNECTOR_LOAD_TSFILE_STRATEGY_ASYNC_VALUE
+ : PipeSinkConstant.CONNECTOR_LOAD_TSFILE_STRATEGY_SYNC_VALUE);
+ return PipeTransferDataNodeHandshakeV2Req.toTPipeTransferReq(params);
+ }
+
+ private static void assertSealAccepted(
+ final File tsFile, final String user, final boolean asyncLoadStrategy)
throws Exception {
+ final TPipeTransferResp resp = sendTsFile(tsFile, user, asyncLoadStrategy);
+ Assert.assertEquals(
+ "The asynchronous load should be accepted and authorized when it is
executed: "
+ + resp.getStatus(),
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ resp.getStatus().getCode());
+ }
+
+ private static void assertSealRejected(
+ final File tsFile, final String user, final boolean asyncLoadStrategy)
throws Exception {
+ final TPipeTransferResp resp = sendTsFile(tsFile, user, asyncLoadStrategy);
+ Assert.assertNotEquals(
+ "The load of an account without grants must not be accepted: " +
resp.getStatus(),
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ resp.getStatus().getCode());
+ }
+
+ private static void waitUntilActiveLoadIsIdle(final long timeoutMs) throws
InterruptedException {
+ final long deadline = System.currentTimeMillis() + timeoutMs;
+ while (System.currentTimeMillis() < deadline) {
+ boolean hasTsFile = false;
+ for (final DataNodeWrapper dataNode :
EnvFactory.getEnv().getDataNodeWrapperList()) {
+ if (containsTsFile(getActiveLoadPendingDir(dataNode))) {
+ hasTsFile = true;
+ break;
+ }
+ }
+ if (!hasTsFile) {
+ return;
+ }
+ Thread.sleep(500L);
+ }
+ Assert.fail("Timed out waiting for the active load pending directories to
become empty");
+ }
+
+ private static File getActiveLoadPendingDir(final DataNodeWrapper dataNode) {
+ return new File(
+ dataNode.getNodePath()
+ + File.separator
+ + "ext"
+ + File.separator
+ + "load"
+ + File.separator
+ + "pending");
+ }
+
+ private static 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;
+ }
+
+ private static void assertCount(final String measurement, final long
expected) throws Exception {
+ try (final Connection connection = EnvFactory.getEnv().getConnection();
+ final Statement statement = connection.createStatement();
+ final ResultSet resultSet =
+ statement.executeQuery(
+ String.format("select count(%s) from %s", measurement,
DEVICE))) {
+ Assert.assertTrue(resultSet.next());
+ Assert.assertEquals(
+ "Unexpected number of points written to " + measurement, expected,
resultSet.getLong(1));
+ }
+ }
+
+ private static void assertCountEventually(
+ final String measurement, final long expected, final long timeoutMs)
throws Exception {
+ final long deadline = System.currentTimeMillis() + timeoutMs;
+ AssertionError lastError = null;
+ while (System.currentTimeMillis() < deadline) {
+ try {
+ assertCount(measurement, expected);
+ return;
+ } catch (final AssertionError e) {
+ lastError = e;
+ }
+ Thread.sleep(500L);
+ }
+ if (lastError != null) {
+ throw lastError;
+ }
+ Assert.fail("Timed out waiting for " + expected + " points of " +
measurement);
+ }
+}
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBShowReceiversIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBShowReceiversIT.java
index b43190e9645..34dd7263d4c 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBShowReceiversIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBShowReceiversIT.java
@@ -452,7 +452,9 @@ public class IoTDBShowReceiversIT extends
AbstractPipeSingleIT {
&& receiverDataNodeId == getInt(resultSet, "ReceiverNodeId",
"receiver_node_id")
&& "thrift".equals(getString(resultSet, "Protocol", "protocol"))
&& "Unknown".equals(getString(resultSet, "SenderClusterId",
"sender_cluster_id"))
- && SessionConfig.DEFAULT_USER.equals(getString(resultSet,
"UserName", "user_name"))
+ // A V1 handshake carries no credentials, so the receiver reports
no user instead of
+ // the privileged default user.
+ && "Unknown".equals(getString(resultSet, "UserName", "user_name"))
&& getString(resultSet, "SenderAddress", "sender_address") != null
&& getString(resultSet, "SenderPorts", "sender_ports") != null
&& getInt(resultSet, "ConnectionCount", "connection_count") >= 1
diff --git
a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java
b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java
index 482da8daad1..6bb6d6ef4cd 100644
---
a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java
+++
b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java
@@ -720,4 +720,7 @@ public final class ManagerMessages {
public static final String
MESSAGE_CONFIGNODE_LEADER_IS_WAITING_FOR_NODE_STATUS_PERSISTENCE_8CA96809 =
"ConfigNode leader is waiting for node status persistence.";
+
+ public static final String
LOG_FAILED_TO_RESOLVE_THE_USER_ID_OF_THE_PIPE_SENDER_ARG_THE_AUDIT_RECORDS_OF_THIS_TRANSFER_WILL_NOT_CONTAIN_A_USER_ID_0F8C4256
=
+ "Failed to resolve the user id of the pipe sender {}, the audit records
of this transfer will not contain a user id.";
}
diff --git
a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java
b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java
index 739ee0029c8..10e46904bb5 100644
---
a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java
+++
b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java
@@ -699,4 +699,7 @@ public final class ManagerMessages {
public static final String
MESSAGE_CONFIGNODE_LEADER_IS_WAITING_FOR_NODE_STATUS_PERSISTENCE_8CA96809 =
"ConfigNode leader 正在等待节点状态持久化。";
+
+ public static final String
LOG_FAILED_TO_RESOLVE_THE_USER_ID_OF_THE_PIPE_SENDER_ARG_THE_AUDIT_RECORDS_OF_THIS_TRANSFER_WILL_NOT_CONTAIN_A_USER_ID_0F8C4256
=
+ "无法解析管道发送端 {} 的用户 ID，本次传输的审计记录将不包含用户 ID。";
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/receiver/protocol/IoTDBConfigNodeReceiver.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/receiver/protocol/IoTDBConfigNodeReceiver.java
index 748c8db9acf..e41ac2351e8 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/receiver/protocol/IoTDBConfigNodeReceiver.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/receiver/protocol/IoTDBConfigNodeReceiver.java
@@ -135,6 +135,7 @@ import
org.apache.iotdb.confignode.rpc.thrift.TDeleteTimeSeriesReq;
import org.apache.iotdb.confignode.rpc.thrift.TDropTriggerReq;
import org.apache.iotdb.confignode.rpc.thrift.TSetSchemaTemplateReq;
import org.apache.iotdb.confignode.rpc.thrift.TUnsetSchemaTemplateReq;
+import org.apache.iotdb.confignode.rpc.thrift.TUserResp;
import org.apache.iotdb.confignode.service.ConfigNode;
import org.apache.iotdb.consensus.exception.ConsensusException;
import org.apache.iotdb.db.protocol.session.IClientSession;
@@ -1395,6 +1396,34 @@ public class IoTDBConfigNodeReceiver extends
IoTDBFileReceiver {
return configManager.login(username, password, false).getStatus();
}
+ @Override
+ protected String getLoggedInSessionUsername() {
+ final IClientSession clientSession = SESSION_MANAGER.getCurrSession();
+ return clientSession != null && clientSession.isLogin() ?
clientSession.getUsername() : null;
+ }
+
+ @Override
+ protected IAuditEntity createAuditEntity(final String username, final String
cliHostname) {
+ // The user id must be resolved from the local permission manager instead
of the handshake,
+ // because the id supplied by the sender is not authoritative.
+ long userId = -1L;
+ if (username != null) {
+ try {
+ final TUserResp user =
configManager.getPermissionManager().getUser(username).getUserInfo();
+ if (user != null) {
+ userId = user.getUserId();
+ }
+ } catch (final Exception e) {
+ LOGGER.warn(
+ ManagerMessages
+
.LOG_FAILED_TO_RESOLVE_THE_USER_ID_OF_THE_PIPE_SENDER_ARG_THE_AUDIT_RECORDS_OF_THIS_TRANSFER_WILL_NOT_CONTAIN_A_USER_ID_0F8C4256,
+ username,
+ e);
+ }
+ }
+ return new UserEntity(userId, username, cliHostname);
+ }
+
@Override
protected String getReceiverFileBaseDir() {
return
ConfigNodeDescriptor.getInstance().getConf().getPipeReceiverFileDir();
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 81f1a1000af..861dbfe4c4a 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
@@ -1762,4 +1762,9 @@ public final class StorageEngineMessages {
"Failed to sweep the staged directories of the finished LOAD tasks.";
public static final String
LOG_FAILED_TO_CREATE_SNAPSHOT_FOR_DATABASE_ARG_REGION_ARG_1A7C8DEA =
"Failed to create snapshot for database {}, region {}";
+
+ public static final String
EXCEPTION_FAILED_TO_DETERMINE_THE_USER_THAT_LOADS_THE_TSFILE_E47F7207 =
+ "Failed to determine the user that loads the TsFile";
+ public static final String
EXCEPTION_FAILED_TO_DETERMINE_THE_USER_FOR_THE_TSFILE_DATA_TYPE_CONVERSION_A221793E
=
+ "Failed to determine the user for the TsFile data type conversion";
}
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 44e3076760b..dc9e4f68afc 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
@@ -1681,4 +1681,9 @@ public final class StorageEngineMessages {
"扫描已完成的 LOAD 任务暂存目录失败。";
public static final String
LOG_FAILED_TO_CREATE_SNAPSHOT_FOR_DATABASE_ARG_REGION_ARG_1A7C8DEA =
"为 database {}、region {} 创建快照失败";
+
+ public static final String
EXCEPTION_FAILED_TO_DETERMINE_THE_USER_THAT_LOADS_THE_TSFILE_E47F7207 =
+ "无法确定加载该 TsFile 的用户";
+ public static final String
EXCEPTION_FAILED_TO_DETERMINE_THE_USER_FOR_THE_TSFILE_DATA_TYPE_CONVERSION_A221793E
=
+ "无法确定执行 TsFile 数据类型转换的用户";
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index 919351473b5..0106164a4fd 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -1240,6 +1240,18 @@ public class IoTDBConfig implements
DefaultEncodingProvider {
private boolean loadActiveListeningVerifyEnable = true;
+ /**
+ * Whether a LOAD requires a logged in session with a user name instead of
falling back to the
+ * default user. Disabled by default to keep the legacy fallback.
+ */
+ private boolean loadRequireUserIdentity = false;
+
+ /**
+ * Whether the pipe receiver rejects an asynchronous LOAD without an
authenticated sender. Enabled
+ * by default; set to false to fall back to the default user.
+ */
+ private boolean pipeReceiverRequireUserIdentity = true;
+
private String loadDiskSelectStrategy =
LoadDiskSelectorType.MIN_IO_FIRST.getValue();
private String loadDiskSelectStrategyForIoTV2AndPipe =
@@ -4365,6 +4377,22 @@ public class IoTDBConfig implements
DefaultEncodingProvider {
this.loadActiveListeningVerifyEnable = loadActiveListeningVerifyEnable;
}
+ public boolean isLoadRequireUserIdentity() {
+ return loadRequireUserIdentity;
+ }
+
+ public void setLoadRequireUserIdentity(boolean loadRequireUserIdentity) {
+ this.loadRequireUserIdentity = loadRequireUserIdentity;
+ }
+
+ public boolean isPipeReceiverRequireUserIdentity() {
+ return pipeReceiverRequireUserIdentity;
+ }
+
+ public void setPipeReceiverRequireUserIdentity(boolean
pipeReceiverRequireUserIdentity) {
+ this.pipeReceiverRequireUserIdentity = pipeReceiverRequireUserIdentity;
+ }
+
public String getLoadDiskSelectStrategy() {
return loadDiskSelectStrategy;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index b22ad2dc969..04fe3381b75 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -2709,6 +2709,17 @@ public class IoTDBDescriptor {
"load_active_listening_verify_enable",
Boolean.toString(conf.isLoadActiveListeningVerifyEnable()))));
+ conf.setLoadRequireUserIdentity(
+ Boolean.parseBoolean(
+ properties.getProperty(
+ "load_require_user_identity",
Boolean.toString(conf.isLoadRequireUserIdentity()))));
+
+ conf.setPipeReceiverRequireUserIdentity(
+ Boolean.parseBoolean(
+ properties.getProperty(
+ "pipe_receiver_require_user_identity",
+ Boolean.toString(conf.isPipeReceiverRequireUserIdentity()))));
+
conf.setLoadDiskSelectStrategy(
properties.getProperty(
"load_disk_select_strategy",
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 59f135d34c3..a7eef8a11f0 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
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.pipe.receiver.protocol.thrift;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.audit.IAuditEntity;
+import org.apache.iotdb.commons.audit.UserEntity;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.disk.FolderManager;
import org.apache.iotdb.commons.disk.strategy.DirectoryStrategyType;
@@ -596,6 +597,37 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
return new TPipeTransferResp(getNotLoggedInStatus());
}
+ /**
+ * Returns the name of the user that has been authenticated for the current
transfer, either by
+ * the handshake or by the login of the underlying session, or {@code null}
if no user can be
+ * determined. Callers must not substitute a privileged default for the
missing user.
+ */
+ private String getAuthenticatedUser() {
+ // A name that was not produced by a completed handshake is not an
authenticated identity, no
+ // matter whether it was left over by a failed handshake or by a default
value.
+ if (hasPipeHandshakeCredential && username != null && !username.isEmpty())
{
+ return username;
+ }
+
+ final IClientSession clientSession = SESSION_MANAGER.getCurrSession();
+ if (!isLoggedInSession(clientSession)) {
+ return null;
+ }
+ final String sessionUsername = clientSession.getUsername();
+ return sessionUsername == null || sessionUsername.isEmpty() ? null :
sessionUsername;
+ }
+
+ /**
+ * Returns whether the given session can authorize the statements of the
current transfer.
+ *
+ * <p>The session of a pipe connection is not necessarily logged in. A
connection that has not
+ * completed a handshake owns a session whose user id still defaults to the
id of the super user,
+ * so the presence of a session alone must never be treated as an
authenticated identity.
+ */
+ private static boolean isLoggedInSession(final IClientSession clientSession)
{
+ return Objects.nonNull(clientSession) && clientSession.isLogin();
+ }
+
private static boolean requiresAuthentication(final PipeRequestType
requestType) {
switch (requestType) {
case TRANSFER_TABLET_INSERT_NODE:
@@ -764,6 +796,21 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
final boolean shouldWaitForSchemaBeforeLoad,
final String conversionTaskId)
throws IOException {
+ // The active load runs on another thread, so capture the identity here.
Without a resolvable
+ // identity, fall back to the default user unless the receiver requires
authentication.
+ String authenticatedUser = getAuthenticatedUser();
+ if
(IoTDBDescriptor.getInstance().getConfig().isPipeReceiverRequireUserIdentity())
{
+ final IClientSession clientSession = SESSION_MANAGER.getCurrSession();
+ if (Objects.isNull(authenticatedUser)
+ || Objects.isNull(clientSession)
+ || !clientSession.isLogin()
+ || !authenticatedUser.equals(clientSession.getUsername())) {
+ return getNotLoggedInStatus();
+ }
+ } else if (Objects.isNull(authenticatedUser)) {
+ authenticatedUser = AuthorityChecker.SUPER_USER;
+ }
+
final Map<String, String> loadAttributes =
buildLoadTsFileAttributesForAsync(
dataBaseName,
@@ -771,7 +818,8 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
validateTsFile.get(),
shouldMarkAsPipeRequest.get(),
shouldWaitForSchemaBeforeLoad,
- conversionTaskId);
+ conversionTaskId,
+ authenticatedUser);
if (!LoadUtil.loadFilesToActiveDir(loadAttributes, absolutePaths, true)) {
throw new
PipeException(DataNodePipeMessages.LOAD_ACTIVE_LISTENING_PIPE_DIR_IS_NOT);
@@ -799,42 +847,14 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
&& task.getState() !=
PipeTsFileConversionTaskManager.State.PAUSED);
}
- static Map<String, String> buildLoadTsFileAttributesForAsync(
- final String dataBaseName,
- final boolean shouldConvertDataTypeOnTypeMismatch,
- final boolean validateTsFile,
- final boolean shouldMarkAsPipeRequest) {
- return buildLoadTsFileAttributesForAsync(
- dataBaseName,
- shouldConvertDataTypeOnTypeMismatch,
- validateTsFile,
- shouldMarkAsPipeRequest,
- false,
- null);
- }
-
- static Map<String, String> buildLoadTsFileAttributesForAsync(
- final String dataBaseName,
- final boolean shouldConvertDataTypeOnTypeMismatch,
- final boolean validateTsFile,
- final boolean shouldMarkAsPipeRequest,
- final boolean shouldWaitForSchemaBeforeLoad) {
- return buildLoadTsFileAttributesForAsync(
- dataBaseName,
- shouldConvertDataTypeOnTypeMismatch,
- validateTsFile,
- shouldMarkAsPipeRequest,
- shouldWaitForSchemaBeforeLoad,
- null);
- }
-
static Map<String, String> buildLoadTsFileAttributesForAsync(
final String dataBaseName,
final boolean shouldConvertDataTypeOnTypeMismatch,
final boolean validateTsFile,
final boolean shouldMarkAsPipeRequest,
final boolean shouldWaitForSchemaBeforeLoad,
- final String conversionTaskId) {
+ final String conversionTaskId,
+ final String userName) {
return ActiveLoadPathHelper.buildAttributes(
dataBaseName,
LoadTsFileStatement.getDatabaseLevelByTreeDatabase(dataBaseName),
@@ -843,7 +863,9 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
!shouldWaitForSchemaBeforeLoad,
null,
shouldMarkAsPipeRequest,
- AuthorityChecker.SUPER_USER,
+ // The user executing the asynchronous load. It must always be the
authenticated sender: a
+ // privileged default here would let any authenticated account load
data as that user.
+ userName,
conversionTaskId);
}
@@ -1486,7 +1508,10 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
}
final IClientSession clientSession = SESSION_MANAGER.getCurrSession();
-
+ if (!isLoggedInSession(clientSession)) {
+ // Without a logged in session the statement can not be authorized, so
fail closed.
+ return getNotLoggedInStatus();
+ }
// For table model, the authority check is done in inner execution. No
need to check here
if (!isTableModelStatement) {
final TSStatus permissionCheckStatus =
@@ -1595,8 +1620,42 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
return AuthorityChecker.checkUser(username, password);
}
+ @Override
+ protected String getLoggedInSessionUsername() {
+ final IClientSession clientSession = SESSION_MANAGER.getCurrSession();
+ return Objects.nonNull(clientSession) && clientSession.isLogin()
+ ? clientSession.getUsername()
+ : null;
+ }
+
+ @Override
+ protected IAuditEntity createAuditEntity(final String username, final String
cliHostname) {
+ // The role of the transfer is the one whose credentials were just
verified. Its id is resolved
+ // from the local user manager and never taken from the handshake: the id
supplied by the sender
+ // is not verified, and some table model authorization paths treat the
super user id as a grant
+ // of every privilege.
+ final IClientSession clientSession = SESSION_MANAGER.getCurrSession();
+ // Only the address of the connection is taken from the session. The
identity comes from the
+ // handshake and its id is resolved locally, because both the hostname and
the id of the
+ // handshake are supplied by the sender.
+ final String clientAddress =
+ Objects.nonNull(clientSession) ? clientSession.getClientAddress() :
null;
+ return new UserEntity(
+ Objects.isNull(username) ? -1L :
AuthorityChecker.getUserId(username).orElse(-1L),
+ username,
+ Objects.nonNull(clientAddress) && !clientAddress.isEmpty() ?
clientAddress : cliHostname);
+ }
+
private TSStatus executeStatementForTableModel(
final Statement statement, final String databaseName) {
+ // The transfer runs on the thread of its connection, so the session here
is the one that the
+ // handshake logged in. A statement can only be authorized by a logged in
session: a session of
+ // a connection without a completed handshake still carries the id of the
super user, so it must
+ // not be treated as authenticated, and statements can not be authorized
without any session.
+ final IClientSession clientSession = SESSION_MANAGER.getCurrSession();
+ if (!isLoggedInSession(clientSession)) {
+ return getNotLoggedInStatus();
+ }
try {
autoCreateDatabaseIfNecessary(databaseName);
@@ -1604,10 +1663,9 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
.executeForTableModel(
shouldMarkAsPipeRequest.get() ? new
PipeEnrichedStatement(statement) : statement,
tableSqlParser,
- SESSION_MANAGER.getCurrSession(),
+ clientSession,
SESSION_MANAGER.requestQueryId(),
- SESSION_MANAGER.getSessionInfoOfPipeReceiver(
- SESSION_MANAGER.getCurrSession(), databaseName),
+ SESSION_MANAGER.getSessionInfoOfPipeReceiver(clientSession,
databaseName),
"",
LocalExecutionPlanner.getInstance().metadata,
IoTDBDescriptor.getInstance().getConfig().getQueryTimeoutThreshold())
@@ -1626,10 +1684,9 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
.executeForTableModel(
shouldMarkAsPipeRequest.get() ? new
PipeEnrichedStatement(statement) : statement,
tableSqlParser,
- SESSION_MANAGER.getCurrSession(),
+ clientSession,
SESSION_MANAGER.requestQueryId(),
- SESSION_MANAGER.getSessionInfoOfPipeReceiver(
- SESSION_MANAGER.getCurrSession(), databaseName),
+ SESSION_MANAGER.getSessionInfoOfPipeReceiver(clientSession,
databaseName),
"",
LocalExecutionPlanner.getInstance().metadata,
IoTDBDescriptor.getInstance().getConfig().getQueryTimeoutThreshold())
@@ -1753,6 +1810,12 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
private TSStatus executeStatementForTreeModel(
final Statement statement, final String databaseName) {
+ final IClientSession clientSession = SESSION_MANAGER.getCurrSession();
+ if (!isLoggedInSession(clientSession)) {
+ // Without a logged in session the statement can not be authorized, so
fail closed.
+ return getNotLoggedInStatus();
+ }
+
if (autoCreateTreeDatabaseIfNecessary(databaseName) ==
TreeDatabaseCreationResult.CONFLICTED) {
// Continue execution, but let partition analysis infer the
receiver-side database.
clearTreeDatabaseName(statement);
@@ -1762,7 +1825,7 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
.executeForTreeModel(
shouldMarkAsPipeRequest.get() ? new
PipeEnrichedStatement(statement) : statement,
SESSION_MANAGER.requestQueryId(),
- SESSION_MANAGER.getSessionInfo(SESSION_MANAGER.getCurrSession()),
+ SESSION_MANAGER.getSessionInfo(clientSession),
"",
ClusterPartitionFetcher.getInstance(),
ClusterSchemaFetcher.getInstance(),
@@ -1828,15 +1891,20 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
return status;
}
+ final IClientSession clientSession = SESSION_MANAGER.getCurrSession();
+ if (!isLoggedInSession(clientSession)) {
+ // Without a logged in session the statement can not be authorized, so
fail closed.
+ return getNotLoggedInStatus();
+ }
+
final TSStatus result =
Coordinator.getInstance()
.executeForTableModel(
shouldMarkAsPipeRequest.get() ? new PipeEnriched(statement)
: statement,
tableSqlParser,
- SESSION_MANAGER.getCurrSession(),
+ clientSession,
SESSION_MANAGER.requestQueryId(),
- SESSION_MANAGER.getSessionInfoOfPipeReceiver(
- SESSION_MANAGER.getCurrSession(), databaseName),
+ SESSION_MANAGER.getSessionInfoOfPipeReceiver(clientSession,
databaseName),
"",
LocalExecutionPlanner.getInstance().metadata,
IoTDBDescriptor.getInstance().getConfig().getQueryTimeoutThreshold(),
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 268ea2a222e..2e6a978937d 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
@@ -323,7 +323,9 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
isAutoCreateSchemaRequested,
tabletConversionThresholdBytes,
isGeneratedByPipe,
- Objects.nonNull(context) ? context.getUsername() : null);
+ Objects.nonNull(context) && Objects.nonNull(context.getSession())
+ ? context.getUsername()
+ : null);
if (LoadUtil.loadTsFileAsyncToActiveDir(tsFiles, activeLoadAttributes,
isDeleteAfterLoad)) {
analysis.setFinishQueryAfterAnalyze(true);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTsFileDataTypeConverter.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTsFileDataTypeConverter.java
index 5382afc6f03..20d2802e0fd 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTsFileDataTypeConverter.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTsFileDataTypeConverter.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.storageengine.load.converter;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.audit.UserEntity;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
@@ -50,6 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.ZoneId;
+import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.Semaphore;
@@ -198,25 +200,56 @@ public class LoadTsFileDataTypeConverter {
}
}
- private TSStatus executeForTableModel(final Statement statement, final
String databaseName) {
- final IClientSession session;
- final boolean needToCreateSession = SESSION_MANAGER.getCurrSession() ==
null;
- if (needToCreateSession) {
- session =
- new InternalClientSession(
- String.format(
- "%s_%s",
- LoadTsFileDataTypeConverter.class.getSimpleName(),
- Thread.currentThread().getName()));
- session.setUsername(AuthorityChecker.SUPER_USER);
- session.setClientVersion(IoTDBConstant.ClientVersion.V_1_0);
- session.setZoneId(ZoneId.systemDefault());
- session.setSqlDialect(SqlDialect.TABLE);
+ /**
+ * Creates and registers an internal session for a conversion that is
executed outside of the
+ * request thread. The session inherits the identity of the load request, so
the conversion is
+ * still authorized and audited as that user. A conversion without any
identity is rejected
+ * instead of being executed as a privileged default user.
+ */
+ private IClientSession createConversionSession(final SqlDialect sqlDialect) {
+ final IClientSession session =
+ new InternalClientSession(
+ String.format(
+ "%s_%s",
+ LoadTsFileDataTypeConverter.class.getSimpleName(),
+ Thread.currentThread().getName()));
- SESSION_MANAGER.registerSession(session);
- } else {
- session = SESSION_MANAGER.getCurrSession();
+ final UserEntity userEntity =
+ Objects.isNull(context) || Objects.isNull(context.getSession())
+ ? null
+ : context.getSession().getUserEntity();
+ if (Objects.isNull(userEntity)
+ || Objects.isNull(userEntity.getUsername())
+ || userEntity.getUsername().isEmpty()) {
+ throw new IllegalStateException(
+ StorageEngineMessages
+
.EXCEPTION_FAILED_TO_DETERMINE_THE_USER_FOR_THE_TSFILE_DATA_TYPE_CONVERSION_A221793E);
}
+ // The user id carried by the session is not trusted: a session that only
carries a user name
+ // keeps the default user id 0, which is the id of the super user. Resolve
the id from the user
+ // name again, so that the conversion can never run as a privileged
default user.
+ final long userId =
+ AuthorityChecker.getUserId(userEntity.getUsername())
+ .orElseThrow(
+ () ->
+ new IllegalStateException(
+ StorageEngineMessages
+
.EXCEPTION_FAILED_TO_DETERMINE_THE_USER_FOR_THE_TSFILE_DATA_TYPE_CONVERSION_A221793E));
+ session.setUserId(userId);
+ session.setUsername(userEntity.getUsername());
+ session.setClientVersion(IoTDBConstant.ClientVersion.V_1_0);
+ session.setZoneId(ZoneId.systemDefault());
+ session.setSqlDialect(sqlDialect);
+
+ SESSION_MANAGER.registerSession(session);
+ return session;
+ }
+
+ private TSStatus executeForTableModel(final Statement statement, final
String databaseName) {
+ final IClientSession currentSession = SESSION_MANAGER.getCurrSession();
+ final boolean needToCreateSession = Objects.isNull(currentSession);
+ final IClientSession session =
+ needToCreateSession ? createConversionSession(SqlDialect.TABLE) :
currentSession;
try {
return Coordinator.getInstance()
.executeForTableModel(
@@ -266,24 +299,10 @@ public class LoadTsFileDataTypeConverter {
}
private TSStatus executeForTreeModel(final Statement statement) {
- final IClientSession session;
- final boolean needToCreateSession = SESSION_MANAGER.getCurrSession() ==
null;
- if (needToCreateSession) {
- session =
- new InternalClientSession(
- String.format(
- "%s_%s",
- LoadTsFileDataTypeConverter.class.getSimpleName(),
- Thread.currentThread().getName()));
- session.setUsername(AuthorityChecker.SUPER_USER);
- session.setClientVersion(IoTDBConstant.ClientVersion.V_1_0);
- session.setZoneId(ZoneId.systemDefault());
- session.setSqlDialect(SqlDialect.TREE);
-
- SESSION_MANAGER.registerSession(session);
- } else {
- session = SESSION_MANAGER.getCurrSession();
- }
+ final IClientSession currentSession = SESSION_MANAGER.getCurrSession();
+ final boolean needToCreateSession = Objects.isNull(currentSession);
+ final IClientSession session =
+ needToCreateSession ? createConversionSession(SqlDialect.TREE) :
currentSession;
try {
return Coordinator.getInstance()
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 1ad0a93b83e..1ba1a840e09 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
@@ -178,18 +178,31 @@ public class LoadUtil {
}
private static Map<String, String> appendCurrentUserIfAbsent(
- final Map<String, String> loadAttributes) {
+ final Map<String, String> loadAttributes) throws IOException {
final Map<String, String> 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());
+ final String username = Objects.isNull(session) ? null :
session.getUsername();
+ if
(IoTDBDescriptor.getInstance().getConfig().isLoadRequireUserIdentity()) {
+ if (Objects.isNull(session)
+ || !session.isLogin()
+ || Objects.isNull(username)
+ || username.isEmpty()) {
+ throw new IOException(
+ StorageEngineMessages
+
.EXCEPTION_FAILED_TO_DETERMINE_THE_USER_THAT_LOADS_THE_TSFILE_E47F7207);
+ }
+ attributes.put(ActiveLoadPathHelper.USER_KEY, username);
+ } else {
+ attributes.put(
+ ActiveLoadPathHelper.USER_KEY,
+ Objects.isNull(username) || username.isEmpty()
+ ? AuthorityChecker.SUPER_USER
+ : username);
+ }
}
return attributes;
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverAuditEntityTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverAuditEntityTest.java
new file mode 100644
index 00000000000..c1127a8a6bd
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverAuditEntityTest.java
@@ -0,0 +1,79 @@
+/*
+ * 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.pipe.receiver.protocol.thrift;
+
+import org.apache.iotdb.commons.audit.IAuditEntity;
+import org.apache.iotdb.db.auth.AuthorityChecker;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import java.util.Optional;
+
+import static org.mockito.Mockito.when;
+import static org.powermock.api.mockito.PowerMockito.mockStatic;
+
+@PowerMockIgnore({"com.sun.org.apache.xerces.*", "javax.xml.*", "org.xml.*",
"javax.management.*"})
+@RunWith(PowerMockRunner.class)
+@PrepareForTest(AuthorityChecker.class)
+public class IoTDBDataNodeReceiverAuditEntityTest {
+
+ @Test
+ public void testAuditEntityTakesTheUserIdFromTheUserManager() {
+ mockStatic(AuthorityChecker.class);
+
when(AuthorityChecker.getUserId("pipe_user")).thenReturn(Optional.of(10001L));
+
+ final IAuditEntity auditEntity =
+ new IoTDBDataNodeReceiver().createAuditEntity("pipe_user",
"sender-host");
+
+ Assert.assertEquals(10001L, auditEntity.getUserId());
+ Assert.assertEquals("pipe_user", auditEntity.getUsername());
+ Assert.assertEquals("sender-host", auditEntity.getCliHostname());
+ }
+
+ @Test
+ public void testAuditEntityOfUnknownUserIsNotTheSuperUser() {
+ mockStatic(AuthorityChecker.class);
+
when(AuthorityChecker.getUserId("unknown_user")).thenReturn(Optional.empty());
+
+ final IAuditEntity auditEntity =
+ new IoTDBDataNodeReceiver().createAuditEntity("unknown_user",
"sender-host");
+
+ // The super user id would grant every table model privilege, so an
unresolved user must never
+ // be mapped to it.
+ Assert.assertEquals(-1L, auditEntity.getUserId());
+ Assert.assertEquals("unknown_user", auditEntity.getUsername());
+ }
+
+ @Test
+ public void testAuditEntityWithoutUsernameIsNotTheSuperUser() {
+ mockStatic(AuthorityChecker.class);
+
+ final IAuditEntity auditEntity =
+ new IoTDBDataNodeReceiver().createAuditEntity(null, "sender-host");
+
+ Assert.assertEquals(-1L, auditEntity.getUserId());
+ Assert.assertNull(auditEntity.getUsername());
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverTest.java
index b71542a0cc0..4f91a311a57 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverTest.java
@@ -52,6 +52,8 @@ import java.util.concurrent.atomic.AtomicReference;
public class IoTDBDataNodeReceiverTest {
+ private static final String ASYNC_LOAD_SENDER = "pipe_async_load_sender";
+
private final PipeReceiverRuntimeRegistry registry =
PipeReceiverRuntimeRegistry.getInstance();
@Before
@@ -153,11 +155,12 @@ public class IoTDBDataNodeReceiverTest {
try {
final Map<String, String> attributes =
IoTDBDataNodeReceiver.buildLoadTsFileAttributesForAsync(
- "root.test.sg_0", true, false, true);
+ "root.test.sg_0", true, false, true, false, null,
ASYNC_LOAD_SENDER);
Assert.assertEquals(
"root.test.sg_0",
attributes.get(LoadTsFileConfigurator.DATABASE_NAME_KEY));
Assert.assertEquals("2",
attributes.get(LoadTsFileConfigurator.DATABASE_LEVEL_KEY));
+ Assert.assertEquals(ASYNC_LOAD_SENDER,
attributes.get(ActiveLoadPathHelper.USER_KEY));
final LoadTsFileStatement statement =
LoadTsFileStatement.createUnchecked(tsFile.toString());
ActiveLoadPathHelper.applyAttributesToStatement(attributes, statement,
false);
@@ -170,6 +173,19 @@ public class IoTDBDataNodeReceiverTest {
}
}
+ @Test
+ public void
testLoadTsFileAsyncAttributesCarryAuthenticatedSenderInsteadOfRoot() {
+ // The asynchronous load is executed by the active loader in another
thread. The attributes must
+ // carry the authenticated sender so that the load is authorized as that
user. Falling back to
+ // the super user here lets any authenticated account load data as root.
+ final Map<String, String> attributes =
+ IoTDBDataNodeReceiver.buildLoadTsFileAttributesForAsync(
+ "root.test.sg_0", false, true, true, false, "conversion-task-id",
ASYNC_LOAD_SENDER);
+
+ Assert.assertEquals(ASYNC_LOAD_SENDER,
attributes.get(ActiveLoadPathHelper.USER_KEY));
+ Assert.assertNotEquals("root",
attributes.get(ActiveLoadPathHelper.USER_KEY));
+ }
+
@Test
public void testLoadTsFileWaitsForSchemaInSyncAndAsyncModes() throws
Exception {
final Path tsFile = Files.createTempFile("pipe-load-wait-for-schema",
".tsfile");
@@ -182,7 +198,7 @@ public class IoTDBDataNodeReceiverTest {
final Map<String, String> asyncAttributes =
IoTDBDataNodeReceiver.buildLoadTsFileAttributesForAsync(
- "root.test.sg_0", false, false, true, true);
+ "root.test.sg_0", false, false, true, true, null,
ASYNC_LOAD_SENDER);
Assert.assertEquals(
Boolean.TRUE.toString(),
asyncAttributes.get(LoadTsFileConfigurator.VERIFY_KEY));
Assert.assertEquals(
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/util/LoadUtilTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/util/LoadUtilTest.java
index 10636116b0b..492d27462d5 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/util/LoadUtilTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/util/LoadUtilTest.java
@@ -19,9 +19,11 @@
package org.apache.iotdb.db.storageengine.load.util;
+import org.apache.iotdb.db.auth.AuthorityChecker;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.i18n.DataNodeQueryMessages;
+import org.apache.iotdb.db.i18n.StorageEngineMessages;
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;
@@ -38,6 +40,8 @@ import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -46,6 +50,7 @@ public class LoadUtilTest {
private final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
private String[] originalListeningDirs;
+ private boolean originalRequireUserIdentity;
private File tempDir;
private File sourceDir;
private File targetDir;
@@ -58,6 +63,7 @@ public class LoadUtilTest {
Assert.assertTrue(sourceDir.mkdirs());
Assert.assertTrue(targetDir.mkdirs());
originalListeningDirs = config.getLoadActiveListeningDirs();
+ originalRequireUserIdentity = config.isLoadRequireUserIdentity();
config.setLoadActiveListeningDirs(new String[]
{targetDir.getAbsolutePath()});
LoadUtil.updateLoadDiskSelector();
}
@@ -65,6 +71,7 @@ public class LoadUtilTest {
@After
public void tearDown() {
config.setLoadActiveListeningDirs(originalListeningDirs);
+ config.setLoadRequireUserIdentity(originalRequireUserIdentity);
LoadUtil.updateLoadDiskSelector();
deleteRecursively(tempDir);
}
@@ -278,6 +285,61 @@ public class LoadUtilTest {
}
}
+ @Test
+ public void testLoadToActiveDirRequiresTheUserWhenConfigured() throws
Exception {
+ config.setLoadRequireUserIdentity(true);
+ final File tsFile = createCompletedTsFile("without-user.tsfile");
+
+ try {
+ LoadUtil.loadFilesToActiveDir(
+ new HashMap<>(),
Collections.singletonList(tsFile.getAbsolutePath()), false);
+ Assert.fail("The load must be rejected when no user is logged in");
+ } catch (final IOException e) {
+ Assert.assertEquals(
+ StorageEngineMessages
+
.EXCEPTION_FAILED_TO_DETERMINE_THE_USER_THAT_LOADS_THE_TSFILE_E47F7207,
+ e.getMessage());
+ }
+
+ Assert.assertTrue("The source must be kept", tsFile.isFile());
+ Assert.assertEquals(0, targetDir.listFiles().length);
+ }
+
+ @Test
+ public void testLoadToActiveDirUsesTheDefaultUserWhenNotConfigured() throws
Exception {
+ config.setLoadRequireUserIdentity(false);
+ final File tsFile = createCompletedTsFile("default-user.tsfile");
+
+ Assert.assertTrue(
+ LoadUtil.loadFilesToActiveDir(
+ new HashMap<>(),
Collections.singletonList(tsFile.getAbsolutePath()), false));
+
+ final File loadedTsFile = findTsFile(targetDir);
+ Assert.assertNotNull(loadedTsFile);
+ Assert.assertEquals(
+ AuthorityChecker.SUPER_USER,
+ ActiveLoadPathHelper.parseAttributes(loadedTsFile, targetDir)
+ .get(ActiveLoadPathHelper.USER_KEY));
+ }
+
+ private static File findTsFile(final File dir) {
+ final File[] files = dir.listFiles();
+ if (files == null) {
+ return null;
+ }
+ for (final File file : files) {
+ if (file.isDirectory()) {
+ final File nested = findTsFile(file);
+ if (nested != null) {
+ return nested;
+ }
+ } else if (file.getName().endsWith(".tsfile")) {
+ return file;
+ }
+ }
+ return null;
+ }
+
private List<File> createTsFileAndCompanions() throws Exception {
return createTsFileAndCompanions("1-0-0-0.tsfile");
}
diff --git
a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java
b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java
index c09c42a188f..437e4dd68de 100644
---
a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java
+++
b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java
@@ -615,6 +615,8 @@ public final class PipeMessages {
"Receiver id = %s: Handshake failed, response status = %s.";
public static final String RECEIVER_HANDSHAKE_FAILED_LOGIN =
"Receiver id = %s: Handshake failed because login failed, response
status = %s.";
+ public static final String
MESSAGE_THE_PIPE_RECEIVER_CONNECTION_IS_ALREADY_AUTHENTICATED_AS_ANOTHER_USER_AND_CANNOT_BE_USED_TO_TRANSFER_DATA_AS_ARG_FFA5DF78
=
+ "The pipe receiver connection is already authenticated as another user
and cannot be used to transfer data as %s.";
public static final String RECEIVER_TEMPORARILY_OUT_OF_MEMORY_FORMAT =
"Temporarily out of memory when %s. Requested memory: %d bytes. Root
cause: %s";
public static final String RECEIVER_USER_LOGIN_SUCCESS =
diff --git
a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java
b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java
index b8fd1db6729..f67f0a351ac 100644
---
a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java
+++
b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java
@@ -595,6 +595,8 @@ public final class PipeMessages {
"接收器 id = %s：握手失败，响应状态 = %s。";
public static final String RECEIVER_HANDSHAKE_FAILED_LOGIN =
"接收器 id = %s：因登录失败导致握手失败，响应状态 = %s。";
+ public static final String
MESSAGE_THE_PIPE_RECEIVER_CONNECTION_IS_ALREADY_AUTHENTICATED_AS_ANOTHER_USER_AND_CANNOT_BE_USED_TO_TRANSFER_DATA_AS_ARG_FFA5DF78
=
+ "管道接收端连接已认证为其他用户，无法以 %s 的身份传输数据。";
public static final String RECEIVER_TEMPORARILY_OUT_OF_MEMORY_FORMAT =
"执行 %s 时暂时内存不足。请求内存：%d bytes。根因：%s";
public static final String RECEIVER_USER_LOGIN_SUCCESS =
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java
index 5b10fb34132..d67d099b46a 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java
@@ -63,8 +63,6 @@ import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_EXCEPTION_DATA_CONVERT_ON_TYPE_MISMATCH_DEFAULT_VALUE;
-import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_PASSWORD_DEFAULT_VALUE;
-import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_USER_DEFAULT_VALUE;
/**
* {@link IoTDBFileReceiver} is the parent class of receiver on both
configNode and DataNode,
@@ -82,8 +80,11 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
// Used to restore the original thread name when the receiver is closed.
private String originalThreadName;
- protected String username = CONNECTOR_IOTDB_USER_DEFAULT_VALUE;
- protected String password = CONNECTOR_IOTDB_PASSWORD_DEFAULT_VALUE;
+ // The identity of the sender is only set after a successful handshake or
taken from a logged in
+ // session. There is deliberately no default user, because a missing
identity must never degrade
+ // to a privileged one.
+ protected String username;
+ protected String password;
protected IAuditEntity userEntity;
protected boolean hasPipeHandshakeCredential = false;
@@ -297,14 +298,11 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
return new TPipeTransferResp(status);
}
- long userId = -1;
+ // The user id carried by the handshake is supplied by the sender and can
be forged, so it is
+ // deliberately ignored here. The audit entity is resolved by the receiver
after the credentials
+ // have been verified, see createAuditEntity.
String cliHostname = "";
- final String userIdString =
-
req.getParams().get(PipeTransferHandshakeConstant.HANDSHAKE_KEY_USER_ID);
- if (userIdString != null) {
- userId = Long.parseLong(userIdString);
- }
final String cliHostnameString =
req.getParams().get(PipeTransferHandshakeConstant.HANDSHAKE_KEY_CLI_HOSTNAME);
if (cliHostnameString != null) {
@@ -321,9 +319,27 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
TSStatusCode.NOT_LOGIN, "Pipe handshake missing username or
password."));
}
+ // A connection may only ever stand for one user. If it is already
authenticated as somebody
+ // else, the credentials of this handshake must not rebind it: the
statements of the transfer
+ // would otherwise be executed with the privileges of the other user.
+ final String loggedInSessionUsername = getLoggedInSessionUsername();
+ if (Objects.nonNull(loggedInSessionUsername)
+ && !loggedInSessionUsername.equals(usernameString)) {
+ final TSStatus status =
+ RpcUtils.getStatus(
+ TSStatusCode.NO_PERMISSION,
+ String.format(
+ PipeMessages
+
.MESSAGE_THE_PIPE_RECEIVER_CONNECTION_IS_ALREADY_AUTHENTICATED_AS_ANOTHER_USER_AND_CANNOT_BE_USED_TO_TRANSFER_DATA_AS_ARG_FFA5DF78,
+ usernameString));
+ PipeLogger.log(
+ LOGGER::warn, PipeMessages.RECEIVER_HANDSHAKE_FAILED_WITH_ID,
receiverId.get(), status);
+ return new TPipeTransferResp(status);
+ }
+
username = usernameString;
password = passwordString;
- userEntity = new UserEntity(userId, username, cliHostname);
+ userEntity = null;
hasPipeHandshakeCredential = true;
final TSStatus status;
@@ -331,15 +347,18 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
status = login();
} catch (final Exception e) {
hasPipeHandshakeCredential = false;
+ userEntity = null;
throw e;
}
if (status.code != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
hasPipeHandshakeCredential = false;
+ userEntity = null;
PipeLogger.log(
LOGGER::warn, PipeMessages.RECEIVER_HANDSHAKE_FAILED_LOGIN,
receiverId.get(), status);
return new TPipeTransferResp(status);
} else {
lastSuccessfulLoginTime = System.currentTimeMillis();
+ userEntity = createAuditEntity(username, cliHostname);
LOGGER.info(PipeMessages.RECEIVER_USER_LOGIN_SUCCESS, receiverId.get(),
username);
}
@@ -399,6 +418,10 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
}
hasPipeHandshakeCredential =
handshakeResp.getStatus().getCode() ==
TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ if (!hasPipeHandshakeCredential) {
+ // The handshake did not complete, so the identity of this transfer can
not be used.
+ userEntity = null;
+ }
return handshakeResp;
}
@@ -444,6 +467,28 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
protected abstract TSStatus login();
+ /**
+ * Returns the user name of the client session that this connection is
authenticated with, or
+ * {@code null} if the connection carries no logged in session.
+ *
+ * <p>A handshake is rejected when it tries to authenticate a connection
with another user, so
+ * that the identity of a transfer is always the one whose credentials were
verified.
+ */
+ protected String getLoggedInSessionUsername() {
+ return null;
+ }
+
+ /**
+ * Creates the audit entity of a sender whose credentials have been verified.
+ *
+ * <p>The identity is resolved on the receiver side. Names and ids carried
by the handshake are
+ * not authoritative, so implementations must look the user up in the user
manager of their own
+ * node instead of trusting the sender.
+ */
+ protected IAuditEntity createAuditEntity(final String username, final String
cliHostname) {
+ return new UserEntity(-1L, username, cliHostname);
+ }
+
protected final TPipeTransferResp handleTransferFilePiece(
final PipeTransferFilePieceReq req,
final boolean isRequestThroughAirGap,
@@ -1133,6 +1178,13 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
// Close the session
closeSession();
+ // Do not keep the credentials or the identity of a finished transfer, so
that a reused
+ // receiver object can never answer an unauthenticated request with them.
+ username = null;
+ password = null;
+ userEntity = null;
+ hasPipeHandshakeCredential = false;
+
LOGGER.info(PipeMessages.RECEIVER_EXITED, receiverId.get());
if (originalThreadName != null) {
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiverTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiverTest.java
index 9cf7a9f233b..6426b4f83d5 100644
---
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiverTest.java
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiverTest.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.commons.pipe.receiver;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.audit.IAuditEntity;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.exception.IllegalPathException;
import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException;
@@ -167,6 +168,114 @@ public class IoTDBFileReceiverTest {
}
}
+ @Test
+ public void testHandshakeV2IgnoresSenderSuppliedUserId() throws Exception {
+ final Path baseDir = Files.createTempDirectory("iotdb-file-receiver-test");
+ final DummyFileReceiver receiver = new DummyFileReceiver(baseDir.toFile());
+ try {
+ final Map<String, String> params = buildHandshakeV2Params(true);
+ // A sender must not be able to choose its own user id. The super user
id is especially
+ // dangerous, because some table model authorization paths treat it as a
grant of every
+ // privilege.
+ params.put(PipeTransferHandshakeConstant.HANDSHAKE_KEY_USER_ID, "0");
+ params.put(PipeTransferHandshakeConstant.HANDSHAKE_KEY_CLI_HOSTNAME,
"sender-host");
+
+ final TPipeTransferResp response = receiver.handshakeV2(params);
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
response.getStatus().getCode());
+ // The identity is created by the receiver, after the credentials have
been verified.
+ Assert.assertEquals(1, receiver.getCreateAuditEntityCallCount());
+ Assert.assertEquals("root", receiver.getCreateAuditEntityUsername());
+ final IAuditEntity auditEntity = receiver.getUserEntity();
+ Assert.assertNotNull(auditEntity);
+ Assert.assertEquals("root", auditEntity.getUsername());
+ Assert.assertEquals("sender-host", auditEntity.getCliHostname());
+ // The base receiver resolves no id at all instead of reusing the id of
the sender.
+ Assert.assertEquals(-1L, auditEntity.getUserId());
+ } finally {
+ receiver.handleExit();
+ }
+ }
+
+ @Test
+ public void testHandshakeV2DoesNotKeepIdentityWhenLoginFails() throws
Exception {
+ final Path baseDir = Files.createTempDirectory("iotdb-file-receiver-test");
+ final DummyFileReceiver receiver = new DummyFileReceiver(baseDir.toFile());
+ try {
+ receiver.setLoginStatus(new
TSStatus(TSStatusCode.WRONG_LOGIN_PASSWORD.getStatusCode()));
+ final Map<String, String> params = buildHandshakeV2Params(true);
+ params.put(PipeTransferHandshakeConstant.HANDSHAKE_KEY_USER_ID, "0");
+
+ final TPipeTransferResp response = receiver.handshakeV2(params);
+
+ Assert.assertEquals(
+ TSStatusCode.WRONG_LOGIN_PASSWORD.getStatusCode(),
response.getStatus().getCode());
+ Assert.assertFalse(receiver.hasPipeHandshakeCredential());
+ Assert.assertNull(receiver.getUserEntity());
+ Assert.assertEquals(0, receiver.getCreateAuditEntityCallCount());
+ } finally {
+ receiver.handleExit();
+ }
+ }
+
+ @Test
+ public void testExitClearsIdentityOfTheTransfer() throws Exception {
+ final Path baseDir = Files.createTempDirectory("iotdb-file-receiver-test");
+ final DummyFileReceiver receiver = new DummyFileReceiver(baseDir.toFile());
+ receiver.handshakeV2(buildHandshakeV2Params(true));
+
+ Assert.assertNotNull(receiver.getUserEntity());
+ Assert.assertTrue(receiver.hasPipeHandshakeCredential());
+
+ receiver.handleExit();
+
+ Assert.assertNull(receiver.getUserEntity());
+ Assert.assertFalse(receiver.hasPipeHandshakeCredential());
+ }
+
+ @Test
+ public void testHandshakeV2RejectsAnotherUserOnALoggedInConnection() throws
Exception {
+ final Path baseDir = Files.createTempDirectory("iotdb-file-receiver-test");
+ final DummyFileReceiver receiver = new DummyFileReceiver(baseDir.toFile());
+ try {
+ // The connection is already authenticated as another user. Accepting
the credentials of this
+ // handshake would execute the statements of the transfer with the
privileges of that user.
+ receiver.setLoggedInSessionUsername("another_user");
+
+ final TPipeTransferResp response =
receiver.handshakeV2(buildHandshakeV2Params(true));
+
+ Assert.assertEquals(
+ TSStatusCode.NO_PERMISSION.getStatusCode(),
response.getStatus().getCode());
+ // The credentials are not even checked, because the connection can not
be rebound anyway.
+ Assert.assertEquals(0, receiver.getLoginCallCount());
+ Assert.assertEquals(0, receiver.getCreateAuditEntityCallCount());
+ Assert.assertFalse(receiver.hasPipeHandshakeCredential());
+ Assert.assertNull(receiver.getUserEntity());
+ } finally {
+ receiver.handleExit();
+ }
+ }
+
+ @Test
+ public void testHandshakeV2AcceptsTheUserOfTheLoggedInConnection() throws
Exception {
+ final Path baseDir = Files.createTempDirectory("iotdb-file-receiver-test");
+ final DummyFileReceiver receiver = new DummyFileReceiver(baseDir.toFile());
+ try {
+ receiver.setLoggedInSessionUsername("root");
+
+ final TPipeTransferResp response =
receiver.handshakeV2(buildHandshakeV2Params(true));
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
response.getStatus().getCode());
+ Assert.assertEquals(1, receiver.getLoginCallCount());
+ Assert.assertEquals(1, receiver.getCreateAuditEntityCallCount());
+ Assert.assertNotNull(receiver.getUserEntity());
+ } finally {
+ receiver.handleExit();
+ }
+ }
+
private Map<String, String> buildHandshakeV2Params(final boolean
includeCredentials) {
final Map<String, String> params = new HashMap<>();
params.put(PipeTransferHandshakeConstant.HANDSHAKE_KEY_CLUSTER_ID,
"sender-cluster");
@@ -250,7 +359,11 @@ public class IoTDBFileReceiverTest {
private final File receiverFileBaseDir;
private TSStatus loadFileV1Status = new
TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ private TSStatus loginStatus = new
TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
private int loginCallCount = 0;
+ private String loggedInSessionUsername = null;
+ private int createAuditEntityCallCount = 0;
+ private String createAuditEntityUsername = null;
private boolean failFilePieceMemoryAllocation = false;
private int filePieceMemoryCloseCount = 0;
@@ -367,10 +480,42 @@ public class IoTDBFileReceiverTest {
return "test-cluster";
}
+ @Override
+ protected String getLoggedInSessionUsername() {
+ return loggedInSessionUsername;
+ }
+
@Override
protected TSStatus login() {
loginCallCount++;
- return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ return loginStatus;
+ }
+
+ @Override
+ protected IAuditEntity createAuditEntity(final String username, final
String cliHostname) {
+ createAuditEntityCallCount++;
+ createAuditEntityUsername = username;
+ return super.createAuditEntity(username, cliHostname);
+ }
+
+ void setLoggedInSessionUsername(final String loggedInSessionUsername) {
+ this.loggedInSessionUsername = loggedInSessionUsername;
+ }
+
+ void setLoginStatus(final TSStatus loginStatus) {
+ this.loginStatus = loginStatus;
+ }
+
+ int getCreateAuditEntityCallCount() {
+ return createAuditEntityCallCount;
+ }
+
+ String getCreateAuditEntityUsername() {
+ return createAuditEntityUsername;
+ }
+
+ IAuditEntity getUserEntity() {
+ return userEntity;
}
@Override
