fix: fix password as hidden when update rule (#2786)

Signed-off-by: yisaer <[email protected]>

internal/server/rest.go

@@ -622,13 +622,18 @@ func ruleHandler(w http.ResponseWriter, r *http.Request) {
 			handleError(w, err, "Invalid body", logger)
 			return
 		}
-		err = updateRule(name, string(body), true)
+		newRuleJson, err := replaceRulePassword(name, string(body))
+		if err != nil {
+			handleError(w, err, "Invalid body", logger)
+			return
+		}
+		err = updateRule(name, newRuleJson, true)
 		if err != nil {
 			handleError(w, err, "Update rule error", logger)
 			return
 		}
 		// Update to db after validation
-		_, err = ruleProcessor.ExecUpdate(name, string(body))
+		_, err = ruleProcessor.ExecUpdate(name, newRuleJson)
 		if err != nil {
 			handleError(w, err, "Update rule error, suggest to delete it and recreate", logger)
 			return

internal/server/rest_test.go

@@ -816,3 +816,31 @@ func (suite *RestTestSuite) TestCreateDuplicateRule() {
 	returnVal, _ = io.ReadAll(w2.Result().Body)
 	require.Equal(suite.T(), `{"error":1000,"message":"rule test12345 already exists"}`+"\n", string(returnVal))
 }
+
+func (suite *RestTestSuite) TestSinkHiddenPassword() {
+	buf1 := bytes.NewBuffer([]byte(`{"sql":"CREATE stream demo78() WITH (DATASOURCE=\"0\", TYPE=\"mqtt\")"}`))
+	req1, _ := http.NewRequest(http.MethodPost, "http://localhost:8080/streams", buf1)
+	w1 := httptest.NewRecorder()
+	suite.r.ServeHTTP(w1, req1)
+
+	ruleJson2 := `{"triggered":false,"id":"rule34","sql":"select * from demo78;","actions":[{"mqtt":{"server":"tcp://broker.emqx.io:1883","topic":"devices/demo_001/messages/events/","qos":0,"clientId":"demo_001","username":"xyz.azure-devices.net/demo_001/?api-version=2018-06-30","password":"12345"}}]}`
+	buf2 := bytes.NewBuffer([]byte(ruleJson2))
+	req2, _ := http.NewRequest(http.MethodPost, "http://localhost:8080/rules", buf2)
+	w2 := httptest.NewRecorder()
+	suite.r.ServeHTTP(w2, req2)
+	require.Equal(suite.T(), http.StatusCreated, w2.Code)
+
+	ruleJson2 = `{"triggered":false,"id":"rule34","sql":"select * from demo78;","actions":[{"mqtt":{"server":"tcp://broker.emqx.io:1883","topic":"devices/demo_001/messages/events/","qos":0,"clientId":"demo_001","username":"xyz.azure-devices.net/demo_001/?api-version=2018-06-30","password":"******"}}]}`
+	buf2 = bytes.NewBuffer([]byte(ruleJson2))
+	req2, _ = http.NewRequest(http.MethodPut, "http://localhost:8080/rules/rule34", buf2)
+	w2 = httptest.NewRecorder()
+	suite.r.ServeHTTP(w2, req2)
+	require.Equal(suite.T(), http.StatusOK, w2.Code)
+
+	ruleJson, err := ruleProcessor.GetRuleJson("rule34")
+	require.NoError(suite.T(), err)
+	r := &api.Rule{}
+	require.NoError(suite.T(), json.Unmarshal([]byte(ruleJson), r))
+	m := r.Actions[0]["mqtt"].(map[string]interface{})
+	require.Equal(suite.T(), "12345", m["password"])
+}

internal/server/rule_manager.go

@@ -30,6 +30,7 @@ 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"
 )
 
@@ -173,6 +174,59 @@ func replacePasswdForConfig(typ string, name string, config map[string]interface
 	return config
 }
 
+func replaceRulePassword(id, ruleJson string) (string, error) {
+	r := &api.Rule{
+		Triggered: true,
+	}
+	if err := json.Unmarshal([]byte(ruleJson), r); err != nil {
+		return "", err
+	}
+	existsRule, err := ruleProcessor.GetRuleById(id)
+	if err != nil {
+		return "", err
+	}
+
+	var replacePassword bool
+	for i, action := range r.Actions {
+		if i >= len(existsRule.Actions) {
+			break
+		}
+		for k, v := range action {
+			if m, ok := v.(map[string]interface{}); ok {
+				for key := range hidden.GetHiddenKeys() {
+					if v, ok := m[key]; ok && v == hidden.PASSWORD {
+						oldAction := existsRule.Actions[i]
+						oldV, ok := oldAction[k]
+						if ok {
+							if oldM, ok := oldV.(map[string]interface{}); ok {
+								oldPasswordValue, ok := oldM[key]
+								if ok {
+									oldPasswordStr, ok := oldPasswordValue.(string)
+									if ok && oldPasswordStr != hidden.PASSWORD {
+										m[key] = oldPasswordStr
+										action[k] = m
+										r.Actions[i] = action
+										replacePassword = true
+										continue
+									}
+								}
+							}
+						}
+					}
+				}
+			}
+		}
+	}
+	if !replacePassword {
+		return ruleJson, nil
+	}
+	b, err := json.Marshal(r)
+	if err != nil {
+		return "", err
+	}
+	return string(b), nil
+}
+
 func updateRule(ruleId, ruleJson string, replacePasswd bool) error {
 	// Validate the rule json
 	r, err := ruleProcessor.GetRuleByJson(ruleId, ruleJson)

pkg/hidden/hidden.go

@@ -126,3 +126,7 @@ func ReplaceUrl(resource, config map[string]interface{}) map[string]interface{}
 	}
 	return config
 }
+
+func GetHiddenKeys() map[string]struct{} {
+	return hiddenPasswdKey
+}