NIFI-4199: Add ProxyConfigurationService to SFTP processors

ijokarumawak · ijokarumawak · commit d56d364ade9b · 2018-05-17T17:55:10.000+09:00
- Fixed check style issue
- Use the same proxy related PropertyDescriptors from FTPTransfer and
SFTPTransfer
- Dropped FlowFile EL evaluation support to make it align with other
processors spec, Now it supports VARIABLE_REGISTRY
- Added ProxyConfigurationService to SFTP processors
- Added SOCKS proxy support to SFTP processors
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/FetchSFTP.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/FetchSFTP.java
@@ -29,8 +29,10 @@
 import org.apache.nifi.annotation.documentation.Tags;
 import org.apache.nifi.components.PropertyDescriptor;
 import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processors.standard.util.FTPTransfer;
 import org.apache.nifi.processors.standard.util.FileTransfer;
 import org.apache.nifi.processors.standard.util.SFTPTransfer;
+import org.apache.nifi.proxy.ProxyConfigurationService;
 
 // Note that we do not use @SupportsBatching annotation. This processor cannot support batching because it must ensure that session commits happen before remote files are deleted.
 @InputRequirement(Requirement.INPUT_REQUIRED)
@@ -76,10 +78,12 @@ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
         properties.add(SFTPTransfer.HOST_KEY_FILE);
         properties.add(SFTPTransfer.STRICT_HOST_KEY_CHECKING);
         properties.add(SFTPTransfer.USE_COMPRESSION);
-        properties.add(SFTPTransfer.PROXY_HOST);
-        properties.add(SFTPTransfer.PROXY_PORT);
-        properties.add(SFTPTransfer.PROXY_USERNAME);
-        properties.add(SFTPTransfer.PROXY_PASSWORD);
+        properties.add(ProxyConfigurationService.PROXY_CONFIGURATION_SERVICE);
+        properties.add(FTPTransfer.PROXY_TYPE);
+        properties.add(FTPTransfer.PROXY_HOST);
+        properties.add(FTPTransfer.PROXY_PORT);
+        properties.add(FTPTransfer.HTTP_PROXY_USERNAME);
+        properties.add(FTPTransfer.HTTP_PROXY_PASSWORD);
         return properties;
     }
 
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GetSFTP.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GetSFTP.java
@@ -33,8 +33,10 @@
 import org.apache.nifi.components.ValidationResult;
 import org.apache.nifi.processor.ProcessContext;
 import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processors.standard.util.FTPTransfer;
 import org.apache.nifi.processors.standard.util.FileTransfer;
 import org.apache.nifi.processors.standard.util.SFTPTransfer;
+import org.apache.nifi.proxy.ProxyConfigurationService;
 
 @InputRequirement(Requirement.INPUT_FORBIDDEN)
 @Tags({"sftp", "get", "retrieve", "files", "fetch", "remote", "ingest", "source", "input"})
@@ -81,10 +83,12 @@ protected void init(final ProcessorInitializationContext context) {
         properties.add(SFTPTransfer.USE_KEEPALIVE_ON_TIMEOUT);
         properties.add(SFTPTransfer.USE_COMPRESSION);
         properties.add(SFTPTransfer.USE_NATURAL_ORDERING);
-        properties.add(SFTPTransfer.PROXY_HOST);
-        properties.add(SFTPTransfer.PROXY_PORT);
-        properties.add(SFTPTransfer.PROXY_USERNAME);
-        properties.add(SFTPTransfer.PROXY_PASSWORD);
+        properties.add(ProxyConfigurationService.PROXY_CONFIGURATION_SERVICE);
+        properties.add(FTPTransfer.PROXY_TYPE);
+        properties.add(FTPTransfer.PROXY_HOST);
+        properties.add(FTPTransfer.PROXY_PORT);
+        properties.add(FTPTransfer.HTTP_PROXY_USERNAME);
+        properties.add(FTPTransfer.HTTP_PROXY_PASSWORD);
         this.properties = Collections.unmodifiableList(properties);
     }
 
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListSFTP.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListSFTP.java
@@ -32,8 +32,10 @@
 import org.apache.nifi.components.PropertyDescriptor;
 import org.apache.nifi.components.state.Scope;
 import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processors.standard.util.FTPTransfer;
 import org.apache.nifi.processors.standard.util.FileTransfer;
 import org.apache.nifi.processors.standard.util.SFTPTransfer;
