Add support for TLS in Redis Connector

Author: nishithakbhaskaran

pom.xml

@@ -2249,7 +2249,7 @@
             <dependency>
                 <groupId>redis.clients</groupId>
                 <artifactId>jedis</artifactId>
-                <version>2.6.2</version>
+                <version>3.8.0</version>
             </dependency>
 
             <dependency>

presto-docs/src/main/sphinx/connector/redis.rst

@@ -51,6 +51,9 @@ Property Name                       Description
 ``redis.hide-internal-columns``     Controls whether internal columns are part of the table schema or not
 ``redis.database-index``            Redis database index
 ``redis.password``                  Redis server password
+``redis.user``                      Redis server username
+``redis.tls.enabled``               Whether TLS security is enabled (defaults to ``false``)
+``redis.tls.truststore-path``       Path to the TLS certificate file
 =================================   ==============================================================
 
 ``redis.table-names``
@@ -130,19 +133,39 @@ show up in ``DESCRIBE <table-name>`` or ``SELECT *``.
 This property is optional; the default is ``true``.
 
 ``redis.database-index``
-^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+^^^^^^^^^^^^^^^^^^^^^^^^
 
 The Redis database to query.
 
 This property is optional; the default is ``0``.
 
 ``redis.password``
-^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+^^^^^^^^^^^^^^^^^^
 
 The password for password-protected Redis server.
 
 This property is optional; the default is ``null``.
 
+``redis.user``
+^^^^^^^^^^^^^^
+
+Redis server username.
+
+This property is required; there is no default.
+
+``redis.tls.enabled``
+^^^^^^^^^^^^^^^^^^^^^
+
+Enable or disable TLS security.
+
+This property is optional; default is ``false``.
+
+``redis.tls.truststore-path``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+Path to the TLS certificate file.
+
+This property is required if ``redis.tls.enabled`` is set to ``true``.
 
 Internal Columns
 ----------------

presto-main/etc/catalog/redis.properties

@@ -0,0 +1,7 @@
+connector.name=redis
+redis.table-names=schema1.table1,schema1.table2
+redis.user=user
+redis.password=password
+redis.tls.enabled=true
+redis.tls.truststore-path=<path>
+redis.table-description-dir=<path>

presto-main/etc/config.properties

@@ -51,7 +51,8 @@ plugin.bundles=\
   ../presto-node-ttl-fetchers/pom.xml,\
   ../presto-hive-function-namespace/pom.xml,\
   ../presto-delta/pom.xml,\
-  ../presto-hudi/pom.xml
+  ../presto-hudi/pom.xml,\
+  ../presto-redis/pom.xml
 
 presto.version=testversion
 node-scheduler.include-coordinator=true

presto-redis/src/main/java/com/facebook/presto/redis/RedisConnectorConfig.java

@@ -58,6 +58,11 @@ public class RedisConnectorConfig
      */
     private String redisPassword;
 
+    /**
+     * user for Redis server
+     */
+    private String redisUser;
+
     /**
      * Timeout to connect to Redis.
      */
@@ -88,6 +93,9 @@ public class RedisConnectorConfig
      */
     private boolean keyPrefixSchemaTable;
 
+    private boolean tlsEnabled;
+    private File truststorePath;
+
     @NotNull
     public File getTableDescriptionDir()
     {
@@ -194,6 +202,11 @@ public String getRedisPassword()
         return redisPassword;
     }
 
+    public String getRedisUser()
+    {
+        return redisUser;
+    }
+
     @Config("redis.password")
     @ConfigSecuritySensitive
     public RedisConnectorConfig setRedisPassword(String redisPassword)
@@ -202,6 +215,14 @@ public RedisConnectorConfig setRedisPassword(String redisPassword)
         return this;
     }
 
+    @Config("redis.user")
+    @ConfigSecuritySensitive
+    public RedisConnectorConfig setRedisUser(String redisUser)
+    {
+        this.redisUser = redisUser;
+        return this;
+    }
+
     public boolean isHideInternalColumns()
     {
         return hideInternalColumns;
@@ -236,4 +257,28 @@ private static HostAddress toHostAddress(String value)
     {
         return HostAddress.fromString(value).withDefaultPort(REDIS_DEFAULT_PORT);
     }
+
+    public boolean isTlsEnabled()
+    {
+        return tlsEnabled;
+    }
+
+    @Config("redis.tls.enabled")
+    public RedisConnectorConfig setTlsEnabled(boolean tlsEnabled)
+    {
+        this.tlsEnabled = tlsEnabled;
+        return this;
+    }
+
+    public File getTruststorePath()
+    {
+        return truststorePath;
+    }
+
+    @Config("redis.tls.truststore-path")
+    public RedisConnectorConfig setTruststorePath(File truststorePath)
+    {
+        this.truststorePath = truststorePath;
+        return this;
+    }
 }

