[v2.9] Refactor ID based partitioning, add unit tests (#310)

* Refactor ID based partitioning, add unit tests

This resolves an issue where the requested namespace filter was not
always honored.

* Correct naming issues to appease the linter

pkg/stores/proxy/rbac_store.go

@@ -9,6 +9,7 @@ import (
 	"github.com/rancher/steve/pkg/attributes"
 	"github.com/rancher/steve/pkg/stores/partition"
 	"github.com/rancher/wrangler/v3/pkg/kv"
+	"github.com/sirupsen/logrus"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/watch"
@@ -64,17 +65,10 @@ func (p *rbacPartitioner) All(apiOp *types.APIRequest, schema *types.APISchema,
 		fallthrough
 	case "watch":
 		if id != "" {
-			ns, name := kv.RSplit(id, "/")
+			partitions := generatePartitionsByID(apiOp, schema, verb, id)
-			return []partition.Partition{
+			return partitions, nil
-				Partition{
-					Namespace:   ns,
-					All:         false,
-					Passthrough: false,
-					Names:       sets.NewString(name),
-				},
-			}, nil
 		}
-		partitions, passthrough := isPassthrough(apiOp, schema, verb)
+		partitions, passthrough := generateAggregatePartitions(apiOp, schema, verb)
 		if passthrough {
 			return passthroughPartitions, nil
 		}
@@ -126,15 +120,92 @@ func (b *byNameOrNamespaceStore) Watch(apiOp *types.APIRequest, schema *types.AP
 	return b.Store.WatchNames(apiOp, schema, wr, b.partition.Names)
 }
 
-// isPassthrough determines whether a request can be passed through directly to the underlying store
+// generatePartitionsById determines whether a requester can access a particular resource
-// or if the results need to be partitioned by namespace and name based on the requester's access.
+// and if so, returns the corresponding partitions
-func isPassthrough(apiOp *types.APIRequest, schema *types.APISchema, verb string) ([]partition.Partition, bool) {
+func generatePartitionsByID(apiOp *types.APIRequest, schema *types.APISchema, verb string, id string) []partition.Partition {
 	accessListByVerb, _ := attributes.Access(schema).(accesscontrol.AccessListByVerb)
+	resources := accessListByVerb.Granted(verb)
+
+	idNamespace, name := kv.RSplit(id, "/")
+	apiNamespace := apiOp.Namespace
+	effectiveNamespace := idNamespace
+
+	// If a non-empty namespace was provided, be sure to select that for filtering and permissions checks
+	if idNamespace == "" && apiNamespace != "" {
+		effectiveNamespace = apiNamespace
+	}
+
+	// The external API is flexible, and permits specifying a namespace as a separate key or embedded
+	// within the ID of the object. Both of these cases should be valid:
+	//   {"namespace": "n1", "id": "r1"}
+	//   {"id": "n1/r1"}
+	// however, the following conflicting request is not valid, but was previously accepted:
+	//   {"namespace": "n1", "id": "n2/r1"}
+	// To avoid breaking UI plugins that may inadvertently rely on the feature, we issue a deprecation
+	// warning for now. We still need to pick one of the namespaces for permission verification purposes.
+	if idNamespace != "" && apiNamespace != "" && idNamespace != apiNamespace {
+		logrus.Warningf("DEPRECATION: Conflicting namespaces '%v' and '%v' requested. "+
+			"Selecting '%v' as the effective namespace. Future steve versions will reject this request.",
+			idNamespace, apiNamespace, effectiveNamespace)
+	}
+
+	if accessListByVerb.All(verb) {
+		return []partition.Partition{
+			Partition{
+				Namespace:   effectiveNamespace,
+				All:         false,
+				Passthrough: false,
+				Names:       sets.NewString(name),
+			},
+		}
+	}
+
+	if effectiveNamespace != "" {
+		if resources[effectiveNamespace].All {
+			return []partition.Partition{
+				Partition{
+					Namespace:   effectiveNamespace,
+					All:         false,
+					Passthrough: false,
+					Names:       sets.NewString(name),
+				},
+			}
+		}
+	}
+
+	// For cluster-scoped resources, we will have parsed a "" out
+	// of the ID field from RSplit, but accessListByVerb specifies "*" for
+	// the namespace, so correct that here
+	resourceNamespace := effectiveNamespace
+	if resourceNamespace == "" {
+		resourceNamespace = accesscontrol.All
+	}
+
+	nameset, ok := resources[resourceNamespace]
+	if ok && nameset.Names.Has(name) {
+		return []partition.Partition{
+			Partition{
+				Namespace:   effectiveNamespace,
+				All:         false,
+				Passthrough: false,
+				Names:       sets.NewString(name),
+			},
+		}
+	}
+
+	return nil
+}
+
+// generateAggregatePartitions determines whether a request can be passed through directly to the underlying store
+// or if the results need to be partitioned by namespace and name based on the requester's access.
+func generateAggregatePartitions(apiOp *types.APIRequest, schema *types.APISchema, verb string) ([]partition.Partition, bool) {
+	accessListByVerb, _ := attributes.Access(schema).(accesscontrol.AccessListByVerb)
+	resources := accessListByVerb.Granted(verb)
+
 	if accessListByVerb.All(verb) {
 		return nil, true
 	}
 
-	resources := accessListByVerb.Granted(verb)
 	if apiOp.Namespace != "" {
 		if resources[apiOp.Namespace].All {
 			return nil, true

pkg/stores/proxy/rbac_store_test.go

@@ -11,7 +11,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 )
 
-func TestAll(t *testing.T) {
+func TestVerbList(t *testing.T) {
 	tests := []struct {
 		name           string
 		apiOp          *types.APIRequest
@@ -223,7 +223,7 @@ func TestAll(t *testing.T) {
 			},
 		},
 		{
-			name:  "by id",
+			name:  "by id fully unauthorized",
 			apiOp: &types.APIRequest{},
 			id:    "n1/r1",
 			schema: &types.APISchema{
@@ -231,6 +231,72 @@ func TestAll(t *testing.T) {
 					ID: "foo",
 				},
 			},
+			wantPartitions: nil,
+		},
+		{
+			name:  "by id missing namespace",
+			apiOp: &types.APIRequest{},
+			id:    "n1/r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"list": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n2",
+									ResourceName: "r2",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: nil,
+		},
+		{
+			name:  "by id missing resource",
+			apiOp: &types.APIRequest{},
+			id:    "n1/r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"list": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "r2",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: nil,
+		},
+		{
+			name:  "by id authorized by name",
+			apiOp: &types.APIRequest{},
+			id:    "n1/r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"list": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "r1",
+								},
+							},
+						},
+					},
+				},
+			},
 			wantPartitions: []partition.Partition{
 				Partition{
 					Namespace: "n1",
@@ -238,8 +304,227 @@ func TestAll(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "by id authorized by namespace",
+			apiOp: &types.APIRequest{
+				Namespace: "n1",
+			},
+			id: "n1/r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"list": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "*",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				Partition{
+					Namespace: "n1",
+					Names:     sets.NewString("r1"),
+				},
+			},
+		},
+		{
+			name: "by namespaced id authorized by name",
+			apiOp: &types.APIRequest{
+				Namespace: "n1",
+			},
+			id: "r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"list": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "r1",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				Partition{
+					Namespace: "n1",
+					Names:     sets.NewString("r1"),
+				},
+			},
+		},
+		{
+			name:  "by id ignores unrequested resources",
+			apiOp: &types.APIRequest{},
+			id:    "n1/r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"list": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "r1",
+								},
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "r2",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				Partition{
+					Namespace: "n1",
+					Names:     sets.NewString("r1"),
+				},
+			},
+		},
+		// Note: this is deprecated fallback behavior. When we remove the behavior,
+		// rewrite this test to expect an error instead.
+		{
+			name: "by id prefers id embedded namespace",
+			apiOp: &types.APIRequest{
+				Namespace: "n2",
+			},
+			id: "n1/r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"list": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "r1",
+								},
+								accesscontrol.Access{
+									Namespace:    "n2",
+									ResourceName: "r2",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				Partition{
+					Namespace: "n1",
+					Names:     sets.NewString("r1"),
+				},
+			},
+		},
+		{
+			name:  "cluster scoped id unauthorized",
+			apiOp: &types.APIRequest{},
+			id:    "c1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+				},
+			},
+			wantPartitions: nil,
+		},
+		{
+			name:  "cluster scoped id authorized by name",
+			apiOp: &types.APIRequest{},
+			id:    "c1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": false,
+						"access": accesscontrol.AccessListByVerb{
+							"list": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "*",
+									ResourceName: "c1",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				Partition{
+					Namespace: "",
+					Names:     sets.NewString("c1"),
+				},
+			},
+		},
+		{
+			name:  "cluster scoped id authorized globally",
+			apiOp: &types.APIRequest{},
+			id:    "c1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": false,
+						"access": accesscontrol.AccessListByVerb{
+							"list": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "*",
+									ResourceName: "*",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				Partition{
+					Namespace: "",
+					Names:     sets.NewString("c1"),
+				},
+			},
+		},
+		{
+			name:  "cluster scoped id ignores unrequested resources",
+			apiOp: &types.APIRequest{},
+			id:    "c1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": false,
+						"access": accesscontrol.AccessListByVerb{
+							"list": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "*",
+									ResourceName: "c1",
+								},
+								accesscontrol.Access{
+									Namespace:    "*",
+									ResourceName: "c2",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				Partition{
+					Namespace: "",
+					Names:     sets.NewString("c1"),
+				},
+			},
+		},
 	}
