Authorize default SA on DataImportCron PVC source

In case of DataImportCron with PVC source, the controller checks the
namespace default ServiceAccount is authorized to clone the source PVC.

This is a minimally partial backport of #3970.

Signed-off-by: Arnon Gilboa <[email protected]>

Author: arnongilboa

pkg/controller/BUILD.bazel

@@ -40,6 +40,7 @@ go_library(
         "//vendor/github.com/pkg/errors:go_default_library",
         "//vendor/github.com/prometheus/client_golang/prometheus:go_default_library",
         "//vendor/github.com/robfig/cron/v3: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/dataimportcron-conditions.go

@@ -30,12 +30,13 @@ import (
 )
 
 const (
-	noDigest   = "NoDigest"
-	noImport   = "NoImport"
-	outdated   = "Outdated"
-	scheduled  = "ImportScheduled"
-	inProgress = "ImportProgressing"
-	upToDate   = "UpToDate"
+	noDigest      = "NoDigest"
+	noImport      = "NoImport"
+	notAuthorized = "NotAuthorized"
+	outdated      = "Outdated"
+	scheduled     = "ImportScheduled"
+	inProgress    = "ImportProgressing"
+	upToDate      = "UpToDate"
 )
 
 func updateDataImportCronCondition(cron *cdiv1.DataImportCron, conditionType cdiv1.DataImportCronConditionType, status corev1.ConditionStatus, message, reason string) {

pkg/controller/dataimportcron-controller.go

@@ -33,6 +33,7 @@ import (
 	"github.com/pkg/errors"
 	cronexpr "github.com/robfig/cron/v3"
 
+	authorizationv1 "k8s.io/api/authorization/v1"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 	storagev1 "k8s.io/api/storage/v1"
@@ -889,15 +890,59 @@ func (r *DataImportCronReconciler) createImportDataVolume(ctx context.Context, d
 			return nil
 		}
 	}
-
 	dv := r.newSourceDataVolume(dataImportCron, dvName)
+	if allowed, err := r.authorizeCloneDataVolume(dataImportCron, dv); err != nil {
+		return err
+	} else if !allowed {
+		updateDataImportCronCondition(dataImportCron, cdiv1.DataImportCronProgressing, corev1.ConditionFalse,
+			"Not authorized to create DataVolume", notAuthorized)
+		return nil
+	}
 	if err := r.client.Create(ctx, dv); err != nil && !k8serrors.IsAlreadyExists(err) {
 		return err
 	}
 
 	return nil
 }
 
+func (r *DataImportCronReconciler) authorizeCloneDataVolume(dataImportCron *cdiv1.DataImportCron, dv *cdiv1.DataVolume) (bool, error) {
+	if !isPvcSource(dataImportCron) {
+		return true, nil
+	}
+
+	if resp, err := dv.AuthorizeSA(dv.Namespace, dv.Name, r, dataImportCron.Namespace, "default"); err != nil {
+		return false, err
+	} else if !resp.Allowed {
+		r.log.Info("Not authorized to create DataVolume", "cron", dataImportCron.Name, "reason", resp.Reason)
+		return false, nil
+	}
+
+	return true, nil
+}
+
+func (r *DataImportCronReconciler) CreateSar(sar *authorizationv1.SubjectAccessReview) (*authorizationv1.SubjectAccessReview, error) {
+	if err := r.client.Create(context.TODO(), sar); err != nil {
+		return nil, err
+	}
+	return sar, nil
+}
+
+func (r *DataImportCronReconciler) GetNamespace(name string) (*corev1.Namespace, error) {
+	ns := &corev1.Namespace{}
+	if err := r.client.Get(context.TODO(), types.NamespacedName{Name: name}, ns); err != nil {
+		return nil, err
+	}
+	return ns, nil
+}
+
+func (r *DataImportCronReconciler) GetDataSource(namespace, name string) (*cdiv1.DataSource, error) {
+	das := &cdiv1.DataSource{}
+	if err := r.client.Get(context.TODO(), types.NamespacedName{Namespace: namespace, Name: name}, das); err != nil {
+		return nil, err
+	}
+	return das, nil
+}
+
 func (r *DataImportCronReconciler) handleStorageClassChange(ctx context.Context, dataImportCron *cdiv1.DataImportCron, desiredStorageClass string) error {
 	digest, ok := dataImportCron.Annotations[AnnSourceDesiredDigest]
 	if !ok {

pkg/controller/dataimportcron-controller_test.go

@@ -920,7 +920,8 @@ var _ = Describe("All DataImportCron Tests", func() {
 					Namespace: pvc.Namespace,
 				},
 			}
-			reconciler = createDataImportCronReconciler(cron, pvc)
+			reconciler = createDataImportCronReconciler(cron, pvc,
+				&corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "default"}})
 			_, err := reconciler.Reconcile(context.TODO(), cronReq)
 			Expect(err).ToNot(HaveOccurred())
 		})

