Apache NiFi: Potential Deserialization of Untrusted Data with JNDI in JMS Components

@@ -18,19 +18,29 @@
 
 import org.apache.nifi.components.PropertyDescriptor;
 import org.apache.nifi.components.PropertyDescriptor.Builder;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
 import org.apache.nifi.components.Validator;
 import org.apache.nifi.components.resource.ResourceCardinality;
 import org.apache.nifi.components.resource.ResourceType;
 import org.apache.nifi.expression.ExpressionLanguageScope;
 import org.apache.nifi.processor.util.StandardValidators;
 
 import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashSet;
 import java.util.List;
+import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 
 import static org.apache.nifi.processor.util.StandardValidators.NON_EMPTY_VALIDATOR;
 
 public class JndiJmsConnectionFactoryProperties {
 
+    public static final String URL_SCHEMES_ALLOWED_PROPERTY = "org.apache.nifi.jms.cf.jndi.provider.url.schemes.allowed";
+
     public static final PropertyDescriptor JNDI_INITIAL_CONTEXT_FACTORY = new Builder()
             .name("java.naming.factory.initial")
             .displayName("JNDI Initial Context Factory Class")
@@ -43,9 +53,9 @@ public class JndiJmsConnectionFactoryProperties {
     public static final PropertyDescriptor JNDI_PROVIDER_URL = new Builder()
             .name("java.naming.provider.url")
             .displayName("JNDI Provider URL")
-            .description("The URL of the JNDI Provider to use (java.naming.provider.url).")
+            .description("The URL of the JNDI Provider to use as the value for java.naming.provider.url. See additional details documentation for allowed URL schemes.")
             .required(true)
-            .addValidator(NON_EMPTY_VALIDATOR)
+            .addValidator(new JndiJmsProviderUrlValidator())
             .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
             .build();
 
@@ -114,4 +124,80 @@ public static PropertyDescriptor getDynamicPropertyDescriptor(final String prope
                 .build();
     }
 
+    static class JndiJmsProviderUrlValidator implements Validator {
+
+        private static final Pattern URL_SCHEME_PATTERN = Pattern.compile("^([^:]+)://.+$");
+
+        private static final int SCHEME_GROUP = 1;
+
+        private static final String SPACE_SEPARATOR = " ";
+
+        private static final Set<String> DEFAULT_ALLOWED_SCHEMES = Collections.unmodifiableSet(new LinkedHashSet<>(Arrays.asList(
+                "file",
+                "jgroups",
+                "t3",
+                "t3s",
+                "tcp",
+                "ssl",
+                "udp",
+                "vm"
+        )));
+
+        private final Set<String> allowedSchemes;
+
+        JndiJmsProviderUrlValidator() {
+            final String allowed = System.getProperty(URL_SCHEMES_ALLOWED_PROPERTY);
+            if (allowed == null || allowed.isEmpty()) {
+                allowedSchemes = DEFAULT_ALLOWED_SCHEMES;
+            } else {
+                allowedSchemes = Arrays.stream(allowed.split(SPACE_SEPARATOR)).collect(Collectors.toSet());
+            }
+        }
+
+        @Override
+        public ValidationResult validate(final String subject, final String input, final ValidationContext context) {
+            final ValidationResult.Builder builder = new ValidationResult.Builder().subject(subject).input(input);
+
+            if (input == null || input.isEmpty()) {
+                builder.valid(false);
+                builder.explanation("URL is required");
+            } else if (isUrlAllowed(input)) {
+                builder.valid(true);
+                builder.explanation("URL scheme allowed");
+            } else {
+                builder.valid(false);
+                final String explanation = String.format("URL scheme not allowed. Allowed URL schemes include %s", allowedSchemes);
+                builder.explanation(explanation);
+            }
+
+            return builder.build();
+        }
+
+        private boolean isUrlAllowed(final String input) {
+            final boolean allowed;
+
+            final Matcher matcher = URL_SCHEME_PATTERN.matcher(input);
+            if (matcher.matches()) {
+                final String scheme = matcher.group(SCHEME_GROUP);
+                allowed = isSchemeAllowed(scheme);
+            } else {
+                allowed = true;
+            }
+
+            return allowed;
+        }
+
+        private boolean isSchemeAllowed(final String scheme) {
+            boolean allowed = false;
+
+            for (final String allowedScheme : allowedSchemes) {
+                if (allowedScheme.contains(scheme)) {
+                    allowed = true;
+                    break;
+                }
+            }
+
+            return allowed;
+        }
+    }
 }

@@ -21,9 +21,9 @@
 </head>
 
 <body>
-<h2>Description:</h2>
+<h2>Capabilities</h2>
 <p>
-    This ControllerService allows users to reference a JMS Connection Factory that has already been established and
+    This Controller Service allows users to reference a JMS Connection Factory that has already been established and
     made available via Java Naming and Directory Interface (JNDI) Server. Please see documentation from your JMS Vendor in order
     to understand the appropriate values to configure for this service.
 </p>
@@ -55,7 +55,7 @@ <h2>Description:</h2>
 </p>
 
 
-<h2>Example:</h2>
+<h2>Example Configuration</h2>
 
 <p>
     As an example, the following configuration may be used to connect to Active MQ's JMS Broker, using the Connection Factory provided via their embedded JNDI server:
@@ -91,5 +91,53 @@ <h2>Example:</h2>
     the jar(s) containing the org.apache.activemq.jndi.ActiveMQInitialContextFactory class and the other JMS client classes can be found within the /opt/apache-activemq-5.15.2/lib/ directory.
 </p>
 
+<h2>Property Validation</h2>
+
+<p>
+  The following component properties include additional validation to restrict allowed values:
+</p>
+
+<ul>
+  <li>JNDI Provider URL</li>
+</ul>
+
+<h3>JNDI Provider URL Validation</h3>
+
+<p>
+  The default validation for <code>JNDI Provider URL</code> allows the following URL schemes:
+</p>
+
+<ul>
+  <li>file</li>
+  <li>jgroups</li>
+  <li>ssl</li>
+  <li>t3</li>
+  <li>t3s</li>
+  <li>tcp</li>
+  <li>udp</li>
+  <li>vm</li>
+</ul>
+
+<p>
+  The following Java System property can be configured to override the default allowed URL schemes:
+</p>
+
+<ul>
+  <li>
+    <code>org.apache.nifi.jms.cf.jndi.provider.url.schemes.allowed</code>
+  </li>
+</ul>
+
+<p>
+  The System property must contain a space-separated list of URL schemes. This property can be configured in the application
+  <code>bootstrap.conf</code> as follows:
+</p>
+
+<ul>
+  <li>
+    <code>java.arg.jndiJmsUrlSchemesAllowed=-Dorg.apache.nifi.jms.cf.jndi.provider.url.schemes.allowed=ssl tcp</code>
+  </li>
+</ul>
+
 </body>
 </html>

@@ -0,0 +1,116 @@
+/*
+ * 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.nifi.jms.cf;
+
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.reporting.InitializationException;
+import org.apache.nifi.util.MockProcessContext;
+import org.apache.nifi.util.MockValidationContext;
+import org.apache.nifi.util.NoOpProcessor;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class JndiJmsConnectionFactoryProviderTest {
+
+    private static final String SERVICE_ID = JndiJmsConnectionFactoryProvider.class.getSimpleName();
+
+    private static final String CONTEXT_FACTORY = "ContextFactory";
+
+    private static final String FACTORY_NAME = "ConnectionFactory";
+
+    private static final String TCP_PROVIDER_URL = "tcp://127.0.0.1";
+
+    private static final String LDAP_PROVIDER_URL = "ldap://127.0.0.1";
+
+    private static final String HOST_PORT_URL = "127.0.0.1:1024";
+
+    private static final String LDAP_ALLOWED_URL_SCHEMES = "ldap";
+
+    private TestRunner runner;
+
+    private JndiJmsConnectionFactoryProvider provider;
+
+    @BeforeEach
+    void setRunner() throws InitializationException {
+        runner = TestRunners.newTestRunner(NoOpProcessor.class);
+        provider = new JndiJmsConnectionFactoryProvider();
+        runner.addControllerService(SERVICE_ID, provider);
+    }
+
+    @Test
+    void testPropertiesValid() {
+        setFactoryProperties();
+
+        runner.setProperty(provider, JndiJmsConnectionFactoryProperties.JNDI_PROVIDER_URL, TCP_PROVIDER_URL);
+
+        runner.assertValid(provider);
+    }
+
+    @Test
+    void testPropertiesInvalidUrlNotConfigured() {
+        setFactoryProperties();
+
+        runner.assertNotValid(provider);
+    }
+
+    @Test
+    void testPropertiesInvalidUrlScheme() {
+        setFactoryProperties();
+
+        runner.setProperty(provider, JndiJmsConnectionFactoryProperties.JNDI_PROVIDER_URL, LDAP_PROVIDER_URL);
+
+        runner.assertNotValid(provider);
+    }
+
+    @Test
+    void testPropertiesHostPortUrl() {
+        setFactoryProperties();
+
+        runner.setProperty(provider, JndiJmsConnectionFactoryProperties.JNDI_PROVIDER_URL, HOST_PORT_URL);
+
+        runner.assertValid(provider);
+    }
+
+
+    @Test
+    void testUrlSchemeValidSystemProperty() {
+        try {
+            System.setProperty(JndiJmsConnectionFactoryProperties.URL_SCHEMES_ALLOWED_PROPERTY, LDAP_ALLOWED_URL_SCHEMES);
+
+            final MockProcessContext processContext = new MockProcessContext(new NoOpProcessor());
+            final MockValidationContext validationContext = new MockValidationContext(processContext);
+
+            final JndiJmsConnectionFactoryProperties.JndiJmsProviderUrlValidator validator = new JndiJmsConnectionFactoryProperties.JndiJmsProviderUrlValidator();
+            final ValidationResult result = validator.validate(JndiJmsConnectionFactoryProperties.JNDI_PROVIDER_URL.getDisplayName(), LDAP_PROVIDER_URL, validationContext);
+
+            assertNotNull(result);
+            assertTrue(result.isValid());
+        } finally {
+            System.clearProperty(JndiJmsConnectionFactoryProperties.URL_SCHEMES_ALLOWED_PROPERTY);
+        }
+    }
+
+    private void setFactoryProperties() {
+        runner.setProperty(provider, JndiJmsConnectionFactoryProperties.JNDI_INITIAL_CONTEXT_FACTORY, CONTEXT_FACTORY);
+        runner.setProperty(provider, JndiJmsConnectionFactoryProperties.JNDI_CONNECTION_FACTORY_NAME, FACTORY_NAME);
+    }
+}