Add support for setting TLS context in MongoDB

docs/src/main/sphinx/connector/mongodb.rst

@@ -60,7 +60,11 @@ Property name                              Description
 ``mongodb.max-connection-idle-time``       The maximum idle time of a pooled connection
 ``mongodb.connection-timeout``             The socket connect timeout
 ``mongodb.socket-timeout``                 The socket timeout
-``mongodb.ssl.enabled``                    Use TLS/SSL for connections to mongod/mongos
+``mongodb.tls.enabled``                    Use TLS/SSL for connections to mongod/mongos
+``mongodb.tls.keystore-path``              Path to the PEM or JKS key store
+``mongodb.tls.truststore-path``            Path to the PEM or JKS trust store
+``mongodb.tls.keystore-password``          Password for the key store
+``mongodb.tls.truststore-password``        Password for the trust store
 ``mongodb.read-preference``                The read preference
 ``mongodb.write-concern``                  The write concern
 ``mongodb.required-replica-set``           The required replica set name
@@ -161,13 +165,41 @@ The socket timeout in milliseconds. It is used for I/O socket read and write ope
 
 This property is optional; the default is ``0`` and means no timeout.
 
-``mongodb.ssl.enabled``
+``mongodb.tls.enabled``
 ^^^^^^^^^^^^^^^^^^^^^^^^
 
-This flag enables SSL connections to MongoDB servers.
+This flag enables TLS connections to MongoDB servers.
 
 This property is optional; the default is ``false``.
 
+``mongodb.tls.keystore-path``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+The path to the PEM or JKS key store. This file must be readable by the operating system user running Trino.
+
+This property is optional.
+
+``mongodb.tls.truststore-path``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+The path to PEM or JKS trust store. This file must be readable by the operating system user running Trino.
+
+This property is optional.
+
+``mongodb.tls.keystore-password``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+The key password for the key store specified by ``mongodb.tls.keystore-path``.
+
+This property is optional.
+
+``mongodb.tls.truststore-password``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+The key password for the trust store specified by ``mongodb.tls.truststore-path``.
+
+This property is optional.
+
 ``mongodb.read-preference``
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^
 

plugin/trino-mongodb/src/main/java/io/trino/plugin/mongodb/MongoClientConfig.java

@@ -20,12 +20,15 @@
 import io.airlift.configuration.Config;
 import io.airlift.configuration.ConfigSecuritySensitive;
 import io.airlift.configuration.DefunctConfig;
+import io.airlift.configuration.LegacyConfig;
+import io.airlift.configuration.validation.FileExists;
 
 import javax.validation.constraints.AssertTrue;
 import javax.validation.constraints.Min;
 import javax.validation.constraints.NotNull;
 import javax.validation.constraints.Pattern;
 
+import java.io.File;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Optional;
@@ -51,7 +54,11 @@ public class MongoClientConfig
     private int connectionTimeout = 10_000;
     private int socketTimeout;
     private int maxConnectionIdleTime;
-    private boolean sslEnabled;
+    private boolean tlsEnabled;
+    private File keystorePath;
+    private String keystorePassword;
+    private File truststorePath;
+    private String truststorePassword;
 
     // query configurations
     private int cursorBatchSize; // use driver default
@@ -71,6 +78,15 @@ public boolean isConnectionPropertyValid()
         return seeds.isEmpty() || connectionUrl.isEmpty();
     }
 
