Skip to content

Commit e94f0e8

Browse files
CaideyipiCaideyipi
authored
[dev/1.3] Pipe: Harden legacy pipe file transfer validation and access checks (#17791)
* Pipe: Harden legacy pipe file transfer validation and access checks (#17741) * Fix * fix * Fix legacy pipe receiver test FileUtils import
1 parent 0507f0b commit e94f0e8

5 files changed

Lines changed: 344 additions & 21 deletions

File tree

‎integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBLegacyPipeReceiverSecurityIT.java‎

Lines changed: 112 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,112 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.apache.iotdb.pipe.it.single;
21+
22+
import org.apache.iotdb.common.rpc.thrift.TSStatus;
23+
import org.apache.iotdb.commons.client.property.ThriftClientProperty;
24+
import org.apache.iotdb.commons.conf.IoTDBConstant;
25+
import org.apache.iotdb.commons.pipe.sink.client.IoTDBSyncClient;
26+
import org.apache.iotdb.it.env.EnvFactory;
27+
import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
28+
import org.apache.iotdb.it.framework.IoTDBTestRunner;
29+
import org.apache.iotdb.itbase.category.LocalStandaloneIT;
30+
import org.apache.iotdb.rpc.TSStatusCode;
31+
import org.apache.iotdb.service.rpc.thrift.TSCloseSessionReq;
32+
import org.apache.iotdb.service.rpc.thrift.TSOpenSessionReq;
33+
import org.apache.iotdb.service.rpc.thrift.TSOpenSessionResp;
34+
import org.apache.iotdb.service.rpc.thrift.TSProtocolVersion;
35+
import org.apache.iotdb.service.rpc.thrift.TSyncIdentityInfo;
36+
import org.apache.iotdb.service.rpc.thrift.TSyncTransportMetaInfo;
37+
38+
import org.junit.AfterClass;
39+
import org.junit.Assert;
40+
import org.junit.BeforeClass;
41+
import org.junit.Test;
42+
import org.junit.experimental.categories.Category;
43+
import org.junit.runner.RunWith;
44+
45+
import java.io.File;
46+
import java.nio.ByteBuffer;
47+
import java.nio.charset.StandardCharsets;
48+
import java.time.ZoneId;
49+
50+
@RunWith(IoTDBTestRunner.class)
51+
@Category({LocalStandaloneIT.class})
52+
public class IoTDBLegacyPipeReceiverSecurityIT {
53+
54+
@BeforeClass
55+
public static void setUp() {
56+
EnvFactory.getEnv().initClusterEnvironment();
57+
}
58+
59+
@AfterClass
60+
public static void tearDown() {
61+
EnvFactory.getEnv().cleanClusterEnvironment();
62+
}
63+
64+
@Test
65+
public void testRejectPathTraversalFileNameInLegacyTransportFile() throws Exception {
66+
final DataNodeWrapper dataNode = EnvFactory.getEnv().getDataNodeWrapper(0);
67+
68+
try (final IoTDBSyncClient client =
69+
new IoTDBSyncClient(
70+
new ThriftClientProperty.Builder().build(),
71+
dataNode.getIp(),
72+
dataNode.getPort(),
73+
false,
74+
null,
75+
null)) {
76+
final TSOpenSessionResp openSessionResp = client.openSession(createOpenSessionReq());
77+
Assert.assertEquals(
78+
TSStatusCode.SUCCESS_STATUS.getStatusCode(), openSessionResp.getStatus().getCode());
79+
80+
try {
81+
final TSStatus handshakeStatus =
82+
client.handshake(
83+
new TSyncIdentityInfo(
84+
"pathTraversalPipe", System.currentTimeMillis(), "UNKNOWN", ""));
85+
Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), handshakeStatus.getCode());
86+
87+
final String maliciousFileName =
88+
".." + File.separator + ".." + File.separator + "pwned.tsfile";
89+
final TSStatus status =
90+
client.sendFile(
91+
new TSyncTransportMetaInfo(maliciousFileName, 0),
92+
ByteBuffer.wrap("pwned".getBytes(StandardCharsets.UTF_8)));
93+
94+
Assert.assertEquals(TSStatusCode.SYNC_FILE_ERROR.getStatusCode(), status.getCode());
95+
Assert.assertTrue(status.getMessage().contains("Illegal fileName"));
96+
} finally {
97+
client.closeSession(new TSCloseSessionReq(openSessionResp.getSessionId()));
98+
}
99+
}
100+
}
101+
102+
private TSOpenSessionReq createOpenSessionReq() {
103+
final TSOpenSessionReq req = new TSOpenSessionReq();
104+
req.setClient_protocol(TSProtocolVersion.IOTDB_SERVICE_PROTOCOL_V3);
105+
req.setUsername("root");
106+
req.setPassword("root");
107+
req.setZoneId(ZoneId.systemDefault().toString());
108+
req.putToConfiguration("version", IoTDBConstant.ClientVersion.V_1_0.toString());
109+
req.putToConfiguration("sql_dialect", "tree");
110+
return req;
111+
}
112+
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/IoTDBLegacyPipeReceiverAgent.java‎

Lines changed: 38 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.apache.iotdb.commons.conf.CommonDescriptor;
2525
import org.apache.iotdb.commons.exception.IllegalPathException;
2626
import org.apache.iotdb.commons.path.PartialPath;
27+
import org.apache.iotdb.commons.pipe.receiver.PipeReceiverFilePathUtils;
2728
import org.apache.iotdb.commons.utils.FileUtils;
2829
import org.apache.iotdb.db.auth.AuthorityChecker;
2930
import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -51,6 +52,7 @@
5152
import java.io.IOException;
5253
import java.io.RandomAccessFile;
5354
import java.nio.ByteBuffer;
55+
import java.nio.file.Paths;
5456
import java.time.ZoneId;
5557
import java.util.Map;
5658
import java.util.Objects;
@@ -249,10 +251,12 @@ private SyncIdentityInfo getCurrentSyncIdentityInfo() {
249251
* @param tsFilePipeData pipeData
250252
* @param fileDir path of file data dir
251253
*/
252-
private void handleTsFilePipeData(TsFilePipeData tsFilePipeData, String fileDir) {
253-
String tsFileName = tsFilePipeData.getTsFileName();
254-
File dir = new File(fileDir);
255-
File[] targetFiles =
254+
private void handleTsFilePipeData(final TsFilePipeData tsFilePipeData, final String fileDir)
255+
throws IOException {
256+
final String tsFileName = tsFilePipeData.getTsFileName();
257+
final File tsFile = resolveFileInFileDataDir(fileDir, tsFileName);
258+
final File dir = tsFile.getParentFile();
259+
final File[] targetFiles =
256260
dir.listFiles((dir1, name) -> name.startsWith(tsFileName) && name.endsWith(PATCH_SUFFIX));
257261
if (targetFiles != null) {
258262
for (File targetFile : targetFiles) {
@@ -289,13 +293,21 @@ public TSStatus transportFile(TSyncTransportMetaInfo metaInfo, ByteBuffer buff)
289293
LOGGER.debug(
290294
"Invoke transportData method from client ip = {}", identityInfo.getRemoteAddress());
291295

292-
String fileDir = getFileDataDir(identityInfo);
293-
String fileName = metaInfo.fileName;
294-
long startIndex = metaInfo.startIndex;
295-
File file = new File(fileDir, fileName + PATCH_SUFFIX);
296+
final String fileDir = getFileDataDir(identityInfo);
297+
final String fileName = metaInfo.fileName;
298+
final long startIndex = metaInfo.startIndex;
299+
final File file;
300+
final File fileWithoutPatch;
301+
try {
302+
fileWithoutPatch = resolveFileInFileDataDir(fileDir, fileName);
303+
file = resolveFileInFileDataDir(fileDir, fileName + PATCH_SUFFIX);
304+
} catch (final IOException e) {
305+
LOGGER.warn(e.getMessage());
306+
return RpcUtils.getStatus(TSStatusCode.SYNC_FILE_ERROR, e.getMessage());
307+
}
296308

297309
// step2. check startIndex
298-
IndexCheckResult result = checkStartIndexValid(new File(fileDir, fileName), startIndex);
310+
final IndexCheckResult result = checkStartIndexValid(fileWithoutPatch, startIndex);
299311
if (!result.isResult()) {
300312
return RpcUtils.getStatus(TSStatusCode.SYNC_FILE_REDIRECTION_ERROR, result.getIndex());
301313
}
@@ -307,17 +319,31 @@ public TSStatus transportFile(TSyncTransportMetaInfo metaInfo, ByteBuffer buff)
307319
byte[] byteArray = new byte[length];
308320
buff.get(byteArray);
309321
randomAccessFile.write(byteArray);
310-
recordStartIndex(new File(fileDir, fileName), startIndex + length);
322+
recordStartIndex(fileWithoutPatch, startIndex + length);
311323
LOGGER.debug("Sync {} start at {} to {} is done.", fileName, startIndex, startIndex + length);
312-
} catch (IOException e) {
324+
} catch (final IOException e) {
313325
LOGGER.error(e.getMessage());
314326
return RpcUtils.getStatus(TSStatusCode.SYNC_FILE_ERROR, e.getMessage());
315327
}
316328

317329
return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS, "");
318330
}
319331

320-
private IndexCheckResult checkStartIndexValid(File file, long startIndex) {
332+
private static File resolveFileInFileDataDir(final String fileDir, final String fileName)
333+
throws IOException {
334+
if (StringUtils.isEmpty(fileName)) {
335+
throw new IOException("Illegal fileName: " + fileName);
336+
}
337+
338+
final String illegalError = FileUtils.getIllegalError4Directory(fileName);
339+
if (Objects.nonNull(illegalError)) {
340+
throw new IOException("Illegal fileName: " + fileName + ", " + illegalError);
341+
}
342+
343+
return PipeReceiverFilePathUtils.resolveFilePath(Paths.get(fileDir), fileName).toFile();
344+
}
345+
346+
private IndexCheckResult checkStartIndexValid(final File file, final long startIndex) {
321347
// get local index from memory map
322348
long localIndex = getCurrentFileStartIndex(file.getAbsolutePath());
323349
// get local index from file

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/legacy/IoTDBLegacyPipeSink.java‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.apache.iotdb.commons.client.property.ThriftClientProperty;
2525
import org.apache.iotdb.commons.conf.CommonConfig;
2626
import org.apache.iotdb.commons.conf.CommonDescriptor;
27+
import org.apache.iotdb.commons.conf.IoTDBConstant;
2728
import org.apache.iotdb.commons.consensus.DataRegionId;
2829
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeCriticalException;
2930
import org.apache.iotdb.commons.pipe.config.PipeConfig;
@@ -50,6 +51,9 @@
5051
import org.apache.iotdb.rpc.IoTDBConnectionException;
5152
import org.apache.iotdb.rpc.StatementExecutionException;
5253
import org.apache.iotdb.rpc.TSStatusCode;
54+
import org.apache.iotdb.service.rpc.thrift.TSOpenSessionReq;
55+
import org.apache.iotdb.service.rpc.thrift.TSOpenSessionResp;
56+
import org.apache.iotdb.service.rpc.thrift.TSProtocolVersion;
5357
import org.apache.iotdb.service.rpc.thrift.TSyncIdentityInfo;
5458
import org.apache.iotdb.service.rpc.thrift.TSyncTransportMetaInfo;
5559
import org.apache.iotdb.session.pool.SessionPool;
@@ -64,6 +68,7 @@
6468
import java.io.IOException;
6569
import java.io.RandomAccessFile;
6670
import java.nio.ByteBuffer;
71+
import java.time.ZoneId;
6772
import java.util.Arrays;
6873
import java.util.HashSet;
6974
import java.util.List;
@@ -224,6 +229,7 @@ public void handshake() throws Exception {
224229
useSSL,
225230
trustStore,
226231
trustStorePwd);
232+
openClientSession();
227233
final TSyncIdentityInfo identityInfo =
228234
new TSyncIdentityInfo(
229235
pipeName, System.currentTimeMillis(), syncConnectorVersion, databaseName);
@@ -254,6 +260,26 @@ public void handshake() throws Exception {
254260
.build();
255261
}
256262

263+
private void openClientSession() throws TException {
264+
final TSOpenSessionReq openSessionReq = new TSOpenSessionReq();
265+
openSessionReq.setClient_protocol(TSProtocolVersion.IOTDB_SERVICE_PROTOCOL_V3);
266+
openSessionReq.setUsername(user);
267+
openSessionReq.setPassword(password);
268+
openSessionReq.setZoneId(ZoneId.systemDefault().toString());
269+
openSessionReq.putToConfiguration("version", IoTDBConstant.ClientVersion.V_1_0.toString());
270+
openSessionReq.putToConfiguration("sql_dialect", "tree");
271+
272+
final TSOpenSessionResp openSessionResp = client.openSession(openSessionReq);
273+
if (openSessionResp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
274+
final String errorMsg =
275+
String.format(
276+
"Failed to login to receiver %s:%s for legacy pipe transfer because %s",
277+
ipAddress, port, openSessionResp.getStatus().getMessage());
278+
LOGGER.warn(errorMsg);
279+
throw new PipeRuntimeCriticalException(errorMsg);
280+
}
281+
}
282+
257283
@Override
258284
public void heartbeat() throws Exception {
259285
// do nothing

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java‎

Lines changed: 45 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
import org.apache.iotdb.common.rpc.thrift.TSStatus;
2626
import org.apache.iotdb.common.rpc.thrift.TShowConfigurationResp;
2727
import org.apache.iotdb.common.rpc.thrift.TShowConfigurationTemplateResp;
28+
import org.apache.iotdb.commons.auth.entity.PrivilegeType;
2829
import org.apache.iotdb.commons.client.exception.ClientManagerException;
2930
import org.apache.iotdb.commons.conf.CommonConfig;
3031
import org.apache.iotdb.commons.conf.CommonDescriptor;
@@ -2746,24 +2747,59 @@ public TSStatus createTimeseriesUsingSchemaTemplate(TCreateTimeseriesUsingSchema
27462747

27472748
@Override
27482749
public TSStatus handshake(final TSyncIdentityInfo info) throws TException {
2749-
return PipeDataNodeAgent.receiver()
2750-
.legacy()
2751-
.handshake(
2752-
info,
2753-
SESSION_MANAGER.getCurrSession().getClientAddress(),
2754-
partitionFetcher,
2755-
schemaFetcher);
2750+
try {
2751+
final TSStatus status = checkLegacyPipeReceiverPermission();
2752+
if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
2753+
return status;
2754+
}
2755+
return PipeDataNodeAgent.receiver()
2756+
.legacy()
2757+
.handshake(
2758+
info,
2759+
SESSION_MANAGER.getCurrSession().getClientAddress(),
2760+
partitionFetcher,
2761+
schemaFetcher);
2762+
} finally {
2763+
SESSION_MANAGER.updateIdleTime();
2764+
}
27562765
}
27572766

27582767
@Override
27592768
public TSStatus sendPipeData(final ByteBuffer buff) throws TException {
2760-
return PipeDataNodeAgent.receiver().legacy().transportPipeData(buff);
2769+
try {
2770+
final TSStatus status = checkLegacyPipeReceiverPermission();
2771+
if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
2772+
return status;
2773+
}
2774+
return PipeDataNodeAgent.receiver().legacy().transportPipeData(buff);
2775+
} finally {
2776+
SESSION_MANAGER.updateIdleTime();
2777+
}
27612778
}
27622779

27632780
@Override
27642781
public TSStatus sendFile(final TSyncTransportMetaInfo metaInfo, final ByteBuffer buff)
27652782
throws TException {
2766-
return PipeDataNodeAgent.receiver().legacy().transportFile(metaInfo, buff);
2783+
try {
2784+
final TSStatus status = checkLegacyPipeReceiverPermission();
2785+
if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
2786+
return status;
2787+
}
2788+
return PipeDataNodeAgent.receiver().legacy().transportFile(metaInfo, buff);
2789+
} finally {
2790+
SESSION_MANAGER.updateIdleTime();
2791+
}
2792+
}
2793+
2794+
private TSStatus checkLegacyPipeReceiverPermission() {
2795+
final IClientSession clientSession = SESSION_MANAGER.getCurrSessionAndUpdateIdleTime();
2796+
if (!SESSION_MANAGER.checkLogin(clientSession)) {
2797+
return getNotLoggedInStatus();
2798+
}
2799+
return AuthorityChecker.getTSStatus(
2800+
AuthorityChecker.checkSystemPermission(
2801+
clientSession.getUsername(), PrivilegeType.USE_PIPE.ordinal()),
2802+
PrivilegeType.USE_PIPE);
27672803
}
27682804

27692805
@Override

0 commit comments

Comments
 (0)