Repository: pingcap/tidb-binlog (go)

Issue:
Support TLS in MySQL drainer
## Feature Request
 
TLS should be supported in drainer connections to downstream MySQL/MariaDB/TiDB instances/cluster.

It should be possible to specify CA, client certificate, and client certificate key.

Pull request description:
Support TLS to connect mysql/TiDB
<!--
Thank you for contributing to TiDB! Please read TiDB's [CONTRIBUTING](https://github.com/pingcap/tidb/blob/master/CONTRIBUTING.md) document **BEFORE** filing this PR.
-->

### What problem does this PR solve? <!--add issue link with summary if exists-->
fix #887 

### What is changed and how it works?
Add an according configure part `syncer.to.security`
use [RegisterTLSConfig](https://godoc.org/github.com/go-sql-driver/mysql#RegisterTLSConfig) to  registers a custom `tls.Config` to be used with sql.Open

### Check List <!--REMOVE the items that are not applicable-->

Tests <!-- At least one of them must be included. -->


 - Manual test (add detailed scripts or steps below)
    * enable TLS in drainer, fails to connect  TiDB without enabling TLS.
    * after enable TLS, drainer success to connect to TiDB.

relate docs: https://pingcap.com/docs-cn/stable/how-to/secure/enable-tls-clients/



Related changes

 - Need to cherry-pick to the release branch
 - Need to update the documentation
 

Patch:
diff --git a/arbiter/server.go b/arbiter/server.go
index fb8d1b29..8abbcffa 100644
--- a/arbiter/server.go
+++ b/arbiter/server.go
@@ -78,7 +78,7 @@ func NewServer(cfg *Config) (srv *Server, err error) {
 	up := cfg.Up
 	down := cfg.Down
 
-	srv.downDB, err = createDB(down.User, down.Password, down.Host, down.Port)
+	srv.downDB, err = createDB(down.User, down.Password, down.Host, down.Port, nil)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
diff --git a/cmd/drainer/drainer.toml b/cmd/drainer/drainer.toml
index 21b4a912..084dd54c 100644
--- a/cmd/drainer/drainer.toml
+++ b/cmd/drainer/drainer.toml
@@ -108,6 +108,16 @@ port = 3306
 # when setting SyncPartialColumn drainer will allow the downstream schema
 # having more or less column numbers and relax sql mode by removing STRICT_TRANS_TABLES.
 # sync-mode = 1
+#
+# Uncomment this part if you need TLS to connecting downstream MySQL/TiDB.
+# You can only specified only `ssl-ca` if there is no client certificate and don't need server to authenticate client.
+# [syncer.to.security]
+# Path of file that contains list of trusted SSL CAs.
+# ssl-ca = "/path/to/ca.pem"
+# Path of file that contains X509 certificate in PEM format.
+# ssl-cert = "/path/to/drainer.pem"
+# Path of file that contains X509 key in PEM format.
+# ssl-key = "/path/to/drainer-key.pem"
 
 [syncer.to.checkpoint]
 # only support mysql or tidb now, you can uncomment this to control where the checkpoint is saved.
diff --git a/drainer/config.go b/drainer/config.go
index f830aa7f..c51eb9ca 100644
--- a/drainer/config.go
+++ b/drainer/config.go
@@ -217,6 +217,13 @@ func (cfg *Config) Parse(args []string) error {
 		return errors.Errorf("tls config %+v error %v", cfg.Security, err)
 	}
 
+	if cfg.SyncerCfg != nil && cfg.SyncerCfg.To != nil {
+		cfg.SyncerCfg.To.TLS, err = cfg.SyncerCfg.To.Security.ToTLSConfig()
+		if err != nil {
+			return errors.Errorf("tls config %+v error %v", cfg.SyncerCfg.To.Security, err)
+		}
+	}
+
 	if err = cfg.adjustConfig(); err != nil {
 		return errors.Trace(err)
 	}
diff --git a/drainer/relay.go b/drainer/relay.go
index 099c2f91..efd4f799 100644
--- a/drainer/relay.go
+++ b/drainer/relay.go
@@ -43,7 +43,7 @@ func feedByRelayLogIfNeed(cfg *Config) error {
 		return errors.Annotate(err, "failed to create reader")
 	}
 
-	db, err := loader.CreateDBWithSQLMode(scfg.To.User, scfg.To.Password, scfg.To.Host, scfg.To.Port, scfg.StrSQLMode)
+	db, err := loader.CreateDBWithSQLMode(scfg.To.User, scfg.To.Password, scfg.To.Host, scfg.To.Port, scfg.To.TLS, scfg.StrSQLMode)
 	if err != nil {
 		return errors.Annotate(err, "failed to create SQL db")
 	}
diff --git a/drainer/sync/mysql.go b/drainer/sync/mysql.go
index eb44af27..2912a246 100644
--- a/drainer/sync/mysql.go
+++ b/drainer/sync/mysql.go
@@ -87,7 +87,11 @@ func NewMysqlSyncer(
 	relayer relay.Relayer,
 	info *loopbacksync.LoopBackSync,
 ) (*MysqlSyncer, error) {
-	db, err := createDB(cfg.User, cfg.Password, cfg.Host, cfg.Port, sqlMode)
+	if cfg.TLS != nil {
+		log.Info("enable TLS to connect downstream MySQL/TiDB")
+	}
+
+	db, err := createDB(cfg.User, cfg.Password, cfg.Host, cfg.Port, cfg.TLS, sqlMode)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
@@ -103,7 +107,7 @@ func NewMysqlSyncer(
 
 		if newMode != oldMode {
 			db.Close()
-			db, err = createDB(cfg.User, cfg.Password, cfg.Host, cfg.Port, &newMode)
+			db, err = createDB(cfg.User, cfg.Password, cfg.Host, cfg.Port, cfg.TLS, &newMode)
 			if err != nil {
 				return nil, errors.Trace(err)
 			}
diff --git a/drainer/sync/util.go b/drainer/sync/util.go
index f569c433..03cea616 100644
--- a/drainer/sync/util.go
+++ b/drainer/sync/util.go
@@ -14,15 +14,20 @@
 package sync
 
 import (
+	"crypto/tls"
+
 	// mysql driver
 	_ "github.com/go-sql-driver/mysql"
+	"github.com/pingcap/tidb-binlog/pkg/security"
 )
 
 // DBConfig is the DB configuration.
 type DBConfig struct {
-	Host     string `toml:"host" json:"host"`
-	User     string `toml:"user" json:"user"`
-	Password string `toml:"password" json:"password"`
+	Host     string          `toml:"host" json:"host"`
+	User     string          `toml:"user" json:"user"`
+	Password string          `toml:"password" json:"password"`
+	Security security.Config `toml:"security" json:"security"`
+	TLS      *tls.Config     `toml:"-" json:"-"`
 	// if EncryptedPassword is not empty, Password will be ignore.
 	EncryptedPassword       string           `toml:"encrypted_password" json:"encrypted_password"`
 	SyncMode                int              `toml:"sync-mode" json:"sync-mode"`
diff --git a/pkg/loader/util.go b/pkg/loader/util.go
index 996312b1..20ef3dd0 100644
--- a/pkg/loader/util.go
+++ b/pkg/loader/util.go
@@ -14,12 +14,16 @@
 package loader
 
 import (
+	"crypto/tls"
 	gosql "database/sql"
 	"fmt"
 	"hash/crc32"
 	"net/url"
+	"strconv"
 	"strings"
+	"sync/atomic"
 
+	"github.com/go-sql-driver/mysql"
 	"github.com/pingcap/errors"
 )
 
@@ -77,14 +81,25 @@ func getTableInfo(db *gosql.DB, schema string, table string) (info *tableInfo, e
 	return
 }
 
+var customID int64
+
 // CreateDBWithSQLMode return sql.DB
-func CreateDBWithSQLMode(user string, password string, host string, port int, sqlMode *string) (db *gosql.DB, err error) {
+func CreateDBWithSQLMode(user string, password string, host string, port int, tlsConfig *tls.Config, sqlMode *string) (db *gosql.DB, err error) {
 	dsn := fmt.Sprintf("%s:%s@tcp(%s:%d)/?charset=utf8mb4,utf8&interpolateParams=true&readTimeout=1m&multiStatements=true", user, password, host, port)
 	if sqlMode != nil {
 		// same as "set sql_mode = '<sqlMode>'"
 		dsn += "&sql_mode='" + url.QueryEscape(*sqlMode) + "'"
 	}
 
+	if tlsConfig != nil {
+		name := "custom_" + strconv.FormatInt(atomic.AddInt64(&customID, 1), 10)
+		err := mysql.RegisterTLSConfig(name, tlsConfig)
+		if err != nil {
+			return nil, errors.Annotate(err, "failed to RegisterTLSConfig")
+		}
+		dsn += "&tls=" + name
+	}
+
 	db, err = gosql.Open("mysql", dsn)
 	if err != nil {
 		return nil, errors.Trace(err)
@@ -93,8 +108,8 @@ func CreateDBWithSQLMode(user string, password string, host string, port int, sq
 }
 
 // CreateDB return sql.DB
-func CreateDB(user string, password string, host string, port int) (db *gosql.DB, err error) {
-	return CreateDBWithSQLMode(user, password, host, port, nil)
+func CreateDB(user string, password string, host string, port int, tls *tls.Config) (db *gosql.DB, err error) {
+	return CreateDBWithSQLMode(user, password, host, port, tls, nil)
 }
 
 func quoteSchema(schema string, table string) string {
diff --git a/reparo/syncer/mysql.go b/reparo/syncer/mysql.go
index b023e905..f318eef9 100644
--- a/reparo/syncer/mysql.go
+++ b/reparo/syncer/mysql.go
@@ -50,7 +50,7 @@ var (
 var createDB = loader.CreateDB
 
 func newMysqlSyncer(cfg *DBConfig, worker int, batchSize int, safemode bool) (*mysqlSyncer, error) {
-	db, err := createDB(cfg.User, cfg.Password, cfg.Host, cfg.Port)
+	db, err := createDB(cfg.User, cfg.Password, cfg.Host, cfg.Port, nil)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}


Test patch:
diff --git a/arbiter/server_test.go b/arbiter/server_test.go
index 8b68a555..a10f502d 100644
--- a/arbiter/server_test.go
+++ b/arbiter/server_test.go
@@ -15,6 +15,7 @@ package arbiter
 
 import (
 	"context"
+	"crypto/tls"
 	"database/sql"
 	"fmt"
 	"sync"
@@ -55,7 +56,7 @@ func (l *dummyLoader) Close() {
 type testNewServerSuite struct {
 	db            *sql.DB
 	dbMock        sqlmock.Sqlmock
-	origCreateDB  func(string, string, string, int) (*sql.DB, error)
+	origCreateDB  func(string, string, string, int, *tls.Config) (*sql.DB, error)
 	origNewReader func(*reader.Config) (*reader.Reader, error)
 	origNewLoader func(*sql.DB, ...loader.Option) (loader.Loader, error)
 }
@@ -71,7 +72,7 @@ func (s *testNewServerSuite) SetUpTest(c *C) {
 	s.dbMock = mock
 
 	s.origCreateDB = createDB
-	createDB = func(user string, password string, host string, port int) (*sql.DB, error) {
+	createDB = func(user string, password string, host string, port int, _ *tls.Config) (*sql.DB, error) {
 		return s.db, nil
 	}
 
@@ -105,7 +106,7 @@ func (s *testNewServerSuite) TestRejectInvalidAddr(c *C) {
 }
 
 func (s *testNewServerSuite) TestStopIfFailedtoConnectDownStream(c *C) {
-	createDB = func(user string, password string, host string, port int) (*sql.DB, error) {
+	createDB = func(user string, password string, host string, port int, _ *tls.Config) (*sql.DB, error) {
 		return nil, fmt.Errorf("Can't create db")
 	}
 
diff --git a/drainer/sync/syncer_test.go b/drainer/sync/syncer_test.go
index c44f09eb..c0c76eb2 100644
--- a/drainer/sync/syncer_test.go
+++ b/drainer/sync/syncer_test.go
@@ -13,6 +13,7 @@
 package sync
 
 import (
+	"crypto/tls"
 	"database/sql"
 	"reflect"
 	"sync/atomic"
@@ -58,7 +59,7 @@ func (s *syncerSuite) SetUpTest(c *check.C) {
 
 	// create mysql syncer
 	oldCreateDB := createDB
-	createDB = func(string, string, string, int, *string) (db *sql.DB, err error) {
+	createDB = func(string, string, string, int, *tls.Config, *string) (db *sql.DB, err error) {
 		db, s.mysqlMock, err = sqlmock.New()
 		return
 	}
diff --git a/pkg/loader/example_loader_test.go b/pkg/loader/example_loader_test.go
index a545bc32..0f76ad5b 100644
--- a/pkg/loader/example_loader_test.go
+++ b/pkg/loader/example_loader_test.go
@@ -17,7 +17,7 @@ import "log"
 
 func Example() {
 	// create sql.DB
-	db, err := CreateDB("root", "", "localhost", 4000)
+	db, err := CreateDB("root", "", "localhost", 4000, nil /* *tls.Config */)
 	if err != nil {
 		log.Fatal(err)
 	}
diff --git a/reparo/syncer/mysql_test.go b/reparo/syncer/mysql_test.go
index 963bc8c4..3c93d87e 100644
--- a/reparo/syncer/mysql_test.go
+++ b/reparo/syncer/mysql_test.go
@@ -1,6 +1,7 @@
 package syncer
 
 import (
+	"crypto/tls"
 	"database/sql"
 	"time"
 
@@ -24,7 +25,7 @@ func (s *testMysqlSuite) testMysqlSyncer(c *check.C, safemode bool) {
 	)
 
 	oldCreateDB := createDB
-	createDB = func(string, string, string, int) (db *sql.DB, err error) {
+	createDB = func(string, string, string, int, *tls.Config) (db *sql.DB, err error) {
 		db, mock, err = sqlmock.New()
 		return
 	}