-
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			partitioner := rbacPartitioner{}
@@ -250,3 +535,325 @@ func TestAll(t *testing.T) {
 		})
 	}
 }
+
+func TestVerbWatch(t *testing.T) {
+	tests := []struct {
+		name           string
+		apiOp          *types.APIRequest
+		id             string
+		schema         *types.APISchema
+		wantPartitions []partition.Partition
+	}{
+		{
+			name:  "by id fully unauthorized",
+			apiOp: &types.APIRequest{},
+			id:    "n1/r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+				},
+			},
+			wantPartitions: nil,
+		},
+		{
+			name:  "by id missing namespace",
+			apiOp: &types.APIRequest{},
+			id:    "n1/r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"watch": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n2",
+									ResourceName: "r2",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: nil,
+		},
+		{
+			name:  "by id missing resource",
+			apiOp: &types.APIRequest{},
+			id:    "n1/r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"watch": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "r2",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: nil,
+		},
+		{
+			name:  "by id authorized by name",
+			apiOp: &types.APIRequest{},
+			id:    "n1/r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"watch": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "r1",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				Partition{
+					Namespace: "n1",
+					Names:     sets.NewString("r1"),
+				},
+			},
+		},
+		{
+			name: "by namespaced id authorized by name",
+			apiOp: &types.APIRequest{
+				Namespace: "n1",
+			},
+			id: "r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"watch": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "r1",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				Partition{
+					Namespace: "n1",
+					Names:     sets.NewString("r1"),
+				},
+			},
+		},
+		{
+			name: "by id authorized by namespace",
+			apiOp: &types.APIRequest{
+				Namespace: "n1",
+			},
+			id: "n1/r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"watch": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "*",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				Partition{
+					Namespace: "n1",
+					Names:     sets.NewString("r1"),
+				},
+			},
+		},
+		{
+			name:  "by id ignores unrequested resources",
+			apiOp: &types.APIRequest{},
+			id:    "n1/r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"watch": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "r1",
+								},
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "r2",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				Partition{
+					Namespace: "n1",
+					Names:     sets.NewString("r1"),
+				},
+			},
+		},
+		// Note: this is deprecated fallback behavior. When we remove the behavior,
+		// rewrite this test to expect an error instead.
+		{
+			name: "by id prefers id embedded namespace",
+			apiOp: &types.APIRequest{
+				Namespace: "n2",
+			},
+			id: "n1/r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"watch": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "r1",
+								},
+								accesscontrol.Access{
+									Namespace:    "n2",
+									ResourceName: "r2",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				Partition{
+					Namespace: "n1",
+					Names:     sets.NewString("r1"),
+				},
+			},
+		},
+		{
+			name:  "cluster scoped id unauthorized",
+			apiOp: &types.APIRequest{},
+			id:    "c1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+				},
+			},
+			wantPartitions: nil,
+		},
+		{
+			name:  "cluster scoped id authorized by name",
+			apiOp: &types.APIRequest{},
+			id:    "c1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": false,
+						"access": accesscontrol.AccessListByVerb{
+							"watch": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "*",
+									ResourceName: "c1",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				Partition{
+					Namespace: "",
+					Names:     sets.NewString("c1"),
+				},
+			},
+		},
+		{
+			name:  "cluster scoped id authorized globally",
+			apiOp: &types.APIRequest{},
+			id:    "c1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": false,
+						"access": accesscontrol.AccessListByVerb{
+							"watch": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "*",
+									ResourceName: "*",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				Partition{
+					Namespace: "",
+					Names:     sets.NewString("c1"),
+				},
+			},
+		},
+		{
+			name:  "cluster scoped id ignores unrequested resources",
+			apiOp: &types.APIRequest{},
+			id:    "c1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": false,
+						"access": accesscontrol.AccessListByVerb{
+							"watch": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "*",
+									ResourceName: "c1",
+								},
+								accesscontrol.Access{
+									Namespace:    "*",
+									ResourceName: "c2",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				Partition{
+					Namespace: "",
+					Names:     sets.NewString("c1"),
+				},
+			},
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			partitioner := rbacPartitioner{}
+			verb := "watch"
+			gotPartitions, gotErr := partitioner.All(test.apiOp, test.schema, verb, test.id)
+			assert.Nil(t, gotErr)
+			assert.Equal(t, test.wantPartitions, gotPartitions)
+		})
+	}
+}