pkg/operator/resources/cluster/controller.go

@@ -114,6 +114,19 @@ func getControllerClusterPolicyRules() []rbacv1.PolicyRule {
 				"delete",
 			},
 		},
+		{
+			APIGroups: []string{
+				"",
+			},
+			Resources: []string{
+				"namespaces",
+			},
+			Verbs: []string{
+				"get",
+				"list",
+				"watch",
+			},
+		},
 		{
 			APIGroups: []string{
 				"",
@@ -268,6 +281,17 @@ func getControllerClusterPolicyRules() []rbacv1.PolicyRule {
 				"update",
 			},
 		},
+		{
+			APIGroups: []string{
+				"authorization.k8s.io",
+			},
+			Resources: []string{
+				"subjectaccessreviews",
+			},
+			Verbs: []string{
+				"create",
+			},
+		},
 	}
 }
 

tests/dataimportcron_test.go

@@ -822,6 +822,46 @@ var _ = Describe("DataImportCron", Serial, func() {
 			Expect(dataSource.Spec.Source).To(Equal(expectedSource))
 		})
 	})
+
+	Context("DataImportCron controller authorization when cloning from PVC source", func() {
+		var sourceDv *cdiv1.DataVolume
+
+		createSourceDv := func(dvNamespace string) {
+			if dvNamespace != ns {
+				sourceNamespace, err := f.CreateNamespace(dvNamespace, nil)
+				Expect(err).ToNot(HaveOccurred())
+				f.AddNamespaceToDelete(sourceNamespace)
+				dvNamespace = sourceNamespace.Name
+			}
+
+			sourceDv = utils.NewDataVolumeWithHTTPImport("source-pvc", "1Gi", fmt.Sprintf(utils.TinyCoreIsoURL, f.CdiInstallNs))
+			sourceDv, err = utils.CreateDataVolumeFromDefinition(f.CdiClient, dvNamespace, sourceDv)
+			Expect(err).ToNot(HaveOccurred())
+			f.ForceBindPvcIfDvIsWaitForFirstConsumer(sourceDv)
+		}
+
+		It("[test_id:12407] should create DataVolume when default ServiceAccount created the DataImportCron with source PVC in the same namespace", func() {
+			createSourceDv(ns)
+			cron := createDataImportCronWithPVCSource(cronName, sourceDv.Namespace, sourceDv.Name)
+			_, err := f.CdiClient.CdiV1beta1().DataImportCrons(ns).Create(context.TODO(), cron, metav1.CreateOptions{})
+			Expect(err).ToNot(HaveOccurred())
+			waitForConditions(corev1.ConditionFalse, corev1.ConditionTrue)
+		})
+
+		It("[test_id:XXXX] should not create DataVolume when default ServiceAccount created the DataImportCron with source PVC in another namespace", func() {
+			createSourceDv("source-ns")
+			cron := createDataImportCronWithPVCSource(cronName, sourceDv.Namespace, sourceDv.Name)
+			_, err := f.CdiClient.CdiV1beta1().DataImportCrons(ns).Create(context.TODO(), cron, metav1.CreateOptions{})
+			Expect(err).ToNot(HaveOccurred())
+			waitForConditions(corev1.ConditionFalse, corev1.ConditionFalse)
+			cron, err = f.CdiClient.CdiV1beta1().DataImportCrons(ns).Get(context.TODO(), cronName, metav1.GetOptions{})
+			Expect(err).ToNot(HaveOccurred())
+			cronCond := controller.FindDataImportCronConditionByType(cron, cdiv1.DataImportCronProgressing)
+			Expect(cronCond).ToNot(BeNil())
+			Expect(cronCond.ConditionState.Message).To(Equal("Not authorized to create DataVolume"))
+			Expect(cronCond.ConditionState.Reason).To(Equal("NotAuthorized"))
+		})
+	})
 })
 
 func getDataVolumeSourceRegistry(f *framework.Framework) (*cdiv1.DataVolumeSourceRegistry, error) {
@@ -875,6 +915,29 @@ func updateDataSource(clientSet *cdiclientset.Clientset, namespace string, dataS
 	}
 }
 
+func createDataImportCronWithPVCSource(name string, pvcNamespace, pvcName string) *cdiv1.DataImportCron {
+	return &cdiv1.DataImportCron{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+		Spec: cdiv1.DataImportCronSpec{
+			Template: cdiv1.DataVolume{
+				Spec: cdiv1.DataVolumeSpec{
+					Source: &cdiv1.DataVolumeSource{
+						PVC: &cdiv1.DataVolumeSourcePVC{
+							Namespace: pvcNamespace,
+							Name:      pvcName,
+						},
+					},
+					Storage: &cdiv1.StorageSpec{},
+				},
+			},
+			Schedule:          scheduleOnceAYear,
+			ManagedDataSource: "datasource-test",
+		},
+	}
+}
+
 func retryOnceOnErr(f func() error) Assertion {
 	err := f()
 	if err != nil {