Merge pull request #116031 from ritazh/kmsv2_mockkms

kmsv2: add mock kms for reference implementation

staging/publishing/rules.yaml

@@ -300,6 +300,10 @@ rules:
     dependencies:
     - repository: apimachinery
       branch: master
+    - repository: api
+      branch: master
+    - repository: client-go
+      branch: master
   - name: release-1.26
     go: 1.19.6
     source:

staging/src/k8s.io/kms/go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/gogo/protobuf v1.3.2
 	google.golang.org/grpc v1.51.0
 	k8s.io/apimachinery v0.0.0
+	k8s.io/client-go v0.0.0
 	k8s.io/klog/v2 v2.80.1
 	k8s.io/utils v0.0.0-20230209194617-a36077c30491
 )
@@ -19,11 +20,14 @@ require (
 	golang.org/x/net v0.7.0 // indirect
 	golang.org/x/sys v0.5.0 // indirect
 	golang.org/x/text v0.7.0 // indirect
+	golang.org/x/time v0.0.0-20220210224613-90d013bbcef8 // indirect
 	google.golang.org/genproto v0.0.0-20220502173005-c8bf987b8c21 // indirect
 	google.golang.org/protobuf v1.28.1 // indirect
 )
 
 replace (
+	k8s.io/api => ../api
 	k8s.io/apimachinery => ../apimachinery
+	k8s.io/client-go => ../client-go
 	k8s.io/kms => ../kms
 )

staging/src/k8s.io/kms/go.sum

@@ -110,6 +110,8 @@ golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.5/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.7.0 h1:4BRB4x83lYWy72KwLD/qYDuTu7q9PjSagHvijDw7cLo=
 golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
+golang.org/x/time v0.0.0-20220210224613-90d013bbcef8 h1:vVKdlvoWBphwdxWKrFZEuM0kGgGLxUOYcY4U/2Vjg44=
+golang.org/x/time v0.0.0-20220210224613-90d013bbcef8/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY=

staging/src/k8s.io/kms/internal/mock_aes_remote_service.go

@@ -0,0 +1,96 @@
+/*
+Copyright 2023 The Kubernetes 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.
+*/
+
+package internal
+
+import (
+	"context"
+	"crypto/aes"
+	"errors"
+
+	aestransformer "k8s.io/kms/pkg/encrypt/aes"
+	"k8s.io/kms/pkg/service"
+	"k8s.io/kms/pkg/value"
+)
+
+var _ service.Service = &mockAESRemoteService{}
+
+const (
+	mockAnnotationKey = "version.encryption.remote.io"
+)
+
+type mockAESRemoteService struct {
+	keyID       string
+	transformer value.Transformer
+	dataCtx     value.DefaultContext
+}
+
+func (s *mockAESRemoteService) Encrypt(ctx context.Context, uid string, plaintext []byte) (*service.EncryptResponse, error) {
+	out, err := s.transformer.TransformToStorage(ctx, plaintext, s.dataCtx)
+	if err != nil {
+		return nil, err
+	}
+
+	return &service.EncryptResponse{
+		KeyID:      s.keyID,
+		Ciphertext: out,
+		Annotations: map[string][]byte{
+			mockAnnotationKey: []byte("1"),
+		},
+	}, nil
+}
+
+func (s *mockAESRemoteService) Decrypt(ctx context.Context, uid string, req *service.DecryptRequest) ([]byte, error) {
+	if len(req.Annotations) != 1 {
+		return nil, errors.New("invalid annotations")
+	}
+	if v, ok := req.Annotations[mockAnnotationKey]; !ok || string(v) != "1" {
+		return nil, errors.New("invalid version in annotations")
+	}
+	if req.KeyID != s.keyID {
+		return nil, errors.New("invalid keyID")
+	}
+	from, _, err := s.transformer.TransformFromStorage(ctx, req.Ciphertext, s.dataCtx)
+	if err != nil {
+		return nil, err
+	}
+	return from, nil
+}
+
+func (s *mockAESRemoteService) Status(ctx context.Context) (*service.StatusResponse, error) {
+	resp := &service.StatusResponse{
+		Version: "v2alpha1",
+		Healthz: "ok",
+		KeyID:   s.keyID,
+	}
+	return resp, nil
+}
+
+// NewMockAESService creates an instance of mockAESRemoteService.
+func NewMockAESService(aesKey string, keyID string) (service.Service, error) {
+	block, err := aes.NewCipher([]byte(aesKey))
+	if err != nil {
+		return nil, err
+	}
+	if len(keyID) == 0 {
+		return nil, errors.New("invalid keyID")
+	}
+	return &mockAESRemoteService{
+		transformer: aestransformer.NewGCMTransformer(block),
+		keyID:       keyID,
+		dataCtx:     value.DefaultContext([]byte{}),
+	}, nil
+}

