Apache Flume vulnerable to a JNDI RCE in JMSSource

@@ -19,6 +19,8 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
 import java.util.List;
 import java.util.Locale;
 import java.util.Properties;
@@ -53,6 +55,7 @@
 @InterfaceStability.Unstable
 public class JMSSource extends AbstractPollableSource implements BatchSizeSupported {
   private static final Logger logger = LoggerFactory.getLogger(JMSSource.class);
+  private static final String JAVA_SCHEME = "java";
 
   // setup by constructor
   private final InitialContextFactory initialContextFactory;
@@ -179,6 +182,14 @@ protected void doConfigure(Context context) throws FlumeException {
     String connectionFactoryName = context.getString(
         JMSSourceConfiguration.CONNECTION_FACTORY,
         JMSSourceConfiguration.CONNECTION_FACTORY_DEFAULT).trim();
+    try {
+      URI uri = new URI(connectionFactoryName);
+      String scheme = uri.getScheme();
+      assertTrue(scheme == null || scheme.equals(JAVA_SCHEME),
+          "Unsupported JNDI URI: " + connectionFactoryName);
+    } catch (URISyntaxException ex) {
+      logger.warn("Invalid JNDI URI - {}", connectionFactoryName);
+    }
 
     assertNotEmpty(initialContextFactoryName, String.format(
         "Initial Context Factory is empty. This is specified by %s",
@@ -272,6 +283,10 @@ private void assertNotEmpty(String arg, String msg) {
     Preconditions.checkArgument(!arg.isEmpty(), msg);
   }
 
+  private void assertTrue(boolean arg, String msg) {
+    Preconditions.checkArgument(arg, msg);
+  }
+
   @Override
   protected synchronized Status doProcess() throws EventDeliveryException {
     boolean error = true;

@@ -125,6 +125,13 @@ public void testConfigureWithoutDestinationName() throws Exception {
     source.configure(context);
   }
 
+  @Test(expected = IllegalArgumentException.class)
+  public void testConfigureWithConnectionFactory() throws Exception {
+    context.put(JMSSourceConfiguration.CONNECTION_FACTORY,
+        "ldap://localhost:319/connectionFactory");
+    source.configure(context);
+  }
+
   @Test(expected = FlumeException.class)
   public void testConfigureWithBadDestinationType() throws Exception {
     context.put(JMSSourceConfiguration.DESTINATION_TYPE, "DUMMY");