Backport: Fix MySQL Plugin password special character escape bug

plugins/database/mysql/mysql_test.go

@@ -321,6 +321,35 @@ func TestMySQL_SetCredentials(t *testing.T) {
 	}
 }
 
+func TestMySQL_Initialize_ReservedChars(t *testing.T) {
+	pw := "#secret!%25#{@}"
+	cleanup, connURL := mysqlhelper.PrepareMySQLTestContainer(t, false, pw)
+	defer cleanup()
+
+	// Revert password set to test replacement by db.Init
+	connURL = strings.ReplaceAll(connURL, pw, "{{password}}")
+
+	connectionDetails := map[string]interface{}{
+		"connection_url": connURL,
+		"password":       pw,
+	}
+
+	db := new(MetadataLen, MetadataLen, UsernameLen)
+	_, err := db.Init(context.Background(), connectionDetails, true)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	if !db.Initialized {
+		t.Fatal("Database should be initialized")
+	}
+
+	err = db.Close()
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+}
+
 func createTestMySQLUser(t *testing.T, connURL, username, password, query string) {
 	t.Helper()
 	db, err := sql.Open("mysql", connURL)

sdk/database/helper/connutil/sql.go

@@ -55,11 +55,17 @@ func (c *SQLConnectionProducer) Init(ctx context.Context, conf map[string]interf
 		return nil, fmt.Errorf("connection_url cannot be empty")
 	}
 
+	// Don't escape special characters for MySQL password
+	password := c.Password
+	if c.Type != "mysql" {
+		password = url.PathEscape(c.Password)
+	}
+
 	// QueryHelper doesn't do any SQL escaping, but if it starts to do so
 	// then maybe we won't be able to use it to do URL substitution any more.
 	c.ConnectionURL = dbutil.QueryHelper(c.ConnectionURL, map[string]string{
 		"username": url.PathEscape(c.Username),
-		"password": url.PathEscape(c.Password),
+		"password": password,
 	})
 
 	if c.MaxOpenConnections == 0 {