Merge pull request #58375 from liggitt/decrypt

Automatic merge from submit-queue. If you want to cherry-pick this change to another branch, please follow the instructions <a href="https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md">here</a>.

Recheck if transformed data is stale when doing live lookup during update

Fixes #49565

Caching storage can pass in a cached object to `GuaranteedUpdate` as a hint for the current object.

If the hint is identical to the data we want to persist, before short-circuiting as a no-op update, we force a live lookup.

We should check two things on the result of that live lookup before short-circuiting as a no-op update:
1. the bytes we want to persist still match the transformed bytes read from etcd
2. the state read from etcd didn't report itself as stale. this would mean the transformer used to read the data would not be the transformer used to write it, and "no-op" writes should still be performed, since transformation will make the underlying content actually different.

After a live lookup, we checked byte equality, but not the stale indicator. This meant that key rotation or encrypted->decrypted, and decrypted->encrypted updates are broken.

Introduced in #54780 and picked back to 1.8 in #55294

```release-note
Fixed encryption key and encryption provider rotation
```

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

@@ -332,13 +332,16 @@ func (s *store) GuaranteedUpdate(
 				if err != nil {
 					return err
 				}
+				mustCheckData = false
 				if !bytes.Equal(data, origState.data) {
 					// original data changed, restart loop
-					mustCheckData = false
 					continue
 				}
 			}
-			return decode(s.codec, s.versioner, origState.data, out, origState.rev)
+			// recheck that the data from etcd is not stale before short-circuiting a write
+			if !origState.stale {
+				return decode(s.codec, s.versioner, origState.data, out, origState.rev)
+			}
 		}
 
 		newData, err := s.transformer.TransformToStorage(data, transformContext)

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

@@ -526,6 +526,8 @@ func TestGuaranteedUpdateChecksStoredData(t *testing.T) {
 		t.Fatal(err)
 	}
 
+	store.transformer = prefixTransformer{prefix: []byte(defaultTestPrefix)}
+
 	// this update should write the canonical value to etcd because the new serialization differs
 	// from the stored serialization
 	input.ResourceVersion = strconv.FormatInt(resp.Header.Revision, 10)
@@ -540,6 +542,36 @@ func TestGuaranteedUpdateChecksStoredData(t *testing.T) {
 	if out.ResourceVersion == strconv.FormatInt(resp.Header.Revision, 10) {
 		t.Errorf("guaranteed update should have updated the serialized data, got %#v", out)
 	}
+
+	lastVersion := out.ResourceVersion
+
+	// this update should not write to etcd because the input matches the stored data
+	input = out
+	out = &example.Pod{}
+	err = store.GuaranteedUpdate(ctx, key, out, true, nil,
+		func(_ runtime.Object, _ storage.ResponseMeta) (runtime.Object, *uint64, error) {
+			return input, nil, nil
+		}, input)
+	if err != nil {
+		t.Fatalf("Update failed: %v", err)
+	}
+	if out.ResourceVersion != lastVersion {
+		t.Errorf("guaranteed update should have short-circuited write, got %#v", out)
+	}
+
+	store.transformer = prefixTransformer{prefix: []byte(defaultTestPrefix), stale: true}
+
+	// this update should write to etcd because the transformer reported stale
+	err = store.GuaranteedUpdate(ctx, key, out, true, nil,
+		func(_ runtime.Object, _ storage.ResponseMeta) (runtime.Object, *uint64, error) {
+			return input, nil, nil
+		}, input)
+	if err != nil {
+		t.Fatalf("Update failed: %v", err)
+	}
+	if out.ResourceVersion == lastVersion {
+		t.Errorf("guaranteed update should have written to etcd when transformer reported stale, got %#v", out)
+	}
 }
 
 func TestGuaranteedUpdateWithConflict(t *testing.T) {