Add get volumeattachments support to Node authorizer

pkg/kubeapiserver/authorizer/config.go

@@ -82,6 +82,7 @@ func (config AuthorizationConfig) New() (authorizer.Authorizer, authorizer.RuleR
 				graph,
 				config.InformerFactory.Core().InternalVersion().Pods(),
 				config.InformerFactory.Core().InternalVersion().PersistentVolumes(),
+				config.VersionedInformerFactory.Storage().V1alpha1().VolumeAttachments(),
 			)
 			nodeAuthorizer := node.NewAuthorizer(graph, nodeidentifier.NewDefaultNodeIdentifier(), bootstrappolicy.NodeRules())
 			authorizers = append(authorizers, nodeAuthorizer)

plugin/pkg/auth/authorizer/node/BUILD

@@ -14,10 +14,13 @@ go_test(
     deps = [
         "//pkg/apis/core:go_default_library",
         "//pkg/auth/nodeidentifier:go_default_library",
+        "//pkg/features:go_default_library",
         "//plugin/pkg/auth/authorizer/rbac/bootstrappolicy:go_default_library",
+        "//vendor/k8s.io/api/storage/v1alpha1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/authentication/user:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/authorization/authorizer:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/util/feature:go_default_library",
     ],
 )
 
@@ -34,6 +37,7 @@ go_library(
         "//pkg/api/pod:go_default_library",
         "//pkg/apis/core:go_default_library",
         "//pkg/apis/rbac:go_default_library",
+        "//pkg/apis/storage:go_default_library",
         "//pkg/auth/nodeidentifier:go_default_library",
         "//pkg/client/informers/informers_generated/internalversion/core/internalversion:go_default_library",
         "//pkg/features:go_default_library",
@@ -42,9 +46,11 @@ go_library(
         "//third_party/forked/gonum/graph/simple:go_default_library",
         "//third_party/forked/gonum/graph/traverse:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",
+        "//vendor/k8s.io/api/storage/v1alpha1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/authorization/authorizer:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/util/feature:go_default_library",
+        "//vendor/k8s.io/client-go/informers/storage/v1alpha1:go_default_library",
         "//vendor/k8s.io/client-go/tools/cache:go_default_library",
     ],
 )

plugin/pkg/auth/authorizer/node/graph.go

@@ -106,6 +106,7 @@ const (
 	pvcVertexType
 	pvVertexType
 	secretVertexType
+	vaVertexType
 )
 
 var vertexTypes = map[vertexType]string{
@@ -115,6 +116,7 @@ var vertexTypes = map[vertexType]string{
 	pvcVertexType:       "pvc",
 	pvVertexType:        "pv",
 	secretVertexType:    "secret",
+	vaVertexType:        "volumeattachment",
 }
 
 // must be called under a write lock
@@ -263,3 +265,26 @@ func (g *Graph) DeletePV(name string) {
 	defer g.lock.Unlock()
 	g.deleteVertex_locked(pvVertexType, "", name)
 }
+
+// AddVolumeAttachment sets up edges for the following relationships:
+//
+//   volume attachment -> node
+func (g *Graph) AddVolumeAttachment(attachmentName, nodeName string) {
+	g.lock.Lock()
+	defer g.lock.Unlock()
+
+	// clear existing edges
+	g.deleteVertex_locked(vaVertexType, "", attachmentName)
+
+	// if we have a node, establish new edges
+	if len(nodeName) > 0 {
+		vaVertex := g.getOrCreateVertex_locked(vaVertexType, "", attachmentName)
+		nodeVertex := g.getOrCreateVertex_locked(nodeVertexType, "", nodeName)
+		g.graph.SetEdge(newDestinationEdge(vaVertex, nodeVertex, nodeVertex))
+	}
+}
+func (g *Graph) DeleteVolumeAttachment(name string) {
+	g.lock.Lock()
+	defer g.lock.Unlock()
+	g.deleteVertex_locked(vaVertexType, "", name)
+}

plugin/pkg/auth/authorizer/node/graph_populator.go

@@ -19,16 +19,25 @@ package node
 import (
 	"github.com/golang/glog"
 
+	storagev1alpha1 "k8s.io/api/storage/v1alpha1"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	storageinformers "k8s.io/client-go/informers/storage/v1alpha1"
 	"k8s.io/client-go/tools/cache"
 	api "k8s.io/kubernetes/pkg/apis/core"
 	coreinformers "k8s.io/kubernetes/pkg/client/informers/informers_generated/internalversion/core/internalversion"
+	"k8s.io/kubernetes/pkg/features"
 )
 
 type graphPopulator struct {
 	graph *Graph
 }
 
-func AddGraphEventHandlers(graph *Graph, pods coreinformers.PodInformer, pvs coreinformers.PersistentVolumeInformer) {
+func AddGraphEventHandlers(
+	graph *Graph,
+	pods coreinformers.PodInformer,
+	pvs coreinformers.PersistentVolumeInformer,
+	attachments storageinformers.VolumeAttachmentInformer,
+) {
 	g := &graphPopulator{
 		graph: graph,
 	}
@@ -44,6 +53,14 @@ func AddGraphEventHandlers(graph *Graph, pods coreinformers.PodInformer, pvs cor
 		UpdateFunc: g.updatePV,
 		DeleteFunc: g.deletePV,
 	})
+
+	if utilfeature.DefaultFeatureGate.Enabled(features.CSIPersistentVolume) {
+		attachments.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+			AddFunc:    g.addVolumeAttachment,
+			UpdateFunc: g.updateVolumeAttachment,
+			DeleteFunc: g.deleteVolumeAttachment,
+		})
+	}
 }
 
 func (g *graphPopulator) addPod(obj interface{}) {
@@ -106,3 +123,31 @@ func (g *graphPopulator) deletePV(obj interface{}) {
 	}
 	g.graph.DeletePV(pv.Name)
 }
