GC only if allowed to update DV owner finalizers

PVC gets its DV ownerRefs, which can be any CR. If BlockOwnerDeletion is
true, we allow GC only if CDI has RBAC to update the owner finalizers
(we explicitly added it for VirtualMachines). BlockOwnerDeletions gets
validated with this admission plugin which is enabled in OpenShift, but
disabled in our kubevirtci clusters:

https://kubernetes.io/docs/reference/access-authn-authz/admission-controllers/#ownerreferencespermissionenforcement

Signed-off-by: Arnon Gilboa <agilboa@redhat.com>

pkg/controller/BUILD.bazel

@@ -43,6 +43,7 @@ go_library(
         "//vendor/github.com/openshift/api/route/v1:go_default_library",
         "//vendor/github.com/pkg/errors:go_default_library",
         "//vendor/github.com/prometheus/client_golang/prometheus:go_default_library",
+        "//vendor/k8s.io/api/authorization/v1:go_default_library",
         "//vendor/k8s.io/api/batch/v1:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/api/networking/v1:go_default_library",

pkg/controller/datavolume-controller.go

@@ -34,6 +34,7 @@ import (
 	snapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
 	"github.com/pkg/errors"
 
+	authorizationv1 "k8s.io/api/authorization/v1"
 	corev1 "k8s.io/api/core/v1"
 	storagev1 "k8s.io/api/storage/v1"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
@@ -2783,6 +2784,10 @@ func (r *DatavolumeReconciler) garbageCollect(dataVolume *cdiv1.DataVolume, pvc
 	if dataVolume.Status.Phase != cdiv1.Succeeded {
 		return nil, nil
 	}
+	if allowed, err := r.isGarbageCollectionAllowed(dataVolume); !allowed || err != nil {
+		log.Info("Garbage Collection is not allowed, due to owner finalizers update RBAC")
+		return nil, err
+	}
 	cdiConfig := &cdiv1.CDIConfig{}
 	if err := r.client.Get(context.TODO(), types.NamespacedName{Name: common.ConfigName}, cdiConfig); err != nil {
 		return nil, err
@@ -2802,6 +2807,41 @@ func (r *DatavolumeReconciler) garbageCollect(dataVolume *cdiv1.DataVolume, pvc
 	return &reconcile.Result{}, nil
 }
 
+func (r *DatavolumeReconciler) isGarbageCollectionAllowed(dv *cdiv1.DataVolume) (bool, error) {
+	for _, ref := range dv.OwnerReferences {
+		if ref.BlockOwnerDeletion == nil || *ref.BlockOwnerDeletion == false {
+			continue
+		}
+		if allowed, err := r.canUpdateFinalizers(ref); !allowed || err != nil {
+			return false, err
+		}
+	}
+	return true, nil
+}
+
+func (r *DatavolumeReconciler) canUpdateFinalizers(ownerRef metav1.OwnerReference) (bool, error) {
+	gvk := schema.FromAPIVersionAndKind(ownerRef.APIVersion, ownerRef.Kind)
+	mapping, err := r.client.RESTMapper().RESTMapping(gvk.GroupKind(), gvk.Version)
+	if err != nil {
+		return false, err
+	}
+	ssar := &authorizationv1.SelfSubjectAccessReview{
+		Spec: authorizationv1.SelfSubjectAccessReviewSpec{
+			ResourceAttributes: &authorizationv1.ResourceAttributes{
+				Group:       gvk.Group,
+				Version:     gvk.Version,
+				Resource:    mapping.Resource.Resource,
+				Subresource: "finalizers",
+				Verb:        "update",
+			},
+		},
+	}
+	if err := r.client.Create(context.TODO(), ssar); err != nil {
+		return false, err
+	}
+	return ssar.Status.Allowed, nil
+}
+
 func (r *DatavolumeReconciler) detachPvcDeleteDv(pvc *corev1.PersistentVolumeClaim, dv *cdiv1.DataVolume, log logr.Logger) error {
 	if dv.Status.Phase != cdiv1.Succeeded {
 		return nil
@@ -2824,13 +2864,24 @@ func (r *DatavolumeReconciler) detachPvcDeleteDv(pvc *corev1.PersistentVolumeCla
 
 func updatePvcOwnerRefs(pvc *corev1.PersistentVolumeClaim, dv *cdiv1.DataVolume) {
 	refs := pvc.OwnerReferences
-	for i, r := range refs {
-		if r.UID == dv.UID {
-			pvc.OwnerReferences = append(refs[:i], refs[i+1:]...)
-			break
+	if i := findOwnerRef(refs, dv.UID); i >= 0 {
+		refs = append(refs[:i], refs[i+1:]...)
+	}
+	for _, ref := range dv.OwnerReferences {
+		if findOwnerRef(refs, ref.UID) == -1 {
+			refs = append(refs, ref)
+		}
+	}
+	pvc.OwnerReferences = refs
+}
+
+func findOwnerRef(refs []metav1.OwnerReference, uid types.UID) int {
+	for i, ref := range refs {
+		if ref.UID == uid {
+			return i
 		}
 	}
-	pvc.OwnerReferences = append(pvc.OwnerReferences, dv.OwnerReferences...)
+	return -1
 }
 
 func getDeltaTTL(dv *cdiv1.DataVolume, ttl int32) time.Duration {

pkg/controller/datavolume-controller_test.go

@@ -2101,6 +2101,26 @@ var _ = Describe("All DataVolume Tests", func() {
 		Entry("40Gi virtual size, default overhead to be 40Gi if <= 1Gi and 41Gi if > 40Gi", 40*Gi, defaultOverhead),
 		Entry("40Gi virtual size, large overhead to be 40Gi if <= 40Gi and 41Gi if > 40Gi", 40*Gi, largeOverhead),
 	)
+
+	Describe("DataVolume garbage collection", func() {
+		It("updatePvcOwnerRefs should correctly update PVC owner refs", func() {
+			ref := func(uid string) metav1.OwnerReference {
+				return metav1.OwnerReference{UID: types.UID(uid)}
+			}
+			dv := newImportDataVolume("test-dv")
+			vmOwnerRef := metav1.OwnerReference{Kind: "VirtualMachine", Name: "test-vm", UID: "test-vm-uid"}
+			dv.OwnerReferences = append(dv.OwnerReferences, ref("3"), vmOwnerRef, ref("2"), ref("1"))
+
+			pvc := createPvc("test-dv", metav1.NamespaceDefault, nil, nil)
+			dvOwnerRef := metav1.OwnerReference{Kind: "DataVolume", Name: "test-dv", UID: dv.UID}
+			pvc.OwnerReferences = append(pvc.OwnerReferences, ref("1"), dvOwnerRef, ref("2"))
+
+			updatePvcOwnerRefs(pvc, dv)
+			Expect(pvc.OwnerReferences).To(HaveLen(4))
+			Expect(pvc.OwnerReferences).To(Equal([]metav1.OwnerReference{ref("1"), ref("2"), ref("3"), vmOwnerRef}))
+		})
+	})
+
 })
 
 func createStorageSpec() *cdiv1.StorageSpec {

pkg/operator/resources/cluster/controller.go

@@ -234,6 +234,17 @@ func getControllerClusterPolicyRules() []rbacv1.PolicyRule {
 				"watch",
 			},
 		},
+		{
+			APIGroups: []string{
+				"kubevirt.io",
+			},
+			Resources: []string{
+				"virtualmachines/finalizers",
+			},
+			Verbs: []string{
+				"update",
+			},
+		},
 	}
 }