feat: support redis TLS (#19666)

* docs: druid.cache.enableTls

* feat: TLS for redis

* docs: update description

* feat: skip tls hostname verification

* fix: avoid using database from user input when cluster mode

* test: coverage

Author: proost

docs/development/extensions-contrib/redis-cache.md

@@ -84,6 +84,8 @@ Except for the properties above, there are some extra properties which can be cu
 | Properties |Description|Default|Required|
 |--------------------|-----------|-------|--------|
 |`druid.cache.password`| Password to access redis server/cluster | None |no|
+|`druid.cache.enableTls`| Whether to connect to the redis server/cluster over TLS | false |no|
+|`druid.cache.skipTlsHostnameVerification`| When TLS is enabled, skip verifying that the server certificate matches the redis hostname | false |no|
 |`druid.cache.expiration`|Expiration for cache entries | P1D |no|
 |`druid.cache.timeout`|Timeout for connecting to Redis and reading entries from Redis|PT2S|no|
 |`druid.cache.maxTotalConnections`|Max total connections to Redis|8|no|

extensions-contrib/redis-cache/src/main/java/org/apache/druid/client/cache/RedisCacheConfig.java

@@ -144,6 +144,19 @@ public long getSeconds()
   @Min(0)
   private int database = Protocol.DEFAULT_DATABASE;
 
+  /**
+   * whether to connect to redis server/cluster over TLS.
+   */
+  @JsonProperty
+  private boolean enableTls = false;
+
+  /**
+   * whether to skip verifying that the server certificate matches the redis hostname when TLS is
+   * enabled. defaults to false (the hostname is verified)
+   */
+  @JsonProperty
+  private boolean skipTlsHostnameVerification = false;
+
   @JsonProperty
   private RedisClusterConfig cluster;
 
@@ -196,4 +209,14 @@ public int getDatabase()
   {
     return database;
   }
+
+  public boolean getEnableTls()
+  {
+    return enableTls;
+  }
+
+  public boolean getSkipTlsHostnameVerification()
+  {
+    return skipTlsHostnameVerification;
+  }
 }

extensions-contrib/redis-cache/src/main/java/org/apache/druid/client/cache/RedisCacheFactory.java

@@ -19,13 +19,18 @@
 
 package org.apache.druid.client.cache;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.druid.java.util.common.IAE;
 import redis.clients.jedis.ConnectionPoolConfig;
+import redis.clients.jedis.DefaultJedisClientConfig;
 import redis.clients.jedis.HostAndPort;
+import redis.clients.jedis.JedisClientConfig;
 import redis.clients.jedis.JedisCluster;
 import redis.clients.jedis.JedisPool;
 import redis.clients.jedis.JedisPoolConfig;
+import redis.clients.jedis.SslOptions;
+import redis.clients.jedis.SslVerifyMode;
 
 import java.util.Arrays;
 import java.util.Set;
@@ -65,24 +70,12 @@ public static Cache create(final RedisCacheConfig config)
       poolConfig.setMaxIdle(config.getMaxIdleConnections());
       poolConfig.setMinIdle(config.getMinIdleConnections());
 
-      JedisCluster cluster;
-      if (config.getPassword() != null) {
-        cluster = new JedisCluster(
-            nodes,
-            config.getTimeout().getMillisecondsAsInt(), //connection timeout
-            config.getTimeout().getMillisecondsAsInt(), //read timeout
-            config.getCluster().getMaxRedirection(),
-            config.getPassword().getPassword(),
-            poolConfig
-        );
-      } else {
-        cluster = new JedisCluster(
-            nodes,
-            config.getTimeout().getMillisecondsAsInt(), //connection timeout and read timeout
-            config.getCluster().getMaxRedirection(),
-            poolConfig
-        );
-      }
+      JedisCluster cluster = new JedisCluster(
+          nodes,
+          buildClientConfig(config, 0),
+          config.getCluster().getMaxRedirection(),
+          poolConfig
+      );
 
       return new RedisClusterCache(cluster, config);
 
@@ -100,15 +93,31 @@ public static Cache create(final RedisCacheConfig config)
       return new RedisStandaloneCache(
           new JedisPool(
               poolConfig,
-              config.getHost(),
-              config.getPort(),
-              config.getTimeout().getMillisecondsAsInt(), //connection timeout and read timeout
-              config.getPassword() == null ? null : config.getPassword().getPassword(),
-              config.getDatabase(),
-              null
+              new HostAndPort(config.getHost(), config.getPort()),
+              buildClientConfig(config, config.getDatabase())
           ),
           config
       );
     }
   }
