shyiko#70 - Secure binlog connection with SSL

Author: shyiko

readme.md

@@ -8,10 +8,12 @@ but ended up as a complete rewrite. Key differences/features:
 - automatic binlog filename/position | GTID resolution
 - resumable disconnects
 - plugable failover strategies
-- JMX exposure (optionally with statistics)
+- binlog_checksum=CRC32 support (for MySQL 5.6.2+ users)
+- secure communication over the TLS
+- JMX-friendly
+- real-time stats
 - availability in Maven Central
 - no third-party dependencies
-- binlog_checksum=CRC32 support (for MySQL 5.6.2+ users)
 - test suite over different versions of MySQL releases
 
 > If you are looking for something similar in other languages - check out 
@@ -30,32 +32,7 @@ Get the latest JAR(s) from [here](http://search.maven.org/#search%7Cga%7C1%7Cg%3
 </dependency>
 ```
 
-The latest development version always available through Sonatype Snapshots repository (as shown below).
-
-```xml
-<dependencies>
-    <dependency>
-        <groupId>com.github.shyiko</groupId>
-        <artifactId>mysql-binlog-connector-java</artifactId>
-        <version>0.3.2-SNAPSHOT</version>
-    </dependency>
-</dependencies>
-
-<repositories>
-    <repository>
-        <id>sonatype-snapshots</id>
-        <url>https://oss.sonatype.org/content/repositories/snapshots</url>
-        <snapshots>
-            <enabled>true</enabled>
-        </snapshots>
-        <releases>
-            <enabled>false</enabled>
-        </releases>
-    </repository>
-</repositories>
-```
-
-### Reading binary log file
+#### Reading binary log file
 
 ```java
 File binlogFile = ...
@@ -69,7 +46,7 @@ try {
 }
 ```
 
-### Tapping into MySQL replication stream
+#### Tapping into MySQL replication stream
 
 > PREREQUISITES: Whichever user you plan to use for the BinaryLogClient, he MUST have [REPLICATION SLAVE](http://dev.mysql.com/doc/refman/5.5/en/privileges-provided.html#priv_replication-slave) privilege. Unless you specify binlogFilename/binlogPosition yourself (in which case automatic resolution won't kick in), you'll need [REPLICATION CLIENT](http://dev.mysql.com/doc/refman/5.5/en/privileges-provided.html#priv_replication-client) granted as well.
 
@@ -91,7 +68,7 @@ kick off from a specific filename or position, use `client.setBinlogFilename(fil
 > `client.connect()` is blocking (meaning that client will listen for events in the current thread). 
 `client.connect(timeout)`, on the other hand, spawns a separate thread.  
 
-### Controlling event deserialization
+#### Controlling event deserialization
 
 > You might need it for several reasons: 
 you don't want to waste time deserializing events you won't need; 
@@ -120,7 +97,7 @@ BinaryLogClient client = ...
 client.setEventDeserializer(eventDeserializer);
 ```
 
-### Exposing BinaryLogClient with JMX
+#### Exposing BinaryLogClient through JMX
 
 ```java
 MBeanServer mBeanServer = ManagementFactory.getPlatformMBeanServer();
@@ -136,10 +113,32 @@ ObjectName statsObjectName = new ObjectName("mysql.binlog:type=BinaryLogClientSt
 mBeanServer.registerMBean(stats, statsObjectName);
 ```
 
+#### Using SSL
+
+> Introduced in 1.0.0.
+
+TLSv1.1 & TLSv1.2 require [JDK 7](http://bugs.java.com/bugdatabase/view_bug.do?bug_id=6916074)+.  
+Prior to MySQL 5.7.10, MySQL supported only TLSv1 
+(see [Secure Connection Protocols and Ciphers](http://dev.mysql.com/doc/refman/5.7/en/secure-connection-protocols-ciphers.html)). 
+
+> To check that MySQL server is [properly configured with SSL support](http://dev.mysql.com/doc/refman/5.7/en/using-secure-connections.html) -
+`mysql -h host -u root -ptypeyourpasswordmaybe -e "show global variables like 'have_%ssl';"` ("Value" 
+should be "YES"). State of the current session can be determined using `\s` ("SSL" should not be blank).
+
+```java
+System.setProperty("javax.net.ssl.keyStore", "/path/to/keystore.jks");
+System.setProperty("javax.net.ssl.keyStorePassword", "keystore.password");
+System.setProperty("javax.net.ssl.trustStore", "/path/to/truststore.jks");
+System.setProperty("javax.net.ssl.trustStorePassword","truststore.password");
+
+BinaryLogClient client = ...
+client.setSSLMode(SSLMode.REQUIRED);
+```
+
 ## Implementation notes
 
-- data of numeric types (tinyint, etc) always returned signed(!) regardless of whether column definition includes "unsigned" keyword or not
-- data of var\*/\*text/\*blob types always returned as a byte array (for var\* this is true starting from mysql-binlog-connector-java@1.0.0). 
+- data of numeric types (tinyint, etc) always returned signed(!) regardless of whether column definition includes "unsigned" keyword or not.
+- data of var\*/\*text/\*blob types always returned as a byte array (for var\* this is true starting from 1.0.0). 
 
 ## Frequently Asked Questions
 

src/main/java/com/github/shyiko/mysql/binlog/BinaryLogClient.java

@@ -30,12 +30,17 @@
 import com.github.shyiko.mysql.binlog.event.deserialization.GtidEventDataDeserializer;
 import com.github.shyiko.mysql.binlog.event.deserialization.QueryEventDataDeserializer;
 import com.github.shyiko.mysql.binlog.event.deserialization.RotateEventDataDeserializer;
-import com.github.shyiko.mysql.binlog.io.BufferedSocketInputStream;
 import com.github.shyiko.mysql.binlog.io.ByteArrayInputStream;
 import com.github.shyiko.mysql.binlog.jmx.BinaryLogClientMXBean;
 import com.github.shyiko.mysql.binlog.network.AuthenticationException;
+import com.github.shyiko.mysql.binlog.network.ClientCapabilities;
+import com.github.shyiko.mysql.binlog.network.DefaultSSLSocketFactory;
+import com.github.shyiko.mysql.binlog.network.DefaultSocketFactory;
+import com.github.shyiko.mysql.binlog.network.SSLMode;
+import com.github.shyiko.mysql.binlog.network.SSLSocketFactory;
 import com.github.shyiko.mysql.binlog.network.ServerException;
 import com.github.shyiko.mysql.binlog.network.SocketFactory;
+import com.github.shyiko.mysql.binlog.network.TLSHostnameVerifier;
 import com.github.shyiko.mysql.binlog.network.protocol.ErrorPacket;
 import com.github.shyiko.mysql.binlog.network.protocol.GreetingPacket;
 import com.github.shyiko.mysql.binlog.network.protocol.Packet;
@@ -47,10 +52,10 @@
 import com.github.shyiko.mysql.binlog.network.protocol.command.DumpBinaryLogGtidCommand;
 import com.github.shyiko.mysql.binlog.network.protocol.command.PingCommand;
 import com.github.shyiko.mysql.binlog.network.protocol.command.QueryCommand;
+import com.github.shyiko.mysql.binlog.network.protocol.command.SSLRequestCommand;
 
 import java.io.EOFException;
 import java.io.IOException;
-import java.io.InputStream;
 import java.net.InetSocketAddress;
 import java.net.Socket;
 import java.net.SocketException;
@@ -78,22 +83,8 @@
  */
 public class BinaryLogClient implements BinaryLogClientMXBean {
 
-    private static final SocketFactory DEFAULT_SOCKET_FACTORY = new SocketFactory() {
-
-        @Override
-        public Socket createSocket() throws SocketException {
-            return new Socket() {
-
-                private InputStream inputStream;
-
-                @Override
-                public synchronized InputStream getInputStream() throws IOException {
-                    return inputStream != null ? inputStream :
-                        (inputStream = new BufferedSocketInputStream(super.getInputStream()));
-                }
-            };
-        }
-    };
+    private static final SocketFactory DEFAULT_SOCKET_FACTORY = new DefaultSocketFactory();
+    private static final SSLSocketFactory DEFAULT_SSL_SOCKET_FACTORY = new DefaultSSLSocketFactory();
 
     private final Logger logger = Logger.getLogger(getClass().getName());
 
@@ -108,6 +99,7 @@ public synchronized InputStream getInputStream() throws IOException {
     private volatile String binlogFilename;
     private volatile long binlogPosition = 4;
     private volatile long connectionId;
+    private SSLMode sslMode = SSLMode.DISABLED;
 
     private volatile GtidSet gtidSet;
     private final Object gtidSetAccessLock = new Object();
@@ -118,6 +110,7 @@ public synchronized InputStream getInputStream() throws IOException {
     private final List<LifecycleListener> lifecycleListeners = new LinkedList<LifecycleListener>();
 
     private SocketFactory socketFactory;
+    private SSLSocketFactory sslSocketFactory;
 
     private PacketChannel channel;
     private volatile boolean connected;
@@ -187,6 +180,17 @@ public void setBlocking(boolean blocking) {
         this.blocking = blocking;
     }
 
+    public SSLMode getSSLMode() {
+        return sslMode;
+    }
+
+    public void setSSLMode(SSLMode sslMode) {
+        if (sslMode == null) {
+            throw new IllegalArgumentException("SSL mode cannot be NULL");
+        }
+        this.sslMode = sslMode;
+    }
+
     /**
      * @return server id (65535 by default)
      * @see #setServerId(long)
@@ -347,6 +351,13 @@ public void setSocketFactory(SocketFactory socketFactory) {
         this.socketFactory = socketFactory;
     }
 
+    /**
+     * @param sslSocketFactory custom ssl socket factory
+     */
+    public void setSslSocketFactory(SSLSocketFactory sslSocketFactory) {
+        this.sslSocketFactory = sslSocketFactory;
+    }
+
     /**
      * @param threadFactory custom thread factory. If not provided, threads will be created using simple "new Thread()".
      */
@@ -367,7 +378,7 @@ public void connect() throws IOException {
         try {
             establishConnection();
             GreetingPacket greetingPacket = receiveGreeting();
-            authenticate(greetingPacket.getScramble(), greetingPacket.getServerCollation());
+            authenticate(greetingPacket);
             connectionId = greetingPacket.getThreadId();
             if (binlogFilename == null && gtidSet == null) {
                 autoPosition();
@@ -474,10 +485,29 @@ private void ensureEventDataDeserializer(EventType eventType,
         }
     }
 
-    private void authenticate(String salt, int collation) throws IOException {
-        AuthenticateCommand authenticateCommand = new AuthenticateCommand(schema, username, password, salt);
+    private void authenticate(GreetingPacket greetingPacket) throws IOException {
+        int collation = greetingPacket.getServerCollation();
+        int packetNumber = 1;
+        if (sslMode != SSLMode.DISABLED) {
+            boolean serverSupportsSSL = (greetingPacket.getServerCapabilities() & ClientCapabilities.SSL) != 0;
+            if (!serverSupportsSSL && (sslMode == SSLMode.REQUIRED || sslMode == SSLMode.VERIFY_CA ||
+                    sslMode == SSLMode.VERIFY_IDENTITY)) {
+                throw new IOException("MySQL server does not support SSL");
+            }
+            if (serverSupportsSSL) {
+                SSLRequestCommand sslRequestCommand = new SSLRequestCommand();
+                sslRequestCommand.setCollation(collation);
+                channel.write(sslRequestCommand, packetNumber++);
+                SSLSocketFactory sslSocketFactory = this.sslSocketFactory != null ? this.sslSocketFactory :
+                    DEFAULT_SSL_SOCKET_FACTORY;
+                channel.upgradeToSSL(sslSocketFactory,
+                    sslMode == SSLMode.VERIFY_IDENTITY ? new TLSHostnameVerifier() : null);
+            }
+        }
+        AuthenticateCommand authenticateCommand = new AuthenticateCommand(schema, username, password,
+            greetingPacket.getScramble());
         authenticateCommand.setCollation(collation);
-        channel.write(authenticateCommand);
+        channel.write(authenticateCommand, packetNumber);
         byte[] authenticationResult = channel.read();
         if (authenticationResult[0] == ErrorPacket.HEADER) {
             ErrorPacket errorPacket = new ErrorPacket(authenticationResult, 1);

src/main/java/com/github/shyiko/mysql/binlog/network/DefaultSSLSocketFactory.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2016 Stanley Shyiko
+ *
+ * Licensed 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 com.github.shyiko.mysql.binlog.network;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocket;
+import java.io.IOException;
+import java.net.Socket;
+import java.net.SocketException;
+import java.security.GeneralSecurityException;
+
+/**
+ * @author <a href="mailto:stanley.shyiko@gmail.com">Stanley Shyiko</a>
+ */
+public class DefaultSSLSocketFactory implements SSLSocketFactory {
+
+    private final String protocol;
+
+    public DefaultSSLSocketFactory() {
+        this("TLSv1");
+    }
+
+    /**
+     * @param protocol TLSv1, TLSv1.1 or TLSv1.2 (the last two require JDK 7+)
+     */
+    public DefaultSSLSocketFactory(String protocol) {
+        this.protocol = protocol;
+    }
+
+    @Override
+    public SSLSocket createSocket(Socket socket) throws SocketException {
+        SSLContext sc;
+        try {
+            sc = SSLContext.getInstance(this.protocol);
+            initSSLContext(sc);
+        } catch (GeneralSecurityException e) {
+            throw new SocketException(e.getMessage());
+        }
+        try {
+            return (SSLSocket) sc.getSocketFactory()
+                .createSocket(socket, socket.getInetAddress().getHostName(), socket.getPort(), true);
+        } catch (IOException e) {
+            throw new SocketException(e.getMessage());
+        }
+    }
+
+    protected void initSSLContext(SSLContext sc) throws GeneralSecurityException {
+        sc.init(null, null, null);
+    }
+
+}

src/main/java/com/github/shyiko/mysql/binlog/network/DefaultSocketFactory.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2016 Stanley Shyiko
+ *
+ * Licensed 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 com.github.shyiko.mysql.binlog.network;
+
+import com.github.shyiko.mysql.binlog.io.BufferedSocketInputStream;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.Socket;
+import java.net.SocketException;
+
+/**
+ * @author <a href="mailto:stanley.shyiko@gmail.com">Stanley Shyiko</a>
+ */
+public class DefaultSocketFactory implements SocketFactory {
+
+    @Override
+    public Socket createSocket() throws SocketException {
+        return new Socket() {
+
+            private InputStream inputStream;
+
+            @Override
+            public synchronized InputStream getInputStream() throws IOException {
+                return inputStream != null ? inputStream :
+                    (inputStream = new BufferedSocketInputStream(super.getInputStream()));
+            }
+        };
+    }
+
+}

src/main/java/com/github/shyiko/mysql/binlog/network/IdentityVerificationException.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2016 Stanley Shyiko
+ *
+ * Licensed 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 com.github.shyiko.mysql.binlog.network;
+
+import javax.net.ssl.SSLException;
+
+/**
+ * @author <a href="mailto:stanley.shyiko@gmail.com">Stanley Shyiko</a>
+ */
+public class IdentityVerificationException extends SSLException {
+
+    public IdentityVerificationException(String message) {
+        super(message);
+    }
+
+}