pkg/stores/sqlpartition/partitioner.go

@@ -9,6 +9,7 @@ import (
 	"github.com/rancher/steve/pkg/accesscontrol"
 	"github.com/rancher/steve/pkg/attributes"
 	"github.com/rancher/wrangler/v3/pkg/kv"
+	"github.com/sirupsen/logrus"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/watch"
@@ -46,17 +47,10 @@ func (p *rbacPartitioner) All(apiOp *types.APIRequest, schema *types.APISchema,
 		fallthrough
 	case "watch":
 		if id != "" {
-			ns, name := kv.RSplit(id, "/")
+			partitions := generatePartitionsByID(apiOp, schema, verb, id)
-			return []partition.Partition{
+			return partitions, nil
-				{
-					Namespace:   ns,
-					All:         false,
-					Passthrough: false,
-					Names:       sets.New[string](name),
-				},
-			}, nil
 		}
-		partitions, passthrough := isPassthrough(apiOp, schema, verb)
+		partitions, passthrough := generateAggregatePartitions(apiOp, schema, verb)
 		if passthrough {
 			return passthroughPartitions, nil
 		}
@@ -74,15 +68,92 @@ func (p *rbacPartitioner) Store() UnstructuredStore {
 	return p.proxyStore
 }
 
-// isPassthrough determines whether a request can be passed through directly to the underlying store
+// generatePartitionsById determines whether a requester can access a particular resource
-// or if the results need to be partitioned by namespace and name based on the requester's access.
+// and if so, returns the corresponding partitions
-func isPassthrough(apiOp *types.APIRequest, schema *types.APISchema, verb string) ([]partition.Partition, bool) {
+func generatePartitionsByID(apiOp *types.APIRequest, schema *types.APISchema, verb string, id string) []partition.Partition {
 	accessListByVerb, _ := attributes.Access(schema).(accesscontrol.AccessListByVerb)
+	resources := accessListByVerb.Granted(verb)
+
+	idNamespace, name := kv.RSplit(id, "/")
+	apiNamespace := apiOp.Namespace
+	effectiveNamespace := idNamespace
+
+	// If a non-empty namespace was provided, be sure to select that for filtering and permissions checks
+	if idNamespace == "" && apiNamespace != "" {
+		effectiveNamespace = apiNamespace
+	}
+
+	// The external API is flexible, and permits specifying a namespace as a separate key or embedded
+	// within the ID of the object. Both of these cases should be valid:
+	//   {"namespace": "n1", "id": "r1"}
+	//   {"id": "n1/r1"}
+	// however, the following conflicting request is not valid, but was previously accepted:
+	//   {"namespace": "n1", "id": "n2/r1"}
+	// To avoid breaking UI plugins that may inadvertently rely on the feature, we issue a deprecation
+	// warning for now. We still need to pick one of the namespaces for permission verification purposes.
+	if idNamespace != "" && apiNamespace != "" && idNamespace != apiNamespace {
+		logrus.Warningf("DEPRECATION: Conflicting namespaces '%v' and '%v' requested. "+
+			"Selecting '%v' as the effective namespace. Future steve versions will reject this request.",
+			idNamespace, apiNamespace, effectiveNamespace)
+	}
+
+	if accessListByVerb.All(verb) {
+		return []partition.Partition{
+			{
+				Namespace:   effectiveNamespace,
+				All:         false,
+				Passthrough: false,
+				Names:       sets.New(name),
+			},
+		}
+	}
+
+	if effectiveNamespace != "" {
+		if resources[effectiveNamespace].All {
+			return []partition.Partition{
+				{
+					Namespace:   effectiveNamespace,
+					All:         false,
+					Passthrough: false,
+					Names:       sets.New(name),
+				},
+			}
+		}
+	}
+
+	// For cluster-scoped resources, we will have parsed a "" out
+	// of the ID field from RSplit, but accessListByVerb specifies "*" for
+	// the nameset, so correct that here
+	resourceNamespace := effectiveNamespace
+	if resourceNamespace == "" {
+		resourceNamespace = accesscontrol.All
+	}
+
+	nameset, ok := resources[resourceNamespace]
+	if ok && nameset.Names.Has(name) {
+		return []partition.Partition{
+			{
+				Namespace:   effectiveNamespace,
+				All:         false,
+				Passthrough: false,
+				Names:       sets.New(name),
+			},
+		}
+	}
+
+	return nil
+}
+
+// generateAggregatePartitions determines whether a request can be passed through directly to the underlying store
+// or if the results need to be partitioned by namespace and name based on the requester's access.
+func generateAggregatePartitions(apiOp *types.APIRequest, schema *types.APISchema, verb string) ([]partition.Partition, bool) {
+	accessListByVerb, _ := attributes.Access(schema).(accesscontrol.AccessListByVerb)
+	resources := accessListByVerb.Granted(verb)
+
 	if accessListByVerb.All(verb) {
 		return nil, true
 	}
 
-	resources := accessListByVerb.Granted(verb)
 	if apiOp.Namespace != "" {
 		if resources[apiOp.Namespace].All {
 			return nil, true

pkg/stores/sqlpartition/partitioner_test.go

@@ -13,7 +13,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 )
 
-func TestAll(t *testing.T) {
+func TestVerbList(t *testing.T) {
 	tests := []struct {
 		name           string
 		apiOp          *types.APIRequest
@@ -225,7 +225,7 @@ func TestAll(t *testing.T) {
 			},
 		},
 		{
-			name:  "by id",
+			name:  "by id fully unauthorized",
 			apiOp: &types.APIRequest{},
 			id:    "n1/r1",
 			schema: &types.APISchema{
@@ -233,6 +233,72 @@ func TestAll(t *testing.T) {
 					ID: "foo",
 				},
 			},
+			wantPartitions: nil,
+		},
+		{
+			name:  "by id missing namespace",
+			apiOp: &types.APIRequest{},
+			id:    "n1/r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"list": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n2",
+									ResourceName: "r2",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: nil,
+		},
+		{
+			name:  "by id missing resource",
+			apiOp: &types.APIRequest{},
+			id:    "n1/r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"list": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "r2",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: nil,
+		},
+		{
+			name:  "by id authorized by name",
+			apiOp: &types.APIRequest{},
+			id:    "n1/r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"list": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "r1",
+								},
+							},
+						},
+					},
+				},
+			},
 			wantPartitions: []partition.Partition{
 				{
 					Namespace: "n1",
@@ -240,6 +306,228 @@ func TestAll(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "by id authorized by namespace",
+			apiOp: &types.APIRequest{
+				Namespace: "n1",
+			},
+			id: "n1/r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"list": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "*",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				{
+					Namespace: "n1",
+					All:       false,
+					Names:     sets.New[string]("r1"),
+				},
+			},
+		},
+		{
+			name: "by namespaced id authorized by name",
+			apiOp: &types.APIRequest{
+				Namespace: "n1",
+			},
+			id: "r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"list": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "r1",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				{
+					Namespace: "n1",
+					All:       false,
+					Names:     sets.New[string]("r1"),
+				},
+			},
+		},
+		{
+			name:  "by id ignores unrequested resources",
+			apiOp: &types.APIRequest{},
+			id:    "n1/r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"list": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "r1",
+								},
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "r2",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				{
+					Namespace: "n1",
+					Names:     sets.New[string]("r1"),
+				},
+			},
+		},
+		// Note: this is deprecated fallback behavior. When we remove the behavior,
+		// rewrite this test to expect an error instead.
+		{
+			name: "by id prefers id embedded namespace",
+			apiOp: &types.APIRequest{
+				Namespace: "n2",
+			},
+			id: "n1/r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"list": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "r1",
+								},
+								accesscontrol.Access{
+									Namespace:    "n2",
+									ResourceName: "r2",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				{
+					Namespace: "n1",
+					Names:     sets.New[string]("r1"),
+				},
+			},
+		},
+		{
+			name:  "cluster scoped id unauthorized",
+			apiOp: &types.APIRequest{},
+			id:    "c1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+				},
+			},
+			wantPartitions: nil,
+		},
+		{
+			name:  "cluster scoped id authorized by name",
+			apiOp: &types.APIRequest{},
+			id:    "c1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": false,
+						"access": accesscontrol.AccessListByVerb{
+							"list": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "*",
+									ResourceName: "c1",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				{
+					Namespace: "",
+					Names:     sets.New[string]("c1"),
+				},
+			},
+		},
+		{
+			name:  "cluster scoped id authorized globally",
+			apiOp: &types.APIRequest{},
+			id:    "c1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": false,
+						"access": accesscontrol.AccessListByVerb{
+							"list": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "*",
+									ResourceName: "*",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				{
+					Namespace: "",
+					Names:     sets.New[string]("c1"),
+				},
+			},
+		},
+		{
+			name:  "cluster scoped id ignores unrequested resources",
+			apiOp: &types.APIRequest{},
+			id:    "c1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": false,
+						"access": accesscontrol.AccessListByVerb{
+							"list": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "*",
+									ResourceName: "c1",
+								},
+								accesscontrol.Access{
+									Namespace:    "*",
+									ResourceName: "c2",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				{
+					Namespace: "",
+					Names:     sets.New[string]("c1"),
+				},
+			},
+		},
 	}
 
 	for _, test := range tests {
@@ -253,6 +541,330 @@ func TestAll(t *testing.T) {
 	}
 }
 
