[Backport] Support for encryption of MySQL connections (#5122)

docs/content/development/extensions-core/mysql.md

@@ -53,3 +53,23 @@ Make sure to [include](../../operations/including-extensions.html) `mysql-metada
   packaged in a separate tarball that can be downloaded from [here](http://druid.io/downloads.html).
   You can also get it using [pull-deps](../../operations/pull-deps.html), or you can build
   it from source code; see [Build from Source](../build.html).
+
+
+## Encrypting MySQL connections
+  This extension provides support for encrypting MySQL connections. To get more information about encrypting MySQL connections using TLS/SSL in general, please refer to this [guide](https://dev.mysql.com/doc/refman/5.7/en/using-encrypted-connections.html).
+
+## Configuration
+
+|Property|Description|Default|Required|
+|--------|-----------|-------|--------|
+|`druid.metadata.mysql.ssl.useSSL`|Enable SSL|`false`|no|
+|`druid.metadata.mysql.ssl.clientCertificateKeyStoreUrl`|The file path URL to the client certificate key store.|none|no|
+|`druid.metadata.mysql.ssl.clientCertificateKeyStoreType`|The type of the key store where the client certificate is stored.|none|no|
+|`druid.metadata.mysql.ssl.clientCertificateKeyStorePassword`|The [Password Provider](../operations/password-provider.html) or String password for the client key store.|none|no|
+|`druid.metadata.mysql.ssl.verifyServerCertificate`|Enables server certificate verification.|false|no|
+|`druid.metadata.mysql.ssl.trustCertificateKeyStoreUrl`|The file path to the trusted root certificate key store.|Default trust store provided by MySQL|yes if `verifyServerCertificate` is set to true and a custom trust store is used|
+|`druid.metadata.mysql.ssl.trustCertificateKeyStoreType`|The type of the key store where trusted root certificates are stored.|JKS|yes if `verifyServerCertificate` is set to true and keystore type is not JKS|
+|`druid.metadata.mysql.ssl.trustCertificateKeyStorePassword`|The [Password Provider](../operations/password-provider.html) or String password for the trust store.|none|yes if `verifyServerCertificate` is set to true and password is not null|
+|`druid.metadata.mysql.ssl.enabledSSLCipherSuites`|Overrides the existing cipher suites with these cipher suites.|none|no|
+|`druid.metadata.mysql.ssl.enabledTLSProtocols`|Overrides the TLS protocols with these protocols.|none|no|
+

extensions-core/mysql-metadata-storage/src/main/java/io/druid/metadata/storage/mysql/MySQLConnector.java

@@ -19,6 +19,7 @@
 
 package io.druid.metadata.storage.mysql;
 
+import com.google.common.base.Joiner;
 import com.google.common.base.Supplier;
 import com.google.common.collect.ImmutableList;
 import com.google.inject.Inject;
@@ -36,6 +37,7 @@
 import org.skife.jdbi.v2.tweak.HandleCallback;
 import org.skife.jdbi.v2.util.BooleanMapper;
 
+import java.io.File;
 import java.sql.SQLException;
 
 public class MySQLConnector extends SQLMetadataConnector
@@ -48,7 +50,11 @@ public class MySQLConnector extends SQLMetadataConnector
   private final DBI dbi;
 
   @Inject
-  public MySQLConnector(Supplier<MetadataStorageConnectorConfig> config, Supplier<MetadataStorageTablesConfig> dbTables)
+  public MySQLConnector(
+      Supplier<MetadataStorageConnectorConfig> config,
+      Supplier<MetadataStorageTablesConfig> dbTables,
+      MySQLConnectorConfig connectorConfig
+  )
   {
     super(config, dbTables);
 
@@ -57,6 +63,68 @@ public MySQLConnector(Supplier<MetadataStorageConnectorConfig> config, Supplier<
     // so we need to help JDBC find the driver
     datasource.setDriverClassLoader(getClass().getClassLoader());
     datasource.setDriverClassName("com.mysql.jdbc.Driver");
+    datasource.addConnectionProperty("useSSL", String.valueOf(connectorConfig.isUseSSL()));
+    if (connectorConfig.isUseSSL()) {
+      log.info("SSL is enabled on this MySQL connection. ");
+
+      datasource.addConnectionProperty(
+          "verifyServerCertificate",
+          String.valueOf(connectorConfig.isVerifyServerCertificate())
+      );
+      if (connectorConfig.isVerifyServerCertificate()) {
+        log.info("Server certificate verification is enabled. ");
+
+        if (connectorConfig.getTrustCertificateKeyStoreUrl() != null) {
+          datasource.addConnectionProperty(
+              "trustCertificateKeyStoreUrl",
+              new File(connectorConfig.getTrustCertificateKeyStoreUrl()).toURI().toString()
+          );
+        }
+        if (connectorConfig.getTrustCertificateKeyStoreType() != null) {
+          datasource.addConnectionProperty(
+              "trustCertificateKeyStoreType",
+              connectorConfig.getTrustCertificateKeyStoreType()
+          );
+        }
+        if (connectorConfig.getTrustCertificateKeyStorePassword() == null) {
+          log.warn(
+              "Trust store password is empty. Ensure that the trust store has been configured with an empty password.");
+        } else {
+          datasource.addConnectionProperty(
+              "trustCertificateKeyStorePassword",
+              connectorConfig.getTrustCertificateKeyStorePassword()
+          );
+        }
+      }
+      if (connectorConfig.getClientCertificateKeyStoreUrl() != null) {
+        datasource.addConnectionProperty(
+            "clientCertificateKeyStoreUrl",
+            new File(connectorConfig.getClientCertificateKeyStoreUrl()).toURI().toString()
+        );
+      }
+      if (connectorConfig.getClientCertificateKeyStoreType() != null) {
+        datasource.addConnectionProperty(
+            "clientCertificateKeyStoreType",
+            connectorConfig.getClientCertificateKeyStoreType()
+        );
+      }
+      if (connectorConfig.getClientCertificateKeyStorePassword() != null) {
+        datasource.addConnectionProperty(
+            "clientCertificateKeyStorePassword",
+            connectorConfig.getClientCertificateKeyStorePassword()
+        );
+      }
+      Joiner joiner = Joiner.on(",").skipNulls();
+      if (connectorConfig.getEnabledSSLCipherSuites() != null) {
+        datasource.addConnectionProperty(
+            "enabledSSLCipherSuites",
+            joiner.join(connectorConfig.getEnabledSSLCipherSuites())
+        );
+      }
+      if (connectorConfig.getEnabledTLSProtocols() != null) {
+        datasource.addConnectionProperty("enabledTLSProtocols", joiner.join(connectorConfig.getEnabledTLSProtocols()));
+      }
+    }
 
     // use double-quotes for quoting columns, so we can write SQL that works with most databases
     datasource.setConnectionInitSqls(ImmutableList.of("SET sql_mode='ANSI_QUOTES'"));
@@ -97,9 +165,9 @@ public boolean tableExists(Handle handle, String tableName)
   {
     // ensure database defaults to utf8, otherwise bail
     boolean isUtf8 = handle
-                         .createQuery("SELECT @@character_set_database = 'utf8'")
-                         .map(BooleanMapper.FIRST)
-                         .first();
+        .createQuery("SELECT @@character_set_database = 'utf8'")
+        .map(BooleanMapper.FIRST)
+        .first();
 
     if (!isUtf8) {
       throw new ISE(

extensions-core/mysql-metadata-storage/src/main/java/io/druid/metadata/storage/mysql/MySQLConnectorConfig.java

@@ -0,0 +1,123 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets 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 io.druid.metadata.storage.mysql;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import io.druid.metadata.PasswordProvider;
+
+import java.util.List;
+
+public class MySQLConnectorConfig
+{
+  @JsonProperty
+  private boolean useSSL = false;
+
+  @JsonProperty
+  private String trustCertificateKeyStoreUrl;
+
+  @JsonProperty
+  private String trustCertificateKeyStoreType;
+
+  @JsonProperty("trustCertificateKeyStorePassword")
+  private PasswordProvider trustCertificateKeyStorePasswordProvider;
+
+  @JsonProperty
+  private String clientCertificateKeyStoreUrl;
+
+  @JsonProperty
+  private String clientCertificateKeyStoreType;
+
+  @JsonProperty("clientCertificateKeyStorePassword")
+  private PasswordProvider clientCertificateKeyStorePasswordProvider;
+
+  @JsonProperty
+  private List<String> enabledSSLCipherSuites;
+
+  @JsonProperty
+  private List<String> enabledTLSProtocols;
+
+  @JsonProperty
+  private boolean verifyServerCertificate = false;
+
+  public boolean isUseSSL()
+  {
+    return useSSL;
+  }
+
+  public String getTrustCertificateKeyStoreUrl()
+  {
+    return trustCertificateKeyStoreUrl;
+  }
+
+  public String getTrustCertificateKeyStoreType()
+  {
+    return trustCertificateKeyStoreType;
+  }
+
+  public String getTrustCertificateKeyStorePassword()
+  {
+    return trustCertificateKeyStorePasswordProvider == null ? null : trustCertificateKeyStorePasswordProvider.getPassword();
+  }
+
+  public String getClientCertificateKeyStoreUrl()
+  {
+    return clientCertificateKeyStoreUrl;
+  }
+
+  public String getClientCertificateKeyStoreType()
+  {
+    return clientCertificateKeyStoreType;
+  }
+
+  public String getClientCertificateKeyStorePassword()
+  {
+    return clientCertificateKeyStorePasswordProvider == null ? null : clientCertificateKeyStorePasswordProvider.getPassword();
+  }
+
+  public List<String> getEnabledSSLCipherSuites()
+  {
+    return enabledSSLCipherSuites;
+  }
+
+  public List<String> getEnabledTLSProtocols()
+  {
+    return enabledTLSProtocols;
+  }
+
+  public boolean isVerifyServerCertificate()
+  {
+    return verifyServerCertificate;
+  }
+
+  @Override
+  public String toString()
+  {
+    return "MySQLConnectorConfig{" +
+           "useSSL='" + useSSL + '\'' +
+           ", clientCertificateKeyStoreUrl='" + clientCertificateKeyStoreUrl + '\'' +
+           ", clientCertificateKeyStoreType='" + clientCertificateKeyStoreType + '\'' +
+           ", verifyServerCertificate='" + verifyServerCertificate + '\'' +
+           ", trustCertificateKeyStoreUrl='" + trustCertificateKeyStoreUrl + '\'' +
+           ", trustCertificateKeyStoreType='" + trustCertificateKeyStoreType + '\'' +
+           ", enabledSSLCipherSuites=" + enabledSSLCipherSuites +
+           ", enabledTLSProtocols=" + enabledTLSProtocols +
+           '}';
+  }
+}

extensions-core/mysql-metadata-storage/src/main/java/io/druid/metadata/storage/mysql/MySQLMetadataStorageModule.java

@@ -23,6 +23,7 @@
 import com.google.common.collect.ImmutableList;
 import com.google.inject.Binder;
 import com.google.inject.Key;
+import io.druid.guice.JsonConfigProvider;
 import io.druid.guice.LazySingleton;
 import io.druid.guice.PolyBind;
 import io.druid.guice.SQLMetadataStorageDruidModule;
@@ -56,6 +57,8 @@ public void configure(Binder binder)
   {
     super.configure(binder);
 
+    JsonConfigProvider.bind(binder, "druid.metadata.mysql.ssl", MySQLConnectorConfig.class);
+
     PolyBind
         .optionBinder(binder, Key.get(MetadataStorageProvider.class))
         .addBinding(TYPE)