+    @AssertTrue(message = "'mongodb.tls.keystore-path', 'mongodb.tls.keystore-password', 'mongodb.tls.truststore-path' and 'mongodb.tls.truststore-password' must be empty when TLS is disabled")
+    public boolean isValidTlsConfig()
+    {
+        if (!tlsEnabled) {
+            return keystorePath == null && keystorePassword == null && truststorePath == null && truststorePassword == null;
+        }
+        return true;
+    }
+
     @NotNull
     public String getSchemaCollection()
     {
@@ -310,15 +326,66 @@ public MongoClientConfig setImplicitRowFieldPrefix(String implicitRowFieldPrefix
         return this;
     }
 
-    public boolean getSslEnabled()
+    public boolean getTlsEnabled()
+    {
+        return this.tlsEnabled;
+    }
+
+    @Config("mongodb.tls.enabled")
+    @LegacyConfig("mongodb.ssl.enabled")
+    public MongoClientConfig setTlsEnabled(boolean tlsEnabled)
+    {
+        this.tlsEnabled = tlsEnabled;
+        return this;
+    }
+
+    public Optional<@FileExists File> getKeystorePath()
+    {
+        return Optional.ofNullable(keystorePath);
+    }
+
+    @Config("mongodb.tls.keystore-path")
+    public MongoClientConfig setKeystorePath(File keystorePath)
+    {
+        this.keystorePath = keystorePath;
+        return this;
+    }
+
+    public Optional<String> getKeystorePassword()
+    {
+        return Optional.ofNullable(keystorePassword);
+    }
+
+    @Config("mongodb.tls.keystore-password")
+    @ConfigSecuritySensitive
+    public MongoClientConfig setKeystorePassword(String keystorePassword)
+    {
+        this.keystorePassword = keystorePassword;
+        return this;
+    }
+
+    public Optional<@FileExists File> getTruststorePath()
     {
-        return this.sslEnabled;
+        return Optional.ofNullable(truststorePath);
     }
 
-    @Config("mongodb.ssl.enabled")
-    public MongoClientConfig setSslEnabled(boolean sslEnabled)
+    @Config("mongodb.tls.truststore-path")
+    public MongoClientConfig setTruststorePath(File truststorePath)
+    {
+        this.truststorePath = truststorePath;
+        return this;
+    }
+
+    public Optional<String> getTruststorePassword()
+    {
+        return Optional.ofNullable(truststorePassword);
+    }
+
+    @Config("mongodb.tls.truststore-password")
+    @ConfigSecuritySensitive
+    public MongoClientConfig setTruststorePassword(String truststorePassword)
     {
-        this.sslEnabled = sslEnabled;
+        this.truststorePassword = truststorePassword;
         return this;
     }
 

plugin/trino-mongodb/src/main/java/io/trino/plugin/mongodb/MongoClientModule.java

@@ -22,13 +22,22 @@
 import com.mongodb.client.MongoClient;
 import com.mongodb.client.MongoClients;
 import io.trino.plugin.mongodb.ptf.Query;
+import io.trino.spi.TrinoException;
 import io.trino.spi.ptf.ConnectorTableFunction;
 import io.trino.spi.type.TypeManager;
 
 import javax.inject.Singleton;
+import javax.net.ssl.SSLContext;
+
+import java.io.File;
+import java.io.IOException;
+import java.security.GeneralSecurityException;
+import java.util.Optional;
 
 import static com.google.inject.multibindings.Multibinder.newSetBinder;
 import static io.airlift.configuration.ConfigBinder.configBinder;
+import static io.trino.plugin.base.ssl.SslUtils.createSSLContext;
+import static io.trino.spi.StandardErrorCode.GENERIC_INTERNAL_ERROR;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 
 public class MongoClientModule
@@ -60,12 +69,18 @@ public static MongoSession createMongoSession(TypeManager typeManager, MongoClie
                         .maxSize(config.getConnectionsPerHost()))
                 .applyToSocketSettings(builder -> builder
                         .connectTimeout(config.getConnectionTimeout(), MILLISECONDS)
-                        .readTimeout(config.getSocketTimeout(), MILLISECONDS))
-                .applyToSslSettings(builder -> builder.enabled(config.getSslEnabled()));
+                        .readTimeout(config.getSocketTimeout(), MILLISECONDS));
 
         if (config.getRequiredReplicaSetName() != null) {
             options.applyToClusterSettings(builder -> builder.requiredReplicaSetName(config.getRequiredReplicaSetName()));
         }
