finagle-mysql: Add Support for SSL/TLS

Problem

MySQL contains support for SSL/TLS via messages sent
during the connection phase of the MySQL protocol. However,
finagle-mysql does not support the sending of the messages
in the correct sequence in order to negotiate SSL/TLS.

Solution

Add a `SecureHandshake` which is capable of sending the proper
sequence of messages to negotiate SSL/TLS with MySQL.

JIRA Issues: CSL-7939

Differential Revision: https://phabricator.twitter.biz/D328077

CHANGELOG.rst

@@ -32,6 +32,10 @@ New Features
   adding and removing cookies in bulk, without triggering a header rewrite on each item.
   ``PHAB_ID=D318013``
 
+* finagle-mysql: finagle-mysql now supports using SSL/TLS with MySQL. SSL/TLS can be turned on by
+  calling `withTransport.tls(sslClientConfiguration)` with a specified
+  `c.t.f.ssl.client.SslClientConfiguration`. ``PHAB_ID=D328077``
+
 Runtime Behavior Changes
 ~~~~~~~~~~~~~~~~~~~~~~~~
 

finagle-mysql/src/main/scala/com/twitter/finagle/Mysql.scala

@@ -102,6 +102,11 @@ object Mysql extends com.twitter.finagle.Client[Request, Result] with MysqlRichC
   private[this] val includeHandshakeInServiceAcquisitionToggle: Toggle[Int] =
     Toggles("com.twitter.finagle.mysql.IncludeHandshakeInServiceAcquisition")
 
