api,sink: mask sink uri secrets (#5093)

.gitignore

@@ -4,6 +4,7 @@ tools/bin
 tools/include
 tools/workload/bin
 
+.design
 .issue
 .vscode
 .idea

api/v2/changefeed.go

@@ -52,6 +52,27 @@ import (
 	"go.uber.org/zap"
 )
 
+// validateChangefeedIDParam extracts and validates the changefeed ID from the
+// URL path parameter. On failure it writes the error to c and returns false.
+func validateChangefeedIDParam(c *gin.Context) (common.ChangeFeedDisplayName, bool) {
+	changefeedDisplayName := common.NewChangeFeedDisplayName(c.Param(api.APIOpVarChangefeedID), GetKeyspaceValueWithDefault(c))
+	if err := common.ValidateChangefeedID(changefeedDisplayName.Name); err != nil {
+		_ = c.Error(errors.ErrAPIInvalidParam.GenWithStack("invalid changefeed_id: %s, %s",
+			changefeedDisplayName.Name, err.Error()))
+		return common.ChangeFeedDisplayName{}, false
+	}
+	return changefeedDisplayName, true
+}
+
+func maskSinkURIForError(sinkURI string) string {
+	return util.MaskSensitiveDataInURIForError(sinkURI)
+}
+
+func genSinkURIInvalidError(sinkURI string, err error) error {
+	return errors.WrapError(
+		errors.ErrSinkURIInvalid, util.MaskSensitiveDataInURLError(err), maskSinkURIForError(sinkURI))
+}
+
 // CreateChangefeed handles create changefeed request,
 // it returns the changefeed's changefeedInfo that it just created
 // CreateChangefeed creates a changefeed
@@ -135,7 +156,7 @@ func (h *OpenAPIV2) CreateChangefeed(c *gin.Context) {
 	}
 	sinkURIParsed, err := url.Parse(cfg.SinkURI)
 	if err != nil {
-		_ = c.Error(errors.WrapError(errors.ErrSinkURIInvalid, err, cfg.SinkURI))
+		_ = c.Error(genSinkURIInvalidError(cfg.SinkURI, err))
 		return
 	}
 
@@ -144,7 +165,7 @@ func (h *OpenAPIV2) CreateChangefeed(c *gin.Context) {
 	if config.IsMQScheme(scheme) {
 		topic, err = helper.GetTopic(sinkURIParsed)
 		if err != nil {
-			_ = c.Error(errors.WrapError(errors.ErrSinkURIInvalid, err, cfg.SinkURI))
+			_ = c.Error(errors.WrapError(errors.ErrSinkURIInvalid, err, maskSinkURIForError(cfg.SinkURI)))
 			return
 		}
 	}
@@ -291,7 +312,7 @@ func (h *OpenAPIV2) CreateChangefeed(c *gin.Context) {
 	}
 	err = sink.Verify(ctx, cfConfig, changefeedID)
 	if err != nil {
-		_ = c.Error(errors.WrapError(errors.ErrSinkURIInvalid, err, cfg.SinkURI))
+		_ = c.Error(errors.WrapError(errors.ErrSinkURIInvalid, err, maskSinkURIForError(cfg.SinkURI)))
 		return
 	}
 
@@ -431,7 +452,7 @@ func (h *OpenAPIV2) VerifyTable(c *gin.Context) {
 	// verify replicaConfig
 	sinkURIParsed, err := url.Parse(cfg.SinkURI)
 	if err != nil {
-		_ = c.Error(errors.WrapError(errors.ErrSinkURIInvalid, err, cfg.SinkURI))
+		_ = c.Error(genSinkURIInvalidError(cfg.SinkURI, err))
 		return
 	}
 	err = replicaCfg.ValidateAndAdjust(sinkURIParsed)
@@ -445,7 +466,7 @@ func (h *OpenAPIV2) VerifyTable(c *gin.Context) {
 	if config.IsMQScheme(scheme) {
 		topic, err = helper.GetTopic(sinkURIParsed)
 		if err != nil {
-			_ = c.Error(errors.WrapError(errors.ErrSinkURIInvalid, err, cfg.SinkURI))
+			_ = c.Error(errors.WrapError(errors.ErrSinkURIInvalid, err, maskSinkURIForError(cfg.SinkURI)))
 			return
 		}
 	}
@@ -822,14 +843,14 @@ func (h *OpenAPIV2) ResumeChangefeed(c *gin.Context) {
 		)
 		sinkURIParsed, err = url.Parse(cfInfo.SinkURI)
 		if err != nil {
-			_ = c.Error(errors.WrapError(errors.ErrSinkURIInvalid, err, cfInfo.SinkURI))
+			_ = c.Error(genSinkURIInvalidError(cfInfo.SinkURI, err))
 			return
 		}
 		scheme := sinkURIParsed.Scheme
 		if config.IsMQScheme(scheme) {
 			topic, err = helper.GetTopic(sinkURIParsed)
 			if err != nil {
-				_ = c.Error(errors.WrapError(errors.ErrSinkURIInvalid, err, cfInfo.SinkURI))
+				_ = c.Error(errors.WrapError(errors.ErrSinkURIInvalid, err, maskSinkURIForError(cfInfo.SinkURI)))
 				return
 			}
 		}
@@ -975,7 +996,7 @@ func (h *OpenAPIV2) UpdateChangefeed(c *gin.Context) {
 		// verify replicaConfig
 		sinkURIParsed, err := url.Parse(oldCfInfo.SinkURI)
 		if err != nil {
-			_ = c.Error(errors.WrapError(errors.ErrSinkURIInvalid, err, oldCfInfo.SinkURI))
+			_ = c.Error(genSinkURIInvalidError(oldCfInfo.SinkURI, err))
 			return
 		}
 		err = oldCfInfo.Config.ValidateAndAdjust(sinkURIParsed)
@@ -989,7 +1010,7 @@ func (h *OpenAPIV2) UpdateChangefeed(c *gin.Context) {
 		if config.IsMQScheme(scheme) {
 			topic, err = helper.GetTopic(sinkURIParsed)
 			if err != nil {
-				_ = c.Error(errors.WrapError(errors.ErrSinkURIInvalid, err, oldCfInfo.SinkURI))
+				_ = c.Error(errors.WrapError(errors.ErrSinkURIInvalid, err, maskSinkURIForError(oldCfInfo.SinkURI)))
 				return
 			}
 		}
@@ -1033,7 +1054,7 @@ func (h *OpenAPIV2) UpdateChangefeed(c *gin.Context) {
 
 	err = sink.Verify(ctx, oldCfInfo.ToChangefeedConfig(), oldCfInfo.ChangefeedID)
 	if err != nil {
-		_ = c.Error(errors.WrapError(errors.ErrSinkURIInvalid, err, oldCfInfo.SinkURI))
+		_ = c.Error(errors.WrapError(errors.ErrSinkURIInvalid, err, maskSinkURIForError(oldCfInfo.SinkURI)))
 		return
 	}
 

api/v2/changefeed_test.go

@@ -14,6 +14,7 @@
 package v2
 
 import (
+	"net/url"
 	"testing"
 
 	"github.com/pingcap/ticdc/pkg/common"
@@ -79,3 +80,32 @@ func TestVerifyRouteConflict(t *testing.T) {
 	require.Contains(t, err.Error(), "source `db1`.`orders`")
 	require.Contains(t, err.Error(), "source `db2`.`orders`")
 }
+
+func TestMaskSinkURIForError(t *testing.T) {
+	sinkURI := "kafka://127.0.0.1:9092/topic?protocol=canal-json" +
+		"&sasl-user=ticdc&sasl-password=verysecure&secret-access-key=rawsecret"
+
+	maskedURI := maskSinkURIForError(sinkURI)
+	require.NotContains(t, maskedURI, "verysecure")
+	require.NotContains(t, maskedURI, "rawsecret")
+	require.Contains(t, maskedURI, "sasl-password=xxxxx")
+	require.Contains(t, maskedURI, "secret-access-key=xxxxx")
+	require.Contains(t, maskedURI, "sasl-user=ticdc")
+
+	invalidURI := "mysql://root:verysecure@127.0.0.1/%zz"
+	require.Equal(t, "<invalid uri>", maskSinkURIForError(invalidURI))
+
+	err := genSinkURIInvalidError(invalidURI, mustParseURLError(t, invalidURI))
+	require.NotContains(t, err.Error(), "verysecure")
+	require.Contains(t, err.Error(), "<invalid uri>")
+	require.Contains(t, err.Error(), `parse "<invalid uri>"`)
+	require.Contains(t, err.Error(), "invalid URL escape")
+}
+
+func mustParseURLError(t *testing.T, rawURL string) error {
+	t.Helper()
+
+	_, err := url.Parse(rawURL)
+	require.Error(t, err)
+	return err
+}

downstreamadapter/sink/sink.go

@@ -26,6 +26,7 @@ import (
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
 	"github.com/pingcap/ticdc/pkg/config"
 	"github.com/pingcap/ticdc/pkg/errors"
+	"github.com/pingcap/ticdc/pkg/util"
 )
 
 type Sink interface {
@@ -50,7 +51,10 @@ type Sink interface {
 func New(ctx context.Context, cfg *config.ChangefeedConfig, changefeedID common.ChangeFeedID) (Sink, error) {
 	sinkURI, err := url.Parse(cfg.SinkURI)
 	if err != nil {
-		return nil, errors.WrapError(errors.ErrSinkURIInvalid, err)
+		return nil, errors.WrapError(
+			errors.ErrSinkURIInvalid,
+			util.MaskSensitiveDataInURLError(err),
+			util.MaskSensitiveDataInURIForError(cfg.SinkURI))
 	}
 	scheme := config.GetScheme(sinkURI)
 	switch scheme {
@@ -65,13 +69,17 @@ func New(ctx context.Context, cfg *config.ChangefeedConfig, changefeedID common.
 	case config.BlackHoleScheme:
 		return blackhole.New(changefeedID)
 	}
-	return nil, errors.ErrSinkURIInvalid.GenWithStackByArgs(sinkURI)
+	return nil, errors.ErrSinkURIInvalid.GenWithStackByArgs(
+		util.MaskSensitiveDataInURIForError(sinkURI.String()))
 }
 
 func Verify(ctx context.Context, cfg *config.ChangefeedConfig, changefeedID common.ChangeFeedID) error {
 	sinkURI, err := url.Parse(cfg.SinkURI)
 	if err != nil {
-		return errors.WrapError(errors.ErrSinkURIInvalid, err)
+		return errors.WrapError(
+			errors.ErrSinkURIInvalid,
+			util.MaskSensitiveDataInURLError(err),
+			util.MaskSensitiveDataInURIForError(cfg.SinkURI))
 	}
 	scheme := config.GetScheme(sinkURI)
 	switch scheme {
@@ -86,5 +94,6 @@ func Verify(ctx context.Context, cfg *config.ChangefeedConfig, changefeedID comm
 	case config.BlackHoleScheme:
 		return nil
 	}
-	return errors.ErrSinkURIInvalid.GenWithStackByArgs(sinkURI)
+	return errors.ErrSinkURIInvalid.GenWithStackByArgs(
+		util.MaskSensitiveDataInURIForError(sinkURI.String()))
 }

downstreamadapter/sink/sink_test.go

@@ -0,0 +1,77 @@
+// Copyright 2026 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package sink
+
+import (
+	"context"
+	"testing"
+
+	"github.com/pingcap/ticdc/pkg/common"
+	"github.com/pingcap/ticdc/pkg/config"
+	"github.com/stretchr/testify/require"
+)
+
+func TestUnknownSchemeMasksSensitiveSinkURI(t *testing.T) {
+	t.Parallel()
+
+	changefeedID := common.NewChangefeedID(common.DefaultKeyspaceName)
+	cfg := &config.ChangefeedConfig{
+		SinkURI: "unknown://127.0.0.1:9092/topic?sasl-password=verysecure&access-key=rawkey",
+	}
+
+	_, err := New(context.Background(), cfg, changefeedID)
+	require.Error(t, err)
+	requireMaskedSinkURIError(t, err)
+
+	err = Verify(context.Background(), cfg, changefeedID)
+	require.Error(t, err)
+	requireMaskedSinkURIError(t, err)
+}
+
+func TestParseErrorMasksSensitiveSinkURI(t *testing.T) {
+	t.Parallel()
+
+	changefeedID := common.NewChangefeedID(common.DefaultKeyspaceName)
+	cfg := &config.ChangefeedConfig{
+		SinkURI: "mysql://root:verysecure@127.0.0.1/%zz",
+	}
+
+	_, err := New(context.Background(), cfg, changefeedID)
+	require.Error(t, err)
+	requireInvalidSinkURIError(t, err)
+
+	err = Verify(context.Background(), cfg, changefeedID)
+	require.Error(t, err)
+	requireInvalidSinkURIError(t, err)
+}
+
+func requireMaskedSinkURIError(t *testing.T, err error) {
+	t.Helper()
+
+	errMsg := err.Error()
+	require.NotContains(t, errMsg, "verysecure")
+	require.NotContains(t, errMsg, "rawkey")
+	require.Contains(t, errMsg, "sasl-password=xxxxx")
+	require.Contains(t, errMsg, "access-key=xxxxx")
+}
+
+func requireInvalidSinkURIError(t *testing.T, err error) {
+	t.Helper()
+
+	errMsg := err.Error()
+	require.NotContains(t, errMsg, "verysecure")
+	require.Contains(t, errMsg, "<invalid uri>")
+	require.Contains(t, errMsg, `parse "<invalid uri>"`)
+	require.Contains(t, errMsg, "invalid URL escape")
+}

pkg/check/cluster.go

@@ -103,7 +103,10 @@ func getClusterIDBySinkURI(
 ) (uint64, string, bool, error) {
 	uri, err := url.Parse(sinkURI)
 	if err != nil {
-		return 0, "", false, cerrors.WrapError(cerrors.ErrSinkURIInvalid, err, sinkURI)
+		return 0, "", false, cerrors.WrapError(
+			cerrors.ErrSinkURIInvalid,
+			util.MaskSensitiveDataInURLError(err),
+			util.MaskSensitiveDataInURIForError(sinkURI))
 	}
 
 	scheme := config.GetScheme(uri)

pkg/util/uri.go

@@ -17,9 +17,6 @@ import (
 	"net"
 	"net/url"
 	"strings"
-
-	"github.com/pingcap/log"
-	"go.uber.org/zap"
 )
 
 // IsValidIPv6AddressFormatInURI reports whether hostPort is a valid IPv6 address in URI.
@@ -70,7 +67,6 @@ func validOptionalPort(port string) bool {
 func MaskSinkURI(uri string) (string, error) {
 	uriParsed, err := url.Parse(uri)
 	if err != nil {
-		log.Error("failed to parse sink URI", zap.Error(err))
 		return "", err
 	}
 	queries := uriParsed.Query()
@@ -99,7 +95,6 @@ var sensitiveQueryParameterNames = []string{
 func MaskSensitiveDataInURI(uri string) string {
 	uriParsed, err := url.Parse(uri)
 	if err != nil {
-		log.Error("failed to parse sink URI", zap.Error(err))
 		return ""
 	}
 	queries := uriParsed.Query()
@@ -114,3 +109,28 @@ func MaskSensitiveDataInURI(uri string) string {
 	uriParsed.RawQuery = queries.Encode()
 	return uriParsed.Redacted()
 }
+
+// MaskSensitiveDataInURIForError masks sensitive data in a URI for error messages.
+func MaskSensitiveDataInURIForError(uri string) string {
+	maskedURI := MaskSensitiveDataInURI(uri)
+	if maskedURI == "" && uri != "" {
+		return "<invalid uri>"
+	}
+	return maskedURI
+}
+
+// MaskSensitiveDataInURLError masks the URL carried by net/url errors.
+func MaskSensitiveDataInURLError(err error) error {
+	if err == nil {
+		return nil
+	}
+	urlErr, ok := err.(*url.Error)
+	if !ok {
+		return err
+	}
+	return &url.Error{
+		Op:  urlErr.Op,
+		URL: MaskSensitiveDataInURIForError(urlErr.URL),
+		Err: urlErr.Err,
+	}
+}