+        if (config.getTlsEnabled()) {
+            options.applyToSslSettings(builder -> {
+                builder.enabled(true);
+                buildSslContext(config.getKeystorePath(), config.getKeystorePassword(), config.getTruststorePath(), config.getTruststorePassword())
+                        .ifPresent(builder::context);
+            });
+        }
 
         if (config.getConnectionUrl().isPresent()) {
             options.applyConnectionString(new ConnectionString(config.getConnectionUrl().get()));
@@ -84,4 +99,23 @@ public static MongoSession createMongoSession(TypeManager typeManager, MongoClie
                 client,
                 config);
     }
+
+    // TODO https://github.com/trinodb/trino/issues/15247 Add test for x.509 certificates
+    private static Optional<SSLContext> buildSslContext(
+            Optional<File> keystorePath,
+            Optional<String> keystorePassword,
+            Optional<File> truststorePath,
+            Optional<String> truststorePassword)
+    {
+        if (keystorePath.isEmpty() && truststorePath.isEmpty()) {
+            return Optional.empty();
+        }
+
+        try {
+            return Optional.of(createSSLContext(keystorePath, keystorePassword, truststorePath, truststorePassword));
+        }
+        catch (GeneralSecurityException | IOException e) {
+            throw new TrinoException(GENERIC_INTERNAL_ERROR, e);
+        }
+    }
 }

plugin/trino-mongodb/src/test/java/io/trino/plugin/mongodb/TestMongoClientConfig.java

@@ -18,10 +18,15 @@
 import io.airlift.configuration.ConfigurationFactory;
 import org.testng.annotations.Test;
 
+import javax.validation.constraints.AssertTrue;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.util.Map;
 
 import static io.airlift.configuration.testing.ConfigAssertions.assertRecordedDefaults;
 import static io.airlift.configuration.testing.ConfigAssertions.recordDefaults;
+import static io.airlift.testing.ValidationAssertions.assertFailsValidation;
 import static org.testng.Assert.assertEquals;
 
 public class TestMongoClientConfig
@@ -40,7 +45,11 @@ public void testDefaults()
                 .setMaxWaitTime(120_000)
                 .setConnectionTimeout(10_000)
                 .setSocketTimeout(0)
-                .setSslEnabled(false)
+                .setTlsEnabled(false)
+                .setKeystorePath(null)
+                .setKeystorePassword(null)
+                .setTruststorePath(null)
+                .setTruststorePassword(null)
                 .setMaxConnectionIdleTime(0)
                 .setCursorBatchSize(0)
                 .setReadPreference(ReadPreferenceType.PRIMARY)
@@ -51,7 +60,11 @@ public void testDefaults()
 
     @Test
     public void testExplicitPropertyMappings()
+            throws Exception
     {
+        Path keystoreFile = Files.createTempFile(null, null);
+        Path truststoreFile = Files.createTempFile(null, null);
+
         Map<String, String> properties = ImmutableMap.<String, String>builder()
                 .put("mongodb.schema-collection", "_my_schema")
                 .put("mongodb.case-insensitive-name-matching", "true")
@@ -63,7 +76,11 @@ public void testExplicitPropertyMappings()
                 .put("mongodb.max-wait-time", "120001")
                 .put("mongodb.connection-timeout", "9999")
                 .put("mongodb.socket-timeout", "1")
-                .put("mongodb.ssl.enabled", "true")
+                .put("mongodb.tls.enabled", "true")
+                .put("mongodb.tls.keystore-path", keystoreFile.toString())
+                .put("mongodb.tls.keystore-password", "keystore-password")
+                .put("mongodb.tls.truststore-path", truststoreFile.toString())
+                .put("mongodb.tls.truststore-password", "truststore-password")
                 .put("mongodb.max-connection-idle-time", "180000")
                 .put("mongodb.cursor-batch-size", "1")
                 .put("mongodb.read-preference", "NEAREST")
@@ -86,7 +103,11 @@ public void testExplicitPropertyMappings()
                 .setMaxWaitTime(120_001)
                 .setConnectionTimeout(9_999)
                 .setSocketTimeout(1)
-                .setSslEnabled(true)
+                .setTlsEnabled(true)
+                .setKeystorePath(keystoreFile.toFile())
+                .setKeystorePassword("keystore-password")
+                .setTruststorePath(truststoreFile.toFile())
+                .setTruststorePassword("truststore-password")
                 .setMaxConnectionIdleTime(180_000)
                 .setCursorBatchSize(1)
                 .setReadPreference(ReadPreferenceType.NEAREST)
@@ -104,7 +125,11 @@ public void testExplicitPropertyMappings()
         assertEquals(config.getMaxWaitTime(), expected.getMaxWaitTime());
         assertEquals(config.getConnectionTimeout(), expected.getConnectionTimeout());
         assertEquals(config.getSocketTimeout(), expected.getSocketTimeout());
-        assertEquals(config.getSslEnabled(), expected.getSslEnabled());
+        assertEquals(config.getTlsEnabled(), expected.getTlsEnabled());
+        assertEquals(config.getKeystorePath(), expected.getKeystorePath());
+        assertEquals(config.getKeystorePassword(), expected.getKeystorePassword());
+        assertEquals(config.getTruststorePath(), expected.getTruststorePath());
+        assertEquals(config.getTruststorePassword(), expected.getTruststorePassword());
         assertEquals(config.getMaxConnectionIdleTime(), expected.getMaxConnectionIdleTime());
         assertEquals(config.getCursorBatchSize(), expected.getCursorBatchSize());
         assertEquals(config.getReadPreference(), expected.getReadPreference());
@@ -123,4 +148,26 @@ public void testSpecialCharacterCredential()
         MongoCredential expected = MongoCredential.createCredential("username", "database", "P@ss:w0rd".toCharArray());
         assertEquals(credential, expected);
     }
