fix: add DataSource reference handling to authorize utils (#3831)

* fix: add DataSource reference to authorize_utils

A previous PR[1] introduced the concept of DataSource references as a way
for a DataSource to reference another DataSource.

A DV that has its sourceRef set to a DataSource reference that resides
in a different namespace would fail as authorize utils did not include
the necessary changes to the clone source handler to accomodate
DataSource references.

This commit fixes that and introduces a functional test for it.

[1] #3760

Signed-off-by: Adi Aloni <[email protected]>

* datasource-controller: prohibit cross-namespace references

Previously a DataSource could reference a DataSource in a namespace
different from its own. This is unwanted behavior and with this commit
we now error when we create such DataSources.

Signed-off-by: Adi Aloni <[email protected]>

---------

Signed-off-by: Adi Aloni <[email protected]>

Author: Acedus

pkg/apiserver/webhooks/datavolume-mutate_test.go

@@ -358,6 +358,62 @@ var _ = Describe("Mutating DataVolume Webhook", func() {
 			Entry("succeed with same (default) namespace", "default"),
 			Entry("succeed with empty namespace", ""),
 		)
+
+		It("should allow update to DataVolume with sourceRef DataSource reference with clone token", func() {
+			dsRef := &cdicorev1.DataSource{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "dsRef",
+					Namespace: "testNamespace",
+				},
+				Spec: cdicorev1.DataSourceSpec{
+					Source: cdicorev1.DataSourceSource{
+						DataSource: &cdicorev1.DataSourceRefSourceDataSource{
+							Namespace: "testNamespace",
+							Name:      "ds",
+						},
+					},
+				},
+				Status: cdicorev1.DataSourceStatus{
+					Source: cdicorev1.DataSourceSource{
+						PVC: &cdicorev1.DataVolumeSourcePVC{
+							Namespace: "testNamespace",
+							Name:      "pvc",
+						},
+					},
+				},
+			}
+			sourceRef := cdicorev1.DataVolumeSourceRef{
+				Kind:      cdicorev1.DataVolumeDataSource,
+				Namespace: &dsRef.Namespace,
+				Name:      dsRef.Name,
+			}
+			dv := newDataVolumeWithSourceRef("dsRefDv", nil, &sourceRef, nil)
+			dvBytes, _ := json.Marshal(&dv)
+			ar := &admissionv1.AdmissionReview{
+				Request: &admissionv1.AdmissionRequest{
+					Operation: admissionv1.Create,
+					Resource: metav1.GroupVersionResource{
+						Group:    cdicorev1.SchemeGroupVersion.Group,
+						Version:  cdicorev1.SchemeGroupVersion.Version,
+						Resource: "datavolumes",
+					},
+					Object: runtime.RawExtension{
+						Raw: dvBytes,
+					},
+				},
+			}
+
+			resp := mutateDVs(key, ar, true, dsRef)
+			Expect(resp.Allowed).To(BeTrue())
+			Expect(resp.Patch).ToNot(BeNil())
+			var patchObjs []jsonpatch.Operation
+			err := json.Unmarshal(resp.Patch, &patchObjs)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(patchObjs).Should(HaveLen(1))
+			Expect(patchObjs[0].Operation).Should(Equal("add"))
+			Expect(patchObjs[0].Path).Should(Equal("/metadata/annotations"))
+			Expect(patchObjs[0].Value).Should(HaveKey(cc.AnnCloneToken))
+		})
 	})
 })
 

pkg/controller/common/util.go

@@ -396,6 +396,7 @@ var (
 
 	ErrDataSourceMaxDepthReached = errors.New("DataSource reference chain exceeds maximum depth of 1")
 	ErrDataSourceSelfReference   = errors.New("DataSource cannot self-reference")
+	ErrDataSourceCrossNamespace  = errors.New("DataSource cannot reference a DataSource in another namespace")
 )
 
 // FakeValidator is a fake token validator
