Fix connection secret owner check for K8s Secret Store

Fixes crossplane/crossplane#3520

Signed-off-by: Hasan Turken <turkenh@gmail.com>
(cherry picked from commit b130752)

pkg/connection/store/kubernetes/store.go

@@ -211,6 +211,19 @@ func applyOptions(wo ...store.WriteOption) []resource.ApplyOption {
 				},
 				Data: currentSecret.Data,
 			}
+
+			// NOTE(turkenh): With External Secret Stores, we are using a special label/tag with key
+			// "secret.crossplane.io/owner-uid" to track the owner of the connection secret. However, different from
+			// other Secret Store implementations, Kubernetes Store uses metadata.OwnerReferences for this purpose and
+			// we don't want it to appear in the labels of the secret additionally.
+			// Here we are adding the owner label to the internal representation of the current secret as part of
+			// converting store.WriteOption's to k8s resource.ApplyOption's, so that our generic store.WriteOptions
+			// checking secret owner could work as expected.
+			// Fixes: https://github.com/crossplane/crossplane/issues/3520
+			if len(currentSecret.GetOwnerReferences()) > 0 {
+				cs.Metadata.SetOwnerUID(currentSecret.GetOwnerReferences()[0].UID)
+			}
+
 			ds := &store.Secret{
 				ScopedName: store.ScopedName{
 					Name:  desiredSecret.Name,

pkg/connection/store/kubernetes/store_test.go

@@ -25,6 +25,7 @@ import (
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	v1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
@@ -39,6 +40,7 @@ var (
 
 	fakeSecretName      = "fake"
 	fakeSecretNamespace = "fake-namespace"
+	fakeOwnerID         = "00000000-0000-0000-0000-000000000000"
 
 	storeTypeKubernetes = v1.SecretStoreKubernetes
 )
@@ -297,6 +299,46 @@ func TestSecretStoreWriteKeyValues(t *testing.T) {
 				changed: true,
 			},
 		},
+		"SecretHasExpectedOwner": {
+			reason: "Should correctly check the owner of the secret.",
+			args: args{
+				client: resource.ClientApplicator{
+					Applicator: resource.ApplyFn(func(ctx context.Context, obj client.Object, option ...resource.ApplyOption) error {
+						if diff := cmp.Diff(fakeConnectionSecret(withData(map[string][]byte{
+							"existing-key": []byte("new-value"),
+						})), obj.(*corev1.Secret)); diff != "" {
+							t.Errorf("r: -want, +got:\n%s", diff)
+						}
+						for _, fn := range option {
+							if err := fn(ctx, fakeConnectionSecret(withData(map[string][]byte{
+								"existing-key": []byte("old-value"),
+							}), withOwnerID(fakeOwnerID)), obj); err != nil {
+								return err
+							}
+						}
+						return nil
+					}),
+				},
+				secret: &store.Secret{
+					ScopedName: store.ScopedName{
+						Name:  fakeSecretName,
+						Scope: fakeSecretNamespace,
+					},
+					Data: store.KeyValues(map[string][]byte{
+						"existing-key": []byte("new-value"),
+					}),
+				},
+				wo: []store.WriteOption{func(ctx context.Context, current, desired *store.Secret) error {
+					if current.Metadata == nil || current.Metadata.GetOwnerUID() != fakeOwnerID {
+						return errors.Errorf("secret not owned by %s", fakeOwnerID)
+					}
+					return nil
+				}},
+			},
+			want: want{
+				changed: true,
+			},
+		},
 		"SecretUpdatedWithNewKey": {
 			reason: "Should update existing secret additively if a new key added",
 			args: args{
@@ -809,6 +851,17 @@ func withAnnotations(a map[string]string) secretOption {
 		s.Annotations = a
 	}
 }
+
+func withOwnerID(id string) secretOption {
+	return func(s *corev1.Secret) {
+		s.SetOwnerReferences([]metav1.OwnerReference{
+			{
+				UID: types.UID(id),
+			},
+		})
+	}
+}
+
 func fakeConnectionSecret(opts ...secretOption) *corev1.Secret {
 	s := &corev1.Secret{
 		ObjectMeta: metav1.ObjectMeta{