+import org.apache.nifi.proxy.ProxyConfigurationService;
 
 @TriggerSerially
 @InputRequirement(Requirement.INPUT_FORBIDDEN)
@@ -83,10 +85,12 @@ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
         properties.add(SFTPTransfer.DATA_TIMEOUT);
         properties.add(SFTPTransfer.USE_KEEPALIVE_ON_TIMEOUT);
         properties.add(TARGET_SYSTEM_TIMESTAMP_PRECISION);
-        properties.add(SFTPTransfer.PROXY_HOST);
-        properties.add(SFTPTransfer.PROXY_PORT);
-        properties.add(SFTPTransfer.PROXY_USERNAME);
-        properties.add(SFTPTransfer.PROXY_PASSWORD);
+        properties.add(ProxyConfigurationService.PROXY_CONFIGURATION_SERVICE);
+        properties.add(FTPTransfer.PROXY_TYPE);
+        properties.add(FTPTransfer.PROXY_HOST);
+        properties.add(FTPTransfer.PROXY_PORT);
+        properties.add(FTPTransfer.HTTP_PROXY_USERNAME);
+        properties.add(FTPTransfer.HTTP_PROXY_PASSWORD);
         return properties;
     }
 
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSFTP.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSFTP.java
@@ -30,7 +30,9 @@
 import org.apache.nifi.components.PropertyDescriptor;
 import org.apache.nifi.processor.ProcessContext;
 import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processors.standard.util.FTPTransfer;
 import org.apache.nifi.processors.standard.util.SFTPTransfer;
+import org.apache.nifi.proxy.ProxyConfigurationService;
 
 @SupportsBatching
 @InputRequirement(Requirement.INPUT_REQUIRED)
@@ -70,10 +72,12 @@ protected void init(final ProcessorInitializationContext context) {
         properties.add(SFTPTransfer.STRICT_HOST_KEY_CHECKING);
         properties.add(SFTPTransfer.USE_KEEPALIVE_ON_TIMEOUT);
         properties.add(SFTPTransfer.USE_COMPRESSION);
-        properties.add(SFTPTransfer.PROXY_HOST);
-        properties.add(SFTPTransfer.PROXY_PORT);
-        properties.add(SFTPTransfer.PROXY_USERNAME);
-        properties.add(SFTPTransfer.PROXY_PASSWORD);
+        properties.add(ProxyConfigurationService.PROXY_CONFIGURATION_SERVICE);
+        properties.add(FTPTransfer.PROXY_TYPE);
+        properties.add(FTPTransfer.PROXY_HOST);
+        properties.add(FTPTransfer.PROXY_PORT);
+        properties.add(FTPTransfer.HTTP_PROXY_USERNAME);
+        properties.add(FTPTransfer.HTTP_PROXY_PASSWORD);
         this.properties = Collections.unmodifiableList(properties);
     }
 
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FTPTransfer.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FTPTransfer.java
@@ -32,13 +32,15 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
 import java.util.regex.Pattern;
 
 import org.apache.commons.net.ftp.FTPClient;
 import org.apache.commons.net.ftp.FTPFile;
 import org.apache.commons.net.ftp.FTPHTTPClient;
 import org.apache.commons.net.ftp.FTPReply;
 import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.context.PropertyContext;
 import org.apache.nifi.expression.ExpressionLanguageScope;
 import org.apache.nifi.flowfile.FlowFile;
 import org.apache.nifi.logging.ComponentLog;
@@ -89,22 +91,26 @@ public class FTPTransfer implements FileTransfer {
         .name("Proxy Host")
         .description("The fully qualified hostname or IP address of the proxy server")
         .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+        .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
         .build();
     public static final PropertyDescriptor PROXY_PORT = new PropertyDescriptor.Builder()
         .name("Proxy Port")
         .description("The port of the proxy server")
         .addValidator(StandardValidators.PORT_VALIDATOR)
+        .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
         .build();
     public static final PropertyDescriptor HTTP_PROXY_USERNAME = new PropertyDescriptor.Builder()
         .name("Http Proxy Username")
         .description("Http Proxy Username")
         .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+        .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
         .required(false)
         .build();
     public static final PropertyDescriptor HTTP_PROXY_PASSWORD = new PropertyDescriptor.Builder()
         .name("Http Proxy Password")
         .description("Http Proxy Password")
         .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+        .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
         .required(false)
         .sensitive(true)
         .build();
@@ -523,15 +529,7 @@ private FTPClient getClient(final FlowFile flowFile) throws IOException {
             }
         }
 
