NIFI-3193: added ability to authenticate using cert common names

This closes #1971.

Signed-off-by: Tony Kurc <tkurc@apache.org>
Also reviewed by Pierre Villard <pierre.villard.fr@gmail.com>

nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/AbstractAMQPProcessor.java

@@ -37,6 +37,8 @@ import org.apache.nifi.ssl.SSLContextService;
 
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.DefaultSaslConfig;
+
 
 /**
  * Base processor that uses RabbitMQ client API
@@ -97,6 +99,15 @@ abstract class AbstractAMQPProcessor<T extends AMQPWorker> extends AbstractProce
             .required(false)
             .identifiesControllerService(SSLContextService.class)
             .build();
+    public static final PropertyDescriptor USE_CERT_AUTHENTICATION = new PropertyDescriptor.Builder()
+            .name("cert-authentication")
+            .displayName("Use Certificate Authentication")
+            .description("Authenticate using the SSL certificate common name rather than user name/password.")
+            .required(false)
+            .defaultValue("false")
+            .allowableValues("true", "false")
+            .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+            .build();
     public static final PropertyDescriptor CLIENT_AUTH = new PropertyDescriptor.Builder()
             .name("ssl-client-auth")
             .displayName("Client Auth")
@@ -122,6 +133,7 @@ abstract class AbstractAMQPProcessor<T extends AMQPWorker> extends AbstractProce
         descriptors.add(PASSWORD);
         descriptors.add(AMQP_VERSION);
         descriptors.add(SSL_CONTEXT_SERVICE);
+        descriptors.add(USE_CERT_AUTHENTICATION);
         descriptors.add(CLIENT_AUTH);
     }
 
@@ -218,9 +230,14 @@ abstract class AbstractAMQPProcessor<T extends AMQPWorker> extends AbstractProce
         }
 
         // handles TLS/SSL aspects
+        final Boolean useCertAuthentication = context.getProperty(USE_CERT_AUTHENTICATION).asBoolean();
         final SSLContextService sslService = context.getProperty(SSL_CONTEXT_SERVICE).asControllerService(SSLContextService.class);
+        // if the property to use cert authentication is set but the SSL service hasn't been configured, throw an exception.
+        if (useCertAuthentication && sslService == null) {
+            throw new ProviderCreationException("This processor is configured to use cert authentication, " +
+                    "but the SSL Context Service hasn't been configured. You need to configure the SSL Context Service.");
+        }
         final String rawClientAuth = context.getProperty(CLIENT_AUTH).getValue();
-        final SSLContext sslContext;
 
         if (sslService != null) {
             final SSLContextService.ClientAuth clientAuth;
@@ -234,14 +251,14 @@ abstract class AbstractAMQPProcessor<T extends AMQPWorker> extends AbstractProce
                             rawClientAuth, StringUtils.join(SslContextFactory.ClientAuth.values(), ", ")));
                 }
             }
-            sslContext = sslService.createSSLContext(clientAuth);
+            final SSLContext sslContext = sslService.createSSLContext(clientAuth);
-        } else {
-            sslContext = null;
-        }
-
-        // check if the ssl context is set and add it to the factory if so
-        if (sslContext != null) {
             cf.useSslProtocol(sslContext);
+
+            if (useCertAuthentication) {
+                // this tells the factory to use the cert common name for authentication and not user name and password
+                // REF: https://github.com/rabbitmq/rabbitmq-auth-mechanism-ssl
+                cf.setSaslConfig(DefaultSaslConfig.EXTERNAL);
+            }
         }
 
         try {

nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/resources/docs/org.apache.nifi.amqp.processors.ConsumeAMQP/additionalDetails.html

@@ -63,6 +63,9 @@
     <li><b>Password</b> - [REQUIRED] password to use with user name to connect to AMQP broker. 
     Usually provided by the administrator. Defaults to 'guest'.
     </li>
+    <li><b>Use Certificate Authentication</b> - [OPTIONAL] whether or not to use the SSL certificate common name for authentication rather than user name/password.
+    This can only be used in conjunction with SSL. Defaults to 'false'.
+    </li>
     <li><b>Virtual Host</b> - [OPTIONAL] Virtual Host name which segregates AMQP system for enhanced security.
     Please refer to <a href="http://blog.dtzq.com/2012/06/rabbitmq-users-and-virtual-hosts.html">this blog</a> for more details on Virtual Host.
     </li>

nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/resources/docs/org.apache.nifi.amqp.processors.PublishAMQP/additionalDetails.html

@@ -84,6 +84,9 @@
     <li><b>Password</b> - [REQUIRED] password to use with user name to connect to AMQP broker. 
     Usually provided by the administrator. Defaults to 'guest'.
     </li>
+    <li><b>Use Certificate Authentication</b> - [OPTIONAL] whether or not to use the SSL certificate common name for authentication rather than user name/password.
+    This can only be used in conjunction with SSL. Defaults to 'false'.
+    </li>
     <li><b>Virtual Host</b> - [OPTIONAL] Virtual Host name which segregates AMQP system for enhanced security.
     Please refer to <a href="http://blog.dtzq.com/2012/06/rabbitmq-users-and-virtual-hosts.html">this blog</a> for more details on Virtual Host.
     </li>

nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/test/java/org/apache/nifi/amqp/processors/AbstractAMQPProcessorTest.java

@@ -29,6 +29,7 @@ import org.apache.nifi.util.TestRunners;
 import org.junit.Before;
 import org.junit.Test;
 
+
 /**
  * Unit tests for the AbstractAMQPProcessor class
  */
@@ -50,6 +51,7 @@ public class AbstractAMQPProcessorTest {
         testRunner.addControllerService("ssl-context", sslService);
         testRunner.enableControllerService(sslService);
         testRunner.setProperty(AbstractAMQPProcessor.SSL_CONTEXT_SERVICE, "ssl-context");
+        testRunner.setProperty(AbstractAMQPProcessor.USE_CERT_AUTHENTICATION, "false");
         testRunner.setProperty(AbstractAMQPProcessor.HOST, "test");
         testRunner.setProperty(AbstractAMQPProcessor.PORT, "9999");
         testRunner.setProperty(AbstractAMQPProcessor.USER, "test");
@@ -59,6 +61,17 @@ public class AbstractAMQPProcessorTest {
         processor.onTrigger(testRunner.getProcessContext(), testRunner.getProcessSessionFactory());
     }
 
+    @Test(expected = ProviderCreationException.class)
+    public void testInvalidSSLConfiguration() throws Exception {
+        // it's invalid to have use_cert_auth enabled and not have the SSL Context Service configured
+        testRunner.setProperty(AbstractAMQPProcessor.USE_CERT_AUTHENTICATION, "true");
+        testRunner.setProperty(AbstractAMQPProcessor.HOST, "test");
+        testRunner.setProperty(AbstractAMQPProcessor.PORT, "9999");
+        testRunner.setProperty(AbstractAMQPProcessor.USER, "test");
+        testRunner.setProperty(AbstractAMQPProcessor.PASSWORD, "test");
+        processor.onTrigger(testRunner.getProcessContext(), testRunner.getProcessSessionFactory());
+    }
+
     /**
      * Provides a stubbed processor instance for testing
      */
@@ -67,6 +80,7 @@ public class AbstractAMQPProcessorTest {
         protected void rendezvousWithAmqp(ProcessContext context, ProcessSession session) throws ProcessException {
             // nothing to do
         }
+
         @Override
         protected AMQPConsumer finishBuildingTargetResource(ProcessContext context) {
             return null;