[WIP] Proof of concept for a generic password provider

flume-ng-core/src/main/java/org/apache/flume/conf/ExternalProcessPasswordProvider.java

@@ -0,0 +1,89 @@
+/*
+ * 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.flume.conf;
+
+import com.google.common.base.Optional;
+import com.google.common.io.ByteStreams;
+import org.apache.flume.Context;
+
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.nio.charset.UnsupportedCharsetException;
+
+public class ExternalProcessPasswordProvider implements PasswordProvider {
+
+  private static final String COMMAND_KEY = "command";
+  private static final String CHARSET_KEY = "charset";
+
+  private Optional<String> password = Optional.absent();
+
+  @Override
+  public String getPassword(Context context, String key) {
+    if (!password.isPresent()) {
+      password = Optional.of(loadPassword(context, key));
+    }
+    return password.get();
+  }
+
+  private String loadPassword(Context context, String key) {
+    String charsetKey = key + "." + CHARSET_KEY;
+    String charsetName = context.getString(charsetKey, "iso-8859-1");
+
+    Charset charset;
+    try {
+      charset = Charset.forName(charsetName);
+    } catch (UnsupportedCharsetException ex) {
+      throw new RuntimeException("Unsupported charset: " + charsetName, ex);
+    }
+
+    String commandKey = key + "." + COMMAND_KEY;
+    String command = context.getString(commandKey);
+    if (command == null) {
+      throw new IllegalArgumentException(commandKey + " must be set for " +
+          "ExternalProcessPasswordProvider");
+    }
+
+    String commandOutput;
+    try {
+      commandOutput = execCommand(command, charset);
+    } catch (InterruptedException | IOException ex) {
+      throw new RuntimeException(ex);
+    }
+
+    return commandOutput;
+  }
+
+  private String execCommand(String command, Charset charset)
+      throws IOException, InterruptedException {
+    Process p = Runtime.getRuntime().exec(command);
+    p.waitFor();
+    if (p.exitValue() != 0) {
+      String stderr;
+      try {
+        stderr = new String(ByteStreams.toByteArray(p.getErrorStream()), charset);
+      } catch (Throwable t) {
+        stderr = null;
+      }
+      throw new RuntimeException(
+          String.format("Process (%s) exited with non-zero (%s) status code. Sterr: %s",
+              command, p.exitValue(), stderr));
+    }
+    return new String(ByteStreams.toByteArray(p.getInputStream()), charset);
+  }
+}

flume-ng-core/src/main/java/org/apache/flume/conf/PasswordConfigurator.java

@@ -0,0 +1,54 @@
+/*
+ * 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.flume.conf;
+
+import org.apache.flume.Context;
+
+import java.lang.reflect.InvocationTargetException;
+
+public final class PasswordConfigurator {
+
+  private static final String PASSWORD_PROVIDER_CLASS_KEY = "passwordProviderClass";
+
+  private PasswordConfigurator() {
+  }
+
+  public static String getPassword(Context context, String passwordKey) {
+    PasswordProvider passwordProvider;
+    try {
+      passwordProvider = getPasswordProvider(context, passwordKey);
+    } catch (ReflectiveOperationException ex) {
+      throw new RuntimeException(ex);
+    }
+
+    return passwordProvider.getPassword(context, passwordKey);
+  }
+
+  private static PasswordProvider getPasswordProvider(Context context, String passwordKey) throws
+      ClassNotFoundException, IllegalAccessException, InstantiationException, NoSuchMethodException,
+      InvocationTargetException {
+
+    String fullKey = passwordKey + "." + PASSWORD_PROVIDER_CLASS_KEY;
+    String className = context.getString(fullKey, PlainTextPasswordProvider.class.getName());
+
+    Class<? extends PasswordProvider> passwordProviderClass =
+        (Class<? extends PasswordProvider>) Class.forName(className);
+    return passwordProviderClass.newInstance();
+  }
+}

flume-ng-core/src/main/java/org/apache/flume/conf/PasswordProvider.java

@@ -0,0 +1,26 @@
+/*
+ * 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.flume.conf;
+
+import org.apache.flume.Context;
+
+public interface PasswordProvider {
+
+  String getPassword(Context context, String key);
+}

flume-ng-core/src/main/java/org/apache/flume/conf/PlainTextPasswordProvider.java

@@ -0,0 +1,32 @@
+/*
+ * 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.flume.conf;
+
+import org.apache.flume.Context;
+
+public class PlainTextPasswordProvider implements PasswordProvider {
+
+  public PlainTextPasswordProvider() {
+  }
+
+  @Override
+  public String getPassword(Context context, String key) {
+    return context.getString(key);
+  }
+}

flume-ng-core/src/main/java/org/apache/flume/source/AvroSource.java

@@ -37,6 +37,7 @@
 import org.apache.flume.conf.Configurable;
 import org.apache.flume.conf.Configurables;
 import org.apache.flume.conf.LogPrivacyUtil;
+import org.apache.flume.conf.PasswordConfigurator;
 import org.apache.flume.event.EventBuilder;
 import org.apache.flume.instrumentation.SourceCounter;
 import org.apache.flume.source.avro.AvroFlumeEvent;
@@ -181,7 +182,7 @@ public void configure(Context context) {
 
     enableSsl = context.getBoolean(SSL_KEY, false);
     keystore = context.getString(KEYSTORE_KEY);
-    keystorePassword = context.getString(KEYSTORE_PASSWORD_KEY);
+    keystorePassword = PasswordConfigurator.getPassword(context, KEYSTORE_PASSWORD_KEY);
     keystoreType = context.getString(KEYSTORE_TYPE_KEY, "JKS");
     String excludeProtocolsStr = context.getString(EXCLUDE_PROTOCOLS);
     if (excludeProtocolsStr == null) {

flume-ng-core/src/test/java/org/apache/flume/conf/TestExternalProcessPasswordProvider.java

@@ -0,0 +1,52 @@
+/*
+ * 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.flume.conf;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.flume.Context;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestExternalProcessPasswordProvider {
+
+  @Test
+  public void testValidCommand() {
+    Context context = new Context(ImmutableMap.of(
+        "password.command", "echo -n applepie"));
+    ExternalProcessPasswordProvider passwordProvider = new ExternalProcessPasswordProvider();
+    Assert.assertEquals("applepie", passwordProvider.getPassword(context, "password"));
+  }
+
+  @Test
+  public void testUTF8() {
+    Context context = new Context(ImmutableMap.of(
+        "password.command", "echo -n applepieéáű",
+        "password.charset", "utf-8"));
+    ExternalProcessPasswordProvider passwordProvider = new ExternalProcessPasswordProvider();
+    Assert.assertEquals("applepieéáű", passwordProvider.getPassword(context, "password"));
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testNoCommand() {
+    Context context = new Context();
+    ExternalProcessPasswordProvider passwordProvider = new ExternalProcessPasswordProvider();
+    passwordProvider.getPassword(context, "password");
+  }
+
+}

flume-ng-core/src/test/java/org/apache/flume/conf/TestPasswordConfigurator.java

@@ -0,0 +1,53 @@
+/*
+ * 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.flume.conf;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.flume.Context;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestPasswordConfigurator {
+
+  @Test
+  public void testBackwardCompatibility() {
+    Context ctx = new Context(ImmutableMap.of("password", "applepie"));
+    Assert.assertEquals("applepie", PasswordConfigurator.getPassword(ctx, "password"));
+  }
+
+  @Test
+  public void testNullPassword() {
+    Context ctx = new Context();
+    Assert.assertNull(PasswordConfigurator.getPassword(ctx, "password"));
+  }
+
+  @Test
+  public void testPlainTextPasswordProvider() {
+    Context ctx = new Context(ImmutableMap.of("password", "applepie",
+        "password.passwordProviderClass", PlainTextPasswordProvider.class.getName()));
+    Assert.assertEquals("applepie", PasswordConfigurator.getPassword(ctx, "password"));
+  }
+
+  @Test
+  public void testExternalProcessPasswordProvider() {
+    Context ctx = new Context(ImmutableMap.of("password.command", "echo -n applepie",
+        "password.passwordProviderClass", ExternalProcessPasswordProvider.class.getName()));
+    Assert.assertEquals("applepie", PasswordConfigurator.getPassword(ctx, "password"));
+  }
+}