finagle-mysql: Add support of opportunistic TLS

Problem

Currently a finagle-mysql client supports only two TLS modes: it is either
enabled and the server must support it too, or it is disabled and then it
doesn't matter whether the server supports TLS or not.

Solution

We introduce a new mode `desired`, when the client will use TLS if
the server supports it, but will remain operational and use plaintext if it doesn't

JIRA Issues: GRAPH-13420

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

CHANGELOG.rst

@@ -15,6 +15,11 @@ New Features
   This response classifier is useful when a client has set a super low `RequestTimeout` and
   receiving a response is seen as 'best-effort'. ``PHAB_ID=D645818``
 
+* finagle-mysql: Introduce support of opportunistic TLS to allow mysql clients
+  with enabled TLS to speak over encrypted connections with MySQL servers where
+  TLS is on, and fallback to plaintext connections if TLS is switched off on
+  the server side. ``PHAB_ID=D644982``
+
 Runtime Behavior Changes
 ~~~~~~~~~~~~~~~~~~~~~~~~
 

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

@@ -12,6 +12,7 @@ import com.twitter.finagle.param.{
   _
 }
 import com.twitter.finagle.service.{ResponseClassifier, RetryBudget}
+import com.twitter.finagle.ssl.OpportunisticTls
 import com.twitter.finagle.stats.{ExceptionStatsHandler, NullStatsReceiver, StatsReceiver}
 import com.twitter.finagle.tracing.Tracer
 import com.twitter.finagle.transport.{Transport, TransportContext}
@@ -152,6 +153,30 @@ object Mysql extends com.twitter.finagle.Client[Request, Result] with MysqlRichC
     def withCredentials(u: String, p: String): Client =
       configured(Credentials(Option(u), Option(p)))
 
+    /**
+     * Configures the client whether to speak TLS or not.
+     *
+     * By default, don't use opportunistic TLS, and instead always speak TLS
+     * if TLS has been configured.
+     *
+     * The valid levels are Off, which indicates this will never speak TLS,
+     * Desired, which indicates it may speak TLS, but may also not speak TLS,
+     * and Required, which indicates it must speak TLS.
+     *
+     * Clients configured with level `Required` cannot speak to MySQL servers where
+     * TLS is switched off.
+     */
+    def withOpportunisticTls(level: OpportunisticTls.Level): Client =
+      configured(OppTls(Some(level)))
+
+    /**
+     * Disables opportunistic TLS.
+     *
+     * If the client is still TLS configured, it will speak with the server over TLS. To instead
+     * configure this to be `Off`, use `withOpportunisticTls(OpportunisticTls.Off)`.
+     */
+    def withNoOpportunisticTls: Client = configured(OppTls(None))
+
     /**
      * Database to use when this client establishes a new session.
      */

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

@@ -95,11 +95,13 @@ private[mysql] object Handshake {
   /**
    * Creates a `Handshake` based on the specific `Stack` params and `Transport` passed in.
    * If the `Transport.ClientSsl` param is set, then a `SecureHandshake` will be returned.
-   * Otherwise a `PlainHandshake is returned.
+   * Otherwise a `PlainHandshake` is returned.
    */
   def apply(params: Stack.Params, transport: Transport[Packet, Packet]): Handshake =
-    if (params[Transport.ClientSsl].sslClientConfiguration.isDefined)
+    if (params[Transport.ClientSsl].sslClientConfiguration.isDefined) {
       new SecureHandshake(params, transport)
-    else new PlainHandshake(params, transport)
+    } else {
+      new PlainHandshake(params, transport)
+    }
 
 }

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

@@ -0,0 +1,19 @@
+package com.twitter.finagle.mysql
+
+import com.twitter.finagle.FailureFlags
+import java.sql.SQLNonTransientException
+
+/**
+ * Indicates that the server lacks required capabilities
+ */
+class InsufficientServerCapabilitiesException(
+  required: Capability,
+  available: Capability,
+  val flags: Long = FailureFlags.NonRetryable)
+    extends SQLNonTransientException(
+      s"Insufficient capabilities: available: '$available', required: '$required'")
+    with FailureFlags[InsufficientServerCapabilitiesException] {
+
+  protected def copyWithFlags(flags: Long): InsufficientServerCapabilitiesException =
+    new InsufficientServerCapabilitiesException(required, available, flags)
+}

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

@@ -6,6 +6,8 @@ import com.twitter.finagle.netty4.ssl.client.Netty4ClientSslChannelInitializer.O
 import com.twitter.finagle.netty4.transport.ChannelTransportContext
 import com.twitter.finagle.transport.{Transport, TransportContext}
 import com.twitter.finagle.Stack
+import com.twitter.finagle.param.OppTls
+import com.twitter.finagle.ssl.OpportunisticTls
 import com.twitter.util.{Future, Promise, Try}
 import io.netty.channel.Channel
 
@@ -14,6 +16,8 @@ private[mysql] final class SecureHandshake(
   transport: Transport[Packet, Packet])
     extends Handshake(params, transport) {
 
+  private[this] val tlsLevel = params[OppTls].level.getOrElse(OpportunisticTls.Required)
+
   private[this] def onHandshakeComplete(p: Promise[Unit])(result: Try[Unit]): Unit =
     p.updateIfEmpty(result)
 
@@ -76,19 +80,47 @@ private[mysql] final class SecureHandshake(
       settings.maxPacketSize.inBytes.toInt
     )
 
+  private[this] def makePlainHandshakeResponse(handshakeInit: HandshakeInit): HandshakeResponse = {
+    PlainHandshakeResponse(
+      settings.username,
+      settings.password,
+      settings.database,
+      settings.calculatedClientCapabilities,
+      handshakeInit.salt,
+      handshakeInit.serverCapabilities,
+      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(handshakeInit))
-          .map(_ => handshakeInit)
+        if (tlsLevel == OpportunisticTls.Off) {
+          Future.value(makePlainHandshakeResponse(handshakeInit))
+        } else {
+          val serverTlsEnabled = handshakeInit.serverCapabilities.has(Capability.SSL)
+          if (serverTlsEnabled) {
+            writeSslConnectionRequest(handshakeInit)
+              .flatMap(_ => negotiateTls(handshakeInit))
+              .map(_ => handshakeInit)
+              .map(makeSecureHandshakeResponse)
+          } else if (tlsLevel == OpportunisticTls.Desired) {
+            Future.value(makePlainHandshakeResponse(handshakeInit))
+          } else {
+            Future.exception(
+              new InsufficientServerCapabilitiesException(
+                required = Capability(Capability.SSL),
+                available = handshakeInit.serverCapabilities
+              )
+            )
+          }
+        }
       }
-      .map(makeSecureHandshakeResponse)
       .flatMap(messageDispatch)
       .onFailure(_ => transport.close())
   }
-
 }