James 3502, add ssl support

backends-common/rabbitmq/pom.xml

@@ -93,6 +93,14 @@
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-pool2</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpclient</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpcore</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>jcl-over-slf4j</artifactId>

backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/RabbitMQConnectionFactory.java

@@ -18,9 +18,26 @@
  ****************************************************************/
 package org.apache.james.backends.rabbitmq;
 
+import static org.apache.james.backends.rabbitmq.RabbitMQConfiguration.SSLConfiguration.HostNameVerifier;
+import static org.apache.james.backends.rabbitmq.RabbitMQConfiguration.SSLConfiguration.SSLKeyStore;
+import static org.apache.james.backends.rabbitmq.RabbitMQConfiguration.SSLConfiguration.SSLTrustStore;
+import static org.apache.james.backends.rabbitmq.RabbitMQConfiguration.SSLConfiguration.SSLValidationStrategy;
+
+import java.io.IOException;
+import java.security.KeyManagementException;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.security.UnrecoverableKeyException;
+import java.security.cert.CertificateException;
 import java.time.Duration;
+import java.util.Optional;
 
 import javax.inject.Inject;
+import javax.net.ssl.SSLContext;
+
+import org.apache.commons.lang3.NotImplementedException;
+import org.apache.http.ssl.SSLContextBuilder;
+import org.apache.http.ssl.TrustStrategy;
 
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
@@ -31,14 +48,16 @@
 
 public class RabbitMQConnectionFactory {
 
+    private static final TrustStrategy TRUST_ALL = (x509Certificates, authType) -> true;
+
     private final ConnectionFactory connectionFactory;
 
     private final RabbitMQConfiguration configuration;
 
     @Inject
     public RabbitMQConnectionFactory(RabbitMQConfiguration rabbitMQConfiguration) {
-        this.connectionFactory = from(rabbitMQConfiguration);
         this.configuration = rabbitMQConfiguration;
+        this.connectionFactory = from(rabbitMQConfiguration);
     }
 
     private ConnectionFactory from(RabbitMQConfiguration rabbitMQConfiguration) {
@@ -47,25 +66,115 @@ private ConnectionFactory from(RabbitMQConfiguration rabbitMQConfiguration) {
             connectionFactory.setUri(rabbitMQConfiguration.getUri());
             connectionFactory.setHandshakeTimeout(rabbitMQConfiguration.getHandshakeTimeoutInMs());
             connectionFactory.setShutdownTimeout(rabbitMQConfiguration.getShutdownTimeoutInMs());
-            connectionFactory.setChannelRpcTimeout(rabbitMQConfiguration.getChannelRpcTimeoutInMs());
-            connectionFactory.setConnectionTimeout(rabbitMQConfiguration.getConnectionTimeoutInMs());
-            connectionFactory.setNetworkRecoveryInterval(rabbitMQConfiguration.getNetworkRecoveryIntervalInMs());
+            connectionFactory
+                .setChannelRpcTimeout(rabbitMQConfiguration.getChannelRpcTimeoutInMs());
+            connectionFactory
+                .setConnectionTimeout(rabbitMQConfiguration.getConnectionTimeoutInMs());
+            connectionFactory
+                .setNetworkRecoveryInterval(rabbitMQConfiguration.getNetworkRecoveryIntervalInMs());
+
+            connectionFactory
+                .setUsername(rabbitMQConfiguration.getManagementCredentials().getUser());
+            connectionFactory.setPassword(
+                String.valueOf(rabbitMQConfiguration.getManagementCredentials().getPassword()));
 
-            connectionFactory.setUsername(rabbitMQConfiguration.getManagementCredentials().getUser());
-            connectionFactory.setPassword(String.valueOf(rabbitMQConfiguration.getManagementCredentials().getPassword()));
+            if (configuration.useSsl()) {
+                setupSslConfiguration(connectionFactory);
+            }
 
             return connectionFactory;
         } catch (Exception e) {
             throw new RuntimeException(e);
         }
     }
 
+    private void setupSslConfiguration(ConnectionFactory connectionFactory)
+        throws CertificateException, UnrecoverableKeyException, NoSuchAlgorithmException,
+        KeyManagementException, KeyStoreException, IOException {
+        connectionFactory.useSslProtocol(sslContext(configuration));
+        setupHostNameVerification(connectionFactory);
+    }
+
+    private SSLContext sslContext(RabbitMQConfiguration configuration)
+        throws KeyManagementException, NoSuchAlgorithmException, CertificateException,
+        KeyStoreException, IOException, UnrecoverableKeyException {
+        SSLContextBuilder sslContextBuilder = new SSLContextBuilder();
+
+        RabbitMQConfiguration.SSLConfiguration sslConfiguration =
+            configuration.getSslConfiguration();
+
+        setupSslValidationStrategy(sslContextBuilder, sslConfiguration);
+
+        setupClientCertificateAuthentication(sslContextBuilder, sslConfiguration);
+
+        return sslContextBuilder.build();
+
+    }
+
+    private void setupClientCertificateAuthentication(SSLContextBuilder sslContextBuilder,
+                                                      RabbitMQConfiguration.SSLConfiguration sslConfiguration)
+        throws NoSuchAlgorithmException, KeyStoreException, UnrecoverableKeyException,
+        CertificateException, IOException {
+        Optional<SSLKeyStore> keyStore = sslConfiguration.getKeyStore();
+
+        if (keyStore.isPresent()) {
+            SSLKeyStore sslKeyStore = keyStore.get();
+
+            sslContextBuilder.loadKeyMaterial(sslKeyStore.getFile(), sslKeyStore.getPassword(),
+                sslKeyStore.getPassword());
+        }
+    }
+
+    private void setupSslValidationStrategy(SSLContextBuilder sslContextBuilder,
+                                            RabbitMQConfiguration.SSLConfiguration sslConfiguration)
+        throws NoSuchAlgorithmException, KeyStoreException, CertificateException, IOException {
+        SSLValidationStrategy strategy = sslConfiguration
+            .getStrategy();
+
+        switch (strategy) {
+            case DEFAULT:
+                break;
+            case IGNORE:
+                sslContextBuilder.loadTrustMaterial(TRUST_ALL);
+                break;
+            case OVERRIDE:
+                applyTrustStore(sslContextBuilder);
+                break;
+            default:
+                throw new NotImplementedException(
+                    String.format("unrecognized strategy '%s'", strategy.name()));
+        }
+    }
+
+    private SSLContextBuilder applyTrustStore(SSLContextBuilder sslContextBuilder)
+        throws CertificateException, NoSuchAlgorithmException,
+        KeyStoreException, IOException {
+
+        SSLTrustStore trustStore = configuration.getSslConfiguration()
+            .getTrustStore()
+            .orElseThrow(() -> new IllegalStateException("SSLTrustStore cannot to be empty"));
+
+        return sslContextBuilder
+            .loadTrustMaterial(trustStore.getFile(), trustStore.getPassword());
+    }
+
+    private void setupHostNameVerification(ConnectionFactory connectionFactory) {
+        HostNameVerifier hostNameVerifier = configuration.getSslConfiguration()
+            .getHostNameVerifier();
+
+        if (hostNameVerifier == HostNameVerifier.DEFAULT) {
+            connectionFactory.enableHostnameVerification();
+        }
+    }
+
     Connection create() {
         return connectionMono().block();
     }
 
     Mono<Connection> connectionMono() {
         return Mono.fromCallable(connectionFactory::newConnection)
-            .retryWhen(Retry.backoff(configuration.getMaxRetries(), Duration.ofMillis(configuration.getMinDelayInMs())).scheduler(Schedulers.elastic()));
+            .retryWhen(Retry.backoff(configuration.getMaxRetries(),
+                Duration.ofMillis(configuration.getMinDelayInMs()))
+                .scheduler(Schedulers.elastic()));
     }
 }

backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/RabbitMQManagementAPI.java

@@ -19,16 +19,32 @@
 
 package org.apache.james.backends.rabbitmq;
 
+import java.io.IOException;
+import java.security.KeyManagementException;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.security.UnrecoverableKeyException;
+import java.security.cert.CertificateException;
 import java.util.Date;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
+
+import javax.net.ssl.HostnameVerifier;
+import javax.net.ssl.SSLContext;
+
+import org.apache.commons.lang3.NotImplementedException;
+import org.apache.http.conn.ssl.DefaultHostnameVerifier;
+import org.apache.http.ssl.SSLContextBuilder;
+import org.apache.http.ssl.TrustStrategy;
 
 import com.fasterxml.jackson.annotation.JsonFormat;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.annotation.JsonValue;
 import com.google.common.base.MoreObjects;
 
+import feign.Client;
 import feign.Feign;
 import feign.Logger;
 import feign.Param;
@@ -299,26 +315,119 @@ public final boolean equals(Object o) {
 
         @Override
         public final int hashCode() {
-            return Objects.hash(source, vhost, destination, destinationType, routingKey, arguments, propertiesKey);
+            return Objects.hash(source, vhost, destination, destinationType, routingKey, arguments,
+                propertiesKey);
         }
     }
 
     static RabbitMQManagementAPI from(RabbitMQConfiguration configuration) {
-        RabbitMQConfiguration.ManagementCredentials credentials = configuration.getManagementCredentials();
-        return Feign.builder()
-            .requestInterceptor(new BasicAuthRequestInterceptor(credentials.getUser(), new String(credentials.getPassword())))
-            .logger(new Slf4jLogger(RabbitMQManagementAPI.class))
-            .logLevel(Logger.Level.FULL)
-            .encoder(new JacksonEncoder())
-            .decoder(new JacksonDecoder())
-            .retryer(new Retryer.Default())
-            .errorDecoder(RETRY_500)
-            .target(RabbitMQManagementAPI.class, configuration.getManagementUri().toString());
+        try {
+            RabbitMQConfiguration.ManagementCredentials credentials =
+                configuration.getManagementCredentials();
+            RabbitMQManagementAPI rabbitMQManagementAPI = Feign.builder()
+                .client(getClient(configuration))
+                .requestInterceptor(new BasicAuthRequestInterceptor(credentials.getUser(),
+                    new String(credentials.getPassword())))
+                .logger(new Slf4jLogger(RabbitMQManagementAPI.class))
+                .logLevel(Logger.Level.FULL)
+                .encoder(new JacksonEncoder())
+                .decoder(new JacksonDecoder())
+                .retryer(new Retryer.Default())
+                .errorDecoder(RETRY_500)
+                .target(RabbitMQManagementAPI.class, configuration.getManagementUri().toString());
+
+            return rabbitMQManagementAPI;
+        } catch (KeyManagementException | NoSuchAlgorithmException | CertificateException | KeyStoreException | IOException | UnrecoverableKeyException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private static Client getClient(RabbitMQConfiguration configuration)
+        throws KeyManagementException, NoSuchAlgorithmException, CertificateException,
+        KeyStoreException, IOException, UnrecoverableKeyException {
+        if (configuration.useSslManagement()) {
+            SSLContextBuilder sslContextBuilder = new SSLContextBuilder();
+
+            setupSslValidationStrategy(sslContextBuilder, configuration);
+
+            setupClientCertificateAuthentication(sslContextBuilder, configuration);
+
+            SSLContext sslContext = sslContextBuilder.build();
+
+            return new Client.Default(sslContext.getSocketFactory(),
+                getHostNameVerifier(configuration));
+        } else {
+            return new Client.Default(null, null);
+        }
+
+    }
+
+    private static HostnameVerifier getHostNameVerifier(RabbitMQConfiguration configuration) {
+        switch (configuration.getSslConfiguration().getHostNameVerifier()) {
+            case ACCEPT_ANY_HOSTNAME:
+                return ((hostname, session) -> true);
+            default:
+                return new DefaultHostnameVerifier();
+        }
+    }
+
+    private static void setupClientCertificateAuthentication(SSLContextBuilder sslContextBuilder,
+                                                             RabbitMQConfiguration configuration)
+        throws NoSuchAlgorithmException, KeyStoreException, UnrecoverableKeyException,
+        CertificateException, IOException {
+        Optional<RabbitMQConfiguration.SSLConfiguration.SSLKeyStore> keyStore =
+            configuration.getSslConfiguration().getKeyStore();
+
+        if (keyStore.isPresent()) {
+            RabbitMQConfiguration.SSLConfiguration.SSLKeyStore sslKeyStore = keyStore.get();
+
+            sslContextBuilder.loadKeyMaterial(sslKeyStore.getFile(), sslKeyStore.getPassword(),
+                sslKeyStore.getPassword());
+        }
+    }
+
+    private static void setupSslValidationStrategy(SSLContextBuilder sslContextBuilder,
+                                                   RabbitMQConfiguration configuration)
+        throws NoSuchAlgorithmException, KeyStoreException, CertificateException, IOException {
+        RabbitMQConfiguration.SSLConfiguration.SSLValidationStrategy strategy = configuration
+            .getSslConfiguration()
+            .getStrategy();
+
+        final TrustStrategy TRUST_ALL = (x509Certificates, authType) -> true;
+
+        switch (strategy) {
+            case DEFAULT:
+                break;
+            case IGNORE:
+                sslContextBuilder.loadTrustMaterial(TRUST_ALL);
+                break;
+            case OVERRIDE:
+                applyTrustStore(sslContextBuilder, configuration);
+                break;
+            default:
+                throw new NotImplementedException(
+                    String.format("unrecognized strategy '%s'", strategy.name()));
+        }
+    }
+
+    private static SSLContextBuilder applyTrustStore(SSLContextBuilder sslContextBuilder,
+                                                     RabbitMQConfiguration configuration)
+        throws CertificateException, NoSuchAlgorithmException,
+        KeyStoreException, IOException {
+
+        RabbitMQConfiguration.SSLConfiguration.SSLTrustStore trustStore =
+            configuration.getSslConfiguration()
+                .getTrustStore()
+                .orElseThrow(() -> new IllegalStateException("SSLTrustStore cannot to be empty"));
+
+        return sslContextBuilder
+            .loadTrustMaterial(trustStore.getFile(), trustStore.getPassword());
     }
 
     ErrorDecoder RETRY_500 = (methodKey, response) -> {
         if (response.status() == 500) {
-            throw new RetryableException(response.status(), "Error encountered, scheduling retry", response.request().httpMethod(), new Date());
+            throw new RetryableException(response.status(), "Error encountered, scheduling retry",
+                response.request().httpMethod(), new Date());
         }
         throw new RuntimeException("Non recoverable exception status: " + response.status());
     };