fix(dataset): enforce inherited reference grants

docs/dataset-repository-share-access-implementation.md

@@ -131,6 +131,25 @@ For C's final namespace:
 
 Do not treat an upstream legacy `status.readOnly=false` as authorization; resolve the current chain again.
 
+Each ancestor must authorize every distinct namespace below it in the reference chain, including the final namespace. For `target/C -> b/B -> a/A`, check B's grant to `target`, A's grant to `target`, and A's grant to `b`. For longer chains, apply the same rule to each intermediate reference's effective permission. Repeated namespaces do not require repeated checks. Validate and parse each source policy once per resolution, then reuse its selectors for these checks.
+
+### 3.3 Upgrade Compatibility
+
+Intermediate-namespace authorization extends the previous final-namespace checks. Existing references can become denied or read-only even when the root source still grants read-write access to the final namespace:
+
+| Reference path | Relevant grants | Result |
+| --- | --- | --- |
+| `target/C -> a/A` | A grants `target` read-write | Read-write; namespace `b` is not on this path |
+| `target/C -> b/B -> a/A` | A and B grant `target` read-write; A denies `b` | Denied |
+| `target/C -> b/B -> a/A` | A and B grant `target` read-write; A grants `b` read-only | Read-only |
+| `target/C -> b/B -> a/A` | All three required grants are read-write | Read-write |
+
+These results assume valid source identities and PVC/PV bindings. A denial sets `MountPolicy=False` and conservatively sets `status.readOnly=true`; this does not permit read-only mounting. Denial messages identify the final Dataset, the ancestor source, and the namespace whose grant failed. B's own cached MountPolicy is not an authorization check for C.
+
+Before upgrading, inspect existing reference chains and their current namespace labels. Ensure every required intermediate grant matches the intended access. If an intermediate namespace should not receive access, create a new direct reference to an authorized source when appropriate; the existing reference URI is immutable. Consumers must call `Verify` before creating Pods and handle a denied or changed grant. This verification does not revoke mounts in already-running Pods.
+
+Namespace label events propagate to all transitive reference dependents because those labels now affect downstream permissions. Deploy this propagation with the intermediate-namespace checks. If those checks are reverted, reassess and normally remove the propagation change too; leaving it in place causes extra reconciliation but does not itself tighten authorization.
+
 ## 4. Dataset Controller
 
 Modify `internal/controller/dataset/dataset_controller.go`.
@@ -183,6 +202,8 @@ Add at least the following tests:
 - When JSON/unstructured representation omits `status.ReadOnly=false`, `Verify` may return writable only with `MountPolicy=True` and matching `observedGeneration`.
 - Read-only wins where overlapping rules match the same namespace.
 - Multi-level references: upstream read-only, upstream not authorizing the final namespace, fully read-write chains, cycles, and excessive depth.
+- Intermediate namespace grants: deny and read-only must propagate despite read-write grants to the final namespace; direct references remain independent of namespaces outside their chain. Cover repeated namespaces and 32-level chains with both shared and distinct namespaces.
+- Namespace label events enqueue references in that namespace and all transitive dependents exactly once, excluding unrelated references and namespaces without references.
 - Effective permission is reconciled again after a Ready PVC when namespace workspace labels or source policy changes.
 - A REFERENCE that enters Failed due to a temporary source/policy error automatically recovers through periodic requeue even without an additional watch event.
 - A PVC-preparation failure (such as a source without a PVC or a PVC not bound to a PV) must set `MountPolicy=False` and must not leave an old condition behind.

internal/controller/dataset/dataset_controller.go

@@ -1315,50 +1315,60 @@ func (r *DatasetReconciler) enqueueReferenceDatasets(ctx context.Context, object
 		requests = append(requests, reconcile.Request{NamespacedName: key})
 	}
 