+
+  @VisibleForTesting
+  static JedisClientConfig buildClientConfig(RedisCacheConfig config, int database)
+  {
+    DefaultJedisClientConfig.Builder builder = DefaultJedisClientConfig
+        .builder()
+        .connectionTimeoutMillis(config.getTimeout().getMillisecondsAsInt())
+        .socketTimeoutMillis(config.getTimeout().getMillisecondsAsInt())
+        .password(config.getPassword() == null ? null : config.getPassword().getPassword())
+        .database(database)
+        .ssl(config.getEnableTls());
+
+    if (config.getEnableTls()) {
+      SslVerifyMode verifyMode =
+          config.getSkipTlsHostnameVerification() ? SslVerifyMode.CA : SslVerifyMode.FULL;
+      builder.sslOptions(SslOptions.builder().sslVerifyMode(verifyMode).build());
+    }
+
+    return builder.build();
+  }
 }

extensions-contrib/redis-cache/src/test/java/org/apache/druid/client/cache/RedisCacheConfigTest.java

@@ -24,12 +24,15 @@
 import com.github.fppt.jedismock.RedisServer;
 import com.github.fppt.jedismock.server.ServiceOptions;
 import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.metadata.PasswordProvider;
 import org.hamcrest.Description;
 import org.hamcrest.Matcher;
 import org.junit.Assert;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
+import redis.clients.jedis.JedisClientConfig;
+import redis.clients.jedis.SslVerifyMode;
 
 import java.io.IOException;
 
@@ -232,4 +235,92 @@ public void testNoClusterAndHost() throws IOException
     ));
     RedisCacheFactory.create(fromJson);
   }
+
+  @Test
+  public void testEnableTls() throws IOException
+  {
+    ObjectMapper mapper = new ObjectMapper();
+
+    RedisCacheConfig defaultConfig = mapper.readValue(
+        "{\"host\": \"localhost\", \"port\": 6379}",
+        RedisCacheConfig.class
+    );
+    Assert.assertFalse(defaultConfig.getEnableTls());
+
+    RedisCacheConfig tlsConfig = mapper.readValue(
+        "{\"host\": \"localhost\", \"port\": 6379, \"enableTls\": true}",
+        RedisCacheConfig.class
+    );
+    Assert.assertTrue(tlsConfig.getEnableTls());
+  }
+
+  @Test
+  public void testSkipTlsHostnameVerification() throws IOException
+  {
+    ObjectMapper mapper = new ObjectMapper();
+
+    RedisCacheConfig defaultConfig = mapper.readValue(
+        "{\"host\": \"localhost\", \"port\": 6379}",
+        RedisCacheConfig.class
+    );
+    Assert.assertFalse(defaultConfig.getSkipTlsHostnameVerification());
+
+    RedisCacheConfig skipConfig = mapper.readValue(
+        "{\"host\": \"localhost\", \"port\": 6379, \"skipTlsHostnameVerification\": true}",
+        RedisCacheConfig.class
+    );
+    Assert.assertTrue(skipConfig.getSkipTlsHostnameVerification());
+  }
+
+  @Test
+  public void testBuildClientConfig()
+  {
+    // TLS disabled: not SSL, no SSL options, database argument passed through, null password.
+    JedisClientConfig plain = RedisCacheFactory.buildClientConfig(new RedisCacheConfig(), 3);
+    Assert.assertFalse(plain.isSsl());
+    Assert.assertNull(plain.getSslOptions());
+    Assert.assertEquals(3, plain.getDatabase());
+    Assert.assertNull(plain.getPassword());
+
+    // TLS enabled with hostname verification (default) maps to SslVerifyMode.FULL, and a
+    // non-null password is forwarded.
+    RedisCacheConfig verifyConfig = new RedisCacheConfig()
+    {
+      @Override
+      public boolean getEnableTls()
+      {
+        return true;
+      }
+
+      @Override
+      public PasswordProvider getPassword()
+      {
+        return () -> "secret";
+      }
+    };
+    JedisClientConfig verify = RedisCacheFactory.buildClientConfig(verifyConfig, 0);
+    Assert.assertTrue(verify.isSsl());
+    Assert.assertEquals(SslVerifyMode.FULL, verify.getSslOptions().getSslVerifyMode());
+    Assert.assertEquals("secret", verify.getPassword());
+
+    // skipTlsHostnameVerification maps to SslVerifyMode.CA (chain verified, hostname skipped).
+    RedisCacheConfig skipConfig = new RedisCacheConfig()
+    {
+      @Override
+      public boolean getEnableTls()
+      {
+        return true;
+      }
+
+      @Override
+      public boolean getSkipTlsHostnameVerification()
+      {
+        return true;
+      }
+    };
+    Assert.assertEquals(
+        SslVerifyMode.CA,
+        RedisCacheFactory.buildClientConfig(skipConfig, 0).getSslOptions().getSslVerifyMode()
+    );
+  }
 }