[DNM]: Ssl content debug

dm/config/security/security.go

@@ -17,6 +17,8 @@ import (
 	"encoding/base64"
 	"fmt"
 	"os"
+
+	certificate "github.com/pingcap/tiflow/pkg/security"
 )
 
 // Security config.
@@ -83,6 +85,9 @@ func (s *Security) ClearSSLBytesData() {
 	s.SSLCABytes = s.SSLCABytes[:0]
 	s.SSLKeyBytes = s.SSLKeyBytes[:0]
 	s.SSLCertBytes = s.SSLCertBytes[:0]
+	s.SSLCA = ""
+	s.SSLCert = ""
+	s.SSLKey = ""
 }
 
 // Clone returns a deep copy of Security.
@@ -95,5 +100,22 @@ func (s *Security) Clone() *Security {
 	clone.SSLCABytes = append([]byte(nil), s.SSLCABytes...)
 	clone.SSLKeyBytes = append([]byte(nil), s.SSLKeyBytes...)
 	clone.SSLCertBytes = append([]byte(nil), s.SSLCertBytes...)
+	clone.SSLCA = s.SSLCA
+	clone.SSLCert = s.SSLCert
+	clone.SSLKey = s.SSLKey
 	return &clone
 }
+
+func (s *Security) WriteFiles(name string) error {
+	var err error
+	if s.SSLCA, err = certificate.WriteFile(fmt.Sprintf("%s_ca.pem", name), s.SSLCABytes); err != nil {
+		return err
+	}
+	if s.SSLCert, err = certificate.WriteFile(fmt.Sprintf("%s_dm.pem", name), s.SSLCertBytes); err != nil {
+		return err
+	}
+	if s.SSLKey, err = certificate.WriteFile(fmt.Sprintf("%s_dm.key", name), s.SSLKeyBytes); err != nil {
+		return err
+	}
+	return nil
+}

dm/config/security_test.go