staging/src/k8s.io/kms/internal/mock_aes_remote_service_test.go

@@ -0,0 +1,117 @@
+/*
+Copyright 2023 The Kubernetes 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.
+*/
+
+package internal
+
+import (
+	"bytes"
+	"context"
+	"testing"
+
+	"k8s.io/kms/pkg/service"
+)
+
+const (
+	version       = "v2alpha1"
+	testAESKey    = "abcdefghijklmnop"
+	testKeyID     = "test-key-id"
+	testPlaintext = "lorem ipsum dolor sit amet"
+)
+
+func testContext(t *testing.T) context.Context {
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	return ctx
+}
+
+func TestMockAESRemoteService(t *testing.T) {
+	t.Parallel()
+	ctx := testContext(t)
+
+	plaintext := []byte(testPlaintext)
+
+	kmsService, err := NewMockAESService(testAESKey, testKeyID)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	t.Run("should be able to encrypt and decrypt", func(t *testing.T) {
+		t.Parallel()
+
+		encRes, err := kmsService.Encrypt(ctx, "", plaintext)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if bytes.Equal(plaintext, encRes.Ciphertext) {
+			t.Fatal("plaintext and ciphertext shouldn't be equal!")
+		}
+
+		decRes, err := kmsService.Decrypt(ctx, "", &service.DecryptRequest{
+			Ciphertext:  encRes.Ciphertext,
+			KeyID:       encRes.KeyID,
+			Annotations: encRes.Annotations,
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if !bytes.Equal(decRes, plaintext) {
+			t.Errorf("want: %q, have: %q", plaintext, decRes)
+		}
+	})
+
+	t.Run("should return error when decrypt with an invalid keyID", func(t *testing.T) {
+		t.Parallel()
+
+		encRes, err := kmsService.Encrypt(ctx, "", plaintext)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if bytes.Equal(plaintext, encRes.Ciphertext) {
+			t.Fatal("plaintext and ciphertext shouldn't be equal!")
+		}
+
+		_, err = kmsService.Decrypt(ctx, "", &service.DecryptRequest{
+			Ciphertext:  encRes.Ciphertext,
+			KeyID:       encRes.KeyID + "1",
+			Annotations: encRes.Annotations,
+		})
+		if err.Error() != "invalid keyID" {
+			t.Errorf("should have returned an invalid keyID error. Got %v, requested keyID: %q, remote service keyID: %q", err, encRes.KeyID+"1", testKeyID)
+		}
+	})
+
+	t.Run("should return status data", func(t *testing.T) {
+		t.Parallel()
+
+		status, err := kmsService.Status(ctx)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if status.Healthz != "ok" {
+			t.Errorf("want: %q, have: %q", "ok", status.Healthz)
+		}
+		if len(status.KeyID) == 0 {
+			t.Errorf("want: len(keyID) > 0, have: %d", len(status.KeyID))
+		}
+		if status.Version != version {
+			t.Errorf("want %q, have: %q", version, status.Version)
+		}
+	})
+}

staging/src/k8s.io/kms/internal/mock_latency_remote_service.go

@@ -0,0 +1,54 @@
+/*
+Copyright 2023 The Kubernetes 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.
+*/
+
+package internal
+
+import (
+	"context"
+	"time"
+
+	"k8s.io/kms/pkg/service"
+)
+
+type mockLatencyRemoteService struct {
+	delegate service.Service
+	latency  time.Duration
+}
+
+var _ service.Service = &mockLatencyRemoteService{}
+
+func (s *mockLatencyRemoteService) Decrypt(ctx context.Context, uid string, req *service.DecryptRequest) ([]byte, error) {
+	time.Sleep(s.latency)
+	return s.delegate.Decrypt(ctx, uid, req)
+}
+
+func (s *mockLatencyRemoteService) Encrypt(ctx context.Context, uid string, data []byte) (*service.EncryptResponse, error) {
+	time.Sleep(s.latency)
+	return s.delegate.Encrypt(ctx, uid, data)
+}
+
+func (s *mockLatencyRemoteService) Status(ctx context.Context) (*service.StatusResponse, error) {
+	// Passthrough here, not adding any delays for status as delays are usually negligible compare to encrypt and decrypt requests.
+	return s.delegate.Status(ctx)
+}
+
+// NewMockLatencyService creates an instance of mockLatencyRemoteService.
+func NewMockLatencyService(delegate service.Service, latency time.Duration) service.Service {
+	return &mockLatencyRemoteService{
+		delegate: delegate,
+		latency:  latency,
+	}
+}

staging/src/k8s.io/kms/internal/mock_latency_remote_service_test.go

@@ -0,0 +1,99 @@
+/*
+Copyright 2023 The Kubernetes 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.
+*/
+
+package internal
+
+import (
+	"bytes"
+	"testing"
+	"time"
+
+	"k8s.io/kms/pkg/service"
+)
+
+const (
+	testLatencyInMillisecond = 100 * time.Millisecond
+)
+
+func TestMockLatencyRemoteService(t *testing.T) {
+	t.Parallel()
+	ctx := testContext(t)
+
+	plaintext := []byte(testPlaintext)
+	aesService, err := NewMockAESService(testAESKey, testKeyID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	kmsService := NewMockLatencyService(aesService, testLatencyInMillisecond)
+
+	t.Run("should be able to encrypt and decrypt with some known latency", func(t *testing.T) {
+		t.Parallel()
+		start := time.Now()
+		encRes, err := kmsService.Encrypt(ctx, "", plaintext)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		duration := time.Since(start)
+
+		if bytes.Equal(plaintext, encRes.Ciphertext) {
+			t.Fatal("plaintext and ciphertext shouldn't be equal!")
+		}
+		// Max is set to 3s to limit the risk of a CPU limited CI node taking a long time to do encryption.
+		if duration < testLatencyInMillisecond || duration > 3*time.Second {
+			t.Errorf("duration for encrypt should be around: %q, have: %q", testLatencyInMillisecond, duration)
+		}
+		start = time.Now()
+		decRes, err := kmsService.Decrypt(ctx, "", &service.DecryptRequest{
+			Ciphertext:  encRes.Ciphertext,
+			KeyID:       encRes.KeyID,
+			Annotations: encRes.Annotations,
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+		duration = time.Since(start)
+
+		if !bytes.Equal(decRes, plaintext) {
+			t.Errorf("want: %q, have: %q", plaintext, decRes)
+		}
+		if duration < testLatencyInMillisecond || duration > 3*time.Second {
+			t.Errorf("duration decrypt should be around: %q, have: %q", testLatencyInMillisecond, duration)
+		}
+	})
+
+	t.Run("should return status data", func(t *testing.T) {
+		t.Parallel()
+		start := time.Now()
+		status, err := kmsService.Status(ctx)
+		if err != nil {
+			t.Fatal(err)
+		}
+		duration := time.Since(start)
+		if status.Healthz != "ok" {
+			t.Errorf("want: %q, have: %q", "ok", status.Healthz)
+		}
+		if len(status.KeyID) == 0 {
+			t.Errorf("want: len(keyID) > 0, have: %d", len(status.KeyID))
+		}
+		if status.Version != version {
+			t.Errorf("want %q, have: %q", version, status.Version)
+		}
+		if duration > 3*time.Second {
+			t.Errorf("duration status should be less than: 3s, have: %q", duration)
+		}
+	})
+}

staging/src/k8s.io/kms/internal/mock_ratelimit_remote_service.go

@@ -0,0 +1,60 @@
+/*
+Copyright 2023 The Kubernetes 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.
+*/
+
+package internal
+
+import (
+	"context"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"k8s.io/client-go/util/flowcontrol"
+	"k8s.io/kms/pkg/service"
+)
+
+type mockRateLimitRemoteService struct {
+	delegate service.Service
+	limiter  flowcontrol.RateLimiter
+}
+
+var _ service.Service = &mockRateLimitRemoteService{}
+
+func (s *mockRateLimitRemoteService) Decrypt(ctx context.Context, uid string, req *service.DecryptRequest) ([]byte, error) {
+	if !s.limiter.TryAccept() {
+		return nil, status.New(codes.ResourceExhausted, "remote decrypt rate limit exceeded").Err()
+	}
+	return s.delegate.Decrypt(ctx, uid, req)
+}
+
+func (s *mockRateLimitRemoteService) Encrypt(ctx context.Context, uid string, data []byte) (*service.EncryptResponse, error) {
+	if !s.limiter.TryAccept() {
+		return nil, status.New(codes.ResourceExhausted, "remote encrypt rate limit exceeded").Err()
+	}
+	return s.delegate.Encrypt(ctx, uid, data)
+}
+
+func (s *mockRateLimitRemoteService) Status(ctx context.Context) (*service.StatusResponse, error) {
+	// Passthrough here, not adding any rate limiting for status as rate limits are usually for encrypt and decrypt requests.
+	return s.delegate.Status(ctx)
+}
+
+// NewMockRateLimitService creates an instance of mockRateLimitRemoteService.
+func NewMockRateLimitService(delegate service.Service, qps float32, burst int) service.Service {
+	return &mockRateLimitRemoteService{
+		delegate: delegate,
+		limiter:  flowcontrol.NewTokenBucketRateLimiter(qps, burst),
+	}
+}

staging/src/k8s.io/kms/internal/mock_ratelimit_remote_service_test.go

@@ -0,0 +1,74 @@
+/*
+Copyright 2023 The Kubernetes 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.
+*/
+
+package internal
+
+import (
+	"bytes"
+	"testing"
+)
+
+const (
+	testQPS = 1
+	// testBurst should be no more than 9 since 9*100millisecond (test latency) = 900ms, which guarantees there is enough bursts per second.
+	testBurst = 5
+)
+
+func TestMockRateLimitRemoteService(t *testing.T) {
+	t.Parallel()
+	ctx := testContext(t)
+	plaintext := []byte(testPlaintext)
+	aesService, err := NewMockAESService(testAESKey, testKeyID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	mockLatencyService := NewMockLatencyService(aesService, testLatencyInMillisecond)
+	kmsService := NewMockRateLimitService(mockLatencyService, testQPS, testBurst)
+
+	t.Run("should hit rate limit", func(t *testing.T) {
+		rateLimitExceeded := false
+		for i := 0; i < 100; i++ {
+			encRes, err := kmsService.Encrypt(ctx, "", plaintext)
+			if i >= testBurst {
+				if err != nil {
+					if err.Error() != "rpc error: code = ResourceExhausted desc = remote encrypt rate limit exceeded" {
+						t.Fatalf("should have failed with rate limit exceeded %d, have err: %v", testBurst, err)
+					}
+					rateLimitExceeded = true
+				} else {
+					if bytes.Equal(plaintext, encRes.Ciphertext) {
+						t.Fatal("plaintext and ciphertext shouldn't be equal!")
+					}
+				}
+			} else {
+				if err != nil {
+					t.Fatalf("err: %v, i: %d", err, i)
+				}
+				if bytes.Equal(plaintext, encRes.Ciphertext) {
+					t.Fatal("plaintext and ciphertext shouldn't be equal!")
+				}
+			}
+			// status should not hit any rate limit
+			_, err = kmsService.Status(ctx)
+			if err != nil {
+				t.Fatal(err)
+			}
+		}
+		if !rateLimitExceeded {
+			t.Errorf("should have reached the rate limit of %d", testBurst)
+		}
+	})
+}

staging/src/k8s.io/kms/pkg/hierarchy/hierarchy_test.go

@@ -33,6 +33,11 @@ import (
 	testingclock "k8s.io/utils/clock/testing"
 )
 
+const (
+	testAnnotationKey        = "version.encryption.remote.io"
+	testAnnotationKeyVersion = "key-version.encryption.remote.io"
+)
+
 func TestCopyResponseAndAddLocalKEKAnnotation(t *testing.T) {
 	t.Parallel()
 	testCases := []struct {
@@ -60,14 +65,14 @@ func TestCopyResponseAndAddLocalKEKAnnotation(t *testing.T) {
 				Ciphertext: []byte("encryptedLocalKEK"),
 				KeyID:      "keyID",
 				Annotations: map[string][]byte{
-					"version.encryption.remote.io": []byte("1"),
+					testAnnotationKey: []byte("1"),
 				},
 			},
 			want: &service.EncryptResponse{
 				KeyID: "keyID",
 				Annotations: map[string][]byte{
-					"version.encryption.remote.io": []byte("1"),
-					referenceKEKAnnotationKey:      []byte("encryptedLocalKEK"),
+					testAnnotationKey:         []byte("1"),
+					referenceKEKAnnotationKey: []byte("encryptedLocalKEK"),
 				},
 			},
 		},
@@ -77,16 +82,16 @@ func TestCopyResponseAndAddLocalKEKAnnotation(t *testing.T) {
 				Ciphertext: []byte("encryptedLocalKEK"),
 				KeyID:      "keyID",
 				Annotations: map[string][]byte{
-					"version.encryption.remote.io":     []byte("1"),
-					"key-version.encryption.remote.io": []byte("2"),
+					testAnnotationKey:        []byte("1"),
+					testAnnotationKeyVersion: []byte("2"),
 				},
 			},
 			want: &service.EncryptResponse{
 				KeyID: "keyID",
 				Annotations: map[string][]byte{
-					"version.encryption.remote.io":     []byte("1"),
-					"key-version.encryption.remote.io": []byte("2"),
-					referenceKEKAnnotationKey:          []byte("encryptedLocalKEK"),
+					testAnnotationKey:         []byte("1"),
+					testAnnotationKeyVersion:  []byte("2"),
+					referenceKEKAnnotationKey: []byte("encryptedLocalKEK"),
 				},
 			},
 		},
@@ -131,11 +136,11 @@ func TestAnnotationsWithoutReferenceKeys(t *testing.T) {
 		{
 			name: "annotations contains 1 reference key and 1 other key",
 			input: map[string][]byte{
-				referenceKEKAnnotationKey:      []byte("encryptedLocalKEK"),
-				"version.encryption.remote.io": []byte("1"),
+				referenceKEKAnnotationKey: []byte("encryptedLocalKEK"),
+				testAnnotationKey:         []byte("1"),
 			},
 			want: map[string][]byte{
-				"version.encryption.remote.io": []byte("1"),
+				testAnnotationKey: []byte("1"),
 			},
 		},
 	}
@@ -177,8 +182,8 @@ func TestValidateRemoteKMSEncryptResponse(t *testing.T) {
 			name: "no annotation key contains reference suffix",
 			input: &service.EncryptResponse{
 				Annotations: map[string][]byte{
-					"version.encryption.remote.io":     []byte("1"),
-					"key-version.encryption.remote.io": []byte("2"),
+					testAnnotationKey:        []byte("1"),
+					testAnnotationKeyVersion: []byte("2"),
 				},
 			},
 			want: nil,
@@ -264,7 +269,7 @@ func (s *testRemoteService) Encrypt(ctx context.Context, uid string, plaintext [
 		KeyID:      s.keyID,
 		Ciphertext: []byte(base64.StdEncoding.EncodeToString(plaintext)),
 		Annotations: map[string][]byte{
-			"version.encryption.remote.io": []byte("1"),
+			testAnnotationKey: []byte("1"),
 		},
 	}, nil
 }
@@ -280,7 +285,7 @@ func (s *testRemoteService) Decrypt(ctx context.Context, uid string, req *servic
 	if len(req.Annotations) != 1 {
 		return nil, errors.New("invalid annotations")
 	}
-	if v, ok := req.Annotations["version.encryption.remote.io"]; !ok || string(v) != "1" {
+	if v, ok := req.Annotations[testAnnotationKey]; !ok || string(v) != "1" {
 		return nil, errors.New("invalid version in annotations")
 	}
 	return base64.StdEncoding.DecodeString(string(req.Ciphertext))