Merge pull request #128077 from aramase/aramase/f/kep_4412_sa_node_aud_restriction

Enforce service account node audience restriction

pkg/features/kube_features.go

@@ -585,6 +585,13 @@ const (
 	// Decouples Taint Eviction Controller, performing taint-based Pod eviction, from Node Lifecycle Controller.
 	SeparateTaintEvictionController featuregate.Feature = "SeparateTaintEvictionController"
 
+	// owner: @aramase
+	// kep: https://kep.k8s.io/4412
+	//
+	// ServiceAccountNodeAudienceRestriction is used to restrict the audience for which the
+	// kubelet can request a service account token for.
+	ServiceAccountNodeAudienceRestriction featuregate.Feature = "ServiceAccountNodeAudienceRestriction"
+
 	// owner: @munnerz
 	// kep: http://kep.k8s.io/4193
 	//

pkg/features/versioned_kube_features.go

@@ -662,6 +662,10 @@ var defaultVersionedKubernetesFeatureGates = map[featuregate.Feature]featuregate
 		{Version: version.MustParse("1.30"), Default: true, PreRelease: featuregate.Beta},
 	},
 
+	ServiceAccountNodeAudienceRestriction: {
+		{Version: version.MustParse("1.32"), Default: true, PreRelease: featuregate.Beta},
+	},
+
 	ServiceAccountTokenJTI: {
 		{Version: version.MustParse("1.29"), Default: false, PreRelease: featuregate.Alpha},
 		{Version: version.MustParse("1.30"), Default: true, PreRelease: featuregate.Beta},

plugin/pkg/admission/noderestriction/admission.go

@@ -24,17 +24,21 @@ import (
 	"strings"
 
 	"github.com/google/go-cmp/cmp"
+
 	v1 "k8s.io/api/core/v1"
 	apiequality "k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/labels"
+	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apiserver/pkg/admission"
 	apiserveradmission "k8s.io/apiserver/pkg/admission/initializer"
 	"k8s.io/client-go/informers"
 	corev1lister "k8s.io/client-go/listers/core/v1"
+	storagelisters "k8s.io/client-go/listers/storage/v1"
 	"k8s.io/component-base/featuregate"
+	"k8s.io/component-helpers/storage/ephemeral"
 	kubeletapis "k8s.io/kubelet/pkg/apis"
 	podutil "k8s.io/kubernetes/pkg/api/pod"
 	authenticationapi "k8s.io/kubernetes/pkg/apis/authentication"
@@ -43,7 +47,7 @@ import (
 	api "k8s.io/kubernetes/pkg/apis/core"
 	"k8s.io/kubernetes/pkg/apis/policy"
 	"k8s.io/kubernetes/pkg/apis/resource"
-	storage "k8s.io/kubernetes/pkg/apis/storage"
+	"k8s.io/kubernetes/pkg/apis/storage"
 	"k8s.io/kubernetes/pkg/auth/nodeidentifier"
 	"k8s.io/kubernetes/pkg/features"
 )
@@ -70,13 +74,17 @@ func NewPlugin(nodeIdentifier nodeidentifier.NodeIdentifier) *Plugin {
 // Plugin holds state for and implements the admission plugin.
 type Plugin struct {
 	*admission.Handler
-	nodeIdentifier nodeidentifier.NodeIdentifier
+	nodeIdentifier  nodeidentifier.NodeIdentifier
-	podsGetter     corev1lister.PodLister
+	podsGetter      corev1lister.PodLister
-	nodesGetter    corev1lister.NodeLister
+	nodesGetter     corev1lister.NodeLister
+	csiDriverGetter storagelisters.CSIDriverLister
+	pvcGetter       corev1lister.PersistentVolumeClaimLister
+	pvGetter        corev1lister.PersistentVolumeLister
 
 	expansionRecoveryEnabled                       bool
 	dynamicResourceAllocationEnabled               bool
 	allowInsecureKubeletCertificateSigningRequests bool
+	serviceAccountNodeAudienceRestriction          bool
 }
 
 var (
@@ -90,12 +98,18 @@ func (p *Plugin) InspectFeatureGates(featureGates featuregate.FeatureGate) {
 	p.expansionRecoveryEnabled = featureGates.Enabled(features.RecoverVolumeExpansionFailure)
 	p.dynamicResourceAllocationEnabled = featureGates.Enabled(features.DynamicResourceAllocation)
 	p.allowInsecureKubeletCertificateSigningRequests = featureGates.Enabled(features.AllowInsecureKubeletCertificateSigningRequests)
+	p.serviceAccountNodeAudienceRestriction = featureGates.Enabled(features.ServiceAccountNodeAudienceRestriction)
 }
 
 // SetExternalKubeInformerFactory registers an informer factory into Plugin
 func (p *Plugin) SetExternalKubeInformerFactory(f informers.SharedInformerFactory) {
 	p.podsGetter = f.Core().V1().Pods().Lister()
 	p.nodesGetter = f.Core().V1().Nodes().Lister()
+	if p.serviceAccountNodeAudienceRestriction {
+		p.csiDriverGetter = f.Storage().V1().CSIDrivers().Lister()
+		p.pvcGetter = f.Core().V1().PersistentVolumeClaims().Lister()
+		p.pvGetter = f.Core().V1().PersistentVolumes().Lister()
+	}
 }
 
 // ValidateInitialization validates the Plugin was initialized properly
@@ -109,6 +123,17 @@ func (p *Plugin) ValidateInitialization() error {
 	if p.nodesGetter == nil {
 		return fmt.Errorf("%s requires a node getter", PluginName)
 	}
+	if p.serviceAccountNodeAudienceRestriction {
+		if p.csiDriverGetter == nil {
+			return fmt.Errorf("%s requires a CSI driver getter", PluginName)
+		}
+		if p.pvcGetter == nil {
+			return fmt.Errorf("%s requires a PVC getter", PluginName)
+		}
+		if p.pvGetter == nil {
+			return fmt.Errorf("%s requires a PV getter", PluginName)
+		}
+	}
 	return nil
 }
 
@@ -594,6 +619,12 @@ func (p *Plugin) admitServiceAccount(nodeName string, a admission.Attributes) er
 		return admission.NewForbidden(a, fmt.Errorf("node requested token bound to a pod scheduled on a different node"))
 	}
 
+	if p.serviceAccountNodeAudienceRestriction {
+		if err := p.validateNodeServiceAccountAudience(tr, pod); err != nil {
+			return admission.NewForbidden(a, err)
+		}
+	}
+
 	// Note: A token may only be bound to one object at a time. By requiring
 	// the Pod binding, noderestriction eliminates the opportunity to spoof
 	// a Node binding. Instead, kube-apiserver automatically infers and sets
@@ -603,6 +634,106 @@ func (p *Plugin) admitServiceAccount(nodeName string, a admission.Attributes) er
 	return nil
 }
 
+func (p *Plugin) validateNodeServiceAccountAudience(tr *authenticationapi.TokenRequest, pod *v1.Pod) error {
+	// ensure all items in tr.Spec.Audiences are present in a volume mount in the pod
+	requestedAudience := ""
+	switch len(tr.Spec.Audiences) {
+	case 0:
+		requestedAudience = ""
+	case 1:
+		requestedAudience = tr.Spec.Audiences[0]
+	default:
+		return fmt.Errorf("node may only request 0 or 1 audiences")
+	}
+
+	foundAudiencesInPodSpec, err := p.podReferencesAudience(pod, requestedAudience)
+	if err != nil {
+		return fmt.Errorf("error validating audience %q: %w", requestedAudience, err)
+	}
+	if !foundAudiencesInPodSpec {
+		return fmt.Errorf("audience %q not found in pod spec volume", requestedAudience)
+	}
+	return nil
+}
+
+func (p *Plugin) podReferencesAudience(pod *v1.Pod, audience string) (bool, error) {
+	var errs []error
+
+	for _, v := range pod.Spec.Volumes {
+		if v.Projected != nil {
+			for _, src := range v.Projected.Sources {
+				if src.ServiceAccountToken != nil && src.ServiceAccountToken.Audience == audience {
+					return true, nil
+				}
+			}
+		}
+
+		// also allow audiences for CSI token requests
+		// - pod --> ephemeral --> pvc --> pv --> csi --> driver --> tokenrequest with audience
+		// - pod --> pvc --> pv --> csi --> driver --> tokenrequest with audience
+		// - pod --> csi --> driver --> tokenrequest with audience
+		var driverName string
+		var err error
+		switch {
+		case v.Ephemeral != nil && v.Ephemeral.VolumeClaimTemplate != nil:
+			pvcName := ephemeral.VolumeClaimName(pod, &v)
+			driverName, err = p.getCSIFromPVC(pod.Namespace, pvcName)
+		case v.PersistentVolumeClaim != nil:
+			driverName, err = p.getCSIFromPVC(pod.Namespace, v.PersistentVolumeClaim.ClaimName)
+		case v.CSI != nil:
+			driverName = v.CSI.Driver
+		}
+
+		if err != nil {
+			errs = append(errs, err)
+			continue
+		}
+
+		if len(driverName) > 0 {
+			hasAudience, hasAudienceErr := p.csiDriverHasAudience(driverName, audience)
+			if hasAudienceErr != nil {
+				errs = append(errs, hasAudienceErr)
+				continue
+			}
+			if hasAudience {
+				return true, nil
+			}
+		}
+	}
+
+	return false, utilerrors.NewAggregate(errs)
+}
+
+// getCSIFromPVC returns the CSI driver name from the PVC->PV->CSI->Driver chain
+func (p *Plugin) getCSIFromPVC(namespace, claimName string) (string, error) {
+	pvc, err := p.pvcGetter.PersistentVolumeClaims(namespace).Get(claimName)
+	if err != nil {
+		return "", err
+	}
+	pv, err := p.pvGetter.Get(pvc.Spec.VolumeName)
+	if err != nil {
+		return "", err
+	}
+	if pv.Spec.CSI != nil {
+		return pv.Spec.CSI.Driver, nil
+	}
+	return "", nil
+}
+
+func (p *Plugin) csiDriverHasAudience(driverName, audience string) (bool, error) {
+	driver, err := p.csiDriverGetter.Get(driverName)
+	if err != nil {
+		return false, err
+	}
+
+	for _, tokenRequest := range driver.Spec.TokenRequests {
+		if tokenRequest.Audience == audience {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
 func (p *Plugin) admitLease(nodeName string, a admission.Attributes) error {
 	// the request must be against the system namespace reserved for node leases
 	if a.GetNamespace() != api.NamespaceNodeLease {

plugin/pkg/admission/noderestriction/admission_test.go

@@ -28,11 +28,8 @@ import (
 	"testing"
 	"time"
 
-	"k8s.io/apiserver/pkg/util/feature"
-	featuregatetesting "k8s.io/component-base/featuregate/testing"
-	"k8s.io/kubernetes/pkg/features"
-
 	corev1 "k8s.io/api/core/v1"
+	storagev1 "k8s.io/api/storage/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -41,9 +38,12 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apiserver/pkg/admission"
 	"k8s.io/apiserver/pkg/authentication/user"
+	"k8s.io/apiserver/pkg/util/feature"
 	corev1lister "k8s.io/client-go/listers/core/v1"
+	storagelisters "k8s.io/client-go/listers/storage/v1"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/component-base/featuregate"
+	featuregatetesting "k8s.io/component-base/featuregate/testing"
 	kubeletapis "k8s.io/kubelet/pkg/apis"
 	authenticationapi "k8s.io/kubernetes/pkg/apis/authentication"
 	certificatesapi "k8s.io/kubernetes/pkg/apis/certificates"
@@ -51,8 +51,9 @@ import (
 	api "k8s.io/kubernetes/pkg/apis/core"
 	"k8s.io/kubernetes/pkg/apis/policy"
 	resourceapi "k8s.io/kubernetes/pkg/apis/resource"
-	storage "k8s.io/kubernetes/pkg/apis/storage"
+	"k8s.io/kubernetes/pkg/apis/storage"
 	"k8s.io/kubernetes/pkg/auth/nodeidentifier"
+	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/utils/pointer"
 )
 
@@ -107,10 +108,10 @@ func makeTestPodEviction(name string) *policy.Eviction {
 	return eviction
 }
 
-func makeTokenRequest(podname string, poduid types.UID) *authenticationapi.TokenRequest {
+func makeTokenRequest(podname string, poduid types.UID, audiences []string) *authenticationapi.TokenRequest {
 	tr := &authenticationapi.TokenRequest{
 		Spec: authenticationapi.TokenRequestSpec{
-			Audiences: []string{"foo"},
+			Audiences: audiences,
 		},
 	}
 	if podname != "" {
@@ -214,13 +215,16 @@ func setForbiddenUpdateLabels(node *api.Node, value string) *api.Node {
 }
 
 type admitTestCase struct {
-	name        string
+	name            string
-	podsGetter  corev1lister.PodLister
+	podsGetter      corev1lister.PodLister
-	nodesGetter corev1lister.NodeLister
+	nodesGetter     corev1lister.NodeLister
-	attributes  admission.Attributes
+	csiDriverGetter storagelisters.CSIDriverLister
-	features    featuregate.FeatureGate
+	pvcGetter       corev1lister.PersistentVolumeClaimLister
-	setupFunc   func(t *testing.T)
+	pvGetter        corev1lister.PersistentVolumeLister
-	err         string
+	attributes      admission.Attributes
+	features        featuregate.FeatureGate
+	setupFunc       func(t *testing.T)
+	err             string
 }
 
 func (a *admitTestCase) run(t *testing.T) {
@@ -234,6 +238,9 @@ func (a *admitTestCase) run(t *testing.T) {
 		}
 		c.podsGetter = a.podsGetter
 		c.nodesGetter = a.nodesGetter
+		c.csiDriverGetter = a.csiDriverGetter
+		c.pvcGetter = a.pvcGetter
+		c.pvGetter = a.pvGetter
 		err := c.Admit(context.TODO(), a.attributes, nil)
 		if (err == nil) != (len(a.err) == 0) {
 			t.Errorf("nodePlugin.Admit() error = %v, expected %v", err, a.err)
@@ -387,8 +394,100 @@ func Test_nodePlugin_Admit(t *testing.T) {
 		abLabeledPod = withLabels(coremypod, labelsAB)
 
 		privKey, _ = rsa.GenerateKey(rand.Reader, 2048)
+
+		csiDriverWithAudience = &storagev1.CSIDriver{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "com.example.csi.mydriver",
+			},
+			Spec: storagev1.CSIDriverSpec{
+				TokenRequests: []storagev1.TokenRequest{
+					{
+						Audience: "foo",
+					},
+				},
+			},
+		}
+
+		csiDriverIndex  = cache.NewIndexer(cache.MetaNamespaceKeyFunc, nil)
+		csiDriverLister = storagelisters.NewCSIDriverLister(csiDriverIndex)
+
+		noexistingCSIDriverIndex  = cache.NewIndexer(cache.MetaNamespaceKeyFunc, nil)
+		noexistingCSIDriverLister = storagelisters.NewCSIDriverLister(noexistingCSIDriverIndex)
+
+		pvcWithCSIDriver = &corev1.PersistentVolumeClaim{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "pvclaim",
+				Namespace: "ns",
+			},
+			Spec: corev1.PersistentVolumeClaimSpec{
+				VolumeName: "pvname",
+			},
+		}
+
+		ephemeralVolumePVCWithCSIDriver = &corev1.PersistentVolumeClaim{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "myephemeralpod-myvol",
+				Namespace: "ns",
+			},
+			Spec: corev1.PersistentVolumeClaimSpec{
+				VolumeName: "pvname",
+			},
+		}
+
+		pvcIndex  = cache.NewIndexer(cache.MetaNamespaceKeyFunc, nil)
+		pvcLister = corev1lister.NewPersistentVolumeClaimLister(pvcIndex)
+
+		noexistingPVCIndex  = cache.NewIndexer(cache.MetaNamespaceKeyFunc, nil)
+		noexistingPVCLister = corev1lister.NewPersistentVolumeClaimLister(noexistingPVCIndex)
+
+		pvWithCSIDriver = &corev1.PersistentVolume{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "pvname",
+			},
+			Spec: corev1.PersistentVolumeSpec{
+				PersistentVolumeSource: corev1.PersistentVolumeSource{
+					CSI: &corev1.CSIPersistentVolumeSource{Driver: "com.example.csi.mydriver"},
+				},
+			},
+		}
+
+		pvIndex  = cache.NewIndexer(cache.MetaNamespaceKeyFunc, nil)
+		pvLister = corev1lister.NewPersistentVolumeLister(pvIndex)
+
+		noexistingPVIndex  = cache.NewIndexer(cache.MetaNamespaceKeyFunc, nil)
+		noexistingPVLister = corev1lister.NewPersistentVolumeLister(noexistingPVIndex)
 	)
 
+	// create pods for validating the service account node audience restriction
+	projectedVolumeSourceEmptyAudience := &corev1.ProjectedVolumeSource{Sources: []corev1.VolumeProjection{{ServiceAccountToken: &corev1.ServiceAccountTokenProjection{Audience: ""}}}}
+	projectedVolumeSource := &corev1.ProjectedVolumeSource{Sources: []corev1.VolumeProjection{{ServiceAccountToken: &corev1.ServiceAccountTokenProjection{Audience: "foo"}}}}
+	csiDriverVolumeSource := &corev1.CSIVolumeSource{Driver: "com.example.csi.mydriver"}
+	persistentVolumeClaimVolumeSource := &corev1.PersistentVolumeClaimVolumeSource{ClaimName: "pvclaim"}
+	ephemeralVolumeSource := &corev1.EphemeralVolumeSource{VolumeClaimTemplate: &corev1.PersistentVolumeClaimTemplate{}}
+
+	coremypodWithProjectedServiceAccountEmptyAudience, v1mypodWithProjectedServiceAccountEmptyAudience := makeTestPod("ns", "mysapod", "mynode", false)
+	v1mypodWithProjectedServiceAccountEmptyAudience.Spec.Volumes = []corev1.Volume{{VolumeSource: corev1.VolumeSource{Projected: projectedVolumeSourceEmptyAudience}}}
+
+	coremypodWithProjectedServiceAccount, v1mypodWithProjectedServiceAccount := makeTestPod("ns", "mysapod", "mynode", false)
+	v1mypodWithProjectedServiceAccount.Spec.Volumes = []corev1.Volume{{VolumeSource: corev1.VolumeSource{Projected: projectedVolumeSource}}}
+
+	coremypodWithCSI, v1mypodWithCSI := makeTestPod("ns", "mycsipod", "mynode", false)
+	v1mypodWithCSI.Spec.Volumes = []corev1.Volume{{VolumeSource: corev1.VolumeSource{CSI: csiDriverVolumeSource}}}
+
+	coremypodWithPVCRefCSI, v1mypodWithPVCRefCSI := makeTestPod("ns", "mypvcpod", "mynode", false)
+	v1mypodWithPVCRefCSI.Spec.Volumes = []corev1.Volume{{VolumeSource: corev1.VolumeSource{PersistentVolumeClaim: persistentVolumeClaimVolumeSource}}}
+
+	coremypodWithEphemeralVolume, v1mypodWithEphemeralVolume := makeTestPod("ns", "myephemeralpod", "mynode", false)
+	v1mypodWithEphemeralVolume.Spec.Volumes = []corev1.Volume{{Name: "myvol", VolumeSource: corev1.VolumeSource{Ephemeral: ephemeralVolumeSource}}}
+
+	coremypodWithPVCAndCSI, v1mypodWithPVCAndCSI := makeTestPod("ns", "mypvcandcsipod", "mynode", false)
+	v1mypodWithPVCAndCSI.Spec.Volumes = []corev1.Volume{{VolumeSource: corev1.VolumeSource{PersistentVolumeClaim: persistentVolumeClaimVolumeSource}}, {VolumeSource: corev1.VolumeSource{CSI: csiDriverVolumeSource}}}
+
+	checkNilError(t, csiDriverIndex.Add(csiDriverWithAudience))
+	checkNilError(t, pvcIndex.Add(pvcWithCSIDriver))
+	checkNilError(t, pvcIndex.Add(ephemeralVolumePVCWithCSIDriver))
+	checkNilError(t, pvIndex.Add(pvWithCSIDriver))
+
 	existingPodsIndex.Add(v1mymirrorpod)
 	existingPodsIndex.Add(v1othermirrorpod)
 	existingPodsIndex.Add(v1unboundmirrorpod)
@@ -396,6 +495,13 @@ func Test_nodePlugin_Admit(t *testing.T) {
 	existingPodsIndex.Add(v1otherpod)
 	existingPodsIndex.Add(v1unboundpod)
 
+	checkNilError(t, existingPodsIndex.Add(v1mypodWithProjectedServiceAccountEmptyAudience))
+	checkNilError(t, existingPodsIndex.Add(v1mypodWithProjectedServiceAccount))
+	checkNilError(t, existingPodsIndex.Add(v1mypodWithCSI))
+	checkNilError(t, existingPodsIndex.Add(v1mypodWithPVCRefCSI))
+	checkNilError(t, existingPodsIndex.Add(v1mypodWithEphemeralVolume))
+	checkNilError(t, existingPodsIndex.Add(v1mypodWithPVCAndCSI))
+
 	existingNodesIndex.Add(&corev1.Node{ObjectMeta: mynodeObjMeta})
 
 	sapod, _ := makeTestPod("ns", "mysapod", "mynode", true)
@@ -1089,31 +1195,240 @@ func Test_nodePlugin_Admit(t *testing.T) {
 		{
 			name:       "forbid create of unbound token",
 			podsGetter: noExistingPods,
-			attributes: admission.NewAttributesRecord(makeTokenRequest("", ""), nil, tokenrequestKind, "ns", "mysa", svcacctResource, "token", admission.Create, &metav1.CreateOptions{}, false, mynode),
+			attributes: admission.NewAttributesRecord(makeTokenRequest("", "", []string{"foo"}), nil, tokenrequestKind, "ns", "mysa", svcacctResource, "token", admission.Create, &metav1.CreateOptions{}, false, mynode),
 			err:        "not bound to a pod",
 		},
 		{
 			name:       "forbid create of token bound to nonexistant pod",
 			podsGetter: noExistingPods,
-			attributes: admission.NewAttributesRecord(makeTokenRequest("nopod", "someuid"), nil, tokenrequestKind, "ns", "mysa", svcacctResource, "token", admission.Create, &metav1.CreateOptions{}, false, mynode),
+			attributes: admission.NewAttributesRecord(makeTokenRequest("nopod", "someuid", []string{"foo"}), nil, tokenrequestKind, "ns", "mysa", svcacctResource, "token", admission.Create, &metav1.CreateOptions{}, false, mynode),
 			err:        "not found",
 		},
 		{
 			name:       "forbid create of token bound to pod without uid",
 			podsGetter: existingPods,
-			attributes: admission.NewAttributesRecord(makeTokenRequest(coremypod.Name, ""), nil, tokenrequestKind, "ns", "mysa", svcacctResource, "token", admission.Create, &metav1.CreateOptions{}, false, mynode),
+			attributes: admission.NewAttributesRecord(makeTokenRequest(coremypod.Name, "", []string{"foo"}), nil, tokenrequestKind, "ns", "mysa", svcacctResource, "token", admission.Create, &metav1.CreateOptions{}, false, mynode),
 			err:        "pod binding without a uid",
 		},
 		{
 			name:       "forbid create of token bound to pod scheduled on another node",
 			podsGetter: existingPods,
-			attributes: admission.NewAttributesRecord(makeTokenRequest(coreotherpod.Name, coreotherpod.UID), nil, tokenrequestKind, coreotherpod.Namespace, "mysa", svcacctResource, "token", admission.Create, &metav1.CreateOptions{}, false, mynode),
+			attributes: admission.NewAttributesRecord(makeTokenRequest(coreotherpod.Name, coreotherpod.UID, []string{"foo"}), nil, tokenrequestKind, coreotherpod.Namespace, "mysa", svcacctResource, "token", admission.Create, &metav1.CreateOptions{}, false, mynode),
 			err:        "pod scheduled on a different node",
 		},
 		{
 			name:       "allow create of token bound to pod scheduled this node",
 			podsGetter: existingPods,
-			attributes: admission.NewAttributesRecord(makeTokenRequest(coremypod.Name, coremypod.UID), nil, tokenrequestKind, coremypod.Namespace, "mysa", svcacctResource, "token", admission.Create, &metav1.CreateOptions{}, false, mynode),
+			attributes: admission.NewAttributesRecord(makeTokenRequest(coremypod.Name, coremypod.UID, []string{"foo"}), nil, tokenrequestKind, coremypod.Namespace, "mysa", svcacctResource, "token", admission.Create, &metav1.CreateOptions{}, false, mynode),
+		},
+		// Service accounts tests with token audience restrictions
+		{
+			name:       "allow create of token when audience in PSAT volume and ServiceAccountNodeAudienceRestriction is enabled, empty audience",
+			podsGetter: existingPods,
+			features:   feature.DefaultFeatureGate,
+			setupFunc: func(t *testing.T) {
+				t.Helper()
+				featuregatetesting.SetFeatureGateDuringTest(t, feature.DefaultFeatureGate, features.ServiceAccountNodeAudienceRestriction, true)
+			},
+			attributes: admission.NewAttributesRecord(makeTokenRequest(coremypodWithProjectedServiceAccountEmptyAudience.Name, coremypodWithProjectedServiceAccountEmptyAudience.UID, []string{"foo"}), nil, tokenrequestKind, coremypod.Namespace, "mysa", svcacctResource, "token", admission.Create, &metav1.CreateOptions{}, false, mynode),
+		},
+		{
+			name:       "allow create of token when audience in PSAT volume and ServiceAccountNodeAudienceRestriction is enabled, single audience",
+			podsGetter: existingPods,
+			features:   feature.DefaultFeatureGate,
+			setupFunc: func(t *testing.T) {
+				t.Helper()
+				featuregatetesting.SetFeatureGateDuringTest(t, feature.DefaultFeatureGate, features.ServiceAccountNodeAudienceRestriction, true)
+			},
+			attributes: admission.NewAttributesRecord(makeTokenRequest(coremypodWithProjectedServiceAccount.Name, coremypodWithProjectedServiceAccount.UID, []string{"foo"}), nil, tokenrequestKind, coremypod.Namespace, "mysa", svcacctResource, "token", admission.Create, &metav1.CreateOptions{}, false, mynode),
+		},
+		{
+			name:       "forbid create of token with multiple audiences in token request and ServiceAccountNodeAudienceRestriction is enabled",
+			podsGetter: existingPods,
+			features:   feature.DefaultFeatureGate,
+			setupFunc: func(t *testing.T) {
+				t.Helper()
+				featuregatetesting.SetFeatureGateDuringTest(t, feature.DefaultFeatureGate, features.ServiceAccountNodeAudienceRestriction, true)
+			},
+			attributes: admission.NewAttributesRecord(makeTokenRequest(coremypodWithProjectedServiceAccount.Name, coremypodWithProjectedServiceAccount.UID, []string{"foo", "bar"}), nil, tokenrequestKind, coremypod.Namespace, "mysa", svcacctResource, "token", admission.Create, &metav1.CreateOptions{}, false, mynode),
+			err:        "node may only request 0 or 1 audiences",
+		},
+		{
+			name:       "forbid create of token when audience not in pod spec and ServiceAccountNodeAudienceRestriction is enabled",
+			podsGetter: existingPods,
+			features:   feature.DefaultFeatureGate,
+			setupFunc: func(t *testing.T) {
+				t.Helper()
+				featuregatetesting.SetFeatureGateDuringTest(t, feature.DefaultFeatureGate, features.ServiceAccountNodeAudienceRestriction, true)
+			},
+			attributes: admission.NewAttributesRecord(makeTokenRequest(coremypod.Name, coremypod.UID, []string{"foo"}), nil, tokenrequestKind, coremypod.Namespace, "mysa", svcacctResource, "token", admission.Create, &metav1.CreateOptions{}, false, mynode),
+			err:        `serviceaccounts "mysa" is forbidden: audience "foo" not found in pod spec volume`,
+		},
+		{
+			name:            "allow create of token when audience in pod --> csi --> driver --> tokenrequest with audience and ServiceAccountNodeAudienceRestriction is enabled",
+			podsGetter:      existingPods,
+			csiDriverGetter: csiDriverLister,
+			features:        feature.DefaultFeatureGate,
+			setupFunc: func(t *testing.T) {
+				t.Helper()
+				featuregatetesting.SetFeatureGateDuringTest(t, feature.DefaultFeatureGate, features.ServiceAccountNodeAudienceRestriction, true)
+			},
+			attributes: admission.NewAttributesRecord(makeTokenRequest(coremypodWithCSI.Name, v1mypodWithCSI.UID, []string{"foo"}), nil, tokenrequestKind, coremypod.Namespace, "mysa", svcacctResource, "token", admission.Create, &metav1.CreateOptions{}, false, mynode),
+		},
+		{
+			name:            "forbid create of token when audience in pod --> csi --> driver --> tokenrequest does not have audience and ServiceAccountNodeAudienceRestriction is enabled",
+			podsGetter:      existingPods,
+			csiDriverGetter: csiDriverLister,
+			features:        feature.DefaultFeatureGate,
+			setupFunc: func(t *testing.T) {
+				t.Helper()
+				featuregatetesting.SetFeatureGateDuringTest(t, feature.DefaultFeatureGate, features.ServiceAccountNodeAudienceRestriction, true)
+			},
+			attributes: admission.NewAttributesRecord(makeTokenRequest(coremypodWithCSI.Name, v1mypodWithCSI.UID, []string{"bar"}), nil, tokenrequestKind, coremypod.Namespace, "mysa", svcacctResource, "token", admission.Create, &metav1.CreateOptions{}, false, mynode),
+			err:        `audience "bar" not found in pod spec volume`,
+		},
+		{
+			name:            "forbid create of token when audience in pod --> csi --> driver --> tokenrequest with audience and ServiceAccountNodeAudienceRestriction is enabled, csidriver not found",
+			podsGetter:      existingPods,
+			csiDriverGetter: noexistingCSIDriverLister,
+			features:        feature.DefaultFeatureGate,
+			setupFunc: func(t *testing.T) {
+				t.Helper()
+				featuregatetesting.SetFeatureGateDuringTest(t, feature.DefaultFeatureGate, features.ServiceAccountNodeAudienceRestriction, true)
+			},
+			attributes: admission.NewAttributesRecord(makeTokenRequest(coremypodWithCSI.Name, v1mypodWithCSI.UID, []string{"foo"}), nil, tokenrequestKind, coremypod.Namespace, "mysa", svcacctResource, "token", admission.Create, &metav1.CreateOptions{}, false, mynode),
+			err:        `error validating audience "foo": csidriver.storage.k8s.io "com.example.csi.mydriver" not found`,
+		},
+		{
+			name:            "allow create of token when audience in pod --> pvc --> pv --> csi --> driver --> tokenrequest with audience and ServiceAccountNodeAudienceRestriction is enabled",
+			podsGetter:      existingPods,
+			csiDriverGetter: csiDriverLister,
+			pvcGetter:       pvcLister,
+			pvGetter:        pvLister,
+			features:        feature.DefaultFeatureGate,
+			setupFunc: func(t *testing.T) {
+				t.Helper()
+				featuregatetesting.SetFeatureGateDuringTest(t, feature.DefaultFeatureGate, features.ServiceAccountNodeAudienceRestriction, true)
+			},
+			attributes: admission.NewAttributesRecord(makeTokenRequest(coremypodWithPVCRefCSI.Name, v1mypodWithPVCRefCSI.UID, []string{"foo"}), nil, tokenrequestKind, coremypod.Namespace, "mysa", svcacctResource, "token", admission.Create, &metav1.CreateOptions{}, false, mynode),
+		},
+		{
+			name:            "forbid create of token when audience in pod --> pvc --> pv --> csi --> driver --> tokenrequest does not have audience and ServiceAccountNodeAudienceRestriction is enabled",
+			podsGetter:      existingPods,
+			csiDriverGetter: csiDriverLister,
+			pvcGetter:       pvcLister,
+			pvGetter:        pvLister,
+			features:        feature.DefaultFeatureGate,
+			setupFunc: func(t *testing.T) {
+				t.Helper()
+				featuregatetesting.SetFeatureGateDuringTest(t, feature.DefaultFeatureGate, features.ServiceAccountNodeAudienceRestriction, true)
+			},
+			attributes: admission.NewAttributesRecord(makeTokenRequest(coremypodWithPVCRefCSI.Name, v1mypodWithPVCRefCSI.UID, []string{"bar"}), nil, tokenrequestKind, coremypod.Namespace, "mysa", svcacctResource, "token", admission.Create, &metav1.CreateOptions{}, false, mynode),
+			err:        `audience "bar" not found in pod spec volume`,
+		},
+		{
+			name:            "forbid create of token when audience in pod --> pvc --> pv --> csi --> driver --> tokenrequest with audience and ServiceAccountNodeAudienceRestriction is enabled, pvc not found",
+			podsGetter:      existingPods,
+			csiDriverGetter: csiDriverLister,
+			pvcGetter:       noexistingPVCLister,
+			pvGetter:        pvLister,
+			features:        feature.DefaultFeatureGate,
+			setupFunc: func(t *testing.T) {
+				t.Helper()
+				featuregatetesting.SetFeatureGateDuringTest(t, feature.DefaultFeatureGate, features.ServiceAccountNodeAudienceRestriction, true)
+			},
+			attributes: admission.NewAttributesRecord(makeTokenRequest(coremypodWithPVCRefCSI.Name, v1mypodWithPVCRefCSI.UID, []string{"foo"}), nil, tokenrequestKind, coremypod.Namespace, "mysa", svcacctResource, "token", admission.Create, &metav1.CreateOptions{}, false, mynode),
+			err:        `error validating audience "foo": persistentvolumeclaim "pvclaim" not found`,
+		},
+		{
+			name:            "forbid create of token when audience in pod --> pvc --> pv --> csi --> driver --> tokenrequest with audience and ServiceAccountNodeAudienceRestriction is enabled, pv not found",
+			podsGetter:      existingPods,
+			csiDriverGetter: csiDriverLister,
+			pvcGetter:       pvcLister,
+			pvGetter:        noexistingPVLister,
+			features:        feature.DefaultFeatureGate,
+			setupFunc: func(t *testing.T) {
+				t.Helper()
+				featuregatetesting.SetFeatureGateDuringTest(t, feature.DefaultFeatureGate, features.ServiceAccountNodeAudienceRestriction, true)
+			},
+			attributes: admission.NewAttributesRecord(makeTokenRequest(coremypodWithPVCRefCSI.Name, v1mypodWithPVCRefCSI.UID, []string{"foo"}), nil, tokenrequestKind, coremypod.Namespace, "mysa", svcacctResource, "token", admission.Create, &metav1.CreateOptions{}, false, mynode),
+			err:        `error validating audience "foo": persistentvolume "pvname" not found`,
+		},
+		{
+			name:            "allow create of token when audience in pod --> ephemeral --> pvc --> pv --> csi --> driver --> tokenrequest with audience and ServiceAccountNodeAudienceRestriction is enabled",
+			podsGetter:      existingPods,
+			csiDriverGetter: csiDriverLister,
+			pvcGetter:       pvcLister,
+			pvGetter:        pvLister,
+			features:        feature.DefaultFeatureGate,
+			setupFunc: func(t *testing.T) {
+				t.Helper()
+				featuregatetesting.SetFeatureGateDuringTest(t, feature.DefaultFeatureGate, features.ServiceAccountNodeAudienceRestriction, true)
+			},
+			attributes: admission.NewAttributesRecord(makeTokenRequest(coremypodWithEphemeralVolume.Name, v1mypodWithEphemeralVolume.UID, []string{"foo"}), nil, tokenrequestKind, coremypod.Namespace, "mysa", svcacctResource, "token", admission.Create, &metav1.CreateOptions{}, false, mynode),
+		},
+		{
+			name:            "forbid create of token when audience in pod --> ephemeral --> pvc --> pv --> csi --> driver --> tokenrequest does not have audience and ServiceAccountNodeAudienceRestriction is enabled",
+			podsGetter:      existingPods,
+			csiDriverGetter: csiDriverLister,
+			pvcGetter:       pvcLister,
+			pvGetter:        pvLister,
+			features:        feature.DefaultFeatureGate,
+			setupFunc: func(t *testing.T) {
+				t.Helper()
+				featuregatetesting.SetFeatureGateDuringTest(t, feature.DefaultFeatureGate, features.ServiceAccountNodeAudienceRestriction, true)
+			},
+			attributes: admission.NewAttributesRecord(makeTokenRequest(coremypodWithEphemeralVolume.Name, v1mypodWithEphemeralVolume.UID, []string{"bar"}), nil, tokenrequestKind, coremypod.Namespace, "mysa", svcacctResource, "token", admission.Create, &metav1.CreateOptions{}, false, mynode),
+			err:        `audience "bar" not found in pod spec volume`,
+		},
+		{
+			name:            "allow create of token when ServiceAccountNodeAudienceRestriction is disabled, pvc not found should not be checked",
+			podsGetter:      existingPods,
+			csiDriverGetter: csiDriverLister,
+			pvcGetter:       noexistingPVCLister,
+			features:        feature.DefaultFeatureGate,
+			setupFunc: func(t *testing.T) {
+				t.Helper()
+				featuregatetesting.SetFeatureGateDuringTest(t, feature.DefaultFeatureGate, features.ServiceAccountNodeAudienceRestriction, true)
+			},
+			attributes: admission.NewAttributesRecord(makeTokenRequest(coremypodWithPVCAndCSI.Name, v1mypodWithPVCAndCSI.UID, []string{"foo"}), nil, tokenrequestKind, coremypod.Namespace, "mysa", svcacctResource, "token", admission.Create, &metav1.CreateOptions{}, false, mynode),
+		},
+		{
+			name:            "allow create of token when audience in pod --> csi --> driver --> tokenrequest with audience and ServiceAccountNodeAudienceRestriction is enabled, pv not found",
+			podsGetter:      existingPods,
+			csiDriverGetter: csiDriverLister,
+			pvcGetter:       pvcLister,
+			pvGetter:        noexistingPVLister,
+			features:        feature.DefaultFeatureGate,
+			setupFunc: func(t *testing.T) {
+				t.Helper()
+				featuregatetesting.SetFeatureGateDuringTest(t, feature.DefaultFeatureGate, features.ServiceAccountNodeAudienceRestriction, true)
+			},
+			attributes: admission.NewAttributesRecord(makeTokenRequest(coremypodWithPVCAndCSI.Name, v1mypodWithPVCAndCSI.UID, []string{"foo"}), nil, tokenrequestKind, coremypod.Namespace, "mysa", svcacctResource, "token", admission.Create, &metav1.CreateOptions{}, false, mynode),
+		},
+		{
+			name:            "allow create of token when ServiceAccountNodeAudienceRestriction is disabled, pv not found should not be checked",
+			podsGetter:      existingPods,
+			csiDriverGetter: csiDriverLister,
+			pvcGetter:       pvcLister,
+			pvGetter:        noexistingPVLister,
+			features:        feature.DefaultFeatureGate,
+			setupFunc: func(t *testing.T) {
+				t.Helper()
+				featuregatetesting.SetFeatureGateDuringTest(t, feature.DefaultFeatureGate, features.ServiceAccountNodeAudienceRestriction, false)
+			},
+			attributes: admission.NewAttributesRecord(makeTokenRequest(coremypodWithPVCAndCSI.Name, v1mypodWithPVCAndCSI.UID, []string{"foo"}), nil, tokenrequestKind, coremypod.Namespace, "mysa", svcacctResource, "token", admission.Create, &metav1.CreateOptions{}, false, mynode),
+		},
+		{
+			name:            "allow create of token when ServiceAccountNodeAudienceRestriction is disabled, csidriver not found should not be checked",
+			podsGetter:      existingPods,
+			csiDriverGetter: noexistingCSIDriverLister,
+			features:        feature.DefaultFeatureGate,
+			setupFunc: func(t *testing.T) {
+				t.Helper()
+				featuregatetesting.SetFeatureGateDuringTest(t, feature.DefaultFeatureGate, features.ServiceAccountNodeAudienceRestriction, false)
+			},
+			attributes: admission.NewAttributesRecord(makeTokenRequest(coremypodWithCSI.Name, v1mypodWithCSI.UID, []string{"foo"}), nil, tokenrequestKind, coremypod.Namespace, "mysa", svcacctResource, "token", admission.Create, &metav1.CreateOptions{}, false, mynode),
 		},
 
 		// Unrelated objects
@@ -1839,3 +2154,10 @@ func TestAdmitResourceSlice(t *testing.T) {
 
 	}
 }
+
+func checkNilError(t *testing.T, err error) {
+	t.Helper()
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+}

test/featuregates_linter/test_data/versioned_feature_list.yaml

@@ -1108,6 +1108,12 @@
     lockToDefault: false
     preRelease: Beta
     version: "1.29"
+- name: ServiceAccountNodeAudienceRestriction
+  versionedSpecs:
+  - default: true
+    lockToDefault: false
+    preRelease: Beta
+    version: "1.32"
 - name: ServiceAccountTokenJTI
   versionedSpecs:
   - default: false

test/integration/auth/main_test.go

@@ -17,7 +17,9 @@ limitations under the License.
 package auth
 
 import (
+	"context"
 	"testing"
+	"time"
 
 	"k8s.io/kubernetes/test/integration/framework"
 )
@@ -25,3 +27,9 @@ import (
 func TestMain(m *testing.M) {
 	framework.EtcdMain(m.Run)
 }
+
+func testContext(t *testing.T) context.Context {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	t.Cleanup(cancel)
+	return ctx
+}

test/integration/auth/node_test.go

@@ -20,13 +20,16 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"path/filepath"
 	"strings"
 	"testing"
 	"time"
 
+	authenticationv1 "k8s.io/api/authentication/v1"
 	coordination "k8s.io/api/coordination/v1"
 	corev1 "k8s.io/api/core/v1"
 	policy "k8s.io/api/policy/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
 	resourceapi "k8s.io/api/resource/v1beta1"
 	storagev1 "k8s.io/api/storage/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -36,10 +39,14 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	clientset "k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
 	featuregatetesting "k8s.io/component-base/featuregate/testing"
 	kubeapiservertesting "k8s.io/kubernetes/cmd/kube-apiserver/app/testing"
+	kubecontrollermanagertesting "k8s.io/kubernetes/cmd/kube-controller-manager/app/testing"
 	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/test/integration/framework"
+	"k8s.io/kubernetes/test/utils/kubeconfig"
 	"k8s.io/utils/pointer"
 	"k8s.io/utils/ptr"
 )
@@ -83,17 +90,6 @@ func TestNodeAuthorizer(t *testing.T) {
 	clientConfig := server.ClientConfig
 	superuserClient, superuserClientExternal := clientsetForToken(tokenMaster, clientConfig)
 
-	// Wait for a healthy server
-	for {
-		result := superuserClient.CoreV1().RESTClient().Get().AbsPath("/healthz").Do(context.TODO())
-		_, err := result.Raw()
-		if err == nil {
-			break
-		}
-		t.Log(err)
-		time.Sleep(time.Second)
-	}
-
 	// Create objects
 	if _, err := superuserClient.CoreV1().Namespaces().Create(context.TODO(), &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "ns"}}, metav1.CreateOptions{}); err != nil {
 		t.Fatal(err)
@@ -629,7 +625,7 @@ func TestNodeAuthorizer(t *testing.T) {
 	// clean up node2
 	expectAllowed(t, deleteNode2(superuserClient))
 
-	//TODO(mikedanese): integration test node restriction of TokenRequest
+	// TODO(mikedanese): integration test node restriction of TokenRequest
 
 	// node1 allowed to operate on its own lease
 	expectAllowed(t, createNode1Lease(node1Client))
@@ -721,3 +717,433 @@ func expectAllowed(t *testing.T, f func() error) {
 		t.Errorf("Expected no error, got %v", err)
 	}
 }
+
+func checkNilError(t *testing.T, err error) {
+	t.Helper()
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+}
+
+func expectedForbiddenMessage(t *testing.T, f func() error, expectedMessage string) {
+	t.Helper()
+	if ok, err := expect(t, f, func(e error) bool { return apierrors.IsForbidden(e) && strings.Contains(e.Error(), expectedMessage) }); !ok {
+		t.Errorf("Expected forbidden error with message %q, got %v", expectedMessage, err)
+	}
+}
+
+// TestNodeRestrictionServiceAccount is an integration test to verify that
+// the NodeRestriction admission plugin
+// - forbids kubelet to request a token for a service account token that's not bound to a pod
+// - forbids kubelet to request a token for a service account token when pod is not found
+// - kubelet successfully requests a token for a service account token when pod is found and uid matches
+// This test is run with the ServiceAccountNodeAudienceRestriction feature disabled
+// to validate the default behavior of the NodeRestriction admission plugin.
+func TestNodeRestrictionServiceAccount(t *testing.T) {
+	const (
+		// Define credentials
+		// Fake values for testing.
+		tokenMaster = "master-token"
+		tokenNode1  = "node1-token"
+		tokenNode2  = "node2-token"
+	)
+
+	tokenFile, err := os.CreateTemp("", "kubeconfig")
+	checkNilError(t, err)
+
+	_, err = tokenFile.WriteString(strings.Join([]string{
+		fmt.Sprintf(`%s,admin,uid1,"system:masters"`, tokenMaster),
+		fmt.Sprintf(`%s,system:node:node1,uid3,"system:nodes"`, tokenNode1),
+		fmt.Sprintf(`%s,system:node:node2,uid4,"system:nodes"`, tokenNode2),
+	}, "\n"))
+	checkNilError(t, err)
+	checkNilError(t, tokenFile.Close())
+
+	featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ServiceAccountNodeAudienceRestriction, false)
+
+	server := kubeapiservertesting.StartTestServerOrDie(t, nil, []string{
+		"--runtime-config=api/all=true",
+		"--authorization-mode", "Node,RBAC",
+		"--token-auth-file", tokenFile.Name(),
+		"--enable-admission-plugins", "NodeRestriction",
+		"--disable-admission-plugins", "ServiceAccount,TaintNodesByCondition",
+	}, framework.SharedEtcd())
+	defer server.TearDownFn()
+
+	// Build client config and superuser clientset
+	clientConfig := server.ClientConfig
+	superuserClient, _ := clientsetForToken(tokenMaster, clientConfig)
+
+	if _, err := superuserClient.CoreV1().Namespaces().Create(context.TODO(), &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "ns"}}, metav1.CreateOptions{}); err != nil {
+		t.Fatal(err)
+	}
+
+	// RBAC permissions are required to test service account token requests
+	// using the node client because we cannot rely on the node authorizer since we have not
+	// configured the references needed for it to work.  This test is focused on exercising
+	// the node admission logic, not the node authorizer logic.  We want to know what happens
+	// in admission if the request was already authorized.
+	configureRBACForServiceAccountToken(t, superuserClient)
+
+	createTokenRequestNodeNotBoundToPod := func(client clientset.Interface) func() error {
+		return func() error {
+			_, err := client.CoreV1().ServiceAccounts("ns").CreateToken(context.TODO(), "default", &authenticationv1.TokenRequest{}, metav1.CreateOptions{})
+			return err
+		}
+	}
+
+	node1Client, _ := clientsetForToken(tokenNode1, clientConfig)
+	createNode(t, node1Client, "node1")
+	node2Client, _ := clientsetForToken(tokenNode2, clientConfig)
+
+	t.Run("service account token request is forbidden when not bound to a pod", func(t *testing.T) {
+		expectedForbiddenMessage(t, createTokenRequestNodeNotBoundToPod(node1Client), "node requested token not bound to a pod")
+	})
+
+	t.Run("service account token request is forbidden when pod is not found", func(t *testing.T) {
+		expectNotFound(t, createTokenRequest(node1Client, "uid1", ""))
+	})
+
+	t.Run("uid in token request does not match pod uid is forbidden", func(t *testing.T) {
+		createPod(t, superuserClient, nil)
+		expectedForbiddenMessage(t, createTokenRequest(node1Client, "random-uid", ""), "the UID in the bound object reference (random-uid) does not match the UID in record. The object might have been deleted and then recreated")
+		deletePod(t, superuserClient, "pod1")
+	})
+
+	t.Run("node requesting token for pod bound to different node is forbidden", func(t *testing.T) {
+		pod := createPod(t, superuserClient, nil)
+		expectedForbiddenMessage(t, createTokenRequest(node2Client, pod.UID, ""), "node requested token bound to a pod scheduled on a different node")
+		deletePod(t, superuserClient, "pod1")
+	})
+
+	t.Run("service account token request is successful", func(t *testing.T) {
+		// create a pod as an admin to add object references
+		pod := createPod(t, superuserClient, nil)
+		createDefaultServiceAccount(t, superuserClient)
+		expectAllowed(t, createTokenRequest(node1Client, pod.UID, ""))
+	})
+}
+
+// TestNodeRestrictionServiceAccountAudience is an integration test to verify that
+// the NodeRestriction admission plugin
+// - allows kubelet to request a token for a service account that's in the pod spec
+//  1. pod --> ephemeral --> pvc --> pv --> csi --> driver --> tokenrequest with audience
+//  2. pod --> pvc --> pv --> csi --> driver --> tokenrequest with audience
+//  3. pod --> csi --> driver --> tokenrequest with audience
+//  4. pod --> projected --> service account token with audience
+//
+// - forbids kubelet to request a token for a service account that's not in the pod spec
+// when the ServiceAccountNodeAudienceRestriction feature is enabled.
+func TestNodeRestrictionServiceAccountAudience(t *testing.T) {
+	const (
+		// Define credentials
+		// Fake values for testing.
+		tokenMaster = "master-token"
+		tokenNode1  = "node1-token"
+	)
+
+	tokenFile, err := os.CreateTemp("", "kubeconfig")
+	checkNilError(t, err)
+
+	_, err = tokenFile.WriteString(strings.Join([]string{
+		fmt.Sprintf(`%s,admin,uid1,"system:masters"`, tokenMaster),
+		fmt.Sprintf(`%s,system:node:node1,uid3,"system:nodes"`, tokenNode1),
+	}, "\n"))
+	checkNilError(t, err)
+	checkNilError(t, tokenFile.Close())
+
+	featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ServiceAccountNodeAudienceRestriction, true)
+
+	server := kubeapiservertesting.StartTestServerOrDie(t, nil, []string{
+		"--runtime-config=api/all=true",
+		"--authorization-mode", "Node,RBAC",
+		"--token-auth-file", tokenFile.Name(),
+		"--enable-admission-plugins", "NodeRestriction",
+		"--disable-admission-plugins", "TaintNodesByCondition",
+	}, framework.SharedEtcd())
+	defer server.TearDownFn()
+
+	kubeConfigFile := createKubeConfigFileForRestConfig(t, server.ClientConfig)
+
+	ctx := testContext(t)
+
+	kcm := kubecontrollermanagertesting.StartTestServerOrDie(ctx, []string{
+		"--kubeconfig=" + kubeConfigFile,
+		"--controllers=ephemeral-volume-controller", // we need this controller to test the ephemeral volume source in the pod
+		"--leader-elect=false",                      // KCM leader election calls os.Exit when it ends, so it is easier to just turn it off altogether
+	})
+	defer kcm.TearDownFn()
+
+	// Build client config and superuser clientset
+	clientConfig := server.ClientConfig
+	superuserClient, _ := clientsetForToken(tokenMaster, clientConfig)
+
+	if _, err := superuserClient.CoreV1().Namespaces().Create(context.TODO(), &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "ns"}}, metav1.CreateOptions{}); err != nil {
+		t.Fatal(err)
+	}
+
+	// RBAC permissions are required to test service account token requests
+	// using the node client because we cannot rely on the node authorizer since we have not
+	// configured the references needed for it to work.  This test is focused on exercising
+	// the node admission logic, not the node authorizer logic.  We want to know what happens
+	// in admission if the request was already authorized.
+	configureRBACForServiceAccountToken(t, superuserClient)
+
+	_, err = superuserClient.CoreV1().PersistentVolumeClaims("ns").Create(context.TODO(), &corev1.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{Name: "mypvc"},
+		Spec: corev1.PersistentVolumeClaimSpec{
+			AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadOnlyMany},
+			Resources:   corev1.VolumeResourceRequirements{Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("1")}},
+			VolumeName:  "mypv",
+		},
+	}, metav1.CreateOptions{})
+	checkNilError(t, err)
+
+	_, err = superuserClient.CoreV1().PersistentVolumes().Create(context.TODO(), &corev1.PersistentVolume{
+		ObjectMeta: metav1.ObjectMeta{Name: "mypv"},
+		Spec: corev1.PersistentVolumeSpec{
+			AccessModes:            []corev1.PersistentVolumeAccessMode{corev1.ReadOnlyMany},
+			Capacity:               corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("1")},
+			ClaimRef:               &corev1.ObjectReference{Namespace: "ns", Name: "mypvc"},
+			PersistentVolumeSource: corev1.PersistentVolumeSource{CSI: &corev1.CSIPersistentVolumeSource{Driver: "com.example.csi.mydriver", VolumeHandle: "handle"}},
+		},
+	}, metav1.CreateOptions{})
+	checkNilError(t, err)
+
+	node1Client, _ := clientsetForToken(tokenNode1, clientConfig)
+	createNode(t, node1Client, "node1")
+	createDefaultServiceAccount(t, superuserClient)
+
+	t.Run("projected volume source with empty audience works", func(t *testing.T) {
+		projectedVolumeSourceEmptyAudience := &corev1.ProjectedVolumeSource{Sources: []corev1.VolumeProjection{{ServiceAccountToken: &corev1.ServiceAccountTokenProjection{Audience: "", Path: "path"}}}}
+		pod := createPod(t, superuserClient, []corev1.Volume{{Name: "foo", VolumeSource: corev1.VolumeSource{Projected: projectedVolumeSourceEmptyAudience}}})
+		expectAllowed(t, createTokenRequest(node1Client, pod.UID, ""))
+		deletePod(t, superuserClient, "pod1")
+	})
+
+	t.Run("projected volume source with non-empty audience works", func(t *testing.T) {
+		projectedVolumeSource := &corev1.ProjectedVolumeSource{Sources: []corev1.VolumeProjection{{ServiceAccountToken: &corev1.ServiceAccountTokenProjection{Audience: "projected-audience", Path: "path"}}}}
+		pod := createPod(t, superuserClient, []corev1.Volume{{Name: "foo", VolumeSource: corev1.VolumeSource{Projected: projectedVolumeSource}}})
+		expectAllowed(t, createTokenRequest(node1Client, pod.UID, "projected-audience"))
+		deletePod(t, superuserClient, "pod1")
+	})
+
+	t.Run("pod --> csi --> driver --> tokenrequest with audience forbidden - CSI driver not found", func(t *testing.T) {
+		csiDriverVolumeSource := &corev1.CSIVolumeSource{Driver: "com.example.csi.mydriver"}
+		pod := createPod(t, superuserClient, []corev1.Volume{{Name: "foo", VolumeSource: corev1.VolumeSource{CSI: csiDriverVolumeSource}}})
+		expectedForbiddenMessage(t, createTokenRequest(node1Client, pod.UID, "csidrivernotfound-audience"), `error validating audience "csidrivernotfound-audience": csidriver.storage.k8s.io "com.example.csi.mydriver" not found`)
+		deletePod(t, superuserClient, "pod1")
+	})
+
+	t.Run("pod --> csi --> driver --> tokenrequest with audience works", func(t *testing.T) {
+		createCSIDriver(t, superuserClient, "csidriver-audience")
+		csiDriverVolumeSource := &corev1.CSIVolumeSource{Driver: "com.example.csi.mydriver"}
+		pod := createPod(t, superuserClient, []corev1.Volume{{Name: "foo", VolumeSource: corev1.VolumeSource{CSI: csiDriverVolumeSource}}})
+		expectAllowed(t, createTokenRequest(node1Client, pod.UID, "csidriver-audience"))
+		deletePod(t, superuserClient, "pod1")
+		deleteCSIDriver(t, superuserClient)
+	})
+
+	t.Run("pod --> pvc --> pv --> csi --> driver --> tokenrequest with audience forbidden - CSI driver not found", func(t *testing.T) {
+		persistentVolumeClaimVolumeSource := &corev1.PersistentVolumeClaimVolumeSource{ClaimName: "mypvc"}
+		pod := createPod(t, superuserClient, []corev1.Volume{{Name: "foo", VolumeSource: corev1.VolumeSource{PersistentVolumeClaim: persistentVolumeClaimVolumeSource}}})
+		expectedForbiddenMessage(t, createTokenRequest(node1Client, pod.UID, "pvc-csidrivernotfound-audience"), `error validating audience "pvc-csidrivernotfound-audience": csidriver.storage.k8s.io "com.example.csi.mydriver" not found`)
+		deletePod(t, superuserClient, "pod1")
+	})
+
+	t.Run("pod --> pvc --> pv --> csi --> driver --> tokenrequest with audience forbidden - pvc not found", func(t *testing.T) {
+		createCSIDriver(t, superuserClient, "pvcnotfound-audience")
+		persistentVolumeClaimVolumeSource := &corev1.PersistentVolumeClaimVolumeSource{ClaimName: "mypvc1"}
+		pod := createPod(t, superuserClient, []corev1.Volume{{Name: "foo", VolumeSource: corev1.VolumeSource{PersistentVolumeClaim: persistentVolumeClaimVolumeSource}}})
+		expectedForbiddenMessage(t, createTokenRequest(node1Client, pod.UID, "pvcnotfound-audience"), `error validating audience "pvcnotfound-audience": persistentvolumeclaim "mypvc1" not found`)
+		deletePod(t, superuserClient, "pod1")
+		deleteCSIDriver(t, superuserClient)
+	})
+
+	t.Run("pod --> pvc --> pv --> csi --> driver --> tokenrequest with audience works", func(t *testing.T) {
+		createCSIDriver(t, superuserClient, "pvccsidriver-audience")
+		persistentVolumeClaimVolumeSource := &corev1.PersistentVolumeClaimVolumeSource{ClaimName: "mypvc"}
+		pod := createPod(t, superuserClient, []corev1.Volume{{Name: "foo", VolumeSource: corev1.VolumeSource{PersistentVolumeClaim: persistentVolumeClaimVolumeSource}}})
+		expectAllowed(t, createTokenRequest(node1Client, pod.UID, "pvccsidriver-audience"))
+		deletePod(t, superuserClient, "pod1")
+		deleteCSIDriver(t, superuserClient)
+	})
+
+	t.Run("pod --> ephemeral --> pvc --> pv --> csi --> driver --> tokenrequest with audience forbidden - CSI driver not found", func(t *testing.T) {
+		ephemeralVolumeSource := &corev1.EphemeralVolumeSource{VolumeClaimTemplate: &corev1.PersistentVolumeClaimTemplate{
+			Spec: corev1.PersistentVolumeClaimSpec{
+				AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadOnlyMany},
+				Resources:   corev1.VolumeResourceRequirements{Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("1")}},
+				VolumeName:  "mypv",
+			}}}
+		pod := createPod(t, superuserClient, []corev1.Volume{{Name: "foo", VolumeSource: corev1.VolumeSource{Ephemeral: ephemeralVolumeSource}}})
+		expectedForbiddenMessage(t, createTokenRequest(node1Client, pod.UID, "ephemeral-csidrivernotfound-audience"), `error validating audience "ephemeral-csidrivernotfound-audience": csidriver.storage.k8s.io "com.example.csi.mydriver" not found`)
+		deletePod(t, superuserClient, "pod1")
+	})
+
+	t.Run("pod --> ephemeral --> pvc --> pv --> csi --> driver --> tokenrequest with audience works", func(t *testing.T) {
+		createCSIDriver(t, superuserClient, "ephemeralcsidriver-audience")
+		ephemeralVolumeSource := &corev1.EphemeralVolumeSource{VolumeClaimTemplate: &corev1.PersistentVolumeClaimTemplate{
+			Spec: corev1.PersistentVolumeClaimSpec{
+				AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadOnlyMany},
+				Resources:   corev1.VolumeResourceRequirements{Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("1")}},
+				VolumeName:  "mypv",
+			}}}
+		pod := createPod(t, superuserClient, []corev1.Volume{{Name: "foo", VolumeSource: corev1.VolumeSource{Ephemeral: ephemeralVolumeSource}}})
+		expectAllowed(t, createTokenRequest(node1Client, pod.UID, "ephemeralcsidriver-audience"))
+		deletePod(t, superuserClient, "pod1")
+		deleteCSIDriver(t, superuserClient)
+	})
+
+	t.Run("csidriver exists but tokenrequest audience not found should be forbidden", func(t *testing.T) {
+		createCSIDriver(t, superuserClient, "csidriver-audience")
+		pod := createPod(t, superuserClient, nil)
+		expectedForbiddenMessage(t, createTokenRequest(node1Client, pod.UID, "csidriver-audience-not-found"), `audience "csidriver-audience-not-found" not found in pod spec volume`)
+		deletePod(t, superuserClient, "pod1")
+		deleteCSIDriver(t, superuserClient)
+	})
+
+	t.Run("pvc and csidriver exists but tokenrequest audience not found should be forbidden", func(t *testing.T) {
+		createCSIDriver(t, superuserClient, "csidriver-audience")
+		persistentVolumeClaimVolumeSource := &corev1.PersistentVolumeClaimVolumeSource{ClaimName: "mypvc"}
+		pod := createPod(t, superuserClient, []corev1.Volume{{Name: "foo", VolumeSource: corev1.VolumeSource{PersistentVolumeClaim: persistentVolumeClaimVolumeSource}}})
+		expectedForbiddenMessage(t, createTokenRequest(node1Client, pod.UID, "csidriver-audience-not-found"), `audience "csidriver-audience-not-found" not found in pod spec volume`)
+		deletePod(t, superuserClient, "pod1")
+		deleteCSIDriver(t, superuserClient)
+	})
+
+	t.Run("ephemeral volume source with audience not found should be forbidden", func(t *testing.T) {
+		createCSIDriver(t, superuserClient, "csidriver-audience")
+		ephemeralVolumeSource := &corev1.EphemeralVolumeSource{VolumeClaimTemplate: &corev1.PersistentVolumeClaimTemplate{
+			Spec: corev1.PersistentVolumeClaimSpec{
+				AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadOnlyMany},
+				Resources:   corev1.VolumeResourceRequirements{Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("1")}},
+				VolumeName:  "mypv",
+			}}}
+		pod := createPod(t, superuserClient, []corev1.Volume{{Name: "foo", VolumeSource: corev1.VolumeSource{Ephemeral: ephemeralVolumeSource}}})
+		expectedForbiddenMessage(t, createTokenRequest(node1Client, pod.UID, "csidriver-audience-not-found"), `audience "csidriver-audience-not-found" not found in pod spec volume`)
+		deletePod(t, superuserClient, "pod1")
+		deleteCSIDriver(t, superuserClient)
+	})
+
+	t.Run("token request with multiple audiences should be forbidden", func(t *testing.T) {
+		pod := createPod(t, superuserClient, nil)
+		expectedForbiddenMessage(t, createTokenRequest(node1Client, pod.UID, "audience1", "audience2"), "node may only request 0 or 1 audiences")
+		deletePod(t, superuserClient, "pod1")
+	})
+}
+
+func createKubeConfigFileForRestConfig(t *testing.T, restConfig *rest.Config) string {
+	t.Helper()
+
+	clientConfig := kubeconfig.CreateKubeConfig(restConfig)
+
+	kubeConfigFile := filepath.Join(t.TempDir(), "kubeconfig.yaml")
+	if err := clientcmd.WriteToFile(*clientConfig, kubeConfigFile); err != nil {
+		t.Fatal(err)
+	}
+	return kubeConfigFile
+}
+
+func createPod(t *testing.T, client clientset.Interface, volumes []corev1.Volume) *corev1.Pod {
+	t.Helper()
+	pod, err := client.CoreV1().Pods("ns").Create(context.TODO(), &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "pod1",
+		},
+		Spec: corev1.PodSpec{
+			NodeName:           "node1",
+			Containers:         []corev1.Container{{Name: "image", Image: "busybox"}},
+			ServiceAccountName: "default",
+			Volumes:            volumes,
+		},
+	}, metav1.CreateOptions{})
+	checkNilError(t, err)
+	return pod
+}
+
+func deletePod(t *testing.T, client clientset.Interface, podName string) {
+	t.Helper()
+
+	checkNilError(t, client.CoreV1().Pods("ns").Delete(context.TODO(), podName, metav1.DeleteOptions{GracePeriodSeconds: ptr.To[int64](0)}))
+}
+
+func createNode(t *testing.T, client clientset.Interface, nodeName string) {
+	t.Helper()
+
+	_, err := client.CoreV1().Nodes().Create(context.TODO(), &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: nodeName}}, metav1.CreateOptions{})
+	checkNilError(t, err)
+}
+
+func createTokenRequest(client clientset.Interface, uid types.UID, audiences ...string) func() error {
+	return func() error {
+		_, err := client.CoreV1().ServiceAccounts("ns").CreateToken(context.TODO(), "default", &authenticationv1.TokenRequest{
+			Spec: authenticationv1.TokenRequestSpec{
+				BoundObjectRef: &authenticationv1.BoundObjectReference{
+					Kind:       "Pod",
+					Name:       "pod1",
+					APIVersion: "v1",
+					UID:        uid,
+				},
+				Audiences: audiences,
+			},
+		}, metav1.CreateOptions{})
+		return err
+	}
+}
+
+func createCSIDriver(t *testing.T, client clientset.Interface, audience string) {
+	t.Helper()
+
+	_, err := client.StorageV1().CSIDrivers().Create(context.TODO(), &storagev1.CSIDriver{
+		ObjectMeta: metav1.ObjectMeta{Name: "com.example.csi.mydriver"},
+		Spec: storagev1.CSIDriverSpec{
+			TokenRequests: []storagev1.TokenRequest{{Audience: audience}},
+		},
+	}, metav1.CreateOptions{})
+	checkNilError(t, err)
+}
+
+func deleteCSIDriver(t *testing.T, client clientset.Interface) {
+	t.Helper()
+
+	checkNilError(t, client.StorageV1().CSIDrivers().Delete(context.TODO(), "com.example.csi.mydriver", metav1.DeleteOptions{GracePeriodSeconds: ptr.To[int64](0)}))
+}
+
+func createDefaultServiceAccount(t *testing.T, client clientset.Interface) {
+	t.Helper()
+
+	_, err := client.CoreV1().ServiceAccounts("ns").Create(context.TODO(), &corev1.ServiceAccount{
+		ObjectMeta: metav1.ObjectMeta{Name: "default"},
+	}, metav1.CreateOptions{})
+	checkNilError(t, err)
+}
+
+func configureRBACForServiceAccountToken(t *testing.T, client clientset.Interface) {
+	t.Helper()
+
+	_, err := client.RbacV1().ClusterRoles().Update(context.TODO(), &rbacv1.ClusterRole{
+		ObjectMeta: metav1.ObjectMeta{Name: "system:node:node1"},
+		Rules: []rbacv1.PolicyRule{{
+			APIGroups: []string{""},
+			Resources: []string{"serviceaccounts/token"},
+			Verbs:     []string{"create"},
+		}},
+	}, metav1.UpdateOptions{})
+	checkNilError(t, err)
+
+	_, err = client.RbacV1().ClusterRoleBindings().Update(context.TODO(), &rbacv1.ClusterRoleBinding{
+		ObjectMeta: metav1.ObjectMeta{Name: "system:node"},
+		Subjects: []rbacv1.Subject{{
+			APIGroup: rbacv1.GroupName,
+			Kind:     "Group",
+			Name:     "system:nodes",
+		}},
+		RoleRef: rbacv1.RoleRef{
+			APIGroup: rbacv1.GroupName,
+			Kind:     "ClusterRole",
+			Name:     "system:node",
+		},
+	}, metav1.UpdateOptions{})
+	checkNilError(t, err)
+}