Encryption config: correctly handle overlapping providers

This change updates NewPrefixTransformers to not short-circuit on
the first transformer that has a matching prefix.  If the same type
of encryption ProviderConfiguration is used more than once, they
will share the same prefix.  A failure in the first one should not
prevent a later match from being attempted.

Added TestCBCKeyRotationWithOverlappingProviders unit test to
prevent regressions.  Note that this test explicitly exercises this
flow using an EncryptionConfiguration object as the structure of the
resulting transformer is an important part of the check.

Signed-off-by: Monis Khan <mkhan@redhat.com>

staging/src/k8s.io/apiserver/pkg/server/options/encryptionconfig/config_test.go

@@ -605,3 +605,206 @@ resources:
 func serviceComparer(_, _ envelope.Service) bool {
 	return true
 }
+
+func TestCBCKeyRotationWithOverlappingProviders(t *testing.T) {
+	testCBCKeyRotationWithProviders(
+		t,
+		`{
+  "kind": "EncryptionConfiguration",
+  "apiVersion": "apiserver.config.k8s.io/v1",
+  "resources": [
+    {
+      "resources": [
+        "ignored"
+      ],
+      "providers": [
+        {
+          "aescbc": {
+            "keys": [
+              {
+                "name": "1",
+                "secret": "Owq7A4JrJpSjrvH8kXkvl4JmOLzvZ6j9BcGRkR8OPQ4="
+              }
+            ]
+          }
+        },
+        {
+          "aescbc": {
+            "keys": [
+              {
+                "name": "2",
+                "secret": "+qcnfOFX3aRXM9PuY7lQXDWYIQ3GWUdBc3nYBo91SCA="
+              }
+            ]
+          }
+        }
+      ]
+    }
+  ]
+}`,
+		"k8s:enc:aescbc:v1:1:",
+		`{
+  "kind": "EncryptionConfiguration",
+  "apiVersion": "apiserver.config.k8s.io/v1",
+  "resources": [
+    {
+      "resources": [
+        "ignored"
+      ],
+      "providers": [
+        {
+          "aescbc": {
+            "keys": [
+              {
+                "name": "2",
+                "secret": "+qcnfOFX3aRXM9PuY7lQXDWYIQ3GWUdBc3nYBo91SCA="
+              }
+            ]
+          }
+        },
+        {
+          "aescbc": {
+            "keys": [
+              {
+                "name": "1",
+                "secret": "Owq7A4JrJpSjrvH8kXkvl4JmOLzvZ6j9BcGRkR8OPQ4="
+              }
+            ]
+          }
+        }
+      ]
+    }
+  ]
+}`,
+		"k8s:enc:aescbc:v1:2:",
+	)
+}
+
+func TestCBCKeyRotationWithoutOverlappingProviders(t *testing.T) {
+	testCBCKeyRotationWithProviders(
+		t,
+		`{
+  "kind": "EncryptionConfiguration",
+  "apiVersion": "apiserver.config.k8s.io/v1",
+  "resources": [
+    {
+      "resources": [
+        "ignored"
+      ],
+      "providers": [
+        {
+          "aescbc": {
+            "keys": [
+              {
+                "name": "A",
+                "secret": "Owq7A4JrJpSjrvH8kXkvl4JmOLzvZ6j9BcGRkR8OPQ4="
+              },
+              {
+                "name": "B",
+                "secret": "+qcnfOFX3aRXM9PuY7lQXDWYIQ3GWUdBc3nYBo91SCA="
+              }
+            ]
+          }
+        }
+      ]
+    }
+  ]
+}`,
+		"k8s:enc:aescbc:v1:A:",
+		`{
+  "kind": "EncryptionConfiguration",
+  "apiVersion": "apiserver.config.k8s.io/v1",
+  "resources": [
+    {
+      "resources": [
+        "ignored"
+      ],
+      "providers": [
+        {
+          "aescbc": {
+            "keys": [
+              {
+                "name": "B",
+                "secret": "+qcnfOFX3aRXM9PuY7lQXDWYIQ3GWUdBc3nYBo91SCA="
+              },
+              {
+                "name": "A",
+                "secret": "Owq7A4JrJpSjrvH8kXkvl4JmOLzvZ6j9BcGRkR8OPQ4="
+              }
+            ]
+          }
+        }
+      ]
+    }
+  ]
+}`,
+		"k8s:enc:aescbc:v1:B:",
+	)
+}
+
+func testCBCKeyRotationWithProviders(t *testing.T, firstEncryptionConfig, firstPrefix, secondEncryptionConfig, secondPrefix string) {
+	p := getTransformerFromEncryptionConfig(t, firstEncryptionConfig)
+
+	context := value.DefaultContext([]byte("authenticated_data"))
+
+	out, err := p.TransformToStorage([]byte("firstvalue"), context)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.HasPrefix(out, []byte(firstPrefix)) {
+		t.Fatalf("unexpected prefix: %q", out)
+	}
+	from, stale, err := p.TransformFromStorage(out, context)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if stale || !bytes.Equal([]byte("firstvalue"), from) {
+		t.Fatalf("unexpected data: %t %q", stale, from)
+	}
+
+	// verify changing the context fails storage
+	_, _, err = p.TransformFromStorage(out, value.DefaultContext([]byte("incorrect_context")))
+	if err != nil {
+		t.Fatalf("CBC mode does not support authentication: %v", err)
+	}
+
+	// reverse the order, use the second key
+	p = getTransformerFromEncryptionConfig(t, secondEncryptionConfig)
+	from, stale, err = p.TransformFromStorage(out, context)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !stale || !bytes.Equal([]byte("firstvalue"), from) {
+		t.Fatalf("unexpected data: %t %q", stale, from)
+	}
+
+	out, err = p.TransformToStorage([]byte("firstvalue"), context)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.HasPrefix(out, []byte(secondPrefix)) {
+		t.Fatalf("unexpected prefix: %q", out)
+	}
+	from, stale, err = p.TransformFromStorage(out, context)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if stale || !bytes.Equal([]byte("firstvalue"), from) {
+		t.Fatalf("unexpected data: %t %q", stale, from)
+	}
+}
+
+func getTransformerFromEncryptionConfig(t *testing.T, encryptionConfig string) value.Transformer {
+	t.Helper()
+	transformers, err := ParseEncryptionConfiguration(strings.NewReader(encryptionConfig))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(transformers) != 1 {
+		t.Fatalf("input config does not have exactly one resource: %s", encryptionConfig)
+	}
+	for _, transformer := range transformers {
+		return transformer
+	}
+	panic("unreachable")
+}