+func TestVerbWatch(t *testing.T) {
+	tests := []struct {
+		name           string
+		apiOp          *types.APIRequest
+		id             string
+		schema         *types.APISchema
+		wantPartitions []partition.Partition
+	}{
+		{
+			name:  "by id fully unauthorized",
+			apiOp: &types.APIRequest{},
+			id:    "n1/r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+				},
+			},
+			wantPartitions: nil,
+		},
+		{
+			name:  "by id missing namespace",
+			apiOp: &types.APIRequest{},
+			id:    "n1/r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"watch": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n2",
+									ResourceName: "r2",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: nil,
+		},
+		{
+			name:  "by id missing resource",
+			apiOp: &types.APIRequest{},
+			id:    "n1/r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"watch": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "r2",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: nil,
+		},
+		{
+			name:  "by id authorized by name",
+			apiOp: &types.APIRequest{},
+			id:    "n1/r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"watch": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "r1",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				{
+					Namespace: "n1",
+					Names:     sets.New[string]("r1"),
+				},
+			},
+		},
+		{
+			name: "by id authorized by namespace",
+			apiOp: &types.APIRequest{
+				Namespace: "n1",
+			},
+			id: "n1/r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"watch": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "*",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				{
+					Namespace: "n1",
+					All:       false,
+					Names:     sets.New[string]("r1"),
+				},
+			},
+		},
+		{
+			name: "by namespaced id authorized by name",
+			apiOp: &types.APIRequest{
+				Namespace: "n1",
+			},
+			id: "r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"watch": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "r1",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				{
+					Namespace: "n1",
+					All:       false,
+					Names:     sets.New[string]("r1"),
+				},
+			},
+		},
+		{
+			name:  "by id ignores unrequested resources",
+			apiOp: &types.APIRequest{},
+			id:    "n1/r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"watch": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "r1",
+								},
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "r2",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				{
+					Namespace: "n1",
+					Names:     sets.New[string]("r1"),
+				},
+			},
+		},
+		// Note: this is deprecated fallback behavior. When we remove the behavior,
+		// rewrite this test to expect an error instead.
+		{
+			name: "by id prefers id embedded namespace",
+			apiOp: &types.APIRequest{
+				Namespace: "n2",
+			},
+			id: "n1/r1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": true,
+						"access": accesscontrol.AccessListByVerb{
+							"watch": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "n1",
+									ResourceName: "r1",
+								},
+								accesscontrol.Access{
+									Namespace:    "n2",
+									ResourceName: "r2",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				{
+					Namespace: "n1",
+					Names:     sets.New[string]("r1"),
+				},
+			},
+		},
+		{
+			name:  "cluster scoped id unauthorized",
+			apiOp: &types.APIRequest{},
+			id:    "c1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+				},
+			},
+			wantPartitions: nil,
+		},
+		{
+			name:  "cluster scoped id authorized by name",
+			apiOp: &types.APIRequest{},
+			id:    "c1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": false,
+						"access": accesscontrol.AccessListByVerb{
+							"watch": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "*",
+									ResourceName: "c1",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				{
+					Namespace: "",
+					Names:     sets.New[string]("c1"),
+				},
+			},
+		},
+		{
+			name:  "cluster scoped id authorized globally",
+			apiOp: &types.APIRequest{},
+			id:    "c1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": false,
+						"access": accesscontrol.AccessListByVerb{
+							"watch": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "*",
+									ResourceName: "*",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				{
+					Namespace: "",
+					Names:     sets.New[string]("c1"),
+				},
+			},
+		},
+		{
+			name:  "cluster scoped id ignores unrequested resources",
+			apiOp: &types.APIRequest{},
+			id:    "c1",
+			schema: &types.APISchema{
+				Schema: &schemas.Schema{
+					ID: "foo",
+					Attributes: map[string]interface{}{
+						"namespaced": false,
+						"access": accesscontrol.AccessListByVerb{
+							"watch": accesscontrol.AccessList{
+								accesscontrol.Access{
+									Namespace:    "*",
+									ResourceName: "c1",
+								},
+								accesscontrol.Access{
+									Namespace:    "*",
+									ResourceName: "c2",
+								},
+							},
+						},
+					},
+				},
+			},
+			wantPartitions: []partition.Partition{
+				{
+					Namespace: "",
+					Names:     sets.New[string]("c1"),
+				},
+			},
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			partitioner := rbacPartitioner{}
+			verb := "watch"
+			gotPartitions, gotErr := partitioner.All(test.apiOp, test.schema, verb, test.id)
+			assert.Nil(t, gotErr)
+			assert.Equal(t, test.wantPartitions, gotPartitions)
+		})
+	}
+}
+
 func TestStore(t *testing.T) {
 	expectedStore := NewMockUnstructuredStore(gomock.NewController(t))
 	rp := rbacPartitioner{