presto-redis/src/main/java/com/facebook/presto/redis/RedisJedisManager.java

@@ -24,11 +24,24 @@
 
 import javax.annotation.PreDestroy;
 import javax.inject.Inject;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManagerFactory;
 
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.security.KeyManagementException;
+import java.security.KeyStore;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.security.cert.CertificateException;
+import java.security.cert.CertificateFactory;
+import java.security.cert.X509Certificate;
 import java.util.Map;
 
 import static java.lang.Math.toIntExact;
 import static java.util.Objects.requireNonNull;
+import static javax.net.ssl.TrustManagerFactory.getDefaultAlgorithm;
 
 /**
  * Manages connections to the Redis nodes
@@ -37,6 +50,12 @@ public class RedisJedisManager
 {
     private static final Logger log = Logger.get(RedisJedisManager.class);
 
+    private static final String TLS_PROTOCOL = "TLS";
+    private static final int JEDIS_CONN_TIMEOUT = 2000;
+    private static final int JEDIS_SO_TIMEOUT = 2000;
+    private static final int JEDIS_MIN_IDLE_CONNECTIONS = 1;
+    private static final int JEDIS_MAX_IDLE_CONNECTIONS = 5;
+
     private final LoadingCache<HostAddress, JedisPool> jedisPoolCache;
 
     private final RedisConnectorConfig redisConnectorConfig;
@@ -48,8 +67,8 @@ public class RedisJedisManager
             NodeManager nodeManager)
     {
         this.redisConnectorConfig = requireNonNull(redisConnectorConfig, "redisConfig is null");
-        this.jedisPoolCache = CacheBuilder.newBuilder().build(CacheLoader.from(this::createConsumer));
-        this.jedisPoolConfig = new JedisPoolConfig();
+        this.jedisPoolCache = CacheBuilder.newBuilder().build(CacheLoader.from(this::createJedisPool));
+        this.jedisPoolConfig = createJedisPoolConfig();
     }
 
     @PreDestroy
@@ -76,14 +95,97 @@ public JedisPool getJedisPool(HostAddress host)
         return jedisPoolCache.getUnchecked(host);
     }
 
-    private JedisPool createConsumer(HostAddress host)
+    /**
+     * Creates a new JedisPool for the specified host.
+     * Chooses between TLS or non-TLS configuration based on redisConnectorConfig.
+     */
+    private JedisPool createJedisPool(HostAddress host)
+    {
+        boolean isTlsEnabled = redisConnectorConfig.isTlsEnabled();
+        SSLContext sslContext = null;
+
+        if (isTlsEnabled) {
+            KeyStore trustStore = loadTrustStore();
+            sslContext = createSslContext(trustStore);
+        }
+
+        return buildJedisPool(host, isTlsEnabled, sslContext);
+    }
+    /**
+     * Creates SSLContext initialized with the given truststore.
+     */
+    private SSLContext createSslContext(KeyStore trustStore)
+    {
+        if (trustStore == null) {
+            throw new IllegalStateException("Truststore must not be null for TLS connections");
+        }
+
+        try {
+            TrustManagerFactory tmf = TrustManagerFactory.getInstance(getDefaultAlgorithm());
+            tmf.init(trustStore);
+
+            SSLContext sslContext = SSLContext.getInstance(TLS_PROTOCOL);
+            sslContext.init(null, tmf.getTrustManagers(), null);
+
+            return sslContext;
+        }
+        catch (NoSuchAlgorithmException | KeyStoreException | KeyManagementException e) {
+            throw new RuntimeException("Failed to initialize SSLContext", e);
+        }
+    }
+
+    private JedisPoolConfig createJedisPoolConfig()
+    {
+        JedisPoolConfig config = new JedisPoolConfig();
+        config.setMinIdle(JEDIS_MIN_IDLE_CONNECTIONS);
+        config.setMaxTotal(JEDIS_MAX_IDLE_CONNECTIONS);
+        return config;
+    }
+    /**
+     * Loads the truststore containing Redis server certificate.
+     * Returns null if truststore path is not configured.
+     */
+    private KeyStore loadTrustStore()
     {
-        log.info("Creating new JedisPool for %s", host);
-        return new JedisPool(jedisPoolConfig,
+        if (redisConnectorConfig.getTruststorePath() == null) {
+            log.info("No truststore path configured, skipping TLS truststore loading");
+            return null;
+        }
+
+        try {
+            KeyStore trustStore = KeyStore.getInstance(KeyStore.getDefaultType());
+            try (InputStream in = Files.newInputStream(redisConnectorConfig.getTruststorePath().toPath())) {
+                trustStore.load(null, null);
+                CertificateFactory cf = CertificateFactory.getInstance("X.509");
+                X509Certificate cert = (X509Certificate) cf.generateCertificate(in);
+                trustStore.setCertificateEntry("redis-server", cert);
+            }
+            log.info("Loaded truststore from %s", redisConnectorConfig.getTruststorePath());
+            return trustStore;
+        }
+        catch (KeyStoreException | IOException | CertificateException | NoSuchAlgorithmException e) {
+            throw new RuntimeException("Failed to load truststore", e);
+        }
+    }
+
+    private JedisPool buildJedisPool(HostAddress host, boolean useTls, SSLContext sslContext)
+    {
+        log.info("Creating new %s JedisPool for %s", useTls ? "TLS" : "non-TLS", host);
+
+        return new JedisPool(
+                jedisPoolConfig,
                 host.getHostText(),
                 host.getPort(),
                 toIntExact(redisConnectorConfig.getRedisConnectTimeout().toMillis()),
+                JEDIS_SO_TIMEOUT,
+                JEDIS_CONN_TIMEOUT,
+                redisConnectorConfig.getRedisUser(),
                 redisConnectorConfig.getRedisPassword(),
-                redisConnectorConfig.getRedisDataBaseIndex());
+                redisConnectorConfig.getRedisDataBaseIndex(),
+                null,
+                useTls,
+                useTls ? sslContext.getSocketFactory() : null,
+                null,
+                null);
     }
 }