-        final ProxyConfiguration proxyConfig = ProxyConfiguration.getConfiguration(ctx, () -> {
-            final ProxyConfiguration componentProxyConfig = new ProxyConfiguration();
-            componentProxyConfig.setProxyType(Proxy.Type.valueOf(ctx.getProperty(PROXY_TYPE).getValue()));
-            componentProxyConfig.setProxyServerHost(ctx.getProperty(PROXY_HOST).getValue());
-            componentProxyConfig.setProxyServerPort(ctx.getProperty(PROXY_PORT).asInteger());
-            componentProxyConfig.setProxyUserName(ctx.getProperty(HTTP_PROXY_USERNAME).getValue());
-            componentProxyConfig.setProxyUserPassword(ctx.getProperty(HTTP_PROXY_PASSWORD).getValue());
-            return componentProxyConfig;
-        });
+        final ProxyConfiguration proxyConfig = ProxyConfiguration.getConfiguration(ctx, createComponentProxyConfigSupplier(ctx));
 
         final Proxy.Type proxyType = proxyConfig.getProxyType();
         final String proxyHost = proxyConfig.getProxyServerHost();
@@ -639,4 +637,17 @@ protected int numberPermissions(String perms) {
         }
         return number;
     }
+
+    public static Supplier<ProxyConfiguration> createComponentProxyConfigSupplier(final PropertyContext ctx) {
+        return () -> {
+            final ProxyConfiguration componentProxyConfig = new ProxyConfiguration();
+            componentProxyConfig.setProxyType(Proxy.Type.valueOf(ctx.getProperty(PROXY_TYPE).getValue()));
+            componentProxyConfig.setProxyServerHost(ctx.getProperty(PROXY_HOST).evaluateAttributeExpressions().getValue());
+            componentProxyConfig.setProxyServerPort(ctx.getProperty(PROXY_PORT).evaluateAttributeExpressions().asInteger());
+            componentProxyConfig.setProxyUserName(ctx.getProperty(HTTP_PROXY_USERNAME).evaluateAttributeExpressions().getValue());
+            componentProxyConfig.setProxyUserPassword(ctx.getProperty(HTTP_PROXY_PASSWORD).evaluateAttributeExpressions().getValue());
+            return componentProxyConfig;
+        };
+    }
+
 }
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FTPUtils.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FTPUtils.java
@@ -297,4 +297,5 @@ public void setTransferMode(final String val) {
         }
 
     }
+
 }
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/SFTPTransfer.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/SFTPTransfer.java
@@ -33,13 +33,15 @@
 import java.util.concurrent.TimeUnit;
 import java.util.regex.Pattern;
 
+import com.jcraft.jsch.ProxySOCKS5;
 import org.apache.nifi.components.PropertyDescriptor;
 import org.apache.nifi.components.PropertyValue;
 import org.apache.nifi.expression.ExpressionLanguageScope;
 import org.apache.nifi.flowfile.FlowFile;
 import org.apache.nifi.logging.ComponentLog;
 import org.apache.nifi.processor.ProcessContext;
 import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.proxy.ProxyConfiguration;
 import org.slf4j.LoggerFactory;
 
 import com.jcraft.jsch.ChannelSftp;
@@ -51,6 +53,8 @@
 import com.jcraft.jsch.Session;
 import com.jcraft.jsch.SftpException;
 
