JAMES-4072 Support Redis TLS/SSL connection

backends-common/redis/pom.xml

@@ -17,6 +17,10 @@
             <groupId>${james.groupId}</groupId>
             <artifactId>james-core</artifactId>
         </dependency>
+        <dependency>
+            <groupId>${james.groupId}</groupId>
+            <artifactId>james-server-filesystem-api</artifactId>
+        </dependency>
         <dependency>
             <groupId>${james.groupId}</groupId>
             <artifactId>james-server-guice-common</artifactId>
@@ -28,6 +32,10 @@
             <artifactId>james-server-testing</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>${james.groupId}</groupId>
+            <artifactId>james-server-util</artifactId>
+        </dependency>
         <dependency>
             <groupId>com.google.inject</groupId>
             <artifactId>guice</artifactId>

backends-common/redis/src/main/java/org/apache/james/backends/redis/RedisClientFactory.scala

@@ -0,0 +1,93 @@
+/****************************************************************
+ * 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.james.backends.redis
+
+import java.time.Duration
+
+import io.lettuce.core.{ClientOptions, RedisClient, SslOptions}
+import io.lettuce.core.cluster.RedisClusterClient
+import io.lettuce.core.resource.ClientResources
+import jakarta.inject.{Inject, Singleton}
+import org.apache.james.filesystem.api.FileSystem
+import org.apache.james.util.concurrent.NamedThreadFactory
+
+import scala.jdk.CollectionConverters._
+
+class RedisClientFactory @Singleton() @Inject()
+(fileSystem: FileSystem) {
+  def createStandaloneClient(redisConfiguration: StandaloneRedisConfiguration): RedisClient =
+    createStandaloneClient(redisConfiguration, Option.empty)
+
+  def createStandaloneClient(redisConfiguration: StandaloneRedisConfiguration, timeout: Duration): RedisClient =
+    createStandaloneClient(redisConfiguration, Option.apply(timeout))
+
+  def createStandaloneClient(redisConfiguration: StandaloneRedisConfiguration, maybeTimeout: Option[Duration]): RedisClient = {
+    maybeTimeout.foreach(timeout => redisConfiguration.redisURI.setTimeout(timeout))
+    val redisClient = RedisClient.create(redisConfiguration.redisURI)
+    redisClient.setOptions(createClientOptions(redisConfiguration.useSSL, redisConfiguration.mayBeSSLConfiguration))
+    redisClient
+  }
+
+  def createClusterClient(redisConfiguration: ClusterRedisConfiguration): RedisClusterClient =
+    createClusterClient(redisConfiguration, Option.empty)
+
+  def createClusterClient(redisConfiguration: ClusterRedisConfiguration, timeout: Duration): RedisClusterClient =
+    createClusterClient(redisConfiguration, Option.apply(timeout))
+
+  def createClusterClient(redisConfiguration: ClusterRedisConfiguration, maybeTimeout: Option[Duration]): RedisClusterClient = {
+    val resourceBuilder: ClientResources.Builder = ClientResources.builder()
+      .threadFactoryProvider(poolName => NamedThreadFactory.withName(s"redis-driver-$poolName"))
+    redisConfiguration.ioThreads.foreach(value => resourceBuilder.ioThreadPoolSize(value))
+    redisConfiguration.workerThreads.foreach(value => resourceBuilder.computationThreadPoolSize(value))
+    RedisClusterClient.create(resourceBuilder.build(),
+      redisConfiguration.redisURI.value
+        .map(rURI => {
+          maybeTimeout.foreach(timeout => rURI.setTimeout(timeout))
+          rURI
+        }).asJava)
+  }
+
+  def createMasterReplicaClient(redisConfiguration: MasterReplicaRedisConfiguration): RedisClient = {
+    val redisClient = RedisClient.create
+    redisClient.setOptions(createClientOptions(redisConfiguration.useSSL, redisConfiguration.mayBeSSLConfiguration))
+    redisClient
+  }
+
+  def createSentinelClient(redisConfiguration: SentinelRedisConfiguration): RedisClient = {
+    val redisClient = RedisClient.create
+    redisClient.setOptions(createClientOptions(redisConfiguration.useSSL, redisConfiguration.mayBeSSLConfiguration))
+    redisClient
+  }
+
+  private def createClientOptions(useSSL: Boolean, mayBeSSLConfiguration: Option[SSLConfiguration]): ClientOptions = {
+    val clientOptionsBuilder = ClientOptions.builder
+    if (useSSL) {
+      mayBeSSLConfiguration.foreach(sslConfig => {
+        if (!sslConfig.ignoreCertificateCheck) {
+          sslConfig.maybeKeyStore.foreach(redisKeyStore => {
+            val sslOptions = SslOptions.builder.jdkSslProvider.keystore(fileSystem.getFile(redisKeyStore.keyStoreFilePath), redisKeyStore.keyStorePassword.toCharArray).build
+            clientOptionsBuilder.sslOptions(sslOptions)
+          })
+        }
+      })
+    }
+    clientOptionsBuilder.build()
+  }
+}

backends-common/redis/src/main/java/org/apache/james/backends/redis/RedisConfiguration.scala

@@ -19,24 +19,36 @@
 
 package org.apache.james.backends.redis
 
+import java.io.File
+
 import com.google.common.base.{MoreObjects, Preconditions}
 import eu.timepit.refined
 import eu.timepit.refined.api.Refined
 import eu.timepit.refined.collection.NonEmpty
 import io.lettuce.core.{ReadFrom, RedisCredentials, RedisCredentialsProvider, RedisURI}
 import org.apache.commons.configuration2.Configuration
-import org.apache.james.backends.redis.RedisConfiguration.{CLUSTER_TOPOLOGY, MASTER_REPLICA_TOPOLOGY, REDIS_READ_FROM_DEFAULT_VALUE, REDIS_READ_FROM_PROPERTY_NAME, REDIS_SENTINEL_PASSWORD, SENTINEL_TOPOLOGY, STANDALONE_TOPOLOGY}
+import org.apache.james.backends.redis.RedisConfiguration.{CLUSTER_TOPOLOGY, KEY_STORE_FILE_PATH_DEFAULT_VALUE, KEY_STORE_PASSWORD_DEFAULT_VALUE, MASTER_REPLICA_TOPOLOGY, REDIS_IGNORE_CERTIFICATE_CHECK, REDIS_IGNORE_CERTIFICATE_CHECK_DEFAULT_VALUE, REDIS_KEY_STORE_FILE_PATH_PROPERTY_NAME, REDIS_KEY_STORE_PASSWORD_PROPERTY_NAME, REDIS_READ_FROM_DEFAULT_VALUE, REDIS_READ_FROM_PROPERTY_NAME, REDIS_SENTINEL_PASSWORD, REDIS_USE_SSL, REDIS_USE_SSL_DEFAULT_VALUE, SENTINEL_TOPOLOGY, STANDALONE_TOPOLOGY}
 import org.apache.james.backends.redis.RedisUris.{REDIS_URL_PROPERTY_NAME, RedisUris}
+import org.apache.james.filesystem.api.FileSystem
 import org.slf4j.{Logger, LoggerFactory}
 
 object RedisConfiguration {
   val REDIS_READ_FROM_PROPERTY_NAME = "redis.readFrom"
   val REDIS_SENTINEL_PASSWORD = "redis.sentinelPassword"
+  val REDIS_USE_SSL = "redis.ssl.enabled"
+  val REDIS_IGNORE_CERTIFICATE_CHECK = "redis.ignoreCertificateCheck"
+  val REDIS_KEY_STORE_FILE_PATH_PROPERTY_NAME: String = "redis.keystore.file.path"
+  val REDIS_KEY_STORE_PASSWORD_PROPERTY_NAME: String = "redis.keystore.password"
+
   val STANDALONE_TOPOLOGY = "standalone"
   val CLUSTER_TOPOLOGY = "cluster"
   val MASTER_REPLICA_TOPOLOGY = "master-replica"
   val SENTINEL_TOPOLOGY = "sentinel"
   val REDIS_READ_FROM_DEFAULT_VALUE = ReadFrom.MASTER
+  val REDIS_USE_SSL_DEFAULT_VALUE = false
+  val REDIS_IGNORE_CERTIFICATE_CHECK_DEFAULT_VALUE = true
+  val KEY_STORE_FILE_PATH_DEFAULT_VALUE = FileSystem.FILE_PROTOCOL + System.getProperty("java.home") + "/lib/security/cacerts".replace('/', File.separatorChar)
+  val KEY_STORE_PASSWORD_DEFAULT_VALUE = ""
 
   val LOGGER: Logger = LoggerFactory.getLogger(classOf[RedisConfiguration])
 
@@ -89,18 +101,22 @@ trait RedisConfiguration {
 }
 
 object StandaloneRedisConfiguration {
-  def from(config: Configuration): StandaloneRedisConfiguration = from(
-    config.getString(REDIS_URL_PROPERTY_NAME),
-    RedisConfiguration.redisIoThreadsFrom(config),
-    RedisConfiguration.redisWorkerThreadsFrom(config))
+  def from(config: Configuration): StandaloneRedisConfiguration =
+    from(config.getString(REDIS_URL_PROPERTY_NAME),
+      config.getBoolean(REDIS_USE_SSL, REDIS_USE_SSL_DEFAULT_VALUE),
+      SSLConfiguration.from(config),
+      RedisConfiguration.redisIoThreadsFrom(config),
+      RedisConfiguration.redisWorkerThreadsFrom(config))
 
-  def from(redisUri: String): StandaloneRedisConfiguration = from(redisUri, None, None)
+  def from(redisUri: String): StandaloneRedisConfiguration = from(redisUri, false, None, None, None)
 
-  def from(redisUri: String, ioThreads: Option[Int] = None, workerThreads: Option[Int] = None): StandaloneRedisConfiguration =
-    StandaloneRedisConfiguration(RedisURI.create(redisUri), ioThreads, workerThreads)
+  def from(redisUri: String, ssLConfiguration: SSLConfiguration): StandaloneRedisConfiguration = from(redisUri, true, Some(ssLConfiguration), None, None)
+
+  def from(redisUri: String, useSSL: Boolean, mayBeSSLConfiguration: Option[SSLConfiguration], ioThreads: Option[Int] = None, workerThreads: Option[Int] = None): StandaloneRedisConfiguration =
+    StandaloneRedisConfiguration(RedisURI.create(redisUri), useSSL, mayBeSSLConfiguration, ioThreads, workerThreads)
 }
 
-case class StandaloneRedisConfiguration(redisURI: RedisURI, ioThreads: Option[Int], workerThreads: Option[Int]) extends RedisConfiguration {
+case class StandaloneRedisConfiguration(redisURI: RedisURI, useSSL: Boolean, mayBeSSLConfiguration: Option[SSLConfiguration], ioThreads: Option[Int], workerThreads: Option[Int]) extends RedisConfiguration {
   override def asString: String = MoreObjects.toStringHelper(this)
     .add("topology", STANDALONE_TOPOLOGY)
     .add("redisURI", redisURI.toString)
@@ -117,23 +133,40 @@ object MasterReplicaRedisConfiguration {
     }
 
     from(redisUris,
+      config.getBoolean(REDIS_USE_SSL, REDIS_USE_SSL_DEFAULT_VALUE),
+      SSLConfiguration.from(config),
       Option(config.getString("redis.readFrom", null)).map(ReadFrom.valueOf).getOrElse(ReadFrom.MASTER),
       RedisConfiguration.redisIoThreadsFrom(config),
       RedisConfiguration.redisWorkerThreadsFrom(config))
   }
 
   def from(redisUris: Array[String],
+           readFrom: ReadFrom,
+           ioThreads: Option[Int],
+           workerThreads: Option[Int]): MasterReplicaRedisConfiguration =
+    from(redisUris, useSSL = false, Option.empty, readFrom, ioThreads, workerThreads)
+
+  def from(redisUris: Array[String],
+           sslConfiguration: SSLConfiguration,
+           readFrom: ReadFrom): MasterReplicaRedisConfiguration =
+    from(redisUris, useSSL = true, Some(sslConfiguration), readFrom, None, None)
+
+  def from(redisUris: Array[String],
+           useSSL: Boolean,
+           mayBeSSLConfiguration: Option[SSLConfiguration],
            readFrom: ReadFrom,
            ioThreads: Option[Int] = None,
            workerThreads: Option[Int] = None): MasterReplicaRedisConfiguration = {
     Preconditions.checkArgument(redisUris != null && redisUris.length > 0)
     MasterReplicaRedisConfiguration(RedisUris.from(redisUris),
+      useSSL,
+      mayBeSSLConfiguration,
       readFrom,
       ioThreads, workerThreads)
   }
 }
 
-case class MasterReplicaRedisConfiguration(redisURI: RedisUris, readFrom: ReadFrom, ioThreads: Option[Int], workerThreads: Option[Int]) extends RedisConfiguration {
+case class MasterReplicaRedisConfiguration(redisURI: RedisUris, useSSL: Boolean, mayBeSSLConfiguration: Option[SSLConfiguration], readFrom: ReadFrom, ioThreads: Option[Int], workerThreads: Option[Int]) extends RedisConfiguration {
   override def asString: String = MoreObjects.toStringHelper(this)
     .add("topology", MASTER_REPLICA_TOPOLOGY)
     .add("redisURI", redisURI.value.map(_.toString).mkString(";"))
@@ -168,25 +201,65 @@ case class ClusterRedisConfiguration(redisURI: RedisUris, ioThreads: Option[Int]
 object SentinelRedisConfiguration {
   def from(config: Configuration): SentinelRedisConfiguration = from(
     config.getStringArray(REDIS_URL_PROPERTY_NAME).mkString(","),
+    config.getBoolean(REDIS_USE_SSL, REDIS_USE_SSL_DEFAULT_VALUE),
+    SSLConfiguration.from(config),
     Option(config.getString(REDIS_READ_FROM_PROPERTY_NAME, null)).map(ReadFrom.valueOf).getOrElse(REDIS_READ_FROM_DEFAULT_VALUE),
     Option(config.getString(REDIS_SENTINEL_PASSWORD, null)),
     RedisConfiguration.redisIoThreadsFrom(config),
     RedisConfiguration.redisWorkerThreadsFrom(config))
 
-  def from(redisUri: String, readFrom: ReadFrom, sentinelPassword: String): SentinelRedisConfiguration = from(redisUri, readFrom, Option.apply(sentinelPassword), None, None)
+  def from(redisUri: String, readFrom: ReadFrom, sentinelPassword: String): SentinelRedisConfiguration =
+    from(redisUri, useSSL = false, None, readFrom, Option.apply(sentinelPassword), None, None)
 
-  def from(redisUriString: String, readFrom: ReadFrom, maybeSentinelPassword: Option[String], ioThreads: Option[Int] = None, workerThreads: Option[Int] = None): SentinelRedisConfiguration = {
+  def from(redisUri: String, sslConfiguration: SSLConfiguration, readFrom: ReadFrom): SentinelRedisConfiguration =
+    from(redisUri, useSSL = true, Some(sslConfiguration), readFrom, None, None, None)
+
+  def from(redisUriString: String,
+           useSSL: Boolean,
+           mayBeSSLConfiguration: Option[SSLConfiguration],
+           readFrom: ReadFrom,
+           maybeSentinelPassword: Option[String],
+           ioThreads: Option[Int] = None,
+           workerThreads: Option[Int] = None): SentinelRedisConfiguration = {
     val redisURI = RedisURI.create(redisUriString)
     maybeSentinelPassword.foreach(password => redisURI.getSentinels.forEach(uri => uri.setCredentialsProvider(RedisCredentialsProvider.from(() => RedisCredentials.just("", password)))))
-    SentinelRedisConfiguration(redisURI, readFrom, ioThreads, workerThreads)
+    SentinelRedisConfiguration(redisURI, useSSL, mayBeSSLConfiguration, readFrom, ioThreads, workerThreads)
   }
 }
 
-case class SentinelRedisConfiguration(redisURI: RedisURI, readFrom: ReadFrom, ioThreads: Option[Int], workerThreads: Option[Int]) extends RedisConfiguration {
+case class SentinelRedisConfiguration(redisURI: RedisURI, useSSL: Boolean, mayBeSSLConfiguration: Option[SSLConfiguration], readFrom: ReadFrom, ioThreads: Option[Int], workerThreads: Option[Int]) extends RedisConfiguration {
   override def asString: String = MoreObjects.toStringHelper(this)
     .add("topology", SENTINEL_TOPOLOGY)
     .add("redisURI", redisURI.toString)
     .add("redis.ioThreads", ioThreads)
     .add("redis.workerThreads", workerThreads)
     .toString
-}
+}
+
+object SSLConfiguration {
+  def from(config: Configuration): Option[SSLConfiguration] = {
+    if (config.getBoolean(REDIS_USE_SSL, REDIS_USE_SSL_DEFAULT_VALUE)) {
+      val ignoreCertificateCheck = config.getBoolean(REDIS_IGNORE_CERTIFICATE_CHECK, REDIS_IGNORE_CERTIFICATE_CHECK_DEFAULT_VALUE)
+      val maybeKeyStore = {
+        if (ignoreCertificateCheck) {
+          None
+        } else {
+          Some(KeyStore(config.getString(REDIS_KEY_STORE_FILE_PATH_PROPERTY_NAME, KEY_STORE_FILE_PATH_DEFAULT_VALUE),
+            config.getString(REDIS_KEY_STORE_PASSWORD_PROPERTY_NAME, KEY_STORE_PASSWORD_DEFAULT_VALUE)))
+        }
+      }
+      Some(SSLConfiguration(ignoreCertificateCheck, maybeKeyStore))
+    } else {
+      None
+    }
+  }
+
+  def from(): SSLConfiguration = SSLConfiguration(true, None)
+
+  def from(keyStoreFilePath: String, keyStorePassword: String): SSLConfiguration =
+    SSLConfiguration(false, Some(KeyStore(keyStoreFilePath, keyStorePassword)))
+}
+
+case class SSLConfiguration(ignoreCertificateCheck: Boolean, maybeKeyStore: Option[KeyStore])
+
+case class KeyStore(keyStoreFilePath: String, keyStorePassword: String)