+
+    @Test
+    public void testValidation()
+            throws Exception
+    {
+        Path keystoreFile = Files.createTempFile(null, null);
+        Path truststoreFile = Files.createTempFile(null, null);
+
+        assertFailsTlsValidation(new MongoClientConfig().setKeystorePath(keystoreFile.toFile()));
+        assertFailsTlsValidation(new MongoClientConfig().setKeystorePassword("keystore password"));
+        assertFailsTlsValidation(new MongoClientConfig().setTruststorePath(truststoreFile.toFile()));
+        assertFailsTlsValidation(new MongoClientConfig().setTruststorePassword("truststore password"));
+    }
+
+    private static void assertFailsTlsValidation(MongoClientConfig config)
+    {
+        assertFailsValidation(
+                config,
+                "validTlsConfig",
+                "'mongodb.tls.keystore-path', 'mongodb.tls.keystore-password', 'mongodb.tls.truststore-path' and 'mongodb.tls.truststore-password' must be empty when TLS is disabled",
+                AssertTrue.class);
+    }
 }

plugin/trino-mongodb/src/test/java/io/trino/plugin/mongodb/TestMongoPlugin.java

@@ -18,8 +18,6 @@
 import io.trino.spi.connector.ConnectorFactory;
 import io.trino.spi.type.Type;
 import io.trino.testing.TestingConnectorContext;
-import org.testng.annotations.AfterClass;
-import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
 
 import static com.google.common.collect.Iterables.getOnlyElement;
@@ -28,33 +26,17 @@
 
 public class TestMongoPlugin
 {
-    private MongoServer server;
-    private String connectionString;
-
-    @BeforeClass
-    public void start()
-    {
-        server = new MongoServer();
-        connectionString = server.getConnectionString().toString();
-    }
-
     @Test
     public void testCreateConnector()
     {
         MongoPlugin plugin = new MongoPlugin();
 
         ConnectorFactory factory = getOnlyElement(plugin.getConnectorFactories());
-        Connector connector = factory.create("test", ImmutableMap.of("mongodb.connection-url", connectionString), new TestingConnectorContext());
+        Connector connector = factory.create("test", ImmutableMap.of("mongodb.connection-url", "mongodb://localhost:27017"), new TestingConnectorContext());
 
         Type type = getOnlyElement(plugin.getTypes());
         assertEquals(type, OBJECT_ID);
 
         connector.shutdown();
     }
-
-    @AfterClass(alwaysRun = true)
-    public void destroy()
-    {
-        server.close();
-    }
 }