+import static org.apache.nifi.processors.standard.util.FTPTransfer.createComponentProxyConfigSupplier;
+
 public class SFTPTransfer implements FileTransfer {
 
     public static final PropertyDescriptor PRIVATE_KEY_PATH = new PropertyDescriptor.Builder()
@@ -96,38 +100,6 @@ public class SFTPTransfer implements FileTransfer {
         .defaultValue("true")
         .required(true)
         .build();
-    public static final PropertyDescriptor PROXY_HOST = new PropertyDescriptor.Builder()
-            .name("PROXY_HOST")
-            .displayName("Proxy Host")
-            .description("The fully qualified hostname or IP address of the proxy server")
-            .expressionLanguageSupported(true)
-            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
-            .build();
-    public static final PropertyDescriptor PROXY_PORT = new PropertyDescriptor.Builder()
-            .name("PROXY_PORT")
-            .displayName("Proxy Port")
-            .description("The port of the proxy server")
-            .expressionLanguageSupported(true)
-            .addValidator(StandardValidators.PORT_VALIDATOR)
-            .build();
-    public static final PropertyDescriptor PROXY_USERNAME = new PropertyDescriptor.Builder()
-            .name("PROXY_USERNAME")
-            .displayName("Proxy Username")
-            .description("Proxy Username")
-            .expressionLanguageSupported(true)
-            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
-            .required(false)
-            .build();
-    public static final PropertyDescriptor PROXY_PASSWORD = new PropertyDescriptor.Builder()
-            .name("PROXY_PASSWORD")
-            .displayName("Proxy Password")
-            .description("Proxy Password")
-            .expressionLanguageSupported(true)
-            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
-            .required(false)
-            .sensitive(true)
-            .build();
-
 
 
     /**
@@ -452,16 +424,24 @@ protected ChannelSftp getChannel(final FlowFile flowFile) throws IOException {
                 ctx.getProperty(HOSTNAME).evaluateAttributeExpressions(flowFile).getValue(),
                 ctx.getProperty(PORT).evaluateAttributeExpressions(flowFile).asInteger().intValue());
 
-            if (ctx.getProperty(PROXY_HOST).evaluateAttributeExpressions(flowFile).isSet()) {
-                final ProxyHTTP proxy = new ProxyHTTP(
-                        ctx.getProperty(PROXY_HOST).evaluateAttributeExpressions(flowFile).getValue(),
-                        ctx.getProperty(PROXY_PORT).evaluateAttributeExpressions(flowFile).asInteger()
-                );
-                // Check if Username is set and populate the proxy accordingly
-                if (ctx.getProperty(PROXY_USERNAME).evaluateAttributeExpressions(flowFile).isSet()) {
-                    proxy.setUserPasswd(ctx.getProperty(PROXY_USERNAME).evaluateAttributeExpressions(flowFile).getValue(), ctx.getProperty(PROXY_PASSWORD).evaluateAttributeExpressions(flowFile).getValue());
-                }
-                session.setProxy(proxy);
+            final ProxyConfiguration proxyConfig = ProxyConfiguration.getConfiguration(ctx, createComponentProxyConfigSupplier(ctx));
+            switch (proxyConfig.getProxyType()) {
+                case HTTP:
+                    final ProxyHTTP proxyHTTP = new ProxyHTTP(proxyConfig.getProxyServerHost(), proxyConfig.getProxyServerPort());
+                    // Check if Username is set and populate the proxy accordingly
+                    if (proxyConfig.hasCredential()) {
+                        proxyHTTP.setUserPasswd(proxyConfig.getProxyUserName(), proxyConfig.getProxyUserPassword());
+                    }
+                    session.setProxy(proxyHTTP);
+                    break;
+                case SOCKS:
+                    final ProxySOCKS5 proxySOCKS5 = new ProxySOCKS5(proxyConfig.getProxyServerHost(), proxyConfig.getProxyServerPort());
+                    if (proxyConfig.hasCredential()) {
+                        proxySOCKS5.setUserPasswd(proxyConfig.getProxyUserName(), proxyConfig.getProxyUserPassword());
+                    }
+                    session.setProxy(proxySOCKS5);
+                    break;
+
             }
 
             final String hostKeyVal = ctx.getProperty(HOST_KEY_FILE).getValue();

Original file line number	Diff line number	Diff line change
`@@ -297,4 +297,5 @@ public void setTransferMode(final String val) {`
`297`	`297`	`}`
`298`	`298`
`299`	`299`	`}`
	`300`	`+`
`300`	`301`	`}`