@@ -106,6 +106,9 @@ func (c *testTLSConfig) TestLoadAndClearContent() {
 	c.Require().Len(s.SSLCABytes, 0)
 	c.Require().Len(s.SSLCertBytes, 0)
 	c.Require().Len(s.SSLKeyBytes, 0)
+	c.Require().Equal(s.SSLCA, "")
+	c.Require().Equal(s.SSLCert, "")
+	c.Require().Equal(s.SSLKey, "")
 
 	s.SSLCABase64 = "MTIz"
 	err = s.LoadTLSContent()

dm/config/task.go

@@ -33,6 +33,7 @@ import (
 	"github.com/pingcap/tidb/pkg/util/filter"
 	router "github.com/pingcap/tidb/pkg/util/table-router"
 	"github.com/pingcap/tiflow/dm/config/dbconfig"
+	"github.com/pingcap/tiflow/dm/config/security"
 	"github.com/pingcap/tiflow/dm/pkg/log"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 	"github.com/pingcap/tiflow/dm/pkg/utils"
@@ -301,6 +302,9 @@ type LoaderConfig struct {
 	RangeConcurrency    int                          `yaml:"range-concurrency" toml:"range-concurrency" json:"range-concurrency"`
 	CompressKVPairs     string                       `yaml:"compress-kv-pairs" toml:"compress-kv-pairs" json:"compress-kv-pairs"`
 	PDAddr              string                       `yaml:"pd-addr" toml:"pd-addr" json:"pd-addr"`
+	// now only creating task by OpenAPI will use the `Security` field to connect PD.
+	// TODO: support setting `Security` by dmctl
+	Security *security.Security `toml:"security" json:"security" yaml:"security"`
 }
 
 // DefaultLoaderConfig return default loader config for task.

dm/config/task_converters.go

@@ -239,6 +239,21 @@ func OpenAPITaskToSubTaskConfigs(task *openapi.Task, toDBCfg *dbconfig.DBConfig,
 			if fullCfg.PdAddr != nil {
 				subTaskCfg.LoaderConfig.PDAddr = *fullCfg.PdAddr
 			}
+			if fullCfg.Security != nil {
+				if fullCfg.Security.SslCaContent == "" || fullCfg.Security.SslCertContent == "" || fullCfg.Security.SslKeyContent == "" {
+					return nil, terror.ErrOpenAPICommonError.Generatef("Invalid security config, full migrate conf's security fields should not be \"\"")
+				}
+				var certAllowedCN []string
+				if fullCfg.Security.CertAllowedCn != nil {
+					certAllowedCN = *fullCfg.Security.CertAllowedCn
+				}
+				subTaskCfg.LoaderConfig.Security = &security.Security{
+					SSLCABytes:    []byte(fullCfg.Security.SslCaContent),
+					SSLCertBytes:  []byte(fullCfg.Security.SslCertContent),
+					SSLKeyBytes:   []byte(fullCfg.Security.SslKeyContent),
+					CertAllowedCN: certAllowedCN,
+				}
+			}
 			if fullCfg.RangeConcurrency != nil {
 				subTaskCfg.LoaderConfig.RangeConcurrency = *fullCfg.RangeConcurrency
 			}
@@ -540,6 +555,14 @@ func SubTaskConfigsToOpenAPITask(subTaskConfigList []*SubTaskConfig) *openapi.Ta
 		DataDir:       &oneSubtaskConfig.LoaderConfig.Dir,
 		ImportThreads: &oneSubtaskConfig.LoaderConfig.PoolSize,
 	}
+	// only load task use physical mode need PD address
+	if oneSubtaskConfig.LoaderConfig.ImportMode == LoadModePhysical {
+		taskSourceConfig.FullMigrateConf.PdAddr = &oneSubtaskConfig.LoaderConfig.PDAddr
+	}
+	importMode := openapi.TaskFullMigrateConfImportMode(oneSubtaskConfig.LoaderConfig.ImportMode)
+	if importMode != "" {
+		taskSourceConfig.FullMigrateConf.ImportMode = &importMode
+	}
 	consistencyInTask := oneSubtaskConfig.MydumperConfig.ExtraArgs
 	consistency := strings.Replace(consistencyInTask, "--consistency ", "", 1)
 	if consistency != "" {
@@ -549,6 +572,18 @@ func SubTaskConfigsToOpenAPITask(subTaskConfigList []*SubTaskConfig) *openapi.Ta
 		ReplBatch:   &oneSubtaskConfig.SyncerConfig.Batch,
 		ReplThreads: &oneSubtaskConfig.SyncerConfig.WorkerCount,
 	}
+	if oneSubtaskConfig.LoaderConfig.Security != nil {
+		var certAllowedCN []string
+		if oneSubtaskConfig.LoaderConfig.Security.CertAllowedCN != nil {
+			certAllowedCN = oneSubtaskConfig.LoaderConfig.Security.CertAllowedCN
+		}
+		taskSourceConfig.FullMigrateConf.Security = &openapi.Security{
+			SslCaContent:   string(oneSubtaskConfig.LoaderConfig.Security.SSLCABytes),
+			SslCertContent: string(oneSubtaskConfig.LoaderConfig.Security.SSLCertBytes),
+			SslKeyContent:  string(oneSubtaskConfig.LoaderConfig.Security.SSLKeyBytes),
+			CertAllowedCn:  &certAllowedCN,
+		}
+	}
 	// set filter rules
 	filterRuleMap := openapi.Task_BinlogFilterRule{}
 	for sourceName, ruleList := range filterMap {
@@ -660,6 +695,18 @@ func SubTaskConfigsToOpenAPITask(subTaskConfigList []*SubTaskConfig) *openapi.Ta
 		ignoreItems := oneSubtaskConfig.IgnoreCheckingItems
 		task.IgnoreCheckingItems = &ignoreItems
 	}
+	if oneSubtaskConfig.To.Security != nil {
+		var certAllowedCN []string
+		if oneSubtaskConfig.To.Security.CertAllowedCN != nil {
+			certAllowedCN = oneSubtaskConfig.To.Security.CertAllowedCN
+		}
+		task.TargetConfig.Security = &openapi.Security{
+			SslCaContent:   string(oneSubtaskConfig.To.Security.SSLCABytes),
+			SslCertContent: string(oneSubtaskConfig.To.Security.SSLCertBytes),
+			SslKeyContent:  string(oneSubtaskConfig.To.Security.SSLKeyBytes),
+			CertAllowedCn:  &certAllowedCN,
+		}
+	}
 	return &task
 }
 

dm/config/task_converters_test.go

@@ -20,6 +20,7 @@ import (
 	"github.com/pingcap/check"
 	"github.com/pingcap/tidb/pkg/util/filter"
 	"github.com/pingcap/tiflow/dm/config/dbconfig"
+	"github.com/pingcap/tiflow/dm/config/security"
 	"github.com/pingcap/tiflow/dm/openapi"
 	"github.com/pingcap/tiflow/dm/openapi/fixtures"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
@@ -65,6 +66,12 @@ func testNoShardTaskToSubTaskConfigs(c *check.C) {
 		Port:     task.TargetConfig.Port,
 		User:     task.TargetConfig.User,
 		Password: task.TargetConfig.Password,
+		Security: &security.Security{
+			SSLCABytes:    []byte(task.TargetConfig.Security.SslCaContent),
+			SSLCertBytes:  []byte(task.TargetConfig.Security.SslCertContent),
+			SSLKeyBytes:   []byte(task.TargetConfig.Security.SslKeyContent),
+			CertAllowedCN: *task.TargetConfig.Security.CertAllowedCn,
+		},
 	}
 	// change meta
 	newMeta := "new_dm_meta"
@@ -125,6 +132,13 @@ func testNoShardTaskToSubTaskConfigs(c *check.C) {
 	c.Assert(subTaskConfig.DumpIOTotalBytes.Load(), check.Equals, uint64(0))
 	c.Assert(subTaskConfig.UUID, check.HasLen, len(uuid.NewString()))
 	c.Assert(subTaskConfig.DumpUUID, check.HasLen, len(uuid.NewString()))
+	// check security items
+	c.Assert(string(subTaskConfig.To.Security.SSLCABytes), check.Equals, task.TargetConfig.Security.SslCaContent)
+	c.Assert(string(subTaskConfig.To.Security.SSLCertBytes), check.Equals, task.TargetConfig.Security.SslCertContent)
+	c.Assert(string(subTaskConfig.To.Security.SSLKeyBytes), check.Equals, task.TargetConfig.Security.SslKeyContent)
+	c.Assert(string(subTaskConfig.LoaderConfig.Security.SSLCABytes), check.Equals, task.SourceConfig.FullMigrateConf.Security.SslCaContent)
+	c.Assert(string(subTaskConfig.LoaderConfig.Security.SSLCertBytes), check.Equals, task.SourceConfig.FullMigrateConf.Security.SslCertContent)
+	c.Assert(string(subTaskConfig.LoaderConfig.Security.SSLKeyBytes), check.Equals, task.SourceConfig.FullMigrateConf.Security.SslKeyContent)
 }
 
 func testShardAndFilterTaskToSubTaskConfigs(c *check.C) {
@@ -284,6 +298,12 @@ func testNoShardSubTaskConfigsToOpenAPITask(c *check.C) {
 		Port:     task.TargetConfig.Port,
 		User:     task.TargetConfig.User,
 		Password: task.TargetConfig.Password,
+		Security: &security.Security{
+			SSLCABytes:    []byte(task.TargetConfig.Security.SslCaContent),
+			SSLCertBytes:  []byte(task.TargetConfig.Security.SslCertContent),
+			SSLKeyBytes:   []byte(task.TargetConfig.Security.SslKeyContent),
+			CertAllowedCN: *task.TargetConfig.Security.CertAllowedCn,
+		},
 	}
 	subTaskConfigList, err := OpenAPITaskToSubTaskConfigs(&task, toDBCfg, sourceCfgMap)
 	c.Assert(err, check.IsNil)
@@ -370,6 +390,12 @@ func TestConvertWithIgnoreCheckItems(t *testing.T) {
 		Port:     task.TargetConfig.Port,
 		User:     task.TargetConfig.User,
 		Password: task.TargetConfig.Password,
+		Security: &security.Security{
+			SSLCABytes:    []byte(task.TargetConfig.Security.SslCaContent),
+			SSLCertBytes:  []byte(task.TargetConfig.Security.SslCertContent),
+			SSLKeyBytes:   []byte(task.TargetConfig.Security.SslKeyContent),
+			CertAllowedCN: *task.TargetConfig.Security.CertAllowedCn,
+		},
 	}
 	subTaskConfigList, err := OpenAPITaskToSubTaskConfigs(&task, toDBCfg, sourceCfgMap)
 	require.NoError(t, err)

dm/config/task_test.go

@@ -532,6 +532,12 @@ func TestGenAndFromSubTaskConfigs(t *testing.T) {
 			"sql_mode":  " NO_AUTO_VALUE_ON_ZERO,ANSI_QUOTES",
 			"time_zone": "+00:00",
 		}
+		security2 = security.Security{
+			SSLCA:         "/path/to/ca2",
+			SSLCert:       "/path/to/cert2",
+			SSLKey:        "/path/to/key2",
+			CertAllowedCN: []string{"allowed-cn"},
+		}
 		security = security.Security{
 			SSLCA:         "/path/to/ca",
 			SSLCert:       "/path/to/cert",
@@ -674,6 +680,7 @@ func TestGenAndFromSubTaskConfigs(t *testing.T) {
 				PDAddr:              "http://test:2379",
 				RangeConcurrency:    32,
 				CompressKVPairs:     "gzip",
+				Security:            &security2,
 			},
 			SyncerConfig: SyncerConfig{
 				WorkerCount:             32,

dm/loader/lightning.go

@@ -279,7 +279,9 @@ func (l *LightningLoader) runLightning(ctx context.Context, cfg *lcfg.Config) (e
 	if l.cfg.LoaderConfig.ImportMode == config.LoadModePhysical {
 		opts = append(opts, lserver.WithDupIndicator(&hasDup))
 	}
-
+	l.logger.Debug("ssl content debug", zap.Any("task cfg", cfg))
+	l.logger.Debug("ssl content debug", zap.String("ca content", string(cfg.Security.CABytes)), zap.String("cert content", string(cfg.Security.CertBytes)), zap.String("key content", string(cfg.Security.KeyBytes)))
+	l.logger.Debug("ssl content debug", zap.String("ca content", string(cfg.TiDB.Security.CABytes)), zap.String("cert content", string(cfg.TiDB.Security.CertBytes)), zap.String("key content", string(cfg.TiDB.Security.KeyBytes)))
 	err = l.core.RunOnceWithOptions(taskCtx, cfg, opts...)
 	failpoint.Inject("LoadDataSlowDown", nil)
 	failpoint.Inject("LoadDataSlowDownByTask", func(val failpoint.Value) {
@@ -329,6 +331,12 @@ func GetLightningConfig(globalCfg *lcfg.GlobalConfig, subtaskCfg *config.SubTask
 	if err := cfg.LoadFromGlobal(globalCfg); err != nil {
 		return nil, err
 	}
+	cfg.TiDB.Security = &globalCfg.Security
+	if subtaskCfg.LoaderConfig.Security != nil {
+		cfg.Security.CABytes = subtaskCfg.LoaderConfig.Security.SSLCABytes
+		cfg.Security.CertBytes = subtaskCfg.LoaderConfig.Security.SSLCertBytes
+		cfg.Security.KeyBytes = subtaskCfg.LoaderConfig.Security.SSLKeyBytes
+	}
 	// TableConcurrency is adjusted to the value of RegionConcurrency
 	// when using TiDB backend.
 	// TODO: should we set the TableConcurrency separately.
@@ -342,6 +350,9 @@ func GetLightningConfig(globalCfg *lcfg.GlobalConfig, subtaskCfg *config.SubTask
 		if err := cfg.Security.BuildTLSConfig(); err != nil {
 			return nil, err
 		}
+		if err := cfg.TiDB.Security.BuildTLSConfig(); err != nil {
+			return nil, err
+		}
 		// To enable the loader worker failover, we need to use jobID+sourceID to isolate the checkpoint schema
 		cfg.Checkpoint.Schema = cputil.LightningCheckpointSchema(subtaskCfg.Name, subtaskCfg.SourceID)
 		cfg.Checkpoint.Driver = lcfg.CheckpointDriverMySQL
@@ -657,7 +668,7 @@ func connParamFromConfig(config *lcfg.Config) *common.MySQLConnectParam {
 		SQLMode:  mysql.DefaultSQLMode,
 		// TODO: keep same as Lightning defaultMaxAllowedPacket later
 		MaxAllowedPacket:         64 * 1024 * 1024,
-		TLSConfig:                config.Security.TLSConfig,
-		AllowFallbackToPlaintext: config.Security.AllowFallbackToPlaintext,
+		TLSConfig:                config.TiDB.Security.TLSConfig,
+		AllowFallbackToPlaintext: config.TiDB.Security.AllowFallbackToPlaintext,
 	}
 }

dm/loader/lightning_test.go

@@ -21,7 +21,10 @@ import (
 	"github.com/pingcap/tidb/pkg/lightning/common"
 	lcfg "github.com/pingcap/tidb/pkg/lightning/config"
 	"github.com/pingcap/tiflow/dm/config"
+	"github.com/pingcap/tiflow/dm/config/dbconfig"
+	"github.com/pingcap/tiflow/dm/config/security"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
+	certificate "github.com/pingcap/tiflow/pkg/security"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/stretchr/testify/require"
@@ -99,6 +102,74 @@ func TestGetLightiningConfig(t *testing.T) {
 		})
 	require.NoError(t, err)
 	require.Equal(t, lcfg.CheckpointDriverMySQL, conf.Checkpoint.Driver)
+
+	ca, err := certificate.NewCA()
+	require.NoError(t, err)
+	cert, key, err := ca.GenerateCerts("dm")
+	require.NoError(t, err)
+	caPath, err := certificate.WriteFile("dm-test-client-cert", ca.CAPEM)
+	require.NoError(t, err)
+	certPath, err := certificate.WriteFile("dm-test-client-cert", cert)
+	require.NoError(t, err)
+	keyPath, err := certificate.WriteFile("dm-test-client-key", key)
+	require.NoError(t, err)
+	ca, err = certificate.NewCA()
+	require.NoError(t, err)
+	cert, key, err = ca.GenerateCerts("dm")
+	require.NoError(t, err)
+	caPath2, err := certificate.WriteFile("dm-test-client-cert2", ca.CAPEM)
+	require.NoError(t, err)
+	certPath2, err := certificate.WriteFile("dm-test-client-cert2", cert)
+	require.NoError(t, err)
+	keyPath2, err := certificate.WriteFile("dm-test-client-key2", key)
+	require.NoError(t, err)
+
+	conf, err = GetLightningConfig(
+		&lcfg.GlobalConfig{Security: lcfg.Security{CAPath: caPath, CertPath: certPath, KeyPath: keyPath}},
+		&config.SubTaskConfig{
+			LoaderConfig: config.LoaderConfig{Security: &security.Security{SSLCA: caPath, SSLCert: certPath, SSLKey: keyPath}},
+			To:           dbconfig.DBConfig{Security: &security.Security{SSLCA: caPath2, SSLCert: certPath2, SSLKey: keyPath2}},
+		})
+	require.NoError(t, err)
+	require.Equal(t, conf.Security.CAPath, caPath)
+	require.Equal(t, conf.Security.CertPath, certPath)
+	require.Equal(t, conf.Security.KeyPath, keyPath)
+	require.Equal(t, conf.TiDB.Security.CAPath, caPath2)
+	require.Equal(t, conf.TiDB.Security.CertPath, certPath2)
+	require.Equal(t, conf.TiDB.Security.KeyPath, keyPath2)
+	conf, err = GetLightningConfig(
+		&lcfg.GlobalConfig{Security: lcfg.Security{CAPath: caPath, CertPath: certPath, KeyPath: keyPath}},
+		&config.SubTaskConfig{
+			LoaderConfig: config.LoaderConfig{Security: &security.Security{SSLCA: caPath, SSLCert: certPath, SSLKey: keyPath}},
+			To:           dbconfig.DBConfig{},
+		})
+	require.NoError(t, err)
+	require.Equal(t, conf.Security.CAPath, caPath)
+	require.Equal(t, conf.Security.CertPath, certPath)
+	require.Equal(t, conf.Security.KeyPath, keyPath)
+	require.Equal(t, conf.TiDB.Security.CAPath, caPath)
+	require.Equal(t, conf.TiDB.Security.CertPath, certPath)
+	require.Equal(t, conf.TiDB.Security.KeyPath, keyPath)
+	conf, err = GetLightningConfig(
+		&lcfg.GlobalConfig{},
+		&config.SubTaskConfig{
+			LoaderConfig: config.LoaderConfig{},
+			To:           dbconfig.DBConfig{Security: &security.Security{SSLCA: caPath2, SSLCert: certPath2, SSLKey: keyPath2}},
+		})
+	require.NoError(t, err)
+	require.Equal(t, conf.Security.CAPath, "")
+	require.Equal(t, conf.Security.CertPath, "")
+	require.Equal(t, conf.Security.KeyPath, "")
+	require.Equal(t, conf.TiDB.Security.CAPath, caPath2)
+	require.Equal(t, conf.TiDB.Security.CertPath, certPath2)
+	require.Equal(t, conf.TiDB.Security.KeyPath, keyPath2)
+	// invalid security file path
+	_, err = GetLightningConfig(
+		&lcfg.GlobalConfig{Security: lcfg.Security{CAPath: "caPath"}},
+		&config.SubTaskConfig{
+			To: dbconfig.DBConfig{Security: &security.Security{SSLCA: "caPath"}},
+		})
+	require.EqualError(t, err, "could not read ca certificate: open caPath: no such file or directory")
 }
 
 func TestMetricProxies(t *testing.T) {

dm/openapi/fixtures/task.go

@@ -31,7 +31,15 @@ var (
 		  "full_migrate_conf": {
 			"data_dir": "./exported_data",
 			"export_threads": 4,
-			"import_threads": 16
+			"import_threads": 16,
+			"import_mode": "physical",
+			"pd_addr": "127.0.0.1:2379",
+			"security": {
+				"ssl_ca_content": "ca1",
+				"ssl_cert_content": "cert1",
+				"ssl_key_content": "key1",
+				"cert_allowed_cn": ["PD1", "PD2"]
+			}
 		  },
 		  "incr_migrate_conf": { "repl_batch": 200, "repl_threads": 32 },
 		  "source_conf": [{ "source_name": "mysql-replica-01" }]
@@ -50,7 +58,12 @@ var (
 		  "host": "root",
 		  "password": "123456",
 		  "port": 4000,
-		  "security": null,
+		  "security": {
+		      "ssl_ca_content": "ca2",
+			  "ssl_cert_content": "cert2",
+			  "ssl_key_content": "key2",
+			  "cert_allowed_cn": ["TiDB1", "TiDB2"]
+		  },
 		  "user": "root"
 		},
 		"task_mode": "all",
@@ -110,6 +123,7 @@ var (
 		  "full_migrate_conf": {
 			"data_dir": "./exported_data",
 			"export_threads": 4,
+			"import_mode": "logical",
 			"import_threads": 16
 		  },
 		  "incr_migrate_conf": { "repl_batch": 200, "repl_threads": 32 },