+	var queue []client.ObjectKey
 	switch changed := object.(type) {
 	case *corev1.Namespace:
-		// Namespace labels affect only references whose final target is this namespace.
+		// Namespace labels affect references in this namespace and every
+		// downstream reference that inherits their grants.
 		for i := range list.Items {
 			ds := &list.Items[i]
 			if ds.Namespace == changed.Name && ds.Spec.Source.Type == datasetv1alpha1.DatasetTypeReference && !kubeutils.IsDeleted(ds) {
-				appendRequest(client.ObjectKeyFromObject(ds))
+				key := client.ObjectKeyFromObject(ds)
+				appendRequest(key)
+				queue = append(queue, key)
 			}
 		}
-		return requests
 	case *datasetv1alpha1.Dataset:
-		// Walk the reverse reference graph so source changes reach direct and
-		// transitive dependents without requeueing unrelated references.
-		reverse := make(map[client.ObjectKey][]client.ObjectKey)
-		for i := range list.Items {
-			ds := &list.Items[i]
-			if ds.Spec.Source.Type != datasetv1alpha1.DatasetTypeReference || kubeutils.IsDeleted(ds) {
-				continue
-			}
-			source, err := mountpolicy.ParseReference(ds.Spec.Source.URI)
-			if err != nil {
-				continue
-			}
-			reverse[source] = append(reverse[source], client.ObjectKeyFromObject(ds))
+		// For() already enqueues the changed Dataset itself; this watch only
+		// enqueues its dependents. Namespace events must enqueue both.
+		queue = append(queue, client.ObjectKeyFromObject(changed))
+	default:
+		return nil
+	}
+
+	if len(queue) == 0 {
+		return requests
+	}
+
+	// Walk the reverse reference graph so Dataset and Namespace changes
+	// reach transitive dependents without requeueing unrelated references.
+	reverse := make(map[client.ObjectKey][]client.ObjectKey)
+	for i := range list.Items {
+		ds := &list.Items[i]
+		if ds.Spec.Source.Type != datasetv1alpha1.DatasetTypeReference || kubeutils.IsDeleted(ds) {
+			continue
+		}
+		source, err := mountpolicy.ParseReference(ds.Spec.Source.URI)
+		if err != nil {
+			continue
 		}
+		reverse[source] = append(reverse[source], client.ObjectKeyFromObject(ds))
+	}
 
-		queue := []client.ObjectKey{client.ObjectKeyFromObject(changed)}
-		visited := make(map[client.ObjectKey]struct{})
-		for len(queue) > 0 {
-			key := queue[0]
-			queue = queue[1:]
-			if _, ok := visited[key]; ok {
-				continue
-			}
-			visited[key] = struct{}{}
-			for _, dependent := range reverse[key] {
-				appendRequest(dependent)
-				queue = append(queue, dependent)
-			}
+	visited := make(map[client.ObjectKey]struct{})
+	for len(queue) > 0 {
+		key := queue[0]
+		queue = queue[1:]
+		if _, ok := visited[key]; ok {
+			continue
+		}
+		visited[key] = struct{}{}
+		for _, dependent := range reverse[key] {
+			appendRequest(dependent)
+			queue = append(queue, dependent)
 		}
-		return requests
-	default:
-		return nil
 	}
+	return requests
 }
 
 func dependencyDatasetChanged(update event.UpdateEvent) bool {

internal/controller/dataset/dataset_controller_test.go

@@ -154,6 +154,47 @@ func TestDatasetReconciler_enqueueReferenceDatasetsScopesDependencies(t *testing
 	require.True(t, dependencyDatasetChanged(event.UpdateEvent{ObjectOld: oldDataset, ObjectNew: newDataset}))
 }
 
+func TestDatasetReconciler_enqueueNamespaceTransitiveReferences(t *testing.T) {
+	scheme := runtime.NewScheme()
+	require.NoError(t, datasetv1alpha1.AddToScheme(scheme))
+	require.NoError(t, corev1.AddToScheme(scheme))
+
+	reference := func(namespace, name, uri string) *datasetv1alpha1.Dataset {
+		return &datasetv1alpha1.Dataset{
+			ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: name},
+			Spec: datasetv1alpha1.DatasetSpec{Source: datasetv1alpha1.DatasetSource{
+				Type: datasetv1alpha1.DatasetTypeReference, URI: uri,
+			}},
+		}
+	}
+	// Both intermediate references are in the changed namespace. Their shared
+	// descendants must be enqueued once, including those multiple hops away.
+	first := reference("middle", "first", "dataset://origin/source")
+	second := reference("middle", "second", "dataset://middle/first")
+	third := reference("target", "third", "dataset://middle/second")
+	leaf := reference("target", "leaf", "dataset://target/third")
+	unrelated := reference("other", "unrelated", "dataset://origin/source")
+	source := &datasetv1alpha1.Dataset{ObjectMeta: metav1.ObjectMeta{Namespace: "origin", Name: "source"}}
+	reconciler := &DatasetReconciler{Client: fake.NewClientBuilder().WithScheme(scheme).
+		WithObjects(source, first, second, third, leaf, unrelated).Build(), Scheme: scheme}
+
+	requests := reconciler.enqueueReferenceDatasets(context.Background(), &corev1.Namespace{
+		ObjectMeta: metav1.ObjectMeta{Name: "middle", Labels: map[string]string{"workspace": "changed"}},
+	})
+	require.ElementsMatch(t, []client.ObjectKey{
+		{Namespace: "middle", Name: "first"},
+		{Namespace: "middle", Name: "second"},
+		{Namespace: "target", Name: "third"},
+		{Namespace: "target", Name: "leaf"},
+	}, requestKeys(requests))
+
+	// A source namespace's labels do not affect grants to other namespaces.
+	requests = reconciler.enqueueReferenceDatasets(context.Background(), &corev1.Namespace{
+		ObjectMeta: metav1.ObjectMeta{Name: "origin"},
+	})
+	require.Empty(t, requests)
+}
+
 func requestKeys(requests []reconcile.Request) []client.ObjectKey {
 	keys := make([]client.ObjectKey, 0, len(requests))
 	for _, request := range requests {

internal/controller/dataset/mount_policy_additional_test.go

@@ -31,7 +31,10 @@ func TestReconcileMountPolicyStoresFullSourceChain(t *testing.T) {
 	c.Spec.Share = false
 	objects := append(referenceTestStorage(a, nil), referenceTestStorage(b, a)...)
 	objects = append(objects, referenceTestStorage(c, b)...)
-	objects = append(objects, &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "target", Labels: map[string]string{"workspace": "one"}}})
+	objects = append(objects,
+		&corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "target", Labels: map[string]string{"workspace": "one"}}},
+		&corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "b", Labels: map[string]string{"workspace": "one"}}},
+	)
 	reconciler := &DatasetReconciler{Client: fake.NewClientBuilder().WithScheme(scheme).WithObjects(objects...).Build(), Scheme: scheme}
 
 	require.NoError(t, reconciler.reconcileMountPolicy(ctx, c))