presto-redis/src/main/java/com/facebook/presto/redis/RedisRecordCursor.java

@@ -118,7 +118,7 @@ public boolean hasUnscannedData()
         // no more keys are unscanned when
         // when redis scan command
         // returns 0 string cursor
-        return (!redisCursor.getStringCursor().equals("0"));
+        return (!redisCursor.getCursor().equals("0"));
     }
 
     @Override
@@ -299,7 +299,7 @@ private boolean fetchKeys()
                 case STRING: {
                     String cursor = SCAN_POINTER_START;
                     if (redisCursor != null) {
-                        cursor = redisCursor.getStringCursor();
+                        cursor = redisCursor.getCursor();
                     }
 
                     log.debug("Scanning new Redis keys from cursor %s . %d values read so far", cursor, totalValues);

presto-redis/src/test/java/com/facebook/presto/redis/TestRedisConnectorConfig.java

@@ -26,6 +26,9 @@ public class TestRedisConnectorConfig
     public void testDefaults()
     {
         ConfigAssertions.assertRecordedDefaults(ConfigAssertions.recordDefaults(RedisConnectorConfig.class)
+                .setRedisUser(null)
+                .setTlsEnabled(false)
+                .setTruststorePath(null)
                 .setNodes("")
                 .setDefaultSchema("default")
                 .setTableNames("")
@@ -53,6 +56,9 @@ public void testExplicitPropertyMappings()
                 .put("redis.hide-internal-columns", "false")
                 .put("redis.connect-timeout", "10s")
                 .put("redis.database-index", "5")
+                .put("redis.user", "nobody")
+                .put("redis.tls.enabled", "true")
+                .put("redis.tls.truststore-path", "/dev/null")
                 .put("redis.password", "secret")
                 .build();
 
@@ -65,6 +71,9 @@ public void testExplicitPropertyMappings()
                 .setRedisScanCount(20)
                 .setRedisConnectTimeout("10s")
                 .setRedisDataBaseIndex(5)
+                .setRedisUser("nobody")
+                .setTlsEnabled(true)
+                .setTruststorePath(new File("/dev/null"))
                 .setRedisPassword("secret")
                 .setRedisKeyDelimiter(",")
                 .setKeyPrefixSchemaTable(true);