Merge pull request #134065 from serathius/storage-recursive

Unify directory protection for recursive requests in storage

staging/src/k8s.io/apiserver/pkg/storage/etcd3/store.go

@@ -230,7 +230,7 @@ func (s *store) getResourceSizeEstimator() *resourceSizeEstimator {
 
 // Get implements storage.Interface.Get.
 func (s *store) Get(ctx context.Context, key string, opts storage.GetOptions, out runtime.Object) error {
-	preparedKey, err := s.prepareKey(key)
+	preparedKey, err := s.prepareKey(key, false)
 	if err != nil {
 		return err
 	}
@@ -266,7 +266,7 @@ func (s *store) Get(ctx context.Context, key string, opts storage.GetOptions, ou
 
 // Create implements storage.Interface.Create.
 func (s *store) Create(ctx context.Context, key string, obj, out runtime.Object, ttl uint64) error {
-	preparedKey, err := s.prepareKey(key)
+	preparedKey, err := s.prepareKey(key, false)
 	if err != nil {
 		return err
 	}
@@ -336,7 +336,7 @@ func (s *store) Create(ctx context.Context, key string, obj, out runtime.Object,
 func (s *store) Delete(
 	ctx context.Context, key string, out runtime.Object, preconditions *storage.Preconditions,
 	validateDeletion storage.ValidateObjectFunc, cachedExistingObject runtime.Object, opts storage.DeleteOptions) error {
-	preparedKey, err := s.prepareKey(key)
+	preparedKey, err := s.prepareKey(key, false)
 	if err != nil {
 		return err
 	}
@@ -457,7 +457,7 @@ func (s *store) conditionalDelete(
 func (s *store) GuaranteedUpdate(
 	ctx context.Context, key string, destination runtime.Object, ignoreNotFound bool,
 	preconditions *storage.Preconditions, tryUpdate storage.UpdateFunc, cachedExistingObject runtime.Object) error {
-	preparedKey, err := s.prepareKey(key)
+	preparedKey, err := s.prepareKey(key, false)
 	if err != nil {
 		return err
 	}
@@ -645,13 +645,10 @@ func (s *store) Stats(ctx context.Context) (storage.Stats, error) {
 	// returning stats without resource size
 
 	startTime := time.Now()
-	prefix, err := s.prepareKey(s.resourcePrefix)
+	prefix, err := s.prepareKey(s.resourcePrefix, true)
 	if err != nil {
 		return storage.Stats{}, err
 	}
-	if !strings.HasSuffix(prefix, "/") {
-		prefix += "/"
-	}
 	count, err := s.client.Kubernetes.Count(ctx, prefix, kubernetes.CountOptions{})
 	metrics.RecordEtcdRequest("listWithCount", s.groupResource, err, startTime)
 	if err != nil {
@@ -677,13 +674,10 @@ func (s *store) EnableResourceSizeEstimation(getKeys storage.KeysFunc) error {
 
 func (s *store) getKeys(ctx context.Context) ([]string, error) {
 	startTime := time.Now()
-	prefix, err := s.prepareKey(s.resourcePrefix)
+	prefix, err := s.prepareKey(s.resourcePrefix, true)
 	if err != nil {
 		return nil, err
 	}
-	if !strings.HasSuffix(prefix, "/") {
-		prefix += "/"
-	}
 	resp, err := s.client.KV.Get(ctx, prefix, clientv3.WithPrefix(), clientv3.WithKeysOnly())
 	metrics.RecordEtcdRequest("listOnlyKeys", s.groupResource, err, startTime)
 	if err != nil {
@@ -734,7 +728,7 @@ func (s *store) GetCurrentResourceVersion(ctx context.Context) (uint64, error) {
 
 // GetList implements storage.Interface.
 func (s *store) GetList(ctx context.Context, key string, opts storage.ListOptions, listObj runtime.Object) error {
-	keyPrefix, err := s.prepareKey(key)
+	keyPrefix, err := s.prepareKey(key, opts.Recursive)
 	if err != nil {
 		return err
 	}
@@ -755,14 +749,6 @@ func (s *store) GetList(ctx context.Context, key string, opts storage.ListOption
 		return fmt.Errorf("need ptr to slice: %v", err)
 	}
 
-	// For recursive lists, we need to make sure the key ended with "/" so that we only
-	// get children "directories". e.g. if we have key "/a", "/a/b", "/ab", getting keys
-	// with prefix "/a" will return all three, while with prefix "/a/" will return only
-	// "/a/b" which is the correct answer.
-	if opts.Recursive && !strings.HasSuffix(keyPrefix, "/") {
-		keyPrefix += "/"
-	}
-
 	// set the appropriate clientv3 options to filter the returned data set
 	limit := opts.Predicate.Limit
 	paging := opts.Predicate.Limit > 0
@@ -971,7 +957,7 @@ func growSlice(v reflect.Value, maxCapacity int, sizes ...int) {
 
 // Watch implements storage.Interface.Watch.
 func (s *store) Watch(ctx context.Context, key string, opts storage.ListOptions) (watch.Interface, error) {
-	preparedKey, err := s.prepareKey(key)
+	preparedKey, err := s.prepareKey(key, opts.Recursive)
 	if err != nil {
 		return nil, err
 	}
@@ -1118,7 +1104,7 @@ func (s *store) validateMinimumResourceVersion(minimumResourceVersion string, ac
 	return nil
 }
 
-func (s *store) prepareKey(key string) (string, error) {
+func (s *store) prepareKey(key string, recursive bool) (string, error) {
 	if key == ".." ||
 		strings.HasPrefix(key, "../") ||
 		strings.HasSuffix(key, "/..") ||
@@ -1139,6 +1125,13 @@ func (s *store) prepareKey(key string) (string, error) {
 	if key[0] == '/' {
 		startIndex = 1
 	}
+	// For recursive requests, we need to make sure the key ended with "/" so that we only
+	// get children "directories". e.g. if we have key "/a", "/a/b", "/ab", getting keys
+	// with prefix "/a" will return all three, while with prefix "/a/" will return only
+	// "/a/b" which is the correct answer.
+	if recursive && !strings.HasSuffix(key, "/") {
+		key += "/"
+	}
 	return s.pathPrefix + key[startIndex:], nil
 }
 

staging/src/k8s.io/apiserver/pkg/storage/etcd3/store_test.go

@@ -655,7 +655,7 @@ func testSetup(t testing.TB, opts ...setupOption) (context.Context, *store, *kub
 	return ctx, store, client
 }
 
-func TestValidateKey(t *testing.T) {
+func TestPrepareKey(t *testing.T) {
 	validKeys := []string{
 		"/foo/bar/baz/a.b.c/",
 		"/foo",
@@ -689,21 +689,46 @@ func TestValidateKey(t *testing.T) {
 	)
 	_, store, _ := testSetup(t, withPrefix(pathPrefix))
 
-	for _, key := range validKeys {
-		k, err := store.prepareKey(key)
-		if err != nil {
-			t.Errorf("key %q should be valid; unexpected error: %v", key, err)
-		} else if !strings.HasPrefix(k, expectPrefix) {
-			t.Errorf("key %q should have prefix %q", k, expectPrefix)
+	t.Run("Non-recursive", func(t *testing.T) {
+		for _, key := range validKeys {
+			preparedKey, err := store.prepareKey(key, false)
+			if err != nil {
+				t.Errorf("preparing key %q should be valid; unexpected error: %v", key, err)
+				continue
+			}
+			if !strings.HasPrefix(preparedKey, expectPrefix) {
+				t.Errorf("prepared key %q should have prefix %q", preparedKey, expectPrefix)
+			}
+			if !strings.HasSuffix(preparedKey, key) {
+				t.Errorf("Prepared non-recursive key %q should have suffix %q", preparedKey, key)
+			}
 		}
-	}
+	})
 
-	for _, key := range invalidKeys {
-		_, err := store.prepareKey(key)
-		if err == nil {
-			t.Errorf("key %q should be invalid", key)
+	t.Run("Recursive", func(t *testing.T) {
+		for _, key := range validKeys {
+			preparedKey, err := store.prepareKey(key, true)
+			if err != nil {
+				t.Errorf("preparing key %q should be valid; unexpected error: %v", key, err)
+				continue
+			}
+			if !strings.HasPrefix(preparedKey, expectPrefix) {
+				t.Errorf("prepared key %q should have prefix %q", preparedKey, expectPrefix)
+			}
+			if !strings.HasSuffix(preparedKey, "/") {
+				t.Errorf("Prepared non-recursive key %q should have suffix '/'", preparedKey)
+			}
 		}
-	}
+	})
+
+	t.Run("Invalid", func(t *testing.T) {
+		for _, key := range invalidKeys {
+			_, err := store.prepareKey(key, false)
+			if err == nil {
+				t.Errorf("key %q should be invalid", key)
+			}
+		}
+	})
 }
 
 func TestInvalidKeys(t *testing.T) {

staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher.go

@@ -104,7 +104,7 @@ type watchChan struct {
 // pred must be non-nil. Only if opts.Predicate matches the change, it will be returned.
 func (w *watcher) Watch(ctx context.Context, key string, rev int64, opts storage.ListOptions) (watch.Interface, error) {
 	if opts.Recursive && !strings.HasSuffix(key, "/") {
-		key += "/"
+		return nil, fmt.Errorf(`recursive key needs to end with "/"`)
 	}
 	if opts.ProgressNotify && w.newFunc == nil {
 		return nil, apierrors.NewInternalError(errors.New("progressNotify for watch is unsupported by the etcd storage because no newFunc was provided"))

staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher_test.go

@@ -227,7 +227,7 @@ func TestTooLargeResourceVersionErrorForWatchList(t *testing.T) {
 		t.Fatalf("Unable to convert NewTooLargeResourceVersionError to apierrors.StatusError")
 	}
 
-	w, err := store.watcher.Watch(ctx, "/abc", int64(102), requestOpts)
+	w, err := store.watcher.Watch(ctx, "/abc/", int64(102), requestOpts)
 	if err != nil {
 		t.Fatal(err)
 	}