staging/src/k8s.io/apiserver/pkg/storage/value/BUILD

@@ -30,6 +30,7 @@ go_library(
     importmap = "k8s.io/kubernetes/vendor/k8s.io/apiserver/pkg/storage/value",
     importpath = "k8s.io/apiserver/pkg/storage/value",
     deps = [
+        "//staging/src/k8s.io/apimachinery/pkg/util/errors:go_default_library",
         "//staging/src/k8s.io/component-base/metrics:go_default_library",
         "//staging/src/k8s.io/component-base/metrics/legacyregistry:go_default_library",
         "//vendor/github.com/prometheus/client_golang/prometheus:go_default_library",

staging/src/k8s.io/apiserver/pkg/storage/value/transformer.go

@@ -22,6 +22,8 @@ import (
 	"fmt"
 	"sync"
 	"time"
+
+	"k8s.io/apimachinery/pkg/util/errors"
 )
 
 func init() {
@@ -129,6 +131,7 @@ func NewPrefixTransformers(err error, transformers ...PrefixTransformer) Transfo
 // the first transformer.
 func (t *prefixTransformers) TransformFromStorage(data []byte, context Context) ([]byte, bool, error) {
 	start := time.Now()
+	var errs []error
 	for i, transformer := range t.transformers {
 		if bytes.HasPrefix(data, transformer.Prefix) {
 			result, stale, err := transformer.Transformer.TransformFromStorage(data[len(transformer.Prefix):], context)
@@ -144,9 +147,48 @@ func (t *prefixTransformers) TransformFromStorage(data []byte, context Context)
 			} else {
 				RecordTransformation("from_storage", string(transformer.Prefix), start, err)
 			}
+
+			// It is valid to have overlapping prefixes when the same encryption provider
+			// is specified multiple times but with different keys (the first provider is
+			// being rotated to and some later provider is being rotated away from).
+			//
+			// Example:
+			//
+			//  {
+			//    "aescbc": {
+			//      "keys": [
+			//        {
+			//          "name": "2",
+			//          "secret": "some key 2"
+			//        }
+			//      ]
+			//    }
+			//  },
+			//  {
+			//    "aescbc": {
+			//      "keys": [
+			//        {
+			//          "name": "1",
+			//          "secret": "some key 1"
+			//        }
+			//      ]
+			//    }
+			//  },
+			//
+			// The transformers for both aescbc configs share the prefix k8s:enc:aescbc:v1:
+			// but a failure in the first one should not prevent a later match from being attempted.
+			// Thus we never short-circuit on a prefix match that results in an error.
+			if err != nil {
+				errs = append(errs, err)
+				continue
+			}
+
 			return result, stale || i != 0, err
 		}
 	}
+	if err := errors.Reduce(errors.NewAggregate(errs)); err != nil {
+		return nil, false, err
+	}
 	RecordTransformation("from_storage", "unknown", start, t.err)
 	return nil, false, t.err
 }

staging/src/k8s.io/apiserver/pkg/storage/value/transformer_test.go

@@ -45,7 +45,7 @@ func (t *testTransformer) TransformToStorage(to []byte, context Context) (data [
 
 func TestPrefixFrom(t *testing.T) {
 	testErr := fmt.Errorf("test error")
-	transformErr := fmt.Errorf("test error")
+	transformErr := fmt.Errorf("transform error")
 	transformers := []PrefixTransformer{
 		{Prefix: []byte("first:"), Transformer: &testTransformer{from: []byte("value1")}},
 		{Prefix: []byte("second:"), Transformer: &testTransformer{from: []byte("value2")}},
@@ -64,7 +64,7 @@ func TestPrefixFrom(t *testing.T) {
 		{[]byte("first:value"), []byte("value1"), false, nil, 0},
 		{[]byte("second:value"), []byte("value2"), true, nil, 1},
 		{[]byte("third:value"), nil, false, testErr, -1},
-		{[]byte("fails:value"), nil, true, transformErr, 2},
+		{[]byte("fails:value"), nil, false, transformErr, 2},
 		{[]byte("stale:value"), []byte("value3"), true, nil, 3},
 	}
 	for i, test := range testCases {