Skip to content

Commit 3fb3245

Browse files
wgzhaowgzhao
committed
[feature][plugin][ftpreader] Add public key authentication support for sftp
The item of password becomes optional
1 parent be6cef6 commit 3fb3245

File tree

6 files changed

+63
-11
lines changed

6 files changed

+63
-11
lines changed

Diff for: ‎docs/reader/ftpreader.md

+4-1
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,10 @@ Ftp Reader 提供了读取远程 FTP/SFTP 文件系统数据存储的能力。
2222
| timeout | 否 | int | 60000 | 连接 ftp 服务器连接超时时间，单位毫秒(ms) |
2323
| connectPattern | 否 | string | PASV | 连接模式，仅支持 `PORT`, `PASV` 模式。该参数仅在 ftp 协议时使用 |
2424
| username | 是 | string | 无 | ftp 服务器访问用户名 |
25-
| password | 是 | string | 无 | ftp 服务器访问密码 |
25+
| password | 否 | string | 无 | ftp 服务器访问密码 |
26+
| useKey | 否 | boolean | false | 是否使用私钥登录，仅针对 sftp 登录有效 |
27+
| keyPath | 否 | string | `~/.ssh/id_rsa` | 私钥地址 |
28+
| keyPass | 否 | string | 无 | 私钥密码，若没有设置私钥密码，则无需配置该项 |
2629
| path | 是 | list | 无 | 远程 FTP 文件系统的路径信息，注意这里可以支持填写多个路径，详细描述见下 |
2730
| column | 是 | `list<map>` | 无 | 读取字段列表，type 指定源数据的类型，详见下文 |
2831
| fieldDelimiter | 是 | string | `,` | 描述：读取的字段分隔符 |

Diff for: ‎plugin/reader/ftpreader/src/main/java/com/wgzhao/addax/plugin/reader/ftpreader/FtpHelper.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,7 @@ public abstract class FtpHelper
3636
* @param timeout 超时时间（单位为秒）
3737
* @param connectMode 连接模式
3838
*/
39-
public abstract void loginFtpServer(String host, String username, String password, int port, int timeout, String connectMode);
39+
public abstract void loginFtpServer(String host, String username, String password, int port, String keyPath, String keyPass, int timeout, String connectMode);
4040

4141
/**
4242
* 断开与ftp服务器的连接

Diff for: ‎plugin/reader/ftpreader/src/main/java/com/wgzhao/addax/plugin/reader/ftpreader/FtpKey.java

+6
Original file line numberDiff line numberDiff line change
@@ -33,4 +33,10 @@ private FtpKey() {}
3333
public static final String TIME_OUT = "timeout";
3434
public static final String CONNECT_PATTERN = "connectPattern";
3535
public static final String MAX_TRAVERSAL_LEVEL = "maxTraversalLevel";
36+
37+
public static final String USE_KEY = "useKey";
38+
// ssh private key
39+
public static final String KEY_PATH = "keyPath";
40+
// ssh private key passphrase
41+
public static final String KEY_PASS = "keyPass";
3642
}

Diff for: ‎plugin/reader/ftpreader/src/main/java/com/wgzhao/addax/plugin/reader/ftpreader/FtpReader.java

+34-6
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
import org.slf4j.Logger;
2929
import org.slf4j.LoggerFactory;
3030

31+
import java.io.File;
3132
import java.io.InputStream;
3233
import java.util.ArrayList;
3334
import java.util.HashSet;
@@ -57,6 +58,8 @@ public static class Job
5758
private String connectPattern;
5859
private int maxTraversalLevel;
5960

61+
private static final String DEFAULT_PRIVATE_KEY = "~/.ssh/id_rsa";
62+
6063
private FtpHelper ftpHelper = null;
6164

6265
@Override
@@ -67,6 +70,8 @@ public void init()
6770

6871
this.validateParameter();
6972
StorageReaderUtil.validateParameter(this.originConfig);
73+
String keyPath = this.originConfig.getString(FtpKey.KEY_PATH, null);
74+
String keyPass = this.originConfig.getString(FtpKey.KEY_PASS, null);
7075

7176
if ("sftp".equals(protocol)) {
7277
//sftp协议
@@ -78,20 +83,20 @@ else if ("ftp".equals(protocol)) {
7883
this.port = originConfig.getInt(FtpKey.PORT, FtpConstant.DEFAULT_FTP_PORT);
7984
this.ftpHelper = new StandardFtpHelper();
8085
}
81-
ftpHelper.loginFtpServer(host, username, password, port, timeout, connectPattern);
86+
ftpHelper.loginFtpServer(host, username, password, port, keyPath, keyPass, timeout, connectPattern);
8287
}
8388

8489
private void validateParameter()
8590
{
86-
this.protocol = this.originConfig.getNecessaryValue(FtpKey.PROTOCOL, FtpReaderErrorCode.REQUIRED_VALUE);
91+
this.protocol = this.originConfig.getNecessaryValue(FtpKey.PROTOCOL, FtpReaderErrorCode.REQUIRED_VALUE).toLowerCase();
8792
boolean protocolTag = "ftp".equals(this.protocol) || "sftp".equals(this.protocol);
8893
if (!protocolTag) {
8994
throw AddaxException.asAddaxException(FtpReaderErrorCode.ILLEGAL_VALUE,
9095
String.format("仅支持 ftp和sftp 传输协议 , 不支持您配置的传输协议: [%s]", protocol));
9196
}
9297
this.host = this.originConfig.getNecessaryValue(FtpKey.HOST, FtpReaderErrorCode.REQUIRED_VALUE);
9398
this.username = this.originConfig.getNecessaryValue(FtpKey.USERNAME, FtpReaderErrorCode.REQUIRED_VALUE);
94-
this.password = this.originConfig.getNecessaryValue(FtpKey.PASSWORD, FtpReaderErrorCode.REQUIRED_VALUE);
99+
this.password = this.originConfig.getString(FtpKey.PASSWORD, null);
95100
this.timeout = originConfig.getInt(FtpKey.TIME_OUT, FtpConstant.DEFAULT_TIMEOUT);
96101
this.maxTraversalLevel = originConfig.getInt(FtpKey.MAX_TRAVERSAL_LEVEL, FtpConstant.DEFAULT_MAX_TRAVERSAL_LEVEL);
97102

@@ -114,7 +119,7 @@ private void validateParameter()
114119
}
115120
else {
116121
path = this.originConfig.getList(Key.PATH, String.class);
117-
if (null == path || path.isEmpty() ) {
122+
if (null == path || path.isEmpty()) {
118123
throw AddaxException.asAddaxException(FtpReaderErrorCode.REQUIRED_VALUE, "您需要指定待读取的源目录或文件");
119124
}
120125
for (String eachPath : path) {
@@ -125,6 +130,28 @@ private void validateParameter()
125130
}
126131
}
127132
}
133+
if ("sftp".equals(protocol)) {
134+
// use ssh private key or not ?
135+
boolean useKey = this.originConfig.getBool(FtpKey.USE_KEY, false);
136+
if (useKey) {
137+
String privateKey = this.originConfig.getString(FtpKey.KEY_PATH, DEFAULT_PRIVATE_KEY);
138+
// check privateKey does exist or not
139+
if (privateKey.startsWith("~")) {
140+
// expand home directory
141+
privateKey = privateKey.replaceFirst("^~", System.getProperty("user.home"));
142+
// does it exist?
143+
boolean isFile = new File(privateKey).isFile();
144+
if (isFile) {
145+
this.originConfig.set(FtpKey.KEY_PATH, privateKey);
146+
}
147+
else {
148+
String msg = "You have configured to use the key, but neither the configured key file nor the default file(" +
149+
DEFAULT_PRIVATE_KEY + " exists";
150+
throw AddaxException.asAddaxException(FtpReaderErrorCode.ILLEGAL_VALUE, msg);
151+
}
152+
}
153+
}
154+
}
128155
}
129156

130157
@Override
@@ -224,7 +251,8 @@ public void init()
224251
this.username = readerSliceConfig.getString(FtpKey.USERNAME);
225252
String password = readerSliceConfig.getString(FtpKey.PASSWORD);
226253
int timeout = readerSliceConfig.getInt(FtpKey.TIME_OUT, FtpConstant.DEFAULT_TIMEOUT);
227-
254+
String keyPath = readerSliceConfig.getString(FtpKey.KEY_PATH, null);
255+
String keyPass = readerSliceConfig.getString(FtpKey.KEY_PASS, null);
228256
this.sourceFiles = this.readerSliceConfig.getList(FtpKey.SOURCE_FILES, String.class);
229257

230258
if ("sftp".equals(protocol)) {
@@ -238,7 +266,7 @@ else if ("ftp".equals(protocol)) {
238266
this.connectPattern = readerSliceConfig.getString(FtpKey.CONNECT_PATTERN, FtpConstant.DEFAULT_FTP_CONNECT_PATTERN);// 默认为被动模式
239267
this.ftpHelper = new StandardFtpHelper();
240268
}
241-
ftpHelper.loginFtpServer(host, username, password, port, timeout, connectPattern);
269+
ftpHelper.loginFtpServer(host, username, password, port, keyPath, keyPass, timeout, connectPattern);
242270
}
243271

244272
@Override

Diff for: ‎plugin/reader/ftpreader/src/main/java/com/wgzhao/addax/plugin/reader/ftpreader/SftpHelper.java

+17-2
Original file line numberDiff line numberDiff line change
@@ -47,10 +47,23 @@ public class SftpHelper
4747
HashSet<String> sourceFiles = new HashSet<>();
4848

4949
@Override
50-
public void loginFtpServer(String host, String username, String password, int port, int timeout,
50+
public void loginFtpServer(String host, String username, String password, int port, String keyPath, String keyPass, int timeout,
5151
String connectMode)
5252
{
5353
JSch jsch = new JSch(); // 创建JSch对象
54+
if (keyPath != null) {
55+
try {
56+
if (keyPass != null) {
57+
jsch.addIdentity(keyPath, keyPass);
58+
}
59+
else {
60+
jsch.addIdentity(keyPath);
61+
}
62+
}
63+
catch (JSchException e) {
64+
throw AddaxException.asAddaxException(FtpReaderErrorCode.ILLEGAL_VALUE, "Failed to use private key", e);
65+
}
66+
}
5467
try {
5568
session = jsch.getSession(username, host, port);
5669
// 根据用户名，主机ip，端口获取一个Session对象
@@ -60,7 +73,9 @@ public void loginFtpServer(String host, String username, String password, int po
6073
"session is null,无法通过sftp与服务器建立链接，请检查主机名和用户名是否正确.");
6174
}
6275

63-
session.setPassword(password); // 设置密码
76+
if (password != null) {
77+
session.setPassword(password); // 设置密码
78+
}
6479
Properties config = new Properties();
6580
config.put("StrictHostKeyChecking", "no");
6681
session.setConfig(config); // 为Session对象设置properties

Diff for: ‎plugin/reader/ftpreader/src/main/java/com/wgzhao/addax/plugin/reader/ftpreader/StandardFtpHelper.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,7 @@ public class StandardFtpHelper
4444
HashSet<String> sourceFiles = new HashSet<>();
4545

4646
@Override
47-
public void loginFtpServer(String host, String username, String password, int port, int timeout,
47+
public void loginFtpServer(String host, String username, String password, int port, String keyPath, String keyPass, int timeout,
4848
String connectMode)
4949
{
5050
ftpClient = new FTPClient();

0 commit comments

Comments
 (0)