@@ -92,3 +95,60 @@ func referenceTestStorage(ds, parent *datasetv1alpha1.Dataset) []client.Object {
 	}
 	return []client.Object{ds, pvc, pv}
 }
+
+func TestReconcileMountPolicyInheritsIntermediateGrants(t *testing.T) {
+	for _, mode := range []mountpolicy.GrantResult{mountpolicy.ReadOnly, mountpolicy.ReadWrite, mountpolicy.Denied} {
+		t.Run(string(mode), func(t *testing.T) {
+			ctx := context.Background()
+			scheme := runtime.NewScheme()
+			require.NoError(t, datasetv1alpha1.AddToScheme(scheme))
+			require.NoError(t, corev1.AddToScheme(scheme))
+
+			a := referenceTestSource("a", "a", "a-uid", "a-pvc", "a-pvc-uid", "a-pv", "a-pv-uid", nil, nil)
+			a.Spec.Share = true
+			// A grants the final namespace RW but can restrict B's namespace.
+			if mode != mountpolicy.Denied {
+				access := datasetv1alpha1.AccessModeReadWrite
+				if mode == mountpolicy.ReadOnly {
+					access = datasetv1alpha1.AccessModeReadOnly
+				}
+				a.Spec.ShareAccess.Rules = append(a.Spec.ShareAccess.Rules, datasetv1alpha1.ShareAccessRule{
+					NamespaceSelector: metav1.LabelSelector{MatchLabels: map[string]string{"workspace": "two"}},
+					AccessMode:        access,
+				})
+			}
+			b := referenceTestSource("b", "b", "b-uid", "b-pvc", "b-pvc-uid", "b-pv", "b-pv-uid", a,
+				&datasetv1alpha1.DatasetSource{Type: datasetv1alpha1.DatasetTypeReference, URI: "dataset://a/a"})
+			c := referenceTestSource("c", "target", "c-uid", "c-pvc", "c-pvc-uid", "c-pv", "c-pv-uid", b,
+				&datasetv1alpha1.DatasetSource{Type: datasetv1alpha1.DatasetTypeReference, URI: "dataset://b/b"})
+			objects := append(referenceTestStorage(a, nil), referenceTestStorage(b, a)...)
+			objects = append(objects, referenceTestStorage(c, b)...)
+			objects = append(objects,
+				&corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "target", Labels: map[string]string{"workspace": "one"}}},
+				&corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "b", Labels: map[string]string{"workspace": "two"}}},
+			)
+			reconciler := &DatasetReconciler{Client: fake.NewClientBuilder().WithScheme(scheme).WithObjects(objects...).Build(), Scheme: scheme}
+			err := reconciler.reconcileMountPolicy(ctx, c)
+			reconciler.setMountPolicyCondition(c, err)
+			if mode == mountpolicy.Denied {
+				require.ErrorContains(t, err, "not shared to namespace b")
+				require.ErrorContains(t, err, "reference dataset target/c")
+				require.ErrorContains(t, err, "source dataset a/a")
+				require.True(t, c.Status.ReadOnly)
+				_, err = mountpolicy.Verify(ctx, reconciler.Client, c)
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+			require.Equal(t, mode == mountpolicy.ReadOnly, c.Status.ReadOnly)
+			verified, err := mountpolicy.Verify(ctx, reconciler.Client, c)
+			require.NoError(t, err)
+			require.Equal(t, c.Status.ReadOnly, verified.ReadOnly)
+			if mode == mountpolicy.ReadOnly {
+				c.Status.ReadOnly = false // stale status from the v0.1.14 controller
+				_, err = mountpolicy.Verify(ctx, reconciler.Client, c)
+				require.ErrorContains(t, err, "mount access mode no longer matches source policy")
+			}
+		})
+	}
+}