@@ -2090,9 +2091,13 @@ func ResolveDataSourceChain(ctx context.Context, client client.Client, dataSourc
 
 	ref := dataSource.Spec.Source.DataSource
 	refNs := GetNamespace(ref.Namespace, dataSource.Namespace)
+	if dataSource.Namespace != refNs {
+		return dataSource, ErrDataSourceCrossNamespace
+	}
 	if ref.Name == dataSource.Name && refNs == dataSource.Namespace {
 		return nil, ErrDataSourceSelfReference
 	}
+
 	resolved := &cdiv1.DataSource{}
 	if err := client.Get(ctx, types.NamespacedName{Name: ref.Name, Namespace: refNs}, resolved); err != nil {
 		return nil, err

pkg/controller/datasource-controller.go

@@ -60,6 +60,7 @@ const (
 	dataSourceControllerName = "datasource-controller"
 	maxReferenceDepthReached = "MaxReferenceDepthReached"
 	selfReference            = "SelfReference"
+	crossNamespaceReference  = "CrossNamespaceReference"
 )
 
 // Reconcile loop for DataSourceReconciler
@@ -181,6 +182,8 @@ func handleDataSourceRefError(dataSource *cdiv1.DataSource, err error) error {
 		reason = maxReferenceDepthReached
 	case errors.Is(err, cc.ErrDataSourceSelfReference):
 		reason = selfReference
+	case errors.Is(err, cc.ErrDataSourceCrossNamespace):
+		reason = crossNamespaceReference
 	case k8serrors.IsNotFound(err):
 		reason = cc.NotFound
 	default:

pkg/controller/datasource-controller_test.go

@@ -243,6 +243,16 @@ var _ = Describe("All DataSource Tests", func() {
 				Expect(dsPointer.Spec.Source.DataSource.Name).To(Equal(dsPointer.Name))
 				Expect(dsPointer.Status.Source.DataSource).To(BeNil())
 			})
+			It("DataSource pointer should fail to resolve cross-namespace DataSource", func() {
+				dsPointer := createDataSource(dsName + "-pointer")
+				dsPointer.Spec.Source = cdiv1.DataSourceSource{DataSource: &cdiv1.DataSourceRefSourceDataSource{Namespace: "differentNamespace", Name: dsPointer.Name}}
+				reconciler := createDataSourceReconciler(dsPointer)
+				dsPointerKey := types.NamespacedName{Name: dsPointer.Name, Namespace: metav1.NamespaceDefault}
+				verifyConditions("DataSource cross-namespace reference", false, crossNamespaceReference, dsPointer, reconciler)
+				Expect(reconciler.client.Get(context.TODO(), dsPointerKey, dsPointer)).To(Succeed())
+				Expect(dsPointer.Spec.Source.DataSource.Name).To(Equal(dsPointer.Name))
+				Expect(dsPointer.Status.Source.DataSource).To(BeNil())
+			})
 		})
 
 	})

staging/src/kubevirt.io/containerized-data-importer-api/pkg/apis/core/v1beta1/authorize_utils.go

@@ -45,11 +45,12 @@ func newCloneSourceHandler(dataVolume *DataVolume, dsGet dsGetFunc) (CloneSource
 		if err != nil {
 			return CloneSourceHandler{}, err
 		}
-		if dataSource.Spec.Source.PVC != nil {
-			pvcSource = dataSource.Spec.Source.PVC
-		} else if dataSource.Spec.Source.Snapshot != nil {
-			snapshotSource = dataSource.Spec.Source.Snapshot
-		}
+		pvcSource = dataSource.Spec.Source.PVC
+		snapshotSource = dataSource.Spec.Source.Snapshot
+		if dataSource.Spec.Source.DataSource != nil {
+			pvcSource = dataSource.Status.Source.PVC
+			snapshotSource = dataSource.Status.Source.Snapshot
+		} 
 	}
 
 	switch {