fix: fix confKey API can't replace url/passwd

internal/meta/yamlConfigMeta.go

@@ -206,7 +206,7 @@
 	if nil != err {
 		return fmt.Errorf(`%s%s.%v`, getMsg(language, source, "type_conversion_fail"), plgName, err)
 	}
-
+	reqField = replacePasswdForConfig("source", plgName, confKey, reqField)
 	var cfgOps conf.ConfigOperator
 	var found bool
 
@@ -238,6 +238,7 @@
 	if nil != err {
 		return fmt.Errorf(`%s%s.%v`, getMsg(language, sink, "type_conversion_fail"), plgName, err)
 	}
+	reqField = replacePasswdForConfig("sink", plgName, confKey, reqField)
 
 	var cfgOps conf.ConfigOperator
 	var found bool
@@ -654,3 +655,30 @@
 	}
 	return configResponse
 }
+
+func ReplacePasswdForConfig(typ, name, resourceID string, config map[string]interface{}) map[string]interface{} {
+	ConfigManager.lock.RLock()
+	defer ConfigManager.lock.RUnlock()
+	return replacePasswdForConfig(typ, name, resourceID, config)
+}
+
+// replacePasswdForConfig reload password from resources if the config both include password(as fake password) and resourceId
+func replacePasswdForConfig(typ, name, resourceID string, config map[string]interface{}) map[string]interface{} {
+	var configOperatorKey string
+	switch typ {
+	case "sink":
+		configOperatorKey = fmt.Sprintf(SinkCfgOperatorKeyTemplate, name)
+	case "source":
+		configOperatorKey = fmt.Sprintf(SourceCfgOperatorKeyTemplate, name)
+	case "connection":
+		configOperatorKey = fmt.Sprintf(ConnectionCfgOperatorKeyTemplate, name)
+	}
+	cfgOp, ok := ConfigManager.cfgOperators[configOperatorKey]
+	if ok {
+		if resource, ok := cfgOp.CopyUpdatableConfContent()[resourceID]; ok {
+			config = hidden.ReplacePasswd(resource, config)
+			config = hidden.ReplaceUrl(resource, config)
+		}
+	}
+	return config
+}

internal/meta/yamlConfigMeta_test.go

@@ -15,10 +15,13 @@
 package meta
 
 import (
+	"encoding/json"
 	"os"
 	"path/filepath"
 	"testing"
 
+	"github.com/stretchr/testify/require"
+
 	"github.com/lf-edge/ekuiper/internal/conf"
 )
 
@@ -72,3 +75,25 @@ func TestYamlConfigMeta_Ops(t *testing.T) {
 		t.Error("should overwrite exist config key")
 	}
 }
+
+func TestConfKeyReplace(t *testing.T) {
+	a1 := map[string]interface{}{
+		"url": "mysql://root:[email protected]:3306/test",
+		"key": "value",
+	}
+	bs, err := json.Marshal(a1)
+	require.NoError(t, err)
+	err = AddSinkConfKey("sql", "mysql", "", bs)
+	require.NoError(t, err)
+	a2 := map[string]interface{}{
+		"url": "mysql://root:******@127.0.0.1:3306/test",
+		"key": "value",
+	}
+	replaced := replacePasswdForConfig("sink", "sql", "mysql", a2)
+	require.Equal(t, a1, replaced)
+
+	err = AddSourceConfKey("sql", "mysql", "", bs)
+	require.NoError(t, err)
+	replaced = replacePasswdForConfig("source", "sql", "mysql", a2)
+	require.Equal(t, a1, replaced)
+}

internal/server/rule_manager.go

@@ -31,7 +31,6 @@ import (
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
 	"github.com/lf-edge/ekuiper/pkg/errorx"
-	"github.com/lf-edge/ekuiper/pkg/hidden"
 	"github.com/lf-edge/ekuiper/pkg/infra"
 )
 
@@ -161,22 +160,7 @@ func recoverRule(r *api.Rule) string {
 func replacePasswdForConfig(typ string, name string, config map[string]interface{}) map[string]interface{} {
 	if r, ok := config["resourceId"]; ok {
 		if resourceId, ok := r.(string); ok {
-			var configOperatorKey string
-			switch typ {
-			case "sink":
-				configOperatorKey = fmt.Sprintf(meta.SinkCfgOperatorKeyTemplate, name)
-			case "source":
-				configOperatorKey = fmt.Sprintf(meta.SourceCfgOperatorKeyTemplate, name)
-			case "connection":
-				configOperatorKey = fmt.Sprintf(meta.ConnectionCfgOperatorKeyTemplate, name)
-			}
-			cfgOp, ok := meta.GetConfOperator(configOperatorKey)
-			if ok {
-				if resource, ok := cfgOp.CopyUpdatableConfContent()[resourceId]; ok {
-					config = hidden.ReplacePasswd(resource, config)
-					config = hidden.ReplaceUrl(resource, config)
-				}
-			}
+			return meta.ReplacePasswdForConfig(typ, name, resourceId, config)
 		}
 	}
 	return config