CVE-2022-3023: Use of Externally-Controlled Format String (CWE-134)
value: Use of Externally-Controlled Format String in GitHub repository pingcap/tidb prior to 6.4.0, 6.1.3.
CWE description: The product uses a function that accepts a format string as an argument, but the format string originates from an external source.
CVSS v3 base score: 9.8
Published: 2022-11-04T12:15Z
Repository: https://github.com/pingcap/tidb
Language: Go
Files changed: br/pkg/lightning/checkpoints/checkpoints.go, br/pkg/lightning/common/util.go, br/pkg/lightning/common/util_test.go, br/pkg/lightning/config/config.go, br/pkg/lightning/config/config_test.go, cmd/importer/db.go, dumpling/export/config.go, dumpling/export/dump.go, dumpling/export/sql.go, dumpling/export/sql_test.go, dumpling/tests/s3/import.go, util/dbutil/common.go
Commit date: 2022-10-19 10:27:52 +0800
Commit message: *: don't use DSN to avoid some security problems (#38342)

 
Diff: diff --git a/br/pkg/lightning/checkpoints/checkpoints.go b/br/pkg/lightning/checkpoints/checkpoints.go
index 44f2349b6..13817e28e 100644
--- a/br/pkg/lightning/checkpoints/checkpoints.go
+++ b/br/pkg/lightning/checkpoints/checkpoints.go
@@ -515,11 +515,19 @@ func OpenCheckpointsDB(ctx context.Context, cfg *config.Config) (DB, error) {
 		return NewNullCheckpointsDB(), nil
 	}
 
 	switch cfg.Checkpoint.Driver {
 	case config.CheckpointDriverMySQL:
-		db, err := common.ConnectMySQL(cfg.Checkpoint.DSN)
+		var (
+			db  *sql.DB
+			err error
+		)
+		if cfg.Checkpoint.MySQLParam != nil {
+			db, err = cfg.Checkpoint.MySQLParam.Connect()
+		} else {
+			db, err = sql.Open("mysql", cfg.Checkpoint.DSN)
+		}
 		if err != nil {
 			return nil, errors.Trace(err)
 		}
 		cpdb, err := NewMySQLCheckpointsDB(ctx, db, cfg.Checkpoint.Schema)
 		if err != nil {
@@ -544,11 +552,19 @@ func IsCheckpointsDBExists(ctx context.Context, cfg *config.Config) (bool, error
 	if !cfg.Checkpoint.Enable {
 		return false, nil
 	}
 	switch cfg.Checkpoint.Driver {
 	case config.CheckpointDriverMySQL:
-		db, err := sql.Open("mysql", cfg.Checkpoint.DSN)
+		var (
+			db  *sql.DB
+			err error
+		)
+		if cfg.Checkpoint.MySQLParam != nil {
+			db, err = cfg.Checkpoint.MySQLParam.Connect()
+		} else {
+			db, err = sql.Open("mysql", cfg.Checkpoint.DSN)
+		}
 		if err != nil {
 			return false, errors.Trace(err)
 		}
 		//nolint: errcheck
 		defer db.Close()
diff --git a/br/pkg/lightning/common/util.go b/br/pkg/lightning/common/util.go
index cc03f0ec6..679ba6cc5 100644
--- a/br/pkg/lightning/common/util.go
+++ b/br/pkg/lightning/common/util.go
@@ -21,11 +21,10 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
 	"net"
 	"net/http"
-	"net/url"
 	"os"
 	"strconv"
 	"strings"
 	"syscall"
 	"time"
@@ -56,68 +55,74 @@ type MySQLConnectParam struct {
 	MaxAllowedPacket uint64
 	TLS              string
 	Vars             map[string]string
 }
 
-func (param *MySQLConnectParam) ToDSN() string {
-	hostPort := net.JoinHostPort(param.Host, strconv.Itoa(param.Port))
-	dsn := fmt.Sprintf("%s:%s@tcp(%s)/?charset=utf8mb4&sql_mode='%s'&maxAllowedPacket=%d&tls=%s",
-		param.User, param.Password, hostPort,
-		param.SQLMode, param.MaxAllowedPacket, param.TLS)
+func (param *MySQLConnectParam) ToDriverConfig() *mysql.Config {
+	cfg := mysql.NewConfig()
+	cfg.Params = make(map[string]string)
+
+	cfg.User = param.User
+	cfg.Passwd = param.Password
+	cfg.Net = "tcp"
+	cfg.Addr = net.JoinHostPort(param.Host, strconv.Itoa(param.Port))
+	cfg.Params["charset"] = "utf8mb4"
+	cfg.Params["sql_mode"] = fmt.Sprintf("'%s'", param.SQLMode)
+	cfg.MaxAllowedPacket = int(param.MaxAllowedPacket)
+	cfg.TLSConfig = param.TLS
 
 	for k, v := range param.Vars {
-		dsn += fmt.Sprintf("&%s='%s'", k, url.QueryEscape(v))
+		cfg.Params[k] = fmt.Sprintf("'%s'", v)
 	}
-
-	return dsn
+	return cfg
 }
 
-func tryConnectMySQL(dsn string) (*sql.DB, error) {
-	driverName := "mysql"
-	failpoint.Inject("MockMySQLDriver", func(val failpoint.Value) {
-		driverName = val.(string)
+func tryConnectMySQL(cfg *mysql.Config) (*sql.DB, error) {
+	failpoint.Inject("MustMySQLPassword", func(val failpoint.Value) {
+		pwd := val.(string)
+		if cfg.Passwd != pwd {
+			failpoint.Return(nil, &mysql.MySQLError{Number: tmysql.ErrAccessDenied, Message: "access denied"})
+		}
+		failpoint.Return(nil, nil)
 	})
-	db, err := sql.Open(driverName, dsn)
+	c, err := mysql.NewConnector(cfg)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
+	db := sql.OpenDB(c)
 	if err = db.Ping(); err != nil {
 		_ = db.Close()
 		return nil, errors.Trace(err)
 	}
 	return db, nil
 }
 
 // ConnectMySQL connects MySQL with the dsn. If access is denied and the password is a valid base64 encoding,
 // we will try to connect MySQL with the base64 decoding of the password.
-func ConnectMySQL(dsn string) (*sql.DB, error) {
-	cfg, err := mysql.ParseDSN(dsn)
-	if err != nil {
-		return nil, errors.Trace(err)
-	}
+func ConnectMySQL(cfg *mysql.Config) (*sql.DB, error) {
 	// Try plain password first.
-	db, firstErr := tryConnectMySQL(dsn)
+	db, firstErr := tryConnectMySQL(cfg)
 	if firstErr == nil {
 		return db, nil
 	}
 	// If access is denied and password is encoded by base64, try the decoded string as well.
 	if mysqlErr, ok := errors.Cause(firstErr).(*mysql.MySQLError); ok && mysqlErr.Number == tmysql.ErrAccessDenied {
 		// If password is encoded by base64, try the decoded string as well.
 		if password, decodeErr := base64.StdEncoding.DecodeString(cfg.Passwd); decodeErr == nil && string(password) != cfg.Passwd {
 			cfg.Passwd = string(password)
-			db, err = tryConnectMySQL(cfg.FormatDSN())
+			db2, err := tryConnectMySQL(cfg)
 			if err == nil {
-				return db, nil
+				return db2, nil
 			}
 		}
 	}
 	// If we can't connect successfully, return the first error.
 	return nil, errors.Trace(firstErr)
 }
 
 func (param *MySQLConnectParam) Connect() (*sql.DB, error) {
-	db, err := ConnectMySQL(param.ToDSN())
+	db, err := ConnectMySQL(param.ToDriverConfig())
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
 	return db, nil
 }
diff --git a/br/pkg/lightning/common/util_test.go b/br/pkg/lightning/common/util_test.go
index c7c95b44f..a192ecea1 100644
--- a/br/pkg/lightning/common/util_test.go
+++ b/br/pkg/lightning/common/util_test.go
@@ -14,30 +14,25 @@
 
 package common_test
 
 import (
 	"context"
-	"database/sql"
-	"database/sql/driver"
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
 	"io"
-	"math/rand"
 	"net/http"
 	"net/http/httptest"
-	"strconv"
 	"testing"
 	"time"
 
 	"github.com/DATA-DOG/go-sqlmock"
 	"github.com/go-sql-driver/mysql"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/br/pkg/lightning/common"
 	"github.com/pingcap/tidb/br/pkg/lightning/log"
-	tmysql "github.com/pingcap/tidb/errno"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
 func TestDirNotExist(t *testing.T) {
@@ -83,87 +78,33 @@ func TestGetJSON(t *testing.T) {
 	err = common.GetJSON(ctx, client, testServer.URL, &response)
 	require.Error(t, err)
 	require.Regexp(t, ".*http status code != 200.*", err.Error())
 }
 
-func TestToDSN(t *testing.T) {
-	param := common.MySQLConnectParam{
-		Host:             "127.0.0.1",
-		Port:             4000,
-		User:             "root",
-		Password:         "123456",
-		SQLMode:          "strict",
-		MaxAllowedPacket: 1234,
-		TLS:              "cluster",
-		Vars: map[string]string{
-			"tidb_distsql_scan_concurrency": "1",
-		},
-	}
-	require.Equal(t, "root:123456@tcp(127.0.0.1:4000)/?charset=utf8mb4&sql_mode='strict'&maxAllowedPacket=1234&tls=cluster&tidb_distsql_scan_concurrency='1'", param.ToDSN())
-
-	param.Host = "::1"
-	require.Equal(t, "root:123456@tcp([::1]:4000)/?charset=utf8mb4&sql_mode='strict'&maxAllowedPacket=1234&tls=cluster&tidb_distsql_scan_concurrency='1'", param.ToDSN())
-}
-
-type mockDriver struct {
-	driver.Driver
-	plainPsw string
-}
-
-func (m *mockDriver) Open(dsn string) (driver.Conn, error) {
-	cfg, err := mysql.ParseDSN(dsn)
-	if err != nil {
-		return nil, err
-	}
-	accessDenied := cfg.Passwd != m.plainPsw
-	return &mockConn{accessDenied: accessDenied}, nil
-}
-
-type mockConn struct {
-	driver.Conn
-	driver.Pinger
-	accessDenied bool
-}
-
-func (c *mockConn) Ping(ctx context.Context) error {
-	if c.accessDenied {
-		return &mysql.MySQLError{Number: tmysql.ErrAccessDenied, Message: "access denied"}
-	}
-	return nil
-}
-
-func (c *mockConn) Close() error {
-	return nil
-}
-
 func TestConnect(t *testing.T) {
 	plainPsw := "dQAUoDiyb1ucWZk7"
-	driverName := "mysql-mock-" + strconv.Itoa(rand.Int())
-	sql.Register(driverName, &mockDriver{plainPsw: plainPsw})
 
 	require.NoError(t, failpoint.Enable(
-		"github.com/pingcap/tidb/br/pkg/lightning/common/MockMySQLDriver",
-		fmt.Sprintf("return(\"%s\")", driverName)))
+		"github.com/pingcap/tidb/br/pkg/lightning/common/MustMySQLPassword",
+		fmt.Sprintf("return(\"%s\")", plainPsw)))
 	defer func() {
-		require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/br/pkg/lightning/common/MockMySQLDriver"))
+		require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/br/pkg/lightning/common/MustMySQLPassword"))
 	}()
 
 	param := common.MySQLConnectParam{
 		Host:             "127.0.0.1",
 		Port:             4000,
 		User:             "root",
 		Password:         plainPsw,
 		SQLMode:          "strict",
 		MaxAllowedPacket: 1234,
 	}
-	db, err := param.Connect()
+	_, err := param.Connect()
 	require.NoError(t, err)
-	require.NoError(t, db.Close())
 	param.Password = base64.StdEncoding.EncodeToString([]byte(plainPsw))
-	db, err = param.Connect()
+	_, err = param.Connect()
 	require.NoError(t, err)
-	require.NoError(t, db.Close())
 }
 
 func TestIsContextCanceledError(t *testing.T) {
 	require.True(t, common.IsContextCanceledError(context.Canceled))
 	require.False(t, common.IsContextCanceledError(io.EOF))
diff --git a/br/pkg/lightning/config/config.go b/br/pkg/lightning/config/config.go
index 4c1af0d2b..638784ff3 100644
--- a/br/pkg/lightning/config/config.go
+++ b/br/pkg/lightning/config/config.go
@@ -551,15 +551,16 @@ type TikvImporter struct {
 	LocalWriterMemCacheSize ByteSize `toml:"local-writer-mem-cache-size" json:"local-writer-mem-cache-size"`
 	StoreWriteBWLimit       ByteSize `toml:"store-write-bwlimit" json:"store-write-bwlimit"`
 }
 
 type Checkpoint struct {
-	Schema           string                 `toml:"schema" json:"schema"`
-	DSN              string                 `toml:"dsn" json:"-"` // DSN may contain password, don't expose this to JSON.
-	Driver           string                 `toml:"driver" json:"driver"`
-	Enable           bool                   `toml:"enable" json:"enable"`
-	KeepAfterSuccess CheckpointKeepStrategy `toml:"keep-after-success" json:"keep-after-success"`
+	Schema           string                    `toml:"schema" json:"schema"`
+	DSN              string                    `toml:"dsn" json:"-"` // DSN may contain password, don't expose this to JSON.
+	MySQLParam       *common.MySQLConnectParam `toml:"-" json:"-"`   // For some security reason, we use MySQLParam instead of DSN.
+	Driver           string                    `toml:"driver" json:"driver"`
+	Enable           bool                      `toml:"enable" json:"enable"`
+	KeepAfterSuccess CheckpointKeepStrategy    `toml:"keep-after-success" json:"keep-after-success"`
 }
 
 type Cron struct {
 	SwitchMode     Duration `toml:"switch-mode" json:"switch-mode"`
 	LogProgress    Duration `toml:"log-progress" json:"log-progress"`
@@ -1140,11 +1141,11 @@ func (cfg *Config) AdjustCheckPoint() {
 				Password:         cfg.TiDB.Psw,
 				SQLMode:          mysql.DefaultSQLMode,
 				MaxAllowedPacket: defaultMaxAllowedPacket,
 				TLS:              cfg.TiDB.TLS,
 			}
-			cfg.Checkpoint.DSN = param.ToDSN()
+			cfg.Checkpoint.MySQLParam = &param
 		case CheckpointDriverFile:
 			cfg.Checkpoint.DSN = "/tmp/" + cfg.Checkpoint.Schema + ".pb"
 		}
 	}
 }
diff --git a/br/pkg/lightning/config/config_test.go b/br/pkg/lightning/config/config_test.go
index 2a4dcbe7c..e74094a6b 100644
--- a/br/pkg/lightning/config/config_test.go
+++ b/br/pkg/lightning/config/config_test.go
@@ -30,11 +30,10 @@ import (
 	"time"
 
 	"github.com/BurntSushi/toml"
 	"github.com/pingcap/tidb/br/pkg/lightning/common"
 	"github.com/pingcap/tidb/br/pkg/lightning/config"
-	"github.com/pingcap/tidb/parser/mysql"
 	"github.com/stretchr/testify/require"
 )
 
 func startMockServer(t *testing.T, statusCode int, content string) (*httptest.Server, string, int) {
 	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -624,11 +623,13 @@ func TestLoadConfig(t *testing.T) {
 	taskCfg.Checkpoint.DSN = ""
 	taskCfg.Checkpoint.Driver = config.CheckpointDriverMySQL
 	taskCfg.TiDB.DistSQLScanConcurrency = 1
 	err = taskCfg.Adjust(context.Background())
 	require.NoError(t, err)
-	require.Equal(t, "guest:12345@tcp(172.16.30.11:4001)/?charset=utf8mb4&sql_mode='"+mysql.DefaultSQLMode+"'&maxAllowedPacket=67108864&tls=false", taskCfg.Checkpoint.DSN)
+	equivalentDSN := taskCfg.Checkpoint.MySQLParam.ToDriverConfig().FormatDSN()
+	expectedDSN := "guest:12345@tcp(172.16.30.11:4001)/?tls=false&maxAllowedPacket=67108864&charset=utf8mb4&sql_mode=%27ONLY_FULL_GROUP_BY%2CSTRICT_TRANS_TABLES%2CNO_ZERO_IN_DATE%2CNO_ZERO_DATE%2CERROR_FOR_DIVISION_BY_ZERO%2CNO_AUTO_CREATE_USER%2CNO_ENGINE_SUBSTITUTION%27"
+	require.Equal(t, expectedDSN, equivalentDSN)
 
 	result := taskCfg.String()
 	require.Regexp(t, `.*"pd-addr":"172.16.30.11:2379,172.16.30.12:2379".*`, result)
 
 	cfg, err = config.LoadGlobalConfig([]string{}, nil)
diff --git a/cmd/importer/db.go b/cmd/importer/db.go
index 8b0d7353b..b8ecf83ab 100644
--- a/cmd/importer/db.go
+++ b/cmd/importer/db.go
@@ -20,11 +20,11 @@ import (
 	"math"
 	"math/rand"
 	"strconv"
 	"strings"
 
-	_ "github.com/go-sql-driver/mysql"
+	mysql2 "github.com/go-sql-driver/mysql"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/parser/mysql"
 	"go.uber.org/zap"
 )
@@ -316,17 +316,22 @@ func execSQL(db *sql.DB, sql string) error {
 
 	return nil
 }
 
 func createDB(cfg DBConfig) (*sql.DB, error) {
-	dbDSN := fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?charset=utf8", cfg.User, cfg.Password, cfg.Host, cfg.Port, cfg.Name)
-	db, err := sql.Open("mysql", dbDSN)
+	driverCfg := mysql2.NewConfig()
+	driverCfg.User = cfg.User
+	driverCfg.Passwd = cfg.Password
+	driverCfg.Net = "tcp"
+	driverCfg.Addr = cfg.Host + ":" + strconv.Itoa(cfg.Port)
+	driverCfg.DBName = cfg.Name
+
+	c, err := mysql2.NewConnector(driverCfg)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
-
-	return db, nil
+	return sql.OpenDB(c), nil
 }
 
 func closeDB(db *sql.DB) error {
 	return errors.Trace(db.Close())
 }
diff --git a/dumpling/export/config.go b/dumpling/export/config.go
index 980de0d88..b92d2922d 100644
--- a/dumpling/export/config.go
+++ b/dumpling/export/config.go
@@ -216,10 +216,35 @@ func (conf *Config) GetDSN(db string) string {
 		dsn += "&allowCleartextPasswords=1"
 	}
 	return dsn
 }
 
+// GetDriverConfig returns the MySQL driver config from Config.
+func (conf *Config) GetDriverConfig(db string) *mysql.Config {
+	driverCfg := mysql.NewConfig()
+	// maxAllowedPacket=0 can be used to automatically fetch the max_allowed_packet variable from server on every connection.
+	// https://github.com/go-sql-driver/mysql#maxallowedpacket
+	hostPort := net.JoinHostPort(conf.Host, strconv.Itoa(conf.Port))
+	driverCfg.User = conf.User
+	driverCfg.Passwd = conf.Password
+	driverCfg.Net = "tcp"
+	driverCfg.Addr = hostPort
+	driverCfg.DBName = db
+	driverCfg.Collation = "utf8mb4_general_ci"
+	driverCfg.ReadTimeout = conf.ReadTimeout
+	driverCfg.WriteTimeout = 30 * time.Second
+	driverCfg.InterpolateParams = true
+	driverCfg.MaxAllowedPacket = 0
+	if conf.Security.DriveTLSName != "" {
+		driverCfg.TLSConfig = conf.Security.DriveTLSName
+	}
+	if conf.AllowCleartextPasswords {
+		driverCfg.AllowCleartextPasswords = true
+	}
+	return driverCfg
+}
+
 func timestampDirName() string {
 	return fmt.Sprintf("./export-%s", time.Now().Format(time.RFC3339))
 }
 
 // DefineFlags defines flags of dumpling's configuration
diff --git a/dumpling/export/dump.go b/dumpling/export/dump.go
index cdc91e6e4..857ef5d74 100644
--- a/dumpling/export/dump.go
+++ b/dumpling/export/dump.go
@@ -35,11 +35,11 @@ import (
 	"go.uber.org/zap"
 	"golang.org/x/exp/slices"
 	"golang.org/x/sync/errgroup"
 )
 
-var openDBFunc = sql.Open
+var openDBFunc = openDB
 
 var errEmptyHandleVals = errors.New("empty handleVals for TiDB table")
 
 // Dumper is the dump progress structure
 type Dumper struct {
@@ -1307,15 +1307,15 @@ func startHTTPService(d *Dumper) error {
 }
 
 // openSQLDB is an initialization step of Dumper.
 func openSQLDB(d *Dumper) error {
 	conf := d.conf
-	pool, err := sql.Open("mysql", conf.GetDSN(""))
+	c, err := mysql.NewConnector(conf.GetDriverConfig(""))
 	if err != nil {
 		return errors.Trace(err)
 	}
-	d.dbHandle = pool
+	d.dbHandle = sql.OpenDB(c)
 	return nil
 }
 
 // detectServerInfo is an initialization step of Dumper.
 func detectServerInfo(d *Dumper) error {
@@ -1508,16 +1508,24 @@ func setSessionParam(d *Dumper) error {
 			if conf.ServerInfo.HasTiKV {
 				sessionParam["tidb_snapshot"] = snapshot
 			}
 		}
 	}
-	if d.dbHandle, err = resetDBWithSessionParams(d.tctx, pool, conf.GetDSN(""), conf.SessionParams); err != nil {
+	if d.dbHandle, err = resetDBWithSessionParams(d.tctx, pool, conf.GetDriverConfig(""), conf.SessionParams); err != nil {
 		return errors.Trace(err)
 	}
 	return nil
 }
 
+func openDB(cfg *mysql.Config) (*sql.DB, error) {
+	c, err := mysql.NewConnector(cfg)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+	return sql.OpenDB(c), nil
+}
+
 func (d *Dumper) renewSelectTableRegionFuncForLowerTiDB(tctx *tcontext.Context) error {
 	conf := d.conf
 	if !(conf.ServerInfo.ServerType == version.ServerTypeTiDB && conf.ServerInfo.ServerVersion != nil && conf.ServerInfo.HasTiKV &&
 		conf.ServerInfo.ServerVersion.Compare(*decodeRegionVersion) >= 0 &&
 		conf.ServerInfo.ServerVersion.Compare(*gcSafePointVersion) < 0) {
@@ -1530,11 +1538,11 @@ func (d *Dumper) renewSelectTableRegionFuncForLowerTiDB(tctx *tcontext.Context)
 	// To avoid this function continuously returning errors and confusing users because we fail to init this function at first,
 	// selectTiDBTableRegionFunc is set to always return an ignorable error at first.
 	d.selectTiDBTableRegionFunc = func(_ *tcontext.Context, _ *BaseConn, meta TableMeta) (pkFields []string, pkVals [][]string, err error) {
 		return nil, nil, errors.Annotatef(errEmptyHandleVals, "table: `%s`.`%s`", escapeString(meta.DatabaseName()), escapeString(meta.TableName()))
 	}
-	dbHandle, err := openDBFunc("mysql", conf.GetDSN(""))
+	dbHandle, err := openDBFunc(conf.GetDriverConfig(""))
 	if err != nil {
 		return errors.Trace(err)
 	}
 	defer func() {
 		_ = dbHandle.Close()
diff --git a/dumpling/export/sql.go b/dumpling/export/sql.go
index 83655df99..837bec568 100644
--- a/dumpling/export/sql.go
+++ b/dumpling/export/sql.go
@@ -8,11 +8,10 @@ import (
 	"database/sql"
 	"encoding/json"
 	"fmt"
 	"io"
 	"math"
-	"net/url"
 	"strconv"
 	"strings"
 
 	"github.com/go-sql-driver/mysql"
 	"github.com/pingcap/errors"
@@ -832,11 +831,11 @@ func isUnknownSystemVariableErr(err error) bool {
 	return strings.Contains(err.Error(), "Unknown system variable")
 }
 
 // resetDBWithSessionParams will return a new sql.DB as a replacement for input `db` with new session parameters.
 // If returned error is nil, the input `db` will be closed.
-func resetDBWithSessionParams(tctx *tcontext.Context, db *sql.DB, dsn string, params map[string]interface{}) (*sql.DB, error) {
+func resetDBWithSessionParams(tctx *tcontext.Context, db *sql.DB, cfg *mysql.Config, params map[string]interface{}) (*sql.DB, error) {
 	support := make(map[string]interface{})
 	for k, v := range params {
 		var pv interface{}
 		if str, ok := v.(string); ok {
 			if pvi, err := strconv.ParseInt(str, 10, 64); err == nil {
@@ -860,32 +859,38 @@ func resetDBWithSessionParams(tctx *tcontext.Context, db *sql.DB, dsn string, pa
 		}
 
 		support[k] = pv
 	}
 
+	if cfg.Params == nil {
+		cfg.Params = make(map[string]string)
+	}
+
 	for k, v := range support {
 		var s string
 		// Wrap string with quote to handle string with space. For example, '2020-10-20 13:41:40'
 		// For --params argument, quote doesn't matter because it doesn't affect the actual value
 		if str, ok := v.(string); ok {
 			s = wrapStringWith(str, "'")
 		} else {
 			s = fmt.Sprintf("%v", v)
 		}
-		dsn += fmt.Sprintf("&%s=%s", k, url.QueryEscape(s))
+		cfg.Params[k] = s
 	}
 
 	db.Close()
-	newDB, err := sql.Open("mysql", dsn)
-	if err == nil {
-		// ping to make sure all session parameters are set correctly
-		err = newDB.PingContext(tctx)
-		if err != nil {
-			newDB.Close()
-		}
+	c, err := mysql.NewConnector(cfg)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+	newDB := sql.OpenDB(c)
+	// ping to make sure all session parameters are set correctly
+	err = newDB.PingContext(tctx)
+	if err != nil {
+		newDB.Close()
 	}
-	return newDB, errors.Trace(err)
+	return newDB, nil
 }
 
 func createConnWithConsistency(ctx context.Context, db *sql.DB, repeatableRead bool) (*sql.Conn, error) {
 	conn, err := db.Conn(ctx)
 	if err != nil {
diff --git a/dumpling/export/sql_test.go b/dumpling/export/sql_test.go
index d98a8a3c7..04615637b 100644
--- a/dumpling/export/sql_test.go
+++ b/dumpling/export/sql_test.go
@@ -1343,11 +1343,11 @@ func TestBuildVersion3RegionQueries(t *testing.T) {
 	tctx, cancel := tcontext.Background().WithLogger(appLogger).WithCancel()
 	oldOpenFunc := openDBFunc
 	defer func() {
 		openDBFunc = oldOpenFunc
 	}()
-	openDBFunc = func(_, _ string) (*sql.DB, error) {
+	openDBFunc = func(*mysql.Config) (*sql.DB, error) {
 		return db, nil
 	}
 
 	conf := DefaultConfig()
 	conf.ServerInfo = version.ServerInfo{
diff --git a/dumpling/tests/s3/import.go b/dumpling/tests/s3/import.go
index 0489be3fa..30dc95fae 100644
--- a/dumpling/tests/s3/import.go
+++ b/dumpling/tests/s3/import.go
@@ -4,11 +4,13 @@ package main
 
 import (
 	"context"
 	"database/sql"
 	"fmt"
+	"net"
 	"os"
+	"strconv"
 
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/pingcap/errors"
 	"github.com/spf13/cobra"
 	"golang.org/x/sync/errgroup"
@@ -46,11 +48,11 @@ func main() {
 		worker, err := cmd.Flags().GetInt(flagWorker)
 		if err != nil {
 			return errors.Trace(err)
 		}
 
-		dsn := fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?charset=utf8mb4", "root", "", "127.0.0.1", port, database)
+		dsn := fmt.Sprintf("%s:%s@tcp(%s)/%s?charset=utf8mb4", "root", "", net.JoinHostPort("127.0.0.1", strconv.Itoa(port)), database)
 		db, err := sql.Open("mysql", dsn)
 		if err != nil {
 			return errors.Trace(err)
 		}
 
diff --git a/util/dbutil/common.go b/util/dbutil/common.go
index 37b6da5fd..df54e18bd 100644
--- a/util/dbutil/common.go
+++ b/util/dbutil/common.go
@@ -17,11 +17,11 @@ package dbutil
 import (
 	"context"
 	"database/sql"
 	"encoding/json"
 	"fmt"
-	"net/url"
+	"net"
 	"os"
 	"strconv"
 	"strings"
 	"time"
 
@@ -105,30 +105,35 @@ func GetDBConfigFromEnv(schema string) DBConfig {
 	}
 }
 
 // OpenDB opens a mysql connection FD
 func OpenDB(cfg DBConfig, vars map[string]string) (*sql.DB, error) {
-	var dbDSN string
+	driverCfg := mysql.NewConfig()
+	driverCfg.Params = make(map[string]string)
+	driverCfg.User = cfg.User
+	driverCfg.Passwd = cfg.Password
+	driverCfg.Net = "tcp"
+	driverCfg.Addr = net.JoinHostPort(cfg.Host, strconv.Itoa(cfg.Port))
+	driverCfg.Params["charset"] = "utf8mb4"
+
 	if len(cfg.Snapshot) != 0 {
 		log.Info("create connection with snapshot", zap.String("snapshot", cfg.Snapshot))
-		dbDSN = fmt.Sprintf("%s:%s@tcp(%s:%d)/?charset=utf8mb4&tidb_snapshot=%s", cfg.User, cfg.Password, cfg.Host, cfg.Port, cfg.Snapshot)
-	} else {
-		dbDSN = fmt.Sprintf("%s:%s@tcp(%s:%d)/?charset=utf8mb4", cfg.User, cfg.Password, cfg.Host, cfg.Port)
+		driverCfg.Params["tidb_snapshot"] = cfg.Snapshot
 	}
 
 	for key, val := range vars {
 		// key='val'. add single quote for better compatibility.
-		dbDSN += fmt.Sprintf("&%s=%%27%s%%27", key, url.QueryEscape(val))
+		driverCfg.Params[key] = fmt.Sprintf("'%s'", val)
 	}
 
-	dbConn, err := sql.Open("mysql", dbDSN)
+	c, err := mysql.NewConnector(driverCfg)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
-
-	err = dbConn.Ping()
-	return dbConn, errors.Trace(err)
+	db := sql.OpenDB(c)
+	err = db.Ping()
+	return db, errors.Trace(err)
 }
 
 // CloseDB closes the mysql fd
 func CloseDB(db *sql.DB) error {
 	if db == nil {

Vulnerable code: 		db, err := common.ConnectMySQL(cfg.Checkpoint.DSN)
		db, err := sql.Open("mysql", cfg.Checkpoint.DSN)
	"net/url"
func (param *MySQLConnectParam) ToDSN() string {
	hostPort := net.JoinHostPort(param.Host, strconv.Itoa(param.Port))
	dsn := fmt.Sprintf("%s:%s@tcp(%s)/?charset=utf8mb4&sql_mode='%s'&maxAllowedPacket=%d&tls=%s",
		param.User, param.Password, hostPort,
		param.SQLMode, param.MaxAllowedPacket, param.TLS)
		dsn += fmt.Sprintf("&%s='%s'", k, url.QueryEscape(v))

	return dsn
func tryConnectMySQL(dsn string) (*sql.DB, error) {
	driverName := "mysql"
	failpoint.Inject("MockMySQLDriver", func(val failpoint.Value) {
		driverName = val.(string)
	db, err := sql.Open(driverName, dsn)
func ConnectMySQL(dsn string) (*sql.DB, error) {
	cfg, err := mysql.ParseDSN(dsn)
	if err != nil {
		return nil, errors.Trace(err)
	}
	db, firstErr := tryConnectMySQL(dsn)
			db, err = tryConnectMySQL(cfg.FormatDSN())
				return db, nil
	db, err := ConnectMySQL(param.ToDSN())
	"database/sql"
	"database/sql/driver"
	"math/rand"
	"strconv"
	tmysql "github.com/pingcap/tidb/errno"
func TestToDSN(t *testing.T) {
	param := common.MySQLConnectParam{
		Host:             "127.0.0.1",
		Port:             4000,
		User:             "root",
		Password:         "123456",
		SQLMode:          "strict",
		MaxAllowedPacket: 1234,
		TLS:              "cluster",
		Vars: map[string]string{
			"tidb_distsql_scan_concurrency": "1",
		},
	}
	require.Equal(t, "root:123456@tcp(127.0.0.1:4000)/?charset=utf8mb4&sql_mode='strict'&maxAllowedPacket=1234&tls=cluster&tidb_distsql_scan_concurrency='1'", param.ToDSN())

	param.Host = "::1"
	require.Equal(t, "root:123456@tcp([::1]:4000)/?charset=utf8mb4&sql_mode='strict'&maxAllowedPacket=1234&tls=cluster&tidb_distsql_scan_concurrency='1'", param.ToDSN())
}

type mockDriver struct {
	driver.Driver
	plainPsw string
}

func (m *mockDriver) Open(dsn string) (driver.Conn, error) {
	cfg, err := mysql.ParseDSN(dsn)
	if err != nil {
		return nil, err
	}
	accessDenied := cfg.Passwd != m.plainPsw
	return &mockConn{accessDenied: accessDenied}, nil
}

type mockConn struct {
	driver.Conn
	driver.Pinger
	accessDenied bool
}

func (c *mockConn) Ping(ctx context.Context) error {
	if c.accessDenied {
		return &mysql.MySQLError{Number: tmysql.ErrAccessDenied, Message: "access denied"}
	}
	return nil
}

func (c *mockConn) Close() error {
	return nil
}

	driverName := "mysql-mock-" + strconv.Itoa(rand.Int())
	sql.Register(driverName, &mockDriver{plainPsw: plainPsw})
		"github.com/pingcap/tidb/br/pkg/lightning/common/MockMySQLDriver",
		fmt.Sprintf("return(\"%s\")", driverName)))
		require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/br/pkg/lightning/common/MockMySQLDriver"))
	db, err := param.Connect()
	require.NoError(t, db.Close())
	db, err = param.Connect()
	require.NoError(t, db.Close())
	Schema           string                 `toml:"schema" json:"schema"`
	DSN              string                 `toml:"dsn" json:"-"` // DSN may contain password, don't expose this to JSON.
	Driver           string                 `toml:"driver" json:"driver"`
	Enable           bool                   `toml:"enable" json:"enable"`
	KeepAfterSuccess CheckpointKeepStrategy `toml:"keep-after-success" json:"keep-after-success"`
			cfg.Checkpoint.DSN = param.ToDSN()
	"github.com/pingcap/tidb/parser/mysql"
	require.Equal(t, "guest:12345@tcp(172.16.30.11:4001)/?charset=utf8mb4&sql_mode='"+mysql.DefaultSQLMode+"'&maxAllowedPacket=67108864&tls=false", taskCfg.Checkpoint.DSN)
	_ "github.com/go-sql-driver/mysql"
	dbDSN := fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?charset=utf8", cfg.User, cfg.Password, cfg.Host, cfg.Port, cfg.Name)
	db, err := sql.Open("mysql", dbDSN)

	return db, nil
var openDBFunc = sql.Open
	pool, err := sql.Open("mysql", conf.GetDSN(""))
	d.dbHandle = pool
	if d.dbHandle, err = resetDBWithSessionParams(d.tctx, pool, conf.GetDSN(""), conf.SessionParams); err != nil {
	dbHandle, err := openDBFunc("mysql", conf.GetDSN(""))
	"net/url"
func resetDBWithSessionParams(tctx *tcontext.Context, db *sql.DB, dsn string, params map[string]interface{}) (*sql.DB, error) {
		dsn += fmt.Sprintf("&%s=%s", k, url.QueryEscape(s))
	newDB, err := sql.Open("mysql", dsn)
	if err == nil {
		// ping to make sure all session parameters are set correctly
		err = newDB.PingContext(tctx)
		if err != nil {
			newDB.Close()
		}
	return newDB, errors.Trace(err)
	openDBFunc = func(_, _ string) (*sql.DB, error) {
		dsn := fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?charset=utf8mb4", "root", "", "127.0.0.1", port, database)
	"net/url"
	var dbDSN string
		dbDSN = fmt.Sprintf("%s:%s@tcp(%s:%d)/?charset=utf8mb4&tidb_snapshot=%s", cfg.User, cfg.Password, cfg.Host, cfg.Port, cfg.Snapshot)
	} else {
		dbDSN = fmt.Sprintf("%s:%s@tcp(%s:%d)/?charset=utf8mb4", cfg.User, cfg.Password, cfg.Host, cfg.Port)
		dbDSN += fmt.Sprintf("&%s=%%27%s%%27", key, url.QueryEscape(val))
	dbConn, err := sql.Open("mysql", dbDSN)

	err = dbConn.Ping()
	return dbConn, errors.Trace(err)
Fixed code: 		var (
			db  *sql.DB
			err error
		)
		if cfg.Checkpoint.MySQLParam != nil {
			db, err = cfg.Checkpoint.MySQLParam.Connect()
		} else {
			db, err = sql.Open("mysql", cfg.Checkpoint.DSN)
		}
		var (
			db  *sql.DB
			err error
		)
		if cfg.Checkpoint.MySQLParam != nil {
			db, err = cfg.Checkpoint.MySQLParam.Connect()
		} else {
			db, err = sql.Open("mysql", cfg.Checkpoint.DSN)
		}
func (param *MySQLConnectParam) ToDriverConfig() *mysql.Config {
	cfg := mysql.NewConfig()
	cfg.Params = make(map[string]string)

	cfg.User = param.User
	cfg.Passwd = param.Password
	cfg.Net = "tcp"
	cfg.Addr = net.JoinHostPort(param.Host, strconv.Itoa(param.Port))
	cfg.Params["charset"] = "utf8mb4"
	cfg.Params["sql_mode"] = fmt.Sprintf("'%s'", param.SQLMode)
	cfg.MaxAllowedPacket = int(param.MaxAllowedPacket)
	cfg.TLSConfig = param.TLS
		cfg.Params[k] = fmt.Sprintf("'%s'", v)
	return cfg
func tryConnectMySQL(cfg *mysql.Config) (*sql.DB, error) {
	failpoint.Inject("MustMySQLPassword", func(val failpoint.Value) {
		pwd := val.(string)
		if cfg.Passwd != pwd {
			failpoint.Return(nil, &mysql.MySQLError{Number: tmysql.ErrAccessDenied, Message: "access denied"})
		}
		failpoint.Return(nil, nil)
	c, err := mysql.NewConnector(cfg)
	db := sql.OpenDB(c)
func ConnectMySQL(cfg *mysql.Config) (*sql.DB, error) {
	db, firstErr := tryConnectMySQL(cfg)
			db2, err := tryConnectMySQL(cfg)
				return db2, nil
	db, err := ConnectMySQL(param.ToDriverConfig())
		"github.com/pingcap/tidb/br/pkg/lightning/common/MustMySQLPassword",
		fmt.Sprintf("return(\"%s\")", plainPsw)))
		require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/br/pkg/lightning/common/MustMySQLPassword"))
	_, err := param.Connect()
	_, err = param.Connect()
	Schema           string                    `toml:"schema" json:"schema"`
	DSN              string                    `toml:"dsn" json:"-"` // DSN may contain password, don't expose this to JSON.
	MySQLParam       *common.MySQLConnectParam `toml:"-" json:"-"`   // For some security reason, we use MySQLParam instead of DSN.
	Driver           string                    `toml:"driver" json:"driver"`
	Enable           bool                      `toml:"enable" json:"enable"`
	KeepAfterSuccess CheckpointKeepStrategy    `toml:"keep-after-success" json:"keep-after-success"`
			cfg.Checkpoint.MySQLParam = &param
	equivalentDSN := taskCfg.Checkpoint.MySQLParam.ToDriverConfig().FormatDSN()
	expectedDSN := "guest:12345@tcp(172.16.30.11:4001)/?tls=false&maxAllowedPacket=67108864&charset=utf8mb4&sql_mode=%27ONLY_FULL_GROUP_BY%2CSTRICT_TRANS_TABLES%2CNO_ZERO_IN_DATE%2CNO_ZERO_DATE%2CERROR_FOR_DIVISION_BY_ZERO%2CNO_AUTO_CREATE_USER%2CNO_ENGINE_SUBSTITUTION%27"
	require.Equal(t, expectedDSN, equivalentDSN)
	mysql2 "github.com/go-sql-driver/mysql"
	driverCfg := mysql2.NewConfig()
	driverCfg.User = cfg.User
	driverCfg.Passwd = cfg.Password
	driverCfg.Net = "tcp"
	driverCfg.Addr = cfg.Host + ":" + strconv.Itoa(cfg.Port)
	driverCfg.DBName = cfg.Name

	c, err := mysql2.NewConnector(driverCfg)
	return sql.OpenDB(c), nil
// GetDriverConfig returns the MySQL driver config from Config.
func (conf *Config) GetDriverConfig(db string) *mysql.Config {
	driverCfg := mysql.NewConfig()
	// maxAllowedPacket=0 can be used to automatically fetch the max_allowed_packet variable from server on every connection.
	// https://github.com/go-sql-driver/mysql#maxallowedpacket
	hostPort := net.JoinHostPort(conf.Host, strconv.Itoa(conf.Port))
	driverCfg.User = conf.User
	driverCfg.Passwd = conf.Password
	driverCfg.Net = "tcp"
	driverCfg.Addr = hostPort
	driverCfg.DBName = db
	driverCfg.Collation = "utf8mb4_general_ci"
	driverCfg.ReadTimeout = conf.ReadTimeout
	driverCfg.WriteTimeout = 30 * time.Second
	driverCfg.InterpolateParams = true
	driverCfg.MaxAllowedPacket = 0
	if conf.Security.DriveTLSName != "" {
		driverCfg.TLSConfig = conf.Security.DriveTLSName
	}
	if conf.AllowCleartextPasswords {
		driverCfg.AllowCleartextPasswords = true
	}
	return driverCfg
}

var openDBFunc = openDB
	c, err := mysql.NewConnector(conf.GetDriverConfig(""))
	d.dbHandle = sql.OpenDB(c)
	if d.dbHandle, err = resetDBWithSessionParams(d.tctx, pool, conf.GetDriverConfig(""), conf.SessionParams); err != nil {
func openDB(cfg *mysql.Config) (*sql.DB, error) {
	c, err := mysql.NewConnector(cfg)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return sql.OpenDB(c), nil
}

	dbHandle, err := openDBFunc(conf.GetDriverConfig(""))
func resetDBWithSessionParams(tctx *tcontext.Context, db *sql.DB, cfg *mysql.Config, params map[string]interface{}) (*sql.DB, error) {
	if cfg.Params == nil {
		cfg.Params = make(map[string]string)
	}

		cfg.Params[k] = s
	c, err := mysql.NewConnector(cfg)
	if err != nil {
		return nil, errors.Trace(err)
	}
	newDB := sql.OpenDB(c)
	// ping to make sure all session parameters are set correctly
	err = newDB.PingContext(tctx)
	if err != nil {
		newDB.Close()
	return newDB, nil
	openDBFunc = func(*mysql.Config) (*sql.DB, error) {
	"net"
	"strconv"
		dsn := fmt.Sprintf("%s:%s@tcp(%s)/%s?charset=utf8mb4", "root", "", net.JoinHostPort("127.0.0.1", strconv.Itoa(port)), database)
	"net"
	driverCfg := mysql.NewConfig()
	driverCfg.Params = make(map[string]string)
	driverCfg.User = cfg.User
	driverCfg.Passwd = cfg.Password
	driverCfg.Net = "tcp"
	driverCfg.Addr = net.JoinHostPort(cfg.Host, strconv.Itoa(cfg.Port))
	driverCfg.Params["charset"] = "utf8mb4"

		driverCfg.Params["tidb_snapshot"] = cfg.Snapshot
		driverCfg.Params[key] = fmt.Sprintf("'%s'", val)
	c, err := mysql.NewConnector(driverCfg)
	db := sql.OpenDB(c)
	err = db.Ping()
	return db, errors.Trace(err)