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,42 @@ 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) {
+	group := ""
+	version := ownerRef.APIVersion
+	if s := strings.Split(version, "/"); len(s) > 1 {
+		group = s[0]
+		version = s[1]
+	}
+	resource := strings.ToLower(ownerRef.Kind) + "s/finalizers"
+	ssar := &authorizationv1.SelfSubjectAccessReview{
+		Spec: authorizationv1.SelfSubjectAccessReviewSpec{
+			ResourceAttributes: &authorizationv1.ResourceAttributes{
+				Group:    group,
+				Version:  version,
+				Resource: resource,
+				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 +2865,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/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",
+			},
+		},
 	}
 }