diff --git a/.gitignore b/.gitignore index 5012978619..e224245024 100644 --- a/.gitignore +++ b/.gitignore @@ -4,6 +4,7 @@ tools/bin tools/include tools/workload/bin +.design .issue .vscode .idea diff --git a/api/v2/changefeed.go b/api/v2/changefeed.go index b06900a00f..bf313e6e8c 100644 --- a/api/v2/changefeed.go +++ b/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 } diff --git a/api/v2/changefeed_test.go b/api/v2/changefeed_test.go index 31a56530fb..346d2d9f27 100644 --- a/api/v2/changefeed_test.go +++ b/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, "", maskSinkURIForError(invalidURI)) + + err := genSinkURIInvalidError(invalidURI, mustParseURLError(t, invalidURI)) + require.NotContains(t, err.Error(), "verysecure") + require.Contains(t, err.Error(), "") + require.Contains(t, err.Error(), `parse ""`) + 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 +} diff --git a/downstreamadapter/sink/sink.go b/downstreamadapter/sink/sink.go index 12a934d2c6..6e797fc448 100644 --- a/downstreamadapter/sink/sink.go +++ b/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())) } diff --git a/downstreamadapter/sink/sink_test.go b/downstreamadapter/sink/sink_test.go new file mode 100644 index 0000000000..0d812aa533 --- /dev/null +++ b/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, "") + require.Contains(t, errMsg, `parse ""`) + require.Contains(t, errMsg, "invalid URL escape") +} diff --git a/pkg/check/cluster.go b/pkg/check/cluster.go index d320297a94..534fdaab41 100644 --- a/pkg/check/cluster.go +++ b/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) diff --git a/pkg/util/uri.go b/pkg/util/uri.go index 18bdb01bd5..c67badd5f5 100644 --- a/pkg/util/uri.go +++ b/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 "" + } + 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, + } +} diff --git a/pkg/util/uri_test.go b/pkg/util/uri_test.go new file mode 100644 index 0000000000..006537e457 --- /dev/null +++ b/pkg/util/uri_test.go @@ -0,0 +1,157 @@ +// Copyright 2022 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 util + +import ( + "net/url" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestIsValidIPv6AddressFormatInURI(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + host string + want bool + }{ + {"valid ipv6 address", "[::1]", true}, + {"valid ipv6 address1 with port", "[::1]:8080", true}, + {"valid ipv6 address2 with port", "[1080:0:0:0:8:800:200C:417A]:8080", true}, + {"valid ipv6 address3 with port", "[::FFFF:129.144.52.38]:8080", true}, + {"invalid ipv6 address", "::1", false}, + {"invalid ipv6 address with port", "::1:8000", false}, + } + for _, tt := range tests { + test := tt + t.Run(test.name, func(t *testing.T) { + t.Parallel() + require.Equal(t, test.want, IsValidIPv6AddressFormatInURI(test.host)) + }) + } +} + +func TestIsIPv6Address(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + host string + want bool + }{ + {"valid ipv6 address1", "::1", true}, + {"valid ipv6 address2", "1080:0:0:0:8:800:200C:417A", true}, + {"ipv4 address", "127.0.0.1", false}, + {"empty address", "", false}, + {"not ip address", "emmmmmmmm", false}, + } + + for _, tt := range tests { + test := tt + t.Run(test.name, func(t *testing.T) { + t.Parallel() + require.Equal(t, test.want, IsIPv6Address(test.host)) + }) + } +} + +func TestMaskSinkURI(t *testing.T) { + tests := []struct { + uri string + masked string + }{ + { + "mysql://root:123456@127.0.0.1:3306/?time-zone=Asia/Shanghai", + "mysql://root:xxxxx@127.0.0.1:3306/?time-zone=Asia/Shanghai", + }, + { + "kafka://127.0.0.1:9093/cdc?sasl-mechanism=SCRAM-SHA-256&sasl-user=ticdc&sasl-password=verysecure", + "kafka://127.0.0.1:9093/cdc?sasl-mechanism=SCRAM-SHA-256&sasl-password=xxxxx&sasl-user=ticdc", + }, + } + + for _, tt := range tests { + maskedURI, err := MaskSinkURI(tt.uri) + require.NoError(t, err) + require.Equal(t, tt.masked, maskedURI) + } +} + +func TestMaskSensitiveDataInURI(t *testing.T) { + tests := []struct { + uri string + masked string + }{ + { + "mysql://root:123456@127.0.0.1:3306/?time-zone=c", + "mysql://root:xxxxx@127.0.0.1:3306/?time-zone=c", + }, + { + "mysql://root:123456@127.0.0.1:3306/?access_key=c", + "mysql://root:xxxxx@127.0.0.1:3306/?access_key=xxxxx", + }, + { + "mysql://root:123456@127.0.0.1:3306/?secret_access_key=c", + "mysql://root:xxxxx@127.0.0.1:3306/?secret_access_key=xxxxx", + }, + { + "mysql://root:123456@127.0.0.1:3306/?client_secret=c", + "mysql://root:xxxxx@127.0.0.1:3306/?client_secret=xxxxx", + }, + { + "", + "", + }, + { + "abc", + "abc", + }, + } + for _, q := range sensitiveQueryParameterNames { + tests = append(tests, struct { + uri string + masked string + }{ + "kafka://127.0.0.1:9093/cdc?" + q + "=verysecure", + "kafka://127.0.0.1:9093/cdc?" + q + "=xxxxx", + }) + } + + for _, tt := range tests { + maskedURI := MaskSensitiveDataInURI(tt.uri) + require.Equal(t, tt.masked, maskedURI) + } +} + +func TestMaskSensitiveDataInURIForError(t *testing.T) { + require.Equal(t, "", MaskSensitiveDataInURIForError("")) + require.Equal(t, "abc", MaskSensitiveDataInURIForError("abc")) + require.Equal(t, + "mysql://root:xxxxx@127.0.0.1:3306/?sasl-password=xxxxx", + MaskSensitiveDataInURIForError("mysql://root:verysecure@127.0.0.1:3306/?sasl-password=rawsecret")) + require.Equal(t, "", MaskSensitiveDataInURIForError("mysql://root:verysecure@127.0.0.1/%zz")) +} + +func TestMaskSensitiveDataInURLError(t *testing.T) { + rawURL := "mysql://root:verysecure@127.0.0.1/%zz" + _, err := url.Parse(rawURL) + require.Error(t, err) + + maskedErr := MaskSensitiveDataInURLError(err) + require.NotContains(t, maskedErr.Error(), "verysecure") + require.Contains(t, maskedErr.Error(), `parse ""`) + require.Contains(t, maskedErr.Error(), "invalid URL escape") +}