Merge pull request #11 from sakama/support-proxy-settings

Support proxy settings

Author: civitaspo

README.md

@@ -14,7 +14,7 @@ Stores files on a SFTP Server
 ## Configuration
 
 - **host**: (string, required)
-- **port**: (string, default: `22`)
+- **port**: (int, default: `22`)
 - **user**: (string, required)
 - **password**: (string, default: `null`)
 - **secret_key_file**: (string, default: `null`) [see below](#secret-keyfile-configuration)
@@ -25,6 +25,19 @@ Stores files on a SFTP Server
 - **file_ext**: Extension of output files (string, required)
 - **sequence_format**: Format for sequence part of output files (string, default: `".%03d.%02d"`)
 
+### Proxy configuration
+
+- **proxy**:
+    - **type**: (string(http | socks | stream), required, default: `null`)
+        - **http**: use HTTP Proxy
+        - **socks**: use SOCKS Proxy
+        - **stream**: Connects to the SFTP server through a remote host reached by SSH
+    - **host**: (string, required)
+    - **port**: (int, default: `22`)
+    - **user**: (string, optional)
+    - **password**: (string, optional, default: `null`)
+    - **command**: (string, optional)
+    
 ## Example
 
 ```yaml
@@ -42,10 +55,31 @@ out:
   sequence_format: ".%01d%01d"
 ```
 
+With proxy
+```yaml
+out:
+  type: sftp
+  host: 127.0.0.1
+  port: 22
+  user: embulk
+  secret_key_file: /Users/embulk/.ssh/id_rsa
+  secret_key_passphrase: secret_pass
+  user_directory_is_root: false
+  timeout: 600
+  path_prefix: /data/sftp
+  proxy:
+    type: http
+    host: proxy_host
+    port: 8080
+    user: proxy_user
+    password: proxy_secret_pass
+    command:
+```
+
 ### Secret Keyfile configuration
 
 Please set path of secret_key_file as follows.
-```
+```yaml
 out:
   type: sftp
   ...
@@ -54,7 +88,7 @@ out:
 ```
 
 You can also embed contents of secret_key_file at config.yml.
-```
+```yaml
 out:
   type: sftp
   ...

build.gradle

@@ -29,6 +29,8 @@ dependencies {
     testCompile "org.embulk:embulk-core:0.8.6:tests"
     testCompile "org.embulk:embulk-standards:0.8.6"
     testCompile "org.apache.sshd:apache-sshd:1.1.0+"
+    testCompile "org.littleshoot:littleproxy:1.1.0-beta1"
+    testCompile "io.netty:netty-all:4.0.34.Final"
 }
 
 jacocoTestReport {

src/main/java/org/embulk/output/sftp/ProxyTask.java

@@ -0,0 +1,85 @@
+package org.embulk.output.sftp;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonValue;
+import com.google.common.base.Optional;
+import org.apache.commons.vfs2.FileSystemOptions;
+import org.apache.commons.vfs2.provider.sftp.SftpFileSystemConfigBuilder;
+import org.embulk.config.Config;
+import org.embulk.config.ConfigDefault;
+import org.embulk.config.ConfigException;
+import org.embulk.config.Task;
+
+import java.util.Locale;
+
+interface ProxyTask
+        extends Task
+{
+    @Config("type")
+    public ProxyType getType();
+
+    @Config("host")
+    public Optional<String> getHost();
+
+    @Config("user")
+    @ConfigDefault("null")
+    public Optional<String> getUser();
+
+    @Config("password")
+    @ConfigDefault("null")
+    public Optional<String> getPassword();
+
+    @Config("port")
+    @ConfigDefault("22")
+    public int getPort();
+
+    @Config("command")
+    @ConfigDefault("null")
+    public Optional<String> getCommand();
+
+    public enum ProxyType
+    {
+        HTTP,
+        SOCKS,
+        STREAM;
+
+        @JsonValue
+        @Override
+        public String toString()
+        {
+            return name().toLowerCase(Locale.ENGLISH);
+        }
+
+        @JsonCreator
+        public static ProxyType fromString(String value)
+        {
+            switch (value) {
+                case "http":
+                    return HTTP;
+                case "socks":
+                    return SOCKS;
+                case "stream":
+                    return STREAM;
+                default:
+                    throw new ConfigException(String.format("Unknown proxy type '%s'. Supported proxy types are http, socks, stream", value));
+            }
+        }
+
+        public static SftpFileSystemConfigBuilder setProxyType(SftpFileSystemConfigBuilder builder, FileSystemOptions fsOptions, ProxyTask.ProxyType type)
+        {
+            SftpFileSystemConfigBuilder.ProxyType setType = null;
+            switch (type) {
+                case HTTP:
+                    setType = SftpFileSystemConfigBuilder.PROXY_HTTP;
+                    break;
+                case SOCKS:
+                    setType = SftpFileSystemConfigBuilder.PROXY_SOCKS5;
+                    break;
+                case STREAM:
+                    setType = SftpFileSystemConfigBuilder.PROXY_STREAM;
+            }
+            builder.setProxyType(fsOptions, setType);
+            return builder;
+        }
+    }
+}

src/main/java/org/embulk/output/sftp/SftpFileOutput.java

@@ -79,17 +79,41 @@ private FileSystemOptions initializeFsOptions(PluginTask task)
         FileSystemOptions fsOptions = new FileSystemOptions();
 
         try {
-            SftpFileSystemConfigBuilder.getInstance().setUserDirIsRoot(fsOptions, task.getUserDirIsRoot());
-            SftpFileSystemConfigBuilder.getInstance().setTimeout(fsOptions, task.getSftpConnectionTimeout());
-            SftpFileSystemConfigBuilder.getInstance().setStrictHostKeyChecking(fsOptions, "no");
+            SftpFileSystemConfigBuilder builder = SftpFileSystemConfigBuilder.getInstance();
+            builder.setUserDirIsRoot(fsOptions, task.getUserDirIsRoot());
+            builder.setTimeout(fsOptions, task.getSftpConnectionTimeout());
+            builder.setStrictHostKeyChecking(fsOptions, "no");
             if (task.getSecretKeyFilePath().isPresent()) {
                 IdentityInfo identityInfo = new IdentityInfo(
                     new File((task.getSecretKeyFilePath().transform(localFileToPathString()).get())),
                     task.getSecretKeyPassphrase().getBytes()
                 );
-                SftpFileSystemConfigBuilder.getInstance().setIdentityInfo(fsOptions, identityInfo);
+                builder.setIdentityInfo(fsOptions, identityInfo);
                 logger.info("set identity: {}", task.getSecretKeyFilePath().get());
             }
+
+            if (task.getProxy().isPresent()) {
+                ProxyTask proxy = task.getProxy().get();
+
+                ProxyTask.ProxyType.setProxyType(builder, fsOptions, proxy.getType());
+
+                if (proxy.getHost().isPresent()) {
+                    builder.setProxyHost(fsOptions, proxy.getHost().get());
+                    builder.setProxyPort(fsOptions, proxy.getPort());
+                }
+
+                if (proxy.getUser().isPresent()) {
+                    builder.setProxyUser(fsOptions, proxy.getUser().get());
+                }
+
+                if (proxy.getPassword().isPresent()) {
+                    builder.setProxyPassword(fsOptions, proxy.getPassword().get());
+                }
+
+                if (proxy.getCommand().isPresent()) {
+                    builder.setProxyCommand(fsOptions, proxy.getCommand().get());
+                }
+            }
         }
         catch (FileSystemException e) {
             logger.error(e.getMessage());

src/main/java/org/embulk/output/sftp/SftpFileOutputPlugin.java

@@ -68,6 +68,10 @@ public interface PluginTask
         @Config("sequence_format")
         @ConfigDefault("\"%03d.%02d.\"")
         public String getSequenceFormat();
+
+        @Config("proxy")
+        @ConfigDefault("null")
+        public Optional<ProxyTask> getProxy();
     }
 
     @Override