+
+func (g *graphPopulator) addVolumeAttachment(obj interface{}) {
+	g.updateVolumeAttachment(nil, obj)
+}
+
+func (g *graphPopulator) updateVolumeAttachment(oldObj, obj interface{}) {
+	attachment := obj.(*storagev1alpha1.VolumeAttachment)
+	if oldObj != nil {
+		// skip add if node name is identical
+		oldAttachment := oldObj.(*storagev1alpha1.VolumeAttachment)
+		if oldAttachment.Spec.NodeName == attachment.Spec.NodeName {
+			return
+		}
+	}
+	g.graph.AddVolumeAttachment(attachment.Name, attachment.Spec.NodeName)
+}
+
+func (g *graphPopulator) deleteVolumeAttachment(obj interface{}) {
+	if tombstone, ok := obj.(cache.DeletedFinalStateUnknown); ok {
+		obj = tombstone.Obj
+	}
+	attachment, ok := obj.(*api.PersistentVolume)
+	if !ok {
+		glog.Infof("unexpected type %T", obj)
+		return
+	}
+	g.graph.DeleteVolumeAttachment(attachment.Name)
+}

plugin/pkg/auth/authorizer/node/node_authorizer.go

@@ -26,6 +26,7 @@ import (
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	api "k8s.io/kubernetes/pkg/apis/core"
 	rbacapi "k8s.io/kubernetes/pkg/apis/rbac"
+	storageapi "k8s.io/kubernetes/pkg/apis/storage"
 	"k8s.io/kubernetes/pkg/auth/nodeidentifier"
 	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/plugin/pkg/auth/authorizer/rbac"
@@ -48,6 +49,9 @@ type NodeAuthorizer struct {
 	graph      *Graph
 	identifier nodeidentifier.NodeIdentifier
 	nodeRules  []rbacapi.PolicyRule
+
+	// allows overriding for testing
+	features utilfeature.FeatureGate
 }
 
 // NewAuthorizer returns a new node authorizer
@@ -56,6 +60,7 @@ func NewAuthorizer(graph *Graph, identifier nodeidentifier.NodeIdentifier, rules
 		graph:      graph,
 		identifier: identifier,
 		nodeRules:  rules,
+		features:   utilfeature.DefaultFeatureGate,
 	}
 }
 