+  // This param having a value other than `None` indicates that
+  // the client is setup to try to use SSL/TLS.
+  private[this] def wantsToUseSsl(params: Stack.Params): Boolean =
+    params[Transport.ClientSsl].sslClientConfiguration.isDefined
+
   protected val supportUnsigned: Boolean = UnsignedColumns.param.default.supported
 
   object Client {
@@ -158,8 +163,14 @@ object Mysql extends com.twitter.finagle.Client[Request, Result] with MysqlRichC
       with WithDefaultLoadBalancer[Client]
       with MysqlRichClient {
 
+    // If the param is set to use SSL/TLS, we force handshaking to
+    // be done as part of service acquisition. We want all handshaking
+    // to be done eventually as part of service acquisition, and SSL/TLS
+    // is functionality that didn't exist previously, so make SSL/TLS take
+    // the preferred path.
     private[this] val includeHandshakeInServiceAcquisition: Boolean =
-      includeHandshakeInServiceAcquisitionToggle(ServerInfo().id.hashCode)
+      wantsToUseSsl(params) ||
+        includeHandshakeInServiceAcquisitionToggle(ServerInfo().id.hashCode)
 
     protected val supportUnsigned: Boolean = params[UnsignedColumns].supported
 

finagle-mysql/src/main/scala/com/twitter/finagle/mysql/Handshake.scala

@@ -36,9 +36,6 @@ case object IncompatibleCharset
  *
  * https://dev.mysql.com/doc/internals/en/connection-phase.html
  *
- * @note At this time, the `Handshake` class only supports a
- * `Plain Handshake` extension, `PlainHandshake`.
- *
  * @param params The collection `Stack` params necessary to create the
  * desired MySQL session.
  *
@@ -97,11 +94,12 @@ private[mysql] object Handshake {
 
   /**
    * Creates a `Handshake` based on the specific `Stack` params and `Transport` passed in.
-   *
-   * @note At this time, only a `PlainHandshake` is supported, but eventually a
-   * `SecureHandshake` will be conditionally returned.
+   * If the `Transport.ClientSsl` param is set, then a `SecureHandshake` will be returned.
+   * Otherwise a `PlainHandshake is returned.
    */
   def apply(params: Stack.Params, transport: Transport[Packet, Packet]): Handshake =
-    new PlainHandshake(params, transport)
+    if (params[Transport.ClientSsl].sslClientConfiguration.isDefined)
+      new SecureHandshake(params, transport)
+    else new PlainHandshake(params, transport)
 
 }

finagle-mysql/src/main/scala/com/twitter/finagle/mysql/PlainHandshake.scala

@@ -23,6 +23,8 @@ private[mysql] final class PlainHandshake(
     )
   }
 
+  // For the `PlainHandshake`, after the init
+  // we just return a handshake response.
   def connectionPhase(): Future[Result] =
     readHandshakeInit()
       .map(makePlainHandshakeResponse)

finagle-mysql/src/main/scala/com/twitter/finagle/mysql/SecureHandshake.scala

@@ -0,0 +1,75 @@
+package com.twitter.finagle.mysql
+
+import com.twitter.finagle.mysql.transport.Packet
+import com.twitter.finagle.netty4.ssl.client.Netty4ClientSslChannelInitializer
+import com.twitter.finagle.netty4.ssl.client.Netty4ClientSslChannelInitializer.OnSslHandshakeComplete
+import com.twitter.finagle.netty4.transport.ChannelTransportContext
+import com.twitter.finagle.transport.{Transport, TransportContext}
+import com.twitter.finagle.Stack
+import com.twitter.util.{Future, Promise, Try}
+import io.netty.channel.Channel
+
+private[mysql] final class SecureHandshake(
+  params: Stack.Params,
+  transport: Transport[Packet, Packet])
+    extends Handshake(params, transport) {
+
+  private[this] def onHandshakeComplete(p: Promise[Unit])(result: Try[Unit]): Unit =
+    p.updateIfEmpty(result)
+
+  // We purposefully do not set an interrupt handler on this promise, since
+  // there is no meaningful way to gracefully interrupt an SSL/TLS handshake.
+  // This is very similar to how SSL/TLS negotiation works in finagle-mux.
+  // See `Negotiation` for more details.
+  private[this] def negotiateTls(): Future[Unit] = {
+    val p = new Promise[Unit]
+    val sslParams = params + OnSslHandshakeComplete(onHandshakeComplete(p))
+    val context: TransportContext = transport.context
+    context match {
+      case ctContext: ChannelTransportContext =>
+        val channel: Channel = ctContext.ch
+        channel.pipeline.addFirst("mysqlSslInit", new Netty4ClientSslChannelInitializer(sslParams))
+        p
+      case other =>
+        Future.exception(
+          new IllegalStateException(
+            s"SecureHandshake requires a channel to negotiate SSL/TLS. Found: $other"))
+    }
+  }
+
+  private[this] def writeSslConnectionRequest(handshakeInit: HandshakeInit): Future[Unit] = {
+    val request = SslConnectionRequest(
+      settings.sslCalculatedClientCap,
+      settings.charset,
+      settings.maxPacketSize.inBytes.toInt)
+    transport.write(request.toPacket)
+  }
+
+  private[this] def makeSecureHandshakeResponse(handshakeInit: HandshakeInit): HandshakeResponse =
+    SecureHandshakeResponse(
+      settings.username,
+      settings.password,
+      settings.database,
+      settings.sslCalculatedClientCap,
+      handshakeInit.salt,
+      handshakeInit.serverCap,
+      settings.charset,
+      settings.maxPacketSize.inBytes.toInt
+    )
+
+  // For the `SecureHandshake`, after the init,
+  // we return an `SslConnectionRequest`,
+  // neogtiate SSL/TLS, and then return a handshake response.
+  def connectionPhase(): Future[Result] = {
+    readHandshakeInit()
+      .flatMap { handshakeInit =>
+        writeSslConnectionRequest(handshakeInit)
+          .flatMap(_ => negotiateTls())
+          .map(_ => handshakeInit)
+      }
+      .map(makeSecureHandshakeResponse)
+      .flatMap(messageDispatch)
+      .onFailure(_ => transport.close())
+  }
+
+}

finagle-mysql/src/test/scala/com/twitter/finagle/mysql/integration/MysqlSslTest.scala

@@ -0,0 +1,36 @@
+package com.twitter.finagle.mysql.integration
+
+import com.twitter.finagle.Mysql
+import com.twitter.finagle.ssl.{Protocols, TrustCredentials}
+import com.twitter.finagle.ssl.client.SslClientConfiguration
+import com.twitter.util.{Await, Duration}
+import org.scalatest.FunSuite
+
+class MysqlSslTest extends FunSuite with IntegrationClient {
+
+  override protected def configureClient(
+    username: String,
+    password: String,
+    db: String
+  ): Mysql.Client = {
+    // We care more about testing the SSL/TLS wiring throughout
+    // MySQL here than verifying certificates and hostnames.
+    //
+    // The community edition of MySQL 5.7 isn't built by default
+    // with OpenSSL, and so uses yaSSL, which only supports up to
+    // TLSv1.1. So we use TLSv1.1 here.
+    val sslClientConfig = SslClientConfiguration(
+      trustCredentials = TrustCredentials.Insecure,
+      protocols = Protocols.Enabled(Seq("TLSv1.1")))
+    super
+      .configureClient(username, password, db)
+      .withTransport.tls(sslClientConfig)
+  }
+
+  test("ping over ssl") {
+    val theClient = client.orNull
+    val result = Await.result(theClient.ping(), Duration.fromSeconds(2))
+    // If we get here, result is Unit, and all is good
+  }
+
+}