From 1c2bb438f1b84f610ee56a978ca86a1f43524f3d Mon Sep 17 00:00:00 2001 From: dongmen <20351731+asddongmen@users.noreply.github.com> Date: Tue, 19 May 2026 12:59:44 +0800 Subject: [PATCH 1/2] This is an automated cherry-pick of #5093 Signed-off-by: ti-chi-bot --- .gitignore | 1 + api/v2/changefeed.go | 44 +++- api/v2/changefeed_test.go | 36 ++++ downstreamadapter/sink/sink.go | 17 +- downstreamadapter/sink/sink_test.go | 77 +++++++ pkg/check/active_active_tso_indexes.go | 281 +++++++++++++++++++++++++ pkg/check/cluster.go | 5 +- pkg/util/uri.go | 30 ++- pkg/util/uri_test.go | 157 ++++++++++++++ 9 files changed, 628 insertions(+), 20 deletions(-) create mode 100644 downstreamadapter/sink/sink_test.go create mode 100644 pkg/check/active_active_tso_indexes.go create mode 100644 pkg/util/uri_test.go 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..ce496769d0 100644 --- a/api/v2/changefeed.go +++ b/api/v2/changefeed.go @@ -52,6 +52,30 @@ import ( "go.uber.org/zap" ) +<<<<<<< HEAD +======= +// 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)) +} + +>>>>>>> d38ea0a5d (api,sink: mask sink uri secrets (#5093)) // CreateChangefeed handles create changefeed request, // it returns the changefeed's changefeedInfo that it just created // CreateChangefeed creates a changefeed @@ -135,7 +159,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 +168,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 +315,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 +455,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 +469,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 +846,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 +999,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 +1013,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 +1057,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..e19b1848ec 100644 --- a/api/v2/changefeed_test.go +++ b/api/v2/changefeed_test.go @@ -14,6 +14,7 @@ package v2 import ( +<<<<<<< HEAD "testing" "github.com/pingcap/ticdc/pkg/common" @@ -78,4 +79,39 @@ func TestVerifyRouteConflict(t *testing.T) { require.Contains(t, err.Error(), "target `db1`.`orders`") require.Contains(t, err.Error(), "source `db1`.`orders`") require.Contains(t, err.Error(), "source `db2`.`orders`") +======= + "net/url" + "testing" + + "github.com/stretchr/testify/require" +) + +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 +>>>>>>> d38ea0a5d (api,sink: mask sink uri secrets (#5093)) } 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/active_active_tso_indexes.go b/pkg/check/active_active_tso_indexes.go new file mode 100644 index 0000000000..9916848b7f --- /dev/null +++ b/pkg/check/active_active_tso_indexes.go @@ -0,0 +1,281 @@ +// 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 check + +import ( + "context" + "fmt" + "math" + "net/url" + "strconv" + "time" + + "github.com/pingcap/ticdc/pkg/config" + "github.com/pingcap/ticdc/pkg/errors" + "github.com/pingcap/ticdc/pkg/pdutil" + "github.com/pingcap/ticdc/pkg/util" + pd "github.com/tikv/pd/client" + pdhttp "github.com/tikv/pd/client/http" +) + +const ( + pdTSOUniqueIndexKey = "tso-unique-index" + pdTSOMaxIndexKey = "tso-max-index" +) + +const showPDConfigQuery = "SHOW CONFIG WHERE type='pd' AND (name='tso-unique-index' OR name='tso-max-index')" + +var newPDHTTPClientFn = func(upPD pd.Client) (pdhttp.Client, error) { + sc := config.GetGlobalServerConfig() + if sc == nil { + return pdutil.NewPDHTTPClient(upPD, nil) + } + return pdutil.NewPDHTTPClient(upPD, sc.Security) +} + +// ValidateActiveActiveTSOIndexes validates the upstream/downstream PD TSO index +// compatibility when active-active mode is enabled and the downstream is TiDB. +// +// The validation is fail-closed: inability to retrieve or parse values is +// treated as a validation failure. +func ValidateActiveActiveTSOIndexes( + ctx context.Context, + upPD pd.Client, + changefeedCfg *config.ChangefeedConfig, +) error { + if changefeedCfg == nil { + return errors.ErrActiveActiveTSOIndexIncompatible.GenWithStackByArgs("changefeed config is nil") + } + if !changefeedCfg.EnableActiveActive { + return nil + } + + sinkURI, err := url.Parse(changefeedCfg.SinkURI) + if err != nil { + return errors.WrapError( + errors.ErrSinkURIInvalid, + util.MaskSensitiveDataInURLError(err), + util.MaskSensitiveDataInURIForError(changefeedCfg.SinkURI)) + } + if !config.IsMySQLCompatibleScheme(config.GetScheme(sinkURI)) { + return nil + } + + downUnique, downMax, err := getDownstreamTSOIndexes(ctx, changefeedCfg, sinkURI) + if err != nil { + return errors.WrapError( + errors.ErrActiveActiveTSOIndexIncompatible, + err, + "failed to read downstream tso index config", + ) + } + + upUnique, upMax, err := getUpstreamTSOIndexes(ctx, upPD) + if err != nil { + return errors.WrapError( + errors.ErrActiveActiveTSOIndexIncompatible, + err, + fmt.Sprintf("failed to read upstream tso index config, downstream unique=%d, downstream max=%d", + downUnique, downMax), + ) + } + + // In active-active mode, `tso-unique-index` must be different between upstream and + // downstream to avoid TSO collisions, while `tso-max-index` must match to guarantee + // the same logical index range. + if upUnique == downUnique { + return errors.ErrActiveActiveTSOIndexIncompatible.GenWithStackByArgs( + fmt.Sprintf("active active tso index mismatch, upstream and downstream share the same tso-unique-index=%d, upstream max=%d, downstream max=%d", + upUnique, upMax, downMax), + ) + } + if upMax != downMax { + return errors.ErrActiveActiveTSOIndexIncompatible.GenWithStackByArgs( + fmt.Sprintf("active active tso index mismatch, upstream unique=%d, upstream max=%d, downstream unique=%d, downstream max=%d", + upUnique, upMax, downUnique, downMax), + ) + } + return nil +} + +// getDownstreamTSOIndexes reads the effective PD TSO index values from the +// downstream TiDB cluster via SQL. +// +// These values are part of PD configuration, and TiDB exposes them through +// `SHOW CONFIG`. The query returns one row per TiDB instance, so this function +// requires all instances to report the same values. +// +// The function is fail-closed: any retrieval, parsing, missing key, or +// cross-instance inconsistency is treated as an error. +func getDownstreamTSOIndexes( + ctx context.Context, + changefeedCfg *config.ChangefeedConfig, + sinkURI *url.URL, +) (unique int64, max int64, err error) { + if changefeedCfg == nil { + return 0, 0, errors.New("changefeed config is nil") + } + + mysqlCfg, db, err := newMySQLConfigAndDBFn(ctx, changefeedCfg.ChangefeedID, sinkURI, changefeedCfg) + if err != nil { + return 0, 0, errors.Trace(err) + } + defer func() { _ = db.Close() }() + + readTimeout, err := time.ParseDuration(mysqlCfg.ReadTimeout) + if err != nil { + return 0, 0, errors.Trace(err) + } + + // Bound the downstream query by the sink read timeout to keep the validation + // latency aligned with downstream connectivity expectations. + queryCtx, cancel := context.WithTimeout(ctx, readTimeout) + defer cancel() + + rows, err := db.QueryContext(queryCtx, showPDConfigQuery) + if err != nil { + return 0, 0, errors.Trace(err) + } + defer func() { _ = rows.Close() }() + + var ( + uniqueSet bool + maxSet bool + ) + // SHOW CONFIG returns one row per TiDB instance. Require all instances to + // report consistent values to avoid validating against a partially rolled-out + // or inconsistent configuration. + for rows.Next() { + // Columns: Type | Instance | Name | Value + var typ, instance, name, value string + if err := rows.Scan(&typ, &instance, &name, &value); err != nil { + return 0, 0, errors.Trace(err) + } + switch name { + case pdTSOUniqueIndexKey: + parsed, err := strconv.ParseInt(value, 10, 64) + if err != nil { + return 0, 0, errors.Trace(err) + } + if !uniqueSet { + unique = parsed + uniqueSet = true + continue + } + if unique != parsed { + return 0, 0, errors.New("downstream TiDB reports inconsistent tso-unique-index across instances") + } + case pdTSOMaxIndexKey: + parsed, err := strconv.ParseInt(value, 10, 64) + if err != nil { + return 0, 0, errors.Trace(err) + } + if !maxSet { + max = parsed + maxSet = true + continue + } + if max != parsed { + return 0, 0, errors.New("downstream TiDB reports inconsistent tso-max-index across instances") + } + default: + } + } + if err := rows.Err(); err != nil { + return 0, 0, errors.Trace(err) + } + + if !uniqueSet { + return 0, 0, errors.Errorf("downstream TiDB does not report %s", pdTSOUniqueIndexKey) + } + if !maxSet { + return 0, 0, errors.Errorf("downstream TiDB does not report %s", pdTSOMaxIndexKey) + } + return unique, max, nil +} + +// getUpstreamTSOIndexes reads the PD TSO index values from the upstream PD via +// PD HTTP API. +// +// The PD HTTP client uses the same service discovery (and TLS configuration) as +// the given gRPC PD client. It also probes leader and followers internally, so a +// single GetConfig call is sufficient here. +// +// The caller controls the request deadline through ctx. +func getUpstreamTSOIndexes( + ctx context.Context, + upPD pd.Client, +) (unique int64, max int64, err error) { + if upPD == nil { + return 0, 0, errors.New("pd client is nil") + } + + httpClient, err := newPDHTTPClientFn(upPD) + if err != nil { + return 0, 0, errors.Trace(err) + } + defer httpClient.Close() + + cfg, err := httpClient.GetConfig(ctx) + if err != nil { + return 0, 0, errors.Trace(err) + } + + unique, err = parsePDConfigInt64(cfg, pdTSOUniqueIndexKey) + if err != nil { + return 0, 0, errors.Trace(err) + } + max, err = parsePDConfigInt64(cfg, pdTSOMaxIndexKey) + if err != nil { + return 0, 0, errors.Trace(err) + } + return unique, max, nil +} + +func parsePDConfigInt64(cfg map[string]any, key string) (int64, error) { + v, ok := cfg[key] + if !ok { + return 0, errors.Errorf("pd config key not found: %s", key) + } + + // PD stores `tso-unique-index` and `tso-max-index` as int64 values. + // The PD HTTP client unmarshals the JSON response into map[string]any, + // and encoding/json decodes JSON numbers as float64 in that case. + // Keep the conversion strict and fail-closed. + switch x := v.(type) { + case int64: + return x, nil + case int: + return int64(x), nil + case float64: + // When a JSON number is decoded into an `any`, it becomes float64. + // float64 cannot precisely represent all int64 values (it is exact only + // for integers up to 2^53). For a strict conversion, reject values beyond + // that range to avoid precision loss and implementation-defined behavior + // on overflow. + const maxExactIntInFloat64 = float64(1 << 53) + if math.IsNaN(x) || math.IsInf(x, 0) { + return 0, errors.New("value is not a finite number") + } + if math.Trunc(x) != x { + return 0, errors.New("value is not an integer") + } + if x > maxExactIntInFloat64 || x < -maxExactIntInFloat64 { + return 0, errors.Errorf("value for %s exceeds exact integer range for float64", key) + } + return int64(x), nil + default: + return 0, errors.Errorf("unexpected value type for %s: %T", key, v) + } +} 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") +} From 374db92e3830c5a2a93ea43eb263d28ba4beb3a7 Mon Sep 17 00:00:00 2001 From: dongmen <414110582@qq.com> Date: Tue, 28 Jul 2026 08:01:05 +0800 Subject: [PATCH 2/2] api,sink: resolve release-8.5 cherry-pick conflicts --- api/v2/changefeed.go | 3 - api/v2/changefeed_test.go | 10 +- pkg/check/active_active_tso_indexes.go | 281 ------------------------- 3 files changed, 2 insertions(+), 292 deletions(-) delete mode 100644 pkg/check/active_active_tso_indexes.go diff --git a/api/v2/changefeed.go b/api/v2/changefeed.go index ce496769d0..bf313e6e8c 100644 --- a/api/v2/changefeed.go +++ b/api/v2/changefeed.go @@ -52,8 +52,6 @@ import ( "go.uber.org/zap" ) -<<<<<<< HEAD -======= // 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) { @@ -75,7 +73,6 @@ func genSinkURIInvalidError(sinkURI string, err error) error { errors.ErrSinkURIInvalid, util.MaskSensitiveDataInURLError(err), maskSinkURIForError(sinkURI)) } ->>>>>>> d38ea0a5d (api,sink: mask sink uri secrets (#5093)) // CreateChangefeed handles create changefeed request, // it returns the changefeed's changefeedInfo that it just created // CreateChangefeed creates a changefeed diff --git a/api/v2/changefeed_test.go b/api/v2/changefeed_test.go index e19b1848ec..346d2d9f27 100644 --- a/api/v2/changefeed_test.go +++ b/api/v2/changefeed_test.go @@ -14,7 +14,7 @@ package v2 import ( -<<<<<<< HEAD + "net/url" "testing" "github.com/pingcap/ticdc/pkg/common" @@ -79,12 +79,7 @@ func TestVerifyRouteConflict(t *testing.T) { require.Contains(t, err.Error(), "target `db1`.`orders`") require.Contains(t, err.Error(), "source `db1`.`orders`") require.Contains(t, err.Error(), "source `db2`.`orders`") -======= - "net/url" - "testing" - - "github.com/stretchr/testify/require" -) +} func TestMaskSinkURIForError(t *testing.T) { sinkURI := "kafka://127.0.0.1:9092/topic?protocol=canal-json" + @@ -113,5 +108,4 @@ func mustParseURLError(t *testing.T, rawURL string) error { _, err := url.Parse(rawURL) require.Error(t, err) return err ->>>>>>> d38ea0a5d (api,sink: mask sink uri secrets (#5093)) } diff --git a/pkg/check/active_active_tso_indexes.go b/pkg/check/active_active_tso_indexes.go deleted file mode 100644 index 9916848b7f..0000000000 --- a/pkg/check/active_active_tso_indexes.go +++ /dev/null @@ -1,281 +0,0 @@ -// 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 check - -import ( - "context" - "fmt" - "math" - "net/url" - "strconv" - "time" - - "github.com/pingcap/ticdc/pkg/config" - "github.com/pingcap/ticdc/pkg/errors" - "github.com/pingcap/ticdc/pkg/pdutil" - "github.com/pingcap/ticdc/pkg/util" - pd "github.com/tikv/pd/client" - pdhttp "github.com/tikv/pd/client/http" -) - -const ( - pdTSOUniqueIndexKey = "tso-unique-index" - pdTSOMaxIndexKey = "tso-max-index" -) - -const showPDConfigQuery = "SHOW CONFIG WHERE type='pd' AND (name='tso-unique-index' OR name='tso-max-index')" - -var newPDHTTPClientFn = func(upPD pd.Client) (pdhttp.Client, error) { - sc := config.GetGlobalServerConfig() - if sc == nil { - return pdutil.NewPDHTTPClient(upPD, nil) - } - return pdutil.NewPDHTTPClient(upPD, sc.Security) -} - -// ValidateActiveActiveTSOIndexes validates the upstream/downstream PD TSO index -// compatibility when active-active mode is enabled and the downstream is TiDB. -// -// The validation is fail-closed: inability to retrieve or parse values is -// treated as a validation failure. -func ValidateActiveActiveTSOIndexes( - ctx context.Context, - upPD pd.Client, - changefeedCfg *config.ChangefeedConfig, -) error { - if changefeedCfg == nil { - return errors.ErrActiveActiveTSOIndexIncompatible.GenWithStackByArgs("changefeed config is nil") - } - if !changefeedCfg.EnableActiveActive { - return nil - } - - sinkURI, err := url.Parse(changefeedCfg.SinkURI) - if err != nil { - return errors.WrapError( - errors.ErrSinkURIInvalid, - util.MaskSensitiveDataInURLError(err), - util.MaskSensitiveDataInURIForError(changefeedCfg.SinkURI)) - } - if !config.IsMySQLCompatibleScheme(config.GetScheme(sinkURI)) { - return nil - } - - downUnique, downMax, err := getDownstreamTSOIndexes(ctx, changefeedCfg, sinkURI) - if err != nil { - return errors.WrapError( - errors.ErrActiveActiveTSOIndexIncompatible, - err, - "failed to read downstream tso index config", - ) - } - - upUnique, upMax, err := getUpstreamTSOIndexes(ctx, upPD) - if err != nil { - return errors.WrapError( - errors.ErrActiveActiveTSOIndexIncompatible, - err, - fmt.Sprintf("failed to read upstream tso index config, downstream unique=%d, downstream max=%d", - downUnique, downMax), - ) - } - - // In active-active mode, `tso-unique-index` must be different between upstream and - // downstream to avoid TSO collisions, while `tso-max-index` must match to guarantee - // the same logical index range. - if upUnique == downUnique { - return errors.ErrActiveActiveTSOIndexIncompatible.GenWithStackByArgs( - fmt.Sprintf("active active tso index mismatch, upstream and downstream share the same tso-unique-index=%d, upstream max=%d, downstream max=%d", - upUnique, upMax, downMax), - ) - } - if upMax != downMax { - return errors.ErrActiveActiveTSOIndexIncompatible.GenWithStackByArgs( - fmt.Sprintf("active active tso index mismatch, upstream unique=%d, upstream max=%d, downstream unique=%d, downstream max=%d", - upUnique, upMax, downUnique, downMax), - ) - } - return nil -} - -// getDownstreamTSOIndexes reads the effective PD TSO index values from the -// downstream TiDB cluster via SQL. -// -// These values are part of PD configuration, and TiDB exposes them through -// `SHOW CONFIG`. The query returns one row per TiDB instance, so this function -// requires all instances to report the same values. -// -// The function is fail-closed: any retrieval, parsing, missing key, or -// cross-instance inconsistency is treated as an error. -func getDownstreamTSOIndexes( - ctx context.Context, - changefeedCfg *config.ChangefeedConfig, - sinkURI *url.URL, -) (unique int64, max int64, err error) { - if changefeedCfg == nil { - return 0, 0, errors.New("changefeed config is nil") - } - - mysqlCfg, db, err := newMySQLConfigAndDBFn(ctx, changefeedCfg.ChangefeedID, sinkURI, changefeedCfg) - if err != nil { - return 0, 0, errors.Trace(err) - } - defer func() { _ = db.Close() }() - - readTimeout, err := time.ParseDuration(mysqlCfg.ReadTimeout) - if err != nil { - return 0, 0, errors.Trace(err) - } - - // Bound the downstream query by the sink read timeout to keep the validation - // latency aligned with downstream connectivity expectations. - queryCtx, cancel := context.WithTimeout(ctx, readTimeout) - defer cancel() - - rows, err := db.QueryContext(queryCtx, showPDConfigQuery) - if err != nil { - return 0, 0, errors.Trace(err) - } - defer func() { _ = rows.Close() }() - - var ( - uniqueSet bool - maxSet bool - ) - // SHOW CONFIG returns one row per TiDB instance. Require all instances to - // report consistent values to avoid validating against a partially rolled-out - // or inconsistent configuration. - for rows.Next() { - // Columns: Type | Instance | Name | Value - var typ, instance, name, value string - if err := rows.Scan(&typ, &instance, &name, &value); err != nil { - return 0, 0, errors.Trace(err) - } - switch name { - case pdTSOUniqueIndexKey: - parsed, err := strconv.ParseInt(value, 10, 64) - if err != nil { - return 0, 0, errors.Trace(err) - } - if !uniqueSet { - unique = parsed - uniqueSet = true - continue - } - if unique != parsed { - return 0, 0, errors.New("downstream TiDB reports inconsistent tso-unique-index across instances") - } - case pdTSOMaxIndexKey: - parsed, err := strconv.ParseInt(value, 10, 64) - if err != nil { - return 0, 0, errors.Trace(err) - } - if !maxSet { - max = parsed - maxSet = true - continue - } - if max != parsed { - return 0, 0, errors.New("downstream TiDB reports inconsistent tso-max-index across instances") - } - default: - } - } - if err := rows.Err(); err != nil { - return 0, 0, errors.Trace(err) - } - - if !uniqueSet { - return 0, 0, errors.Errorf("downstream TiDB does not report %s", pdTSOUniqueIndexKey) - } - if !maxSet { - return 0, 0, errors.Errorf("downstream TiDB does not report %s", pdTSOMaxIndexKey) - } - return unique, max, nil -} - -// getUpstreamTSOIndexes reads the PD TSO index values from the upstream PD via -// PD HTTP API. -// -// The PD HTTP client uses the same service discovery (and TLS configuration) as -// the given gRPC PD client. It also probes leader and followers internally, so a -// single GetConfig call is sufficient here. -// -// The caller controls the request deadline through ctx. -func getUpstreamTSOIndexes( - ctx context.Context, - upPD pd.Client, -) (unique int64, max int64, err error) { - if upPD == nil { - return 0, 0, errors.New("pd client is nil") - } - - httpClient, err := newPDHTTPClientFn(upPD) - if err != nil { - return 0, 0, errors.Trace(err) - } - defer httpClient.Close() - - cfg, err := httpClient.GetConfig(ctx) - if err != nil { - return 0, 0, errors.Trace(err) - } - - unique, err = parsePDConfigInt64(cfg, pdTSOUniqueIndexKey) - if err != nil { - return 0, 0, errors.Trace(err) - } - max, err = parsePDConfigInt64(cfg, pdTSOMaxIndexKey) - if err != nil { - return 0, 0, errors.Trace(err) - } - return unique, max, nil -} - -func parsePDConfigInt64(cfg map[string]any, key string) (int64, error) { - v, ok := cfg[key] - if !ok { - return 0, errors.Errorf("pd config key not found: %s", key) - } - - // PD stores `tso-unique-index` and `tso-max-index` as int64 values. - // The PD HTTP client unmarshals the JSON response into map[string]any, - // and encoding/json decodes JSON numbers as float64 in that case. - // Keep the conversion strict and fail-closed. - switch x := v.(type) { - case int64: - return x, nil - case int: - return int64(x), nil - case float64: - // When a JSON number is decoded into an `any`, it becomes float64. - // float64 cannot precisely represent all int64 values (it is exact only - // for integers up to 2^53). For a strict conversion, reject values beyond - // that range to avoid precision loss and implementation-defined behavior - // on overflow. - const maxExactIntInFloat64 = float64(1 << 53) - if math.IsNaN(x) || math.IsInf(x, 0) { - return 0, errors.New("value is not a finite number") - } - if math.Trunc(x) != x { - return 0, errors.New("value is not an integer") - } - if x > maxExactIntInFloat64 || x < -maxExactIntInFloat64 { - return 0, errors.Errorf("value for %s exceeds exact integer range for float64", key) - } - return int64(x), nil - default: - return 0, errors.Errorf("unexpected value type for %s: %T", key, v) - } -}