@@ -64,6 +69,7 @@ var (
 	secretResource    = api.Resource("secrets")
 	pvcResource       = api.Resource("persistentvolumeclaims")
 	pvResource        = api.Resource("persistentvolumes")
+	vaResource        = storageapi.Resource("volumeattachments")
 )
 
 func (r *NodeAuthorizer) Authorize(attrs authorizer.Attributes) (authorizer.Decision, string, error) {
@@ -87,7 +93,7 @@ func (r *NodeAuthorizer) Authorize(attrs authorizer.Attributes) (authorizer.Deci
 		case configMapResource:
 			return r.authorizeGet(nodeName, configMapVertexType, attrs)
 		case pvcResource:
-			if utilfeature.DefaultFeatureGate.Enabled(features.ExpandPersistentVolumes) {
+			if r.features.Enabled(features.ExpandPersistentVolumes) {
 				if attrs.GetSubresource() == "status" {
 					return r.authorizeStatusUpdate(nodeName, pvcVertexType, attrs)
 				}
@@ -95,6 +101,11 @@ func (r *NodeAuthorizer) Authorize(attrs authorizer.Attributes) (authorizer.Deci
 			return r.authorizeGet(nodeName, pvcVertexType, attrs)
 		case pvResource:
 			return r.authorizeGet(nodeName, pvVertexType, attrs)
+		case vaResource:
+			if r.features.Enabled(features.CSIPersistentVolume) {
+				return r.authorizeGet(nodeName, vaVertexType, attrs)
+			}
+			return authorizer.DecisionNoOpinion, fmt.Sprintf("disabled by feature gate %s", features.CSIPersistentVolume), nil
 		}
 	}
 

plugin/pkg/auth/authorizer/node/node_authorizer_test.go

@@ -24,14 +24,31 @@ import (
 
 	"os"
 
+	storagev1alpha1 "k8s.io/api/storage/v1alpha1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apiserver/pkg/authentication/user"
 	"k8s.io/apiserver/pkg/authorization/authorizer"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	api "k8s.io/kubernetes/pkg/apis/core"
 	"k8s.io/kubernetes/pkg/auth/nodeidentifier"
+	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/plugin/pkg/auth/authorizer/rbac/bootstrappolicy"
 )
 
+var (
+	csiEnabledFeature  = utilfeature.NewFeatureGate()
+	csiDisabledFeature = utilfeature.NewFeatureGate()
+)
+
+func init() {
+	if err := csiEnabledFeature.Add(map[utilfeature.Feature]utilfeature.FeatureSpec{features.CSIPersistentVolume: {Default: true}}); err != nil {
+		panic(err)
+	}
+	if err := csiDisabledFeature.Add(map[utilfeature.Feature]utilfeature.FeatureSpec{features.CSIPersistentVolume: {Default: false}}); err != nil {
+		panic(err)
+	}
+}
+
 func TestAuthorizer(t *testing.T) {
 	g := NewGraph()
 
@@ -39,6 +56,7 @@ func TestAuthorizer(t *testing.T) {
 		nodes:                  2,
 		namespaces:             2,
 		podsPerNode:            2,
+		attachmentsPerNode:     1,
 		sharedConfigMapsPerPod: 0,
 		uniqueConfigMapsPerPod: 1,
 		sharedSecretsPerPod:    1,
@@ -46,18 +64,19 @@ func TestAuthorizer(t *testing.T) {
 		sharedPVCsPerPod:       0,
 		uniquePVCsPerPod:       1,
 	}
-	pods, pvs := generate(opts)
-	populate(g, pods, pvs)
+	pods, pvs, attachments := generate(opts)
+	populate(g, pods, pvs, attachments)
 
 	identifier := nodeidentifier.NewDefaultNodeIdentifier()
-	authz := NewAuthorizer(g, identifier, bootstrappolicy.NodeRules())
+	authz := NewAuthorizer(g, identifier, bootstrappolicy.NodeRules()).(*NodeAuthorizer)
 
 	node0 := &user.DefaultInfo{Name: "system:node:node0", Groups: []string{"system:nodes"}}
 
 	tests := []struct {
-		name   string
-		attrs  authorizer.AttributesRecord
-		expect authorizer.Decision
+		name     string
+		attrs    authorizer.AttributesRecord
+		expect   authorizer.Decision
+		features utilfeature.FeatureGate
 	}{
 		{
 			name:   "allowed configmap",
@@ -115,10 +134,33 @@ func TestAuthorizer(t *testing.T) {
 			attrs:  authorizer.AttributesRecord{User: node0, ResourceRequest: true, Verb: "get", Resource: "persistentvolumes", Name: "pv0-pod0-node1-ns0", Namespace: ""},
 			expect: authorizer.DecisionNoOpinion,
 		},
+		{
+			name:     "disallowed attachment - no relationship",
+			attrs:    authorizer.AttributesRecord{User: node0, ResourceRequest: true, Verb: "get", Resource: "volumeattachments", APIGroup: "storage.k8s.io", Name: "attachment0-node1"},
+			features: csiEnabledFeature,
+			expect:   authorizer.DecisionNoOpinion,
+		},
+		{
+			name:     "disallowed attachment - feature disabled",
+			attrs:    authorizer.AttributesRecord{User: node0, ResourceRequest: true, Verb: "get", Resource: "volumeattachments", APIGroup: "storage.k8s.io", Name: "attachment0-node0"},
+			features: csiDisabledFeature,
+			expect:   authorizer.DecisionNoOpinion,
+		},
+		{
+			name:     "allowed attachment - feature enabled",
+			attrs:    authorizer.AttributesRecord{User: node0, ResourceRequest: true, Verb: "get", Resource: "volumeattachments", APIGroup: "storage.k8s.io", Name: "attachment0-node0"},
+			features: csiEnabledFeature,
+			expect:   authorizer.DecisionAllow,
+		},
 	}
 
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
+			if tc.features == nil {
+				authz.features = utilfeature.DefaultFeatureGate
+			} else {
+				authz.features = tc.features
+			}
 			decision, _, _ := authz.Authorize(tc.attrs)
 			if decision != tc.expect {
 				t.Errorf("expected %v, got %v", tc.expect, decision)
@@ -204,6 +246,8 @@ type sampleDataOpts struct {
 
 	podsPerNode int
 
+	attachmentsPerNode int
+
 	sharedConfigMapsPerPod int
 	sharedSecretsPerPod    int
 	sharedPVCsPerPod       int
@@ -218,6 +262,7 @@ func BenchmarkPopulationAllocation(b *testing.B) {
 		nodes:                  500,
 		namespaces:             200,
 		podsPerNode:            200,
+		attachmentsPerNode:     20,
 		sharedConfigMapsPerPod: 0,
 		uniqueConfigMapsPerPod: 1,
 		sharedSecretsPerPod:    1,
@@ -226,12 +271,12 @@ func BenchmarkPopulationAllocation(b *testing.B) {
 		uniquePVCsPerPod:       1,
 	}
 
-	pods, pvs := generate(opts)
+	pods, pvs, attachments := generate(opts)
 	b.ResetTimer()
 
 	for i := 0; i < b.N; i++ {
 		g := NewGraph()
-		populate(g, pods, pvs)
+		populate(g, pods, pvs, attachments)
 	}
 }
 
@@ -248,6 +293,7 @@ func BenchmarkPopulationRetention(b *testing.B) {
 		nodes:                  500,
 		namespaces:             200,
 		podsPerNode:            200,
+		attachmentsPerNode:     20,
 		sharedConfigMapsPerPod: 0,
 		uniqueConfigMapsPerPod: 1,
 		sharedSecretsPerPod:    1,
@@ -256,14 +302,14 @@ func BenchmarkPopulationRetention(b *testing.B) {
 		uniquePVCsPerPod:       1,
 	}
 
-	pods, pvs := generate(opts)
+	pods, pvs, attachments := generate(opts)
 	// Garbage collect before the first iteration
 	runtime.GC()
 	b.ResetTimer()
 
 	for i := 0; i < b.N; i++ {
 		g := NewGraph()
-		populate(g, pods, pvs)
+		populate(g, pods, pvs, attachments)
 
 		if i == 0 {
 			f, _ := os.Create("BenchmarkPopulationRetention.profile")
@@ -283,6 +329,7 @@ func BenchmarkAuthorization(b *testing.B) {
 		nodes:                  500,
 		namespaces:             200,
 		podsPerNode:            200,
+		attachmentsPerNode:     20,
 		sharedConfigMapsPerPod: 0,
 		uniqueConfigMapsPerPod: 1,
 		sharedSecretsPerPod:    1,
@@ -290,18 +337,19 @@ func BenchmarkAuthorization(b *testing.B) {
 		sharedPVCsPerPod:       0,
 		uniquePVCsPerPod:       1,
 	}
-	pods, pvs := generate(opts)
-	populate(g, pods, pvs)
+	pods, pvs, attachments := generate(opts)
+	populate(g, pods, pvs, attachments)
 
 	identifier := nodeidentifier.NewDefaultNodeIdentifier()
-	authz := NewAuthorizer(g, identifier, bootstrappolicy.NodeRules())
+	authz := NewAuthorizer(g, identifier, bootstrappolicy.NodeRules()).(*NodeAuthorizer)
 
 	node0 := &user.DefaultInfo{Name: "system:node:node0", Groups: []string{"system:nodes"}}
 
 	tests := []struct {
-		name   string
-		attrs  authorizer.AttributesRecord
-		expect authorizer.Decision
+		name     string
+		attrs    authorizer.AttributesRecord
+		expect   authorizer.Decision
+		features utilfeature.FeatureGate
 	}{
 		{
 			name:   "allowed configmap",
@@ -343,10 +391,33 @@ func BenchmarkAuthorization(b *testing.B) {
 			attrs:  authorizer.AttributesRecord{User: node0, ResourceRequest: true, Verb: "get", Resource: "persistentvolumes", Name: "pv0-pod0-node1-ns0", Namespace: ""},
 			expect: authorizer.DecisionNoOpinion,
 		},
+		{
+			name:     "disallowed attachment - no relationship",
+			attrs:    authorizer.AttributesRecord{User: node0, ResourceRequest: true, Verb: "get", Resource: "volumeattachments", APIGroup: "storage.k8s.io", Name: "attachment0-node1"},
+			features: csiEnabledFeature,
+			expect:   authorizer.DecisionNoOpinion,
+		},
+		{
+			name:     "disallowed attachment - feature disabled",
+			attrs:    authorizer.AttributesRecord{User: node0, ResourceRequest: true, Verb: "get", Resource: "volumeattachments", APIGroup: "storage.k8s.io", Name: "attachment0-node0"},
+			features: csiDisabledFeature,
+			expect:   authorizer.DecisionNoOpinion,
+		},
+		{
+			name:     "allowed attachment - feature enabled",
+			attrs:    authorizer.AttributesRecord{User: node0, ResourceRequest: true, Verb: "get", Resource: "volumeattachments", APIGroup: "storage.k8s.io", Name: "attachment0-node0"},
+			features: csiEnabledFeature,
+			expect:   authorizer.DecisionAllow,
+		},
 	}
 
 	b.ResetTimer()
 	for _, tc := range tests {
+		if tc.features == nil {
+			authz.features = utilfeature.DefaultFeatureGate
+		} else {
+			authz.features = tc.features
+		}
 		b.Run(tc.name, func(b *testing.B) {
 			for i := 0; i < b.N; i++ {
 				decision, _, _ := authz.Authorize(tc.attrs)
@@ -358,7 +429,7 @@ func BenchmarkAuthorization(b *testing.B) {
 	}
 }
 
-func populate(graph *Graph, pods []*api.Pod, pvs []*api.PersistentVolume) {
+func populate(graph *Graph, pods []*api.Pod, pvs []*api.PersistentVolume, attachments []*storagev1alpha1.VolumeAttachment) {
 	p := &graphPopulator{}
 	p.graph = graph
 	for _, pod := range pods {
@@ -367,15 +438,19 @@ func populate(graph *Graph, pods []*api.Pod, pvs []*api.PersistentVolume) {
 	for _, pv := range pvs {
 		p.addPV(pv)
 	}
+	for _, attachment := range attachments {
+		p.addVolumeAttachment(attachment)
+	}
 }
 
 // generate creates sample pods and persistent volumes based on the provided options.
 // the secret/configmap/pvc/node references in the pod and pv objects are named to indicate the connections between the objects.
 // for example, secret0-pod0-node0 is a secret referenced by pod0 which is bound to node0.
 // when populated into the graph, the node authorizer should allow node0 to access that secret, but not node1.
-func generate(opts sampleDataOpts) ([]*api.Pod, []*api.PersistentVolume) {
+func generate(opts sampleDataOpts) ([]*api.Pod, []*api.PersistentVolume, []*storagev1alpha1.VolumeAttachment) {
 	pods := make([]*api.Pod, 0, opts.nodes*opts.podsPerNode)
 	pvs := make([]*api.PersistentVolume, 0, (opts.nodes*opts.podsPerNode*opts.uniquePVCsPerPod)+(opts.sharedPVCsPerPod*opts.namespaces))
+	attachments := make([]*storagev1alpha1.VolumeAttachment, 0, opts.nodes*opts.attachmentsPerNode)
 
 	for n := 0; n < opts.nodes; n++ {
 		nodeName := fmt.Sprintf("node%d", n)
@@ -432,6 +507,12 @@ func generate(opts sampleDataOpts) ([]*api.Pod, []*api.PersistentVolume) {
 
 			pods = append(pods, pod)
 		}
+		for a := 0; a < opts.attachmentsPerNode; a++ {
+			attachment := &storagev1alpha1.VolumeAttachment{}
+			attachment.Name = fmt.Sprintf("attachment%d-%s", a, nodeName)
+			attachment.Spec.NodeName = nodeName
+			attachments = append(attachments, attachment)
+		}
 	}
-	return pods, pvs
+	return pods, pvs, attachments
 }

test/integration/auth/BUILD

@@ -52,6 +52,7 @@ go_test(
         "//test/integration/framework:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",
         "//vendor/k8s.io/api/authentication/v1beta1:go_default_library",
+        "//vendor/k8s.io/api/storage/v1alpha1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/api/errors:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/api/resource:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
@@ -69,6 +70,7 @@ go_test(
         "//vendor/k8s.io/apiserver/pkg/authorization/authorizer:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/authorization/authorizerfactory:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/registry/generic:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/server/storage:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/util/feature:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/util/feature/testing:go_default_library",
         "//vendor/k8s.io/apiserver/plugin/pkg/authenticator/token/tokentest:go_default_library",

test/integration/auth/node_test.go

@@ -23,6 +23,7 @@ import (
 	"testing"
 	"time"
 
+	storagev1alpha1 "k8s.io/api/storage/v1alpha1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -31,9 +32,11 @@ import (
 	"k8s.io/apiserver/pkg/authentication/request/bearertoken"
 	"k8s.io/apiserver/pkg/authentication/token/tokenfile"
 	"k8s.io/apiserver/pkg/authentication/user"
+	serverstorage "k8s.io/apiserver/pkg/server/storage"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	utilfeaturetesting "k8s.io/apiserver/pkg/util/feature/testing"
 	versionedinformers "k8s.io/client-go/informers"
+	externalclientset "k8s.io/client-go/kubernetes"
 	restclient "k8s.io/client-go/rest"
 	"k8s.io/kubernetes/pkg/api/legacyscheme"
 	api "k8s.io/kubernetes/pkg/apis/core"
@@ -76,6 +79,9 @@ func TestNodeAuthorizer(t *testing.T) {
 	informerFactory := informers.NewSharedInformerFactory(superuserClient, time.Minute)
 	versionedInformerFactory := versionedinformers.NewSharedInformerFactory(superuserClientExternal, time.Minute)
 
+	// Enabled CSIPersistentVolume feature at startup so volumeattachments get watched
+	defer utilfeaturetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.CSIPersistentVolume, true)()
+
 	// Set up Node+RBAC authorizer
 	authorizerConfig := &authorizer.AuthorizationConfig{
 		AuthorizationModes:       []string{"Node", "RBAC"},
@@ -99,6 +105,10 @@ func TestNodeAuthorizer(t *testing.T) {
 	masterConfig.GenericConfig.Authenticator = authenticator
 	masterConfig.GenericConfig.Authorizer = nodeRBACAuthorizer
 	masterConfig.GenericConfig.AdmissionControl = nodeRestrictionAdmission
+
+	// enable testing volume attachments
+	masterConfig.ExtraConfig.APIResourceConfigSource.(*serverstorage.ResourceConfig).EnableVersions(storagev1alpha1.SchemeGroupVersion)
+
 	_, _, closeFn := framework.RunAMasterUsingServer(masterConfig, apiServer, h)
 	defer closeFn()
 
@@ -129,6 +139,17 @@ func TestNodeAuthorizer(t *testing.T) {
 	if _, err := superuserClient.Core().ConfigMaps("ns").Create(&api.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "myconfigmap"}}); err != nil {
 		t.Fatal(err)
 	}
+	pvName := "mypv"
+	if _, err := superuserClientExternal.StorageV1alpha1().VolumeAttachments().Create(&storagev1alpha1.VolumeAttachment{
+		ObjectMeta: metav1.ObjectMeta{Name: "myattachment"},
+		Spec: storagev1alpha1.VolumeAttachmentSpec{
+			Attacher: "foo",
+			Source:   storagev1alpha1.VolumeAttachmentSource{PersistentVolumeName: &pvName},
+			NodeName: "node2",
+		},
+	}); err != nil {
+		t.Fatal(err)
+	}
 	if _, err := superuserClient.Core().PersistentVolumeClaims("ns").Create(&api.PersistentVolumeClaim{
 		ObjectMeta: metav1.ObjectMeta{Name: "mypvc"},
 		Spec: api.PersistentVolumeClaimSpec{
@@ -181,6 +202,12 @@ func TestNodeAuthorizer(t *testing.T) {
 			return err
 		}
 	}
+	getVolumeAttachment := func(client externalclientset.Interface) func() error {
+		return func() error {
+			_, err := client.StorageV1alpha1().VolumeAttachments().Get("myattachment", metav1.GetOptions{})
+			return err
+		}
+	}
 
 	createNode2NormalPod := func(client clientset.Interface) func() error {
 		return func() error {
@@ -307,8 +334,8 @@ func TestNodeAuthorizer(t *testing.T) {
 	}
 
 	nodeanonClient, _ := clientsetForToken(tokenNodeUnknown, clientConfig)
-	node1Client, _ := clientsetForToken(tokenNode1, clientConfig)
-	node2Client, _ := clientsetForToken(tokenNode2, clientConfig)
+	node1Client, node1ClientExternal := clientsetForToken(tokenNode1, clientConfig)
+	node2Client, node2ClientExternal := clientsetForToken(tokenNode2, clientConfig)
 
 	// all node requests from node1 and unknown node fail
 	expectForbidden(t, getSecret(nodeanonClient))
@@ -405,32 +432,40 @@ func TestNodeAuthorizer(t *testing.T) {
 
 	// re-create a pod as an admin to add object references
 	expectAllowed(t, createNode2NormalPod(superuserClient))
-	// With ExpandPersistentVolumes feature disabled
+
+	// ExpandPersistentVolumes feature disabled
 	defer utilfeaturetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ExpandPersistentVolumes, false)()
-	// node->pvc relationship not established
 	expectForbidden(t, updatePVCCapacity(node1Client))
-	// node->pvc relationship established but feature is disabled
 	expectForbidden(t, updatePVCCapacity(node2Client))
 
-	//Enabled ExpandPersistentVolumes feature
+	// ExpandPersistentVolumes feature enabled
 	defer utilfeaturetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ExpandPersistentVolumes, true)()
-	// Node->pvc relationship not established
 	expectForbidden(t, updatePVCCapacity(node1Client))
-	// node->pvc relationship established and feature is enabled
 	expectAllowed(t, updatePVCCapacity(node2Client))
-	// node->pvc relationship established but updating phase
 	expectForbidden(t, updatePVCPhase(node2Client))
+
+	// Disabled CSIPersistentVolume feature
+	defer utilfeaturetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.CSIPersistentVolume, false)()
+	expectForbidden(t, getVolumeAttachment(node1ClientExternal))
+	expectForbidden(t, getVolumeAttachment(node2ClientExternal))
+	// Enabled CSIPersistentVolume feature
+	defer utilfeaturetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.CSIPersistentVolume, true)()
+	expectForbidden(t, getVolumeAttachment(node1ClientExternal))
+	expectAllowed(t, getVolumeAttachment(node2ClientExternal))
 }
 
 // expect executes a function a set number of times until it either returns the
 // expected error or executes too many times. It returns if the retries timed
 // out and the last error returned by the method.
-func expect(f func() error, wantErr func(error) bool) (timeout bool, lastErr error) {
+func expect(t *testing.T, f func() error, wantErr func(error) bool) (timeout bool, lastErr error) {
+	t.Helper()
 	err := wait.PollImmediate(time.Second, 30*time.Second, func() (bool, error) {
+		t.Helper()
 		lastErr = f()
 		if wantErr(lastErr) {
 			return true, nil
 		}
+		t.Logf("unexpected response, will retry: %v", lastErr)
 		return false, nil
 	})
 	return err == nil, lastErr
@@ -438,21 +473,21 @@ func expect(f func() error, wantErr func(error) bool) (timeout bool, lastErr err
 
 func expectForbidden(t *testing.T, f func() error) {
 	t.Helper()
-	if ok, err := expect(f, errors.IsForbidden); !ok {
+	if ok, err := expect(t, f, errors.IsForbidden); !ok {
 		t.Errorf("Expected forbidden error, got %v", err)
 	}
 }
 
 func expectNotFound(t *testing.T, f func() error) {
 	t.Helper()
-	if ok, err := expect(f, errors.IsNotFound); !ok {
+	if ok, err := expect(t, f, errors.IsNotFound); !ok {
 		t.Errorf("Expected notfound error, got %v", err)
 	}
 }
 
 func expectAllowed(t *testing.T, f func() error) {
 	t.Helper()
-	if ok, err := expect(f, func(e error) bool { return e == nil }); !ok {
+	if ok, err := expect(t, f, func(e error) bool { return e == nil }); !ok {
 		t.Errorf("Expected no error, got %v", err)
 	}
 }

test/integration/framework/master_utils.go

@@ -300,7 +300,10 @@ func NewMasterConfig() *master.Config {
 	// FIXME (soltysh): this GroupVersionResource override should be configurable
 	// we need to set both for the whole group and for cronjobs, separately
 	resourceEncoding.SetVersionEncoding(batch.GroupName, *testapi.Batch.GroupVersion(), schema.GroupVersion{Group: batch.GroupName, Version: runtime.APIVersionInternal})
-	resourceEncoding.SetResourceEncoding(schema.GroupResource{Group: "batch", Resource: "cronjobs"}, schema.GroupVersion{Group: batch.GroupName, Version: "v1beta1"}, schema.GroupVersion{Group: batch.GroupName, Version: runtime.APIVersionInternal})
+	resourceEncoding.SetResourceEncoding(schema.GroupResource{Group: batch.GroupName, Resource: "cronjobs"}, schema.GroupVersion{Group: batch.GroupName, Version: "v1beta1"}, schema.GroupVersion{Group: batch.GroupName, Version: runtime.APIVersionInternal})
+	// volumeattachments only exist in storage.k8s.io/v1alpha1
+	resourceEncoding.SetVersionEncoding(storage.GroupName, *testapi.Storage.GroupVersion(), schema.GroupVersion{Group: storage.GroupName, Version: runtime.APIVersionInternal})
+	resourceEncoding.SetResourceEncoding(schema.GroupResource{Group: storage.GroupName, Resource: "volumeattachments"}, schema.GroupVersion{Group: storage.GroupName, Version: "v1alpha1"}, schema.GroupVersion{Group: storage.GroupName, Version: runtime.APIVersionInternal})
 
 	storageFactory := serverstorage.NewDefaultStorageFactory(etcdOptions.StorageConfig, runtime.ContentTypeJSON, ns, resourceEncoding, master.DefaultAPIResourceConfigSource(), nil)
 	storageFactory.SetSerializer(