From 5d44f2242fd5637b202cf43cf99793e58577ceb7 Mon Sep 17 00:00:00 2001 From: Danielle Lancashire Date: Mon, 11 Apr 2022 13:42:03 +0000 Subject: [PATCH 1/2] cv3/mirror: Fetch the most recent prefix revision When a user sets up a Mirror with a restricted user that doesn't have access to the `foo` path, we will fail to get the most recent revision due to permissions issues. With this change, when a prefix is provided we will get the initial revision from the prefix rather than /foo. This allows restricted users to setup sync. --- client/v3/mirror/syncer.go | 10 ++- tests/integration/clientv3/mirror_test.go | 74 +++++++++++++++++++++++ 2 files changed, 82 insertions(+), 2 deletions(-) diff --git a/client/v3/mirror/syncer.go b/client/v3/mirror/syncer.go index c25ee9b0c55..8a9ad3faf12 100644 --- a/client/v3/mirror/syncer.go +++ b/client/v3/mirror/syncer.go @@ -18,7 +18,7 @@ package mirror import ( "context" - "go.etcd.io/etcd/client/v3" + clientv3 "go.etcd.io/etcd/client/v3" ) const ( @@ -52,7 +52,13 @@ func (s *syncer) SyncBase(ctx context.Context) (<-chan clientv3.GetResponse, cha // if rev is not specified, we will choose the most recent revision. if s.rev == 0 { - resp, err := s.c.Get(ctx, "foo") + // If len(s.prefix) == 0, we will check a random key to fetch the most recent + // revision (foo), otherwise we use the provided prefix. + checkPath := "foo" + if len(s.prefix) != 0 { + checkPath = s.prefix + } + resp, err := s.c.Get(ctx, checkPath) if err != nil { errchan <- err close(respchan) diff --git a/tests/integration/clientv3/mirror_test.go b/tests/integration/clientv3/mirror_test.go index c9246e0f2d3..53b1109be67 100644 --- a/tests/integration/clientv3/mirror_test.go +++ b/tests/integration/clientv3/mirror_test.go @@ -23,8 +23,10 @@ import ( "time" "go.etcd.io/etcd/api/v3/mvccpb" + clientv3 "go.etcd.io/etcd/client/v3" "go.etcd.io/etcd/client/v3/mirror" "go.etcd.io/etcd/tests/v3/integration" + "google.golang.org/grpc" ) func TestMirrorSync(t *testing.T) { @@ -124,3 +126,75 @@ func TestMirrorSyncBase(t *testing.T) { t.Errorf("unexpected kv count: %d", count) } } + +func TestMirrorSync_Authenticated(t *testing.T) { + integration.BeforeTest(t) + clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1}) + defer clus.Terminate(t) + + initialClient := clus.Client(0) + + // Create a user to run the mirror process that only has access to /syncpath + initialClient.RoleAdd(context.Background(), "syncer") + initialClient.RoleGrantPermission(context.Background(), "syncer", "/syncpath", clientv3.GetPrefixRangeEnd("/syncpath"), clientv3.PermissionType(clientv3.PermReadWrite)) + initialClient.UserAdd(context.Background(), "syncer", "syncfoo") + initialClient.UserGrantRole(context.Background(), "syncer", "syncer") + + // Seed /syncpath with some initial data + _, err := initialClient.KV.Put(context.TODO(), "/syncpath/foo", "bar") + if err != nil { + t.Fatal(err) + } + + // Require authentication + authSetupRoot(t, initialClient.Auth) + + // Create a client as the `syncer` user. + cfg := clientv3.Config{ + Endpoints: initialClient.Endpoints(), + DialTimeout: 5 * time.Second, + DialOptions: []grpc.DialOption{grpc.WithBlock()}, + Username: "syncer", + Password: "syncfoo", + } + syncClient, err := integration.NewClient(t, cfg) + if err != nil { + t.Fatal(err) + } + defer syncClient.Close() + + // Now run the sync process, create changes, and get the initial sync state + syncer := mirror.NewSyncer(syncClient, "/syncpath", 0) + gch, ech := syncer.SyncBase(context.TODO()) + wkvs := []*mvccpb.KeyValue{{Key: []byte("/syncpath/foo"), Value: []byte("bar"), CreateRevision: 2, ModRevision: 2, Version: 1}} + + for g := range gch { + if !reflect.DeepEqual(g.Kvs, wkvs) { + t.Fatalf("kv = %v, want %v", g.Kvs, wkvs) + } + } + + for e := range ech { + t.Fatalf("unexpected error %v", e) + } + + // Start a continuous sync + wch := syncer.SyncUpdates(context.TODO()) + + // Update state + _, err = syncClient.KV.Put(context.TODO(), "/syncpath/foo", "baz") + if err != nil { + t.Fatal(err) + } + + // Wait for the updated state to sync + select { + case r := <-wch: + wkv := &mvccpb.KeyValue{Key: []byte("/syncpath/foo"), Value: []byte("baz"), CreateRevision: 2, ModRevision: 3, Version: 2} + if !reflect.DeepEqual(r.Events[0].Kv, wkv) { + t.Fatalf("kv = %v, want %v", r.Events[0].Kv, wkv) + } + case <-time.After(time.Second): + t.Fatal("failed to receive update in one second") + } +} From 50978d5b25391bc4a9972ab9196a333428a30f8f Mon Sep 17 00:00:00 2001 From: Danielle Lancashire Date: Tue, 12 Apr 2022 13:42:13 +0000 Subject: [PATCH 2/2] clientv3: disable mirror auth test with proxy --- .../integration/clientv3/mirror_auth_test.go | 103 ++++++++++++++++++ tests/integration/clientv3/mirror_test.go | 74 ------------- 2 files changed, 103 insertions(+), 74 deletions(-) create mode 100644 tests/integration/clientv3/mirror_auth_test.go diff --git a/tests/integration/clientv3/mirror_auth_test.go b/tests/integration/clientv3/mirror_auth_test.go new file mode 100644 index 00000000000..95d1e153008 --- /dev/null +++ b/tests/integration/clientv3/mirror_auth_test.go @@ -0,0 +1,103 @@ +// Copyright 2022 The etcd Authors +// +// 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, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//go:build !cluster_proxy +// +build !cluster_proxy + +package clientv3test + +import ( + "context" + "reflect" + "testing" + "time" + + "go.etcd.io/etcd/api/v3/mvccpb" + clientv3 "go.etcd.io/etcd/client/v3" + "go.etcd.io/etcd/client/v3/mirror" + "go.etcd.io/etcd/tests/v3/integration" + "google.golang.org/grpc" +) + +func TestMirrorSync_Authenticated(t *testing.T) { + integration.BeforeTest(t) + clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1}) + defer clus.Terminate(t) + + initialClient := clus.Client(0) + + // Create a user to run the mirror process that only has access to /syncpath + initialClient.RoleAdd(context.Background(), "syncer") + initialClient.RoleGrantPermission(context.Background(), "syncer", "/syncpath", clientv3.GetPrefixRangeEnd("/syncpath"), clientv3.PermissionType(clientv3.PermReadWrite)) + initialClient.UserAdd(context.Background(), "syncer", "syncfoo") + initialClient.UserGrantRole(context.Background(), "syncer", "syncer") + + // Seed /syncpath with some initial data + _, err := initialClient.KV.Put(context.TODO(), "/syncpath/foo", "bar") + if err != nil { + t.Fatal(err) + } + + // Require authentication + authSetupRoot(t, initialClient.Auth) + + // Create a client as the `syncer` user. + cfg := clientv3.Config{ + Endpoints: initialClient.Endpoints(), + DialTimeout: 5 * time.Second, + DialOptions: []grpc.DialOption{grpc.WithBlock()}, + Username: "syncer", + Password: "syncfoo", + } + syncClient, err := integration.NewClient(t, cfg) + if err != nil { + t.Fatal(err) + } + defer syncClient.Close() + + // Now run the sync process, create changes, and get the initial sync state + syncer := mirror.NewSyncer(syncClient, "/syncpath", 0) + gch, ech := syncer.SyncBase(context.TODO()) + wkvs := []*mvccpb.KeyValue{{Key: []byte("/syncpath/foo"), Value: []byte("bar"), CreateRevision: 2, ModRevision: 2, Version: 1}} + + for g := range gch { + if !reflect.DeepEqual(g.Kvs, wkvs) { + t.Fatalf("kv = %v, want %v", g.Kvs, wkvs) + } + } + + for e := range ech { + t.Fatalf("unexpected error %v", e) + } + + // Start a continuous sync + wch := syncer.SyncUpdates(context.TODO()) + + // Update state + _, err = syncClient.KV.Put(context.TODO(), "/syncpath/foo", "baz") + if err != nil { + t.Fatal(err) + } + + // Wait for the updated state to sync + select { + case r := <-wch: + wkv := &mvccpb.KeyValue{Key: []byte("/syncpath/foo"), Value: []byte("baz"), CreateRevision: 2, ModRevision: 3, Version: 2} + if !reflect.DeepEqual(r.Events[0].Kv, wkv) { + t.Fatalf("kv = %v, want %v", r.Events[0].Kv, wkv) + } + case <-time.After(time.Second): + t.Fatal("failed to receive update in one second") + } +} diff --git a/tests/integration/clientv3/mirror_test.go b/tests/integration/clientv3/mirror_test.go index 53b1109be67..c9246e0f2d3 100644 --- a/tests/integration/clientv3/mirror_test.go +++ b/tests/integration/clientv3/mirror_test.go @@ -23,10 +23,8 @@ import ( "time" "go.etcd.io/etcd/api/v3/mvccpb" - clientv3 "go.etcd.io/etcd/client/v3" "go.etcd.io/etcd/client/v3/mirror" "go.etcd.io/etcd/tests/v3/integration" - "google.golang.org/grpc" ) func TestMirrorSync(t *testing.T) { @@ -126,75 +124,3 @@ func TestMirrorSyncBase(t *testing.T) { t.Errorf("unexpected kv count: %d", count) } } - -func TestMirrorSync_Authenticated(t *testing.T) { - integration.BeforeTest(t) - clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1}) - defer clus.Terminate(t) - - initialClient := clus.Client(0) - - // Create a user to run the mirror process that only has access to /syncpath - initialClient.RoleAdd(context.Background(), "syncer") - initialClient.RoleGrantPermission(context.Background(), "syncer", "/syncpath", clientv3.GetPrefixRangeEnd("/syncpath"), clientv3.PermissionType(clientv3.PermReadWrite)) - initialClient.UserAdd(context.Background(), "syncer", "syncfoo") - initialClient.UserGrantRole(context.Background(), "syncer", "syncer") - - // Seed /syncpath with some initial data - _, err := initialClient.KV.Put(context.TODO(), "/syncpath/foo", "bar") - if err != nil { - t.Fatal(err) - } - - // Require authentication - authSetupRoot(t, initialClient.Auth) - - // Create a client as the `syncer` user. - cfg := clientv3.Config{ - Endpoints: initialClient.Endpoints(), - DialTimeout: 5 * time.Second, - DialOptions: []grpc.DialOption{grpc.WithBlock()}, - Username: "syncer", - Password: "syncfoo", - } - syncClient, err := integration.NewClient(t, cfg) - if err != nil { - t.Fatal(err) - } - defer syncClient.Close() - - // Now run the sync process, create changes, and get the initial sync state - syncer := mirror.NewSyncer(syncClient, "/syncpath", 0) - gch, ech := syncer.SyncBase(context.TODO()) - wkvs := []*mvccpb.KeyValue{{Key: []byte("/syncpath/foo"), Value: []byte("bar"), CreateRevision: 2, ModRevision: 2, Version: 1}} - - for g := range gch { - if !reflect.DeepEqual(g.Kvs, wkvs) { - t.Fatalf("kv = %v, want %v", g.Kvs, wkvs) - } - } - - for e := range ech { - t.Fatalf("unexpected error %v", e) - } - - // Start a continuous sync - wch := syncer.SyncUpdates(context.TODO()) - - // Update state - _, err = syncClient.KV.Put(context.TODO(), "/syncpath/foo", "baz") - if err != nil { - t.Fatal(err) - } - - // Wait for the updated state to sync - select { - case r := <-wch: - wkv := &mvccpb.KeyValue{Key: []byte("/syncpath/foo"), Value: []byte("baz"), CreateRevision: 2, ModRevision: 3, Version: 2} - if !reflect.DeepEqual(r.Events[0].Kv, wkv) { - t.Fatalf("kv = %v, want %v", r.Events[0].Kv, wkv) - } - case <-time.After(time.Second): - t.Fatal("failed to receive update in one second") - } -}