diff --git a/server/grpc_service.go b/server/grpc_service.go index 87e1c3c97b..2361c3af00 100644 --- a/server/grpc_service.go +++ b/server/grpc_service.go @@ -18,7 +18,6 @@ import ( "context" "fmt" "io" - "path" "runtime" "runtime/trace" "strconv" @@ -30,6 +29,8 @@ import ( clientv3 "go.etcd.io/etcd/client/v3" "go.uber.org/zap" "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" "github.com/pingcap/errors" "github.com/pingcap/failpoint" @@ -2376,9 +2377,34 @@ func (*GrpcServer) GetDCLocationInfo(_ context.Context, _ *pdpb.GetDCLocationInf }, nil } -// for CDC compatibility, we need to initialize config path to `globalConfigPath` +// For CDC compatibility, we need to initialize an empty config path to +// `globalConfigPath`. const globalConfigPath = "/global/config/" +func normalizeGlobalConfigPath(configPath string) (string, error) { + if configPath == "" { + return globalConfigPath, nil + } + rootPath := strings.TrimSuffix(globalConfigPath, "/") + if configPath == rootPath { + return globalConfigPath, nil + } + if !strings.HasPrefix(configPath, globalConfigPath) { + return "", status.Errorf(codes.InvalidArgument, "global config path %q is not allowed", configPath) + } + return configPath, nil +} + +func resolveGlobalConfigKey(configPath, name string) string { + if name == "" { + return strings.TrimSuffix(configPath, "/") + } + if strings.HasSuffix(configPath, "/") { + return configPath + name + } + return configPath + "/" + name +} + // StoreGlobalConfig store global config into etcd by transaction // Since item value needs to support marshal of different struct types, // it should be set to `Payload bytes` instead of `Value string` @@ -2393,13 +2419,13 @@ func (s *GrpcServer) StoreGlobalConfig(_ context.Context, request *pdpb.StoreGlo if done != nil { defer done() } - configPath := request.GetConfigPath() - if configPath == "" { - configPath = globalConfigPath + configPath, err := normalizeGlobalConfigPath(request.GetConfigPath()) + if err != nil { + return nil, err } ops := make([]clientv3.Op, len(request.Changes)) for i, item := range request.Changes { - name := path.Join(configPath, item.GetName()) + key := resolveGlobalConfigKey(configPath, item.GetName()) switch item.GetKind() { case pdpb.EventType_PUT: // For CDC compatibility, we need to check the Value field firstly. @@ -2407,9 +2433,9 @@ func (s *GrpcServer) StoreGlobalConfig(_ context.Context, request *pdpb.StoreGlo if value == "" { value = string(item.GetPayload()) } - ops[i] = clientv3.OpPut(name, value) + ops[i] = clientv3.OpPut(key, value) case pdpb.EventType_DELETE: - ops[i] = clientv3.OpDelete(name) + ops[i] = clientv3.OpDelete(key) } } res, err := @@ -2438,15 +2464,21 @@ func (s *GrpcServer) LoadGlobalConfig(ctx context.Context, request *pdpb.LoadGlo defer done() } configPath := request.GetConfigPath() - if configPath == "" { - configPath = globalConfigPath + // Keep the legacy prefix-load behavior used by cloud-storage-engine, but + // only for the exact controller path. Other operations and direct sibling + // paths must remain within the global config namespace. + if request.Names != nil || configPath != keypath.ControllerConfigPath() { + configPath, err = normalizeGlobalConfigPath(configPath) + if err != nil { + return nil, err + } } // Since item value needs to support marshal of different struct types, // it should be set to `Payload bytes` instead of `Value string`. if request.Names != nil { res := make([]*pdpb.GlobalConfigItem, len(request.Names)) for i, name := range request.Names { - r, err := s.client.Get(ctx, path.Join(configPath, name)) + r, err := s.client.Get(ctx, resolveGlobalConfigKey(configPath, name)) if err != nil { res[i] = &pdpb.GlobalConfigItem{Name: name, Error: &pdpb.Error{Type: pdpb.ErrorType_UNKNOWN, Message: err.Error()}} } else if len(r.Kvs) == 0 { @@ -2483,12 +2515,12 @@ func (s *GrpcServer) WatchGlobalConfig(req *pdpb.WatchGlobalConfigRequest, serve if done != nil { defer done() } + configPath, err := normalizeGlobalConfigPath(req.GetConfigPath()) + if err != nil { + return err + } ctx, cancel := context.WithCancel(server.Context()) defer cancel() - configPath := req.GetConfigPath() - if configPath == "" { - configPath = globalConfigPath - } revision := req.GetRevision() // If the revision is compacted, will meet required revision has been compacted error. // - If required revision < CompactRevision, we need to reload all configs to avoid losing data. diff --git a/tests/integrations/client/global_config_test.go b/tests/integrations/client/global_config_test.go index 2d3ef7707b..89e5c3d0a7 100644 --- a/tests/integrations/client/global_config_test.go +++ b/tests/integrations/client/global_config_test.go @@ -16,7 +16,6 @@ package client_test import ( "context" - "path" "strconv" "testing" "time" @@ -25,12 +24,15 @@ import ( "github.com/stretchr/testify/suite" "go.uber.org/zap" "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" "github.com/pingcap/kvproto/pkg/pdpb" "github.com/pingcap/log" pd "github.com/tikv/pd/client" "github.com/tikv/pd/client/pkg/caller" + "github.com/tikv/pd/pkg/utils/keypath" "github.com/tikv/pd/pkg/utils/syncutil" "github.com/tikv/pd/server" "github.com/tikv/pd/tests" @@ -137,40 +139,279 @@ func (suite *globalConfigTestSuite) TestLoadWithoutConfigPath() { re.Equal([]byte("1"), res.Items[0].Payload) } -func (suite *globalConfigTestSuite) TestLoadOtherConfigPath() { +func (suite *globalConfigTestSuite) TestRejectInvalidConfigPath() { re := suite.Require() + invalidPaths := []string{ + "OtherConfigPath", + "/tmp/codex-repro/", + "/global/configuration/", + "/", + keypath.ControllerConfigPath() + "-other", + keypath.ControllerConfigPath() + "s", + keypath.ControllerConfigPath() + "/child", + } + for _, configPath := range invalidPaths { + _, err := suite.server.StoreGlobalConfig(suite.server.Context(), &pdpb.StoreGlobalConfigRequest{ + ConfigPath: configPath, + Changes: []*pdpb.GlobalConfigItem{{ + Kind: pdpb.EventType_PUT, + Name: "", + Payload: []byte("1"), + }}, + }) + re.Equal(codes.InvalidArgument, status.Code(err), configPath) + + _, err = suite.server.LoadGlobalConfig(suite.server.Context(), &pdpb.LoadGlobalConfigRequest{ + Names: []string{"source_id"}, + ConfigPath: configPath, + }) + re.Equal(codes.InvalidArgument, status.Code(err), configPath) + + _, err = suite.server.LoadGlobalConfig(suite.server.Context(), &pdpb.LoadGlobalConfigRequest{ + ConfigPath: configPath, + }) + re.Equal(codes.InvalidArgument, status.Code(err), configPath) + + err = suite.server.WatchGlobalConfig(&pdpb.WatchGlobalConfigRequest{ + ConfigPath: configPath, + }, testReceiver{re: re, ctx: suite.server.Context()}) + re.Equal(codes.InvalidArgument, status.Code(err), configPath) + } +} + +func (suite *globalConfigTestSuite) TestLiteralConfigPath() { + re := suite.Require() + configPaths := []string{ + "/global/config/../pd/", + "/global/config/child/../../pd/", + "/global/config/./child/", + "/global/config//child/", + "/global/config//", + `/global/config/child\secret/`, + } defer func() { - for i := range 3 { - _, err := suite.server.GetClient().Delete(suite.server.Context(), getEtcdPath(strconv.Itoa(i))) + for _, configPath := range configPaths { + _, err := suite.server.GetClient().Delete(suite.server.Context(), configPath+"source_id") re.NoError(err) } }() - for i := range 3 { - _, err := suite.server.GetClient().Put(suite.server.Context(), path.Join("OtherConfigPath", strconv.Itoa(i)), strconv.Itoa(i)) + + for _, configPath := range configPaths { + _, err := suite.server.StoreGlobalConfig(suite.server.Context(), &pdpb.StoreGlobalConfigRequest{ + ConfigPath: configPath, + Changes: []*pdpb.GlobalConfigItem{{ + Kind: pdpb.EventType_PUT, + Name: "source_id", + Payload: []byte(configPath), + }}, + }) + re.NoError(err, configPath) + + res, err := suite.server.GetClient().Get(suite.server.Context(), configPath+"source_id") + re.NoError(err, configPath) + re.Len(res.Kvs, 1, configPath) + re.Equal([]byte(configPath), res.Kvs[0].Value, configPath) + } +} + +func (suite *globalConfigTestSuite) TestResourceGroupControllerPrefixLoadCompatibility() { + re := suite.Require() + controllerPath := keypath.ControllerConfigPath() + siblingKey := controllerPath + "-other" + defer func() { + for _, key := range []string{controllerPath, siblingKey} { + _, err := suite.server.GetClient().Delete(suite.server.Context(), key) + re.NoError(err) + } + }() + + _, err := suite.server.GetClient().Put(suite.server.Context(), controllerPath, "controller") + re.NoError(err) + _, err = suite.server.GetClient().Put(suite.server.Context(), siblingKey, "other") + re.NoError(err) + + // LoadGlobalConfig without Names is a prefix query. Keep that legacy result + // behavior for the exact controller path while rejecting sibling paths as + // direct requests. + res, err := suite.server.LoadGlobalConfig(suite.server.Context(), &pdpb.LoadGlobalConfigRequest{ + ConfigPath: controllerPath, + }) + re.NoError(err) + re.Equal([]*pdpb.GlobalConfigItem{{ + Kind: pdpb.EventType_PUT, + Name: controllerPath, + Payload: []byte("controller"), + }, { + Kind: pdpb.EventType_PUT, + Name: siblingKey, + Payload: []byte("other"), + }}, res.Items) + + _, err = suite.server.StoreGlobalConfig(suite.server.Context(), &pdpb.StoreGlobalConfigRequest{ + ConfigPath: controllerPath, + Changes: []*pdpb.GlobalConfigItem{{ + Kind: pdpb.EventType_PUT, + Name: "settings", + Payload: []byte("1"), + }}, + }) + re.Equal(codes.InvalidArgument, status.Code(err)) + + _, err = suite.server.LoadGlobalConfig(suite.server.Context(), &pdpb.LoadGlobalConfigRequest{ + Names: []string{"settings"}, + ConfigPath: controllerPath, + }) + re.Equal(codes.InvalidArgument, status.Code(err)) + + err = suite.server.WatchGlobalConfig(&pdpb.WatchGlobalConfigRequest{ + ConfigPath: controllerPath, + }, testReceiver{re: re, ctx: suite.server.Context()}) + re.Equal(codes.InvalidArgument, status.Code(err)) +} + +func (suite *globalConfigTestSuite) TestNestedConfigPath() { + re := suite.Require() + nestedPath := "/global/config/tidb" + nestedKey := nestedPath + "/source_id" + siblingKey := "/global/config/tidb-other/source_id" + defer func() { + for _, key := range []string{nestedKey, siblingKey} { + _, err := suite.server.GetClient().Delete(suite.server.Context(), key) + re.NoError(err) + } + }() + + _, err := suite.server.StoreGlobalConfig(suite.server.Context(), &pdpb.StoreGlobalConfigRequest{ + ConfigPath: nestedPath, + Changes: []*pdpb.GlobalConfigItem{{ + Kind: pdpb.EventType_PUT, + Name: "source_id", + Payload: []byte("1"), + }}, + }) + re.NoError(err) + _, err = suite.server.GetClient().Put(suite.server.Context(), siblingKey, "2") + re.NoError(err) + + res, err := suite.server.LoadGlobalConfig(suite.server.Context(), &pdpb.LoadGlobalConfigRequest{ + ConfigPath: nestedPath + "/", + }) + re.NoError(err) + re.Equal([]*pdpb.GlobalConfigItem{{ + Kind: pdpb.EventType_PUT, + Name: nestedKey, + Payload: []byte("1"), + }}, res.Items) + + res, err = suite.server.LoadGlobalConfig(suite.server.Context(), &pdpb.LoadGlobalConfigRequest{ + Names: []string{"source_id"}, + ConfigPath: nestedPath + "/", + }) + re.NoError(err) + re.Equal([]*pdpb.GlobalConfigItem{{ + Kind: pdpb.EventType_PUT, + Name: "source_id", + Payload: []byte("1"), + }}, res.Items) +} + +func (suite *globalConfigTestSuite) TestEmptyConfigName() { + re := suite.Require() + rootPath := "/global/config" + defer func() { + _, err := suite.server.GetClient().Delete(suite.server.Context(), rootPath) re.NoError(err) + }() + + _, err := suite.server.StoreGlobalConfig(suite.server.Context(), &pdpb.StoreGlobalConfigRequest{ + Changes: []*pdpb.GlobalConfigItem{{ + Kind: pdpb.EventType_PUT, + Name: "", + Payload: []byte("root"), + }}, + }) + re.NoError(err) + getRes, err := suite.server.GetClient().Get(suite.server.Context(), rootPath) + re.NoError(err) + re.Len(getRes.Kvs, 1) + re.Equal([]byte("root"), getRes.Kvs[0].Value) + + res, err := suite.server.LoadGlobalConfig(suite.server.Context(), &pdpb.LoadGlobalConfigRequest{ + Names: []string{""}, + }) + re.NoError(err) + re.Equal([]*pdpb.GlobalConfigItem{{ + Kind: pdpb.EventType_PUT, + Name: "", + Payload: []byte("root"), + }}, res.Items) +} + +func (suite *globalConfigTestSuite) TestCompatibleConfigName() { + re := suite.Require() + names := []string{ + ".", + "..", + "nested/source_id", + "nested/..", + "nested/../source_id", + "../source_id", + "nested/../../pd", + `nested\source_id`, + "source id", + "source:id", + "source..id", + "配置", + "/absolute", + } + defer func() { + for _, name := range names { + _, err := suite.server.GetClient().Delete(suite.server.Context(), getEtcdPath(name)) + re.NoError(err) + } + }() + + changes := make([]*pdpb.GlobalConfigItem, 0, len(names)) + for _, name := range names { + changes = append(changes, &pdpb.GlobalConfigItem{ + Kind: pdpb.EventType_PUT, + Name: name, + Payload: []byte(name), + }) } + _, err := suite.server.StoreGlobalConfig(suite.server.Context(), &pdpb.StoreGlobalConfigRequest{ + Changes: changes, + }) + re.NoError(err) + res, err := suite.server.LoadGlobalConfig(suite.server.Context(), &pdpb.LoadGlobalConfigRequest{ - Names: []string{"0", "1"}, - ConfigPath: "OtherConfigPath", + Names: names, }) re.NoError(err) - re.Len(res.Items, 2) + re.Len(res.Items, len(names)) for i, item := range res.Items { - re.Equal(&pdpb.GlobalConfigItem{Kind: pdpb.EventType_PUT, Name: strconv.Itoa(i), Payload: []byte(strconv.Itoa(i))}, item) + re.Equal(names[i], item.Name) + re.Equal([]byte(names[i]), item.Payload) + + expectedKey := getEtcdPath(names[i]) + getRes, err := suite.server.GetClient().Get(suite.server.Context(), expectedKey) + re.NoError(err) + re.Len(getRes.Kvs, 1) + re.Equal(expectedKey, string(getRes.Kvs[0].Key)) } } func (suite *globalConfigTestSuite) TestLoadAndStore() { re := suite.Require() defer func() { - for range 3 { - _, err := suite.server.GetClient().Delete(suite.server.Context(), getEtcdPath("test")) + for i := range 3 { + _, err := suite.server.GetClient().Delete(suite.server.Context(), getEtcdPath(strconv.Itoa(i))) re.NoError(err) } }() changes := []*pdpb.GlobalConfigItem{{Kind: pdpb.EventType_PUT, Name: "0", Payload: []byte("0")}, {Kind: pdpb.EventType_PUT, Name: "1", Payload: []byte("1")}, {Kind: pdpb.EventType_PUT, Name: "2", Payload: []byte("2")}} _, err := suite.server.StoreGlobalConfig(suite.server.Context(), &pdpb.StoreGlobalConfigRequest{ - ConfigPath: globalConfigPath, + ConfigPath: "/global/config", Changes: changes, }) re.NoError(err) @@ -187,15 +428,14 @@ func (suite *globalConfigTestSuite) TestLoadAndStore() { func (suite *globalConfigTestSuite) TestStore() { re := suite.Require() defer func() { - for range 3 { - _, err := suite.server.GetClient().Delete(suite.server.Context(), getEtcdPath("test")) + for i := range 3 { + _, err := suite.server.GetClient().Delete(suite.server.Context(), getEtcdPath(strconv.Itoa(i))) re.NoError(err) } }() changes := []*pdpb.GlobalConfigItem{{Kind: pdpb.EventType_PUT, Name: "0", Payload: []byte("0")}, {Kind: pdpb.EventType_PUT, Name: "1", Payload: []byte("1")}, {Kind: pdpb.EventType_PUT, Name: "2", Payload: []byte("2")}} _, err := suite.server.StoreGlobalConfig(suite.server.Context(), &pdpb.StoreGlobalConfigRequest{ - ConfigPath: globalConfigPath, - Changes: changes, + Changes: changes, }) re.NoError(err) for i := range 3 { @@ -273,24 +513,10 @@ func (suite *globalConfigTestSuite) TestClientLoadWithoutConfigPath() { re.Equal(pd.GlobalConfigItem{EventType: pdpb.EventType_PUT, Name: "source_id", PayLoad: []byte("1"), Value: "1"}, res[0]) } -func (suite *globalConfigTestSuite) TestClientLoadOtherConfigPath() { +func (suite *globalConfigTestSuite) TestClientRejectOtherConfigPath() { re := suite.Require() - defer func() { - for i := range 3 { - _, err := suite.server.GetClient().Delete(suite.server.Context(), getEtcdPath(strconv.Itoa(i))) - re.NoError(err) - } - }() - for i := range 3 { - _, err := suite.server.GetClient().Put(suite.server.Context(), path.Join("OtherConfigPath", strconv.Itoa(i)), strconv.Itoa(i)) - re.NoError(err) - } - res, _, err := suite.client.LoadGlobalConfig(suite.server.Context(), []string{"0", "1"}, "OtherConfigPath") - re.NoError(err) - re.Len(res, 2) - for i, item := range res { - re.Equal(pd.GlobalConfigItem{EventType: pdpb.EventType_PUT, Name: strconv.Itoa(i), PayLoad: []byte(strconv.Itoa(i)), Value: strconv.Itoa(i)}, item) - } + _, _, err := suite.client.LoadGlobalConfig(suite.server.Context(), []string{"source_id"}, "OtherConfigPath") + re.Equal(codes.InvalidArgument, status.Code(err)) } func (suite *globalConfigTestSuite) TestClientStore() {