KEP-5284: Implement Constrained Impersonation

Signed-off-by: Jian Qiu <jqiu@redhat.com>
Signed-off-by: Monis Khan <mok@microsoft.com>

Co-authored-by: Jian Qiu <jqiu@redhat.com>
Co-authored-by: Monis Khan <mok@microsoft.com>

pkg/features/kube_features.go

@@ -1872,6 +1872,10 @@ var defaultVersionedKubernetesFeatureGates = map[featuregate.Feature]featuregate
 		{Version: version.MustParse("1.34"), Default: true, PreRelease: featuregate.GA, LockToDefault: true},
 	},
 
+	genericfeatures.ConstrainedImpersonation: {
+		{Version: version.MustParse("1.35"), Default: false, PreRelease: featuregate.Alpha},
+	},
+
 	genericfeatures.CoordinatedLeaderElection: {
 		{Version: version.MustParse("1.31"), Default: false, PreRelease: featuregate.Alpha},
 		{Version: version.MustParse("1.33"), Default: false, PreRelease: featuregate.Beta},
@@ -2346,6 +2350,8 @@ var defaultKubernetesFeatureGateDependencies = map[featuregate.Feature][]feature
 
 	genericfeatures.ConsistentListFromCache: {},
 
+	genericfeatures.ConstrainedImpersonation: {},
+
 	genericfeatures.DeclarativeValidation: {},
 
 	genericfeatures.DeclarativeValidationTakeover: {genericfeatures.DeclarativeValidation},

pkg/generated/openapi/zz_generated.openapi.go

@@ -1331,6 +1331,7 @@ func GetOpenAPIDefinitions(ref common.ReferenceCallback) map[string]common.OpenA
 		runtime.Unknown{}.OpenAPIModelName():                                                                            schema_k8sio_apimachinery_pkg_runtime_Unknown(ref),
 		intstr.IntOrString{}.OpenAPIModelName():                                                                         schema_apimachinery_pkg_util_intstr_IntOrString(ref),
 		version.Info{}.OpenAPIModelName():                                                                               schema_k8sio_apimachinery_pkg_version_Info(ref),
+		auditv1.AuthenticationMetadata{}.OpenAPIModelName():                                                             schema_pkg_apis_audit_v1_AuthenticationMetadata(ref),
 		auditv1.Event{}.OpenAPIModelName():                                                                              schema_pkg_apis_audit_v1_Event(ref),
 		auditv1.EventList{}.OpenAPIModelName():                                                                          schema_pkg_apis_audit_v1_EventList(ref),
 		auditv1.GroupResources{}.OpenAPIModelName():                                                                     schema_pkg_apis_audit_v1_GroupResources(ref),
@@ -64283,6 +64284,25 @@ func schema_k8sio_apimachinery_pkg_version_Info(ref common.ReferenceCallback) co
 	}
 }
 
+func schema_pkg_apis_audit_v1_AuthenticationMetadata(ref common.ReferenceCallback) common.OpenAPIDefinition {
+	return common.OpenAPIDefinition{
+		Schema: spec.Schema{
+			SchemaProps: spec.SchemaProps{
+				Type: []string{"object"},
+				Properties: map[string]spec.Schema{
+					"impersonationConstraint": {
+						SchemaProps: spec.SchemaProps{
+							Description: "ImpersonationConstraint is the verb associated with the constrained impersonation mode that was used to authorize the ImpersonatedUser associated with this audit event.  It is only set when constrained impersonation was used.",
+							Type:        []string{"string"},
+							Format:      "",
+						},
+					},
+				},
+			},
+		},
+	}
+}
+
 func schema_pkg_apis_audit_v1_Event(ref common.ReferenceCallback) common.OpenAPIDefinition {
 	return common.OpenAPIDefinition{
 		Schema: spec.Schema{
@@ -64357,6 +64377,12 @@ func schema_pkg_apis_audit_v1_Event(ref common.ReferenceCallback) common.OpenAPI
 							Ref:         ref(authenticationv1.UserInfo{}.OpenAPIModelName()),
 						},
 					},
+					"authenticationMetadata": {
+						SchemaProps: spec.SchemaProps{
+							Description: "AuthenticationMetadata contains details about how the request was authenticated.",
+							Ref:         ref(auditv1.AuthenticationMetadata{}.OpenAPIModelName()),
+						},
+					},
 					"sourceIPs": {
 						VendorExtensible: spec.VendorExtensible{
 							Extensions: spec.Extensions{
@@ -64441,7 +64467,7 @@ func schema_pkg_apis_audit_v1_Event(ref common.ReferenceCallback) common.OpenAPI
 			},
 		},
 		Dependencies: []string{
-			authenticationv1.UserInfo{}.OpenAPIModelName(), metav1.MicroTime{}.OpenAPIModelName(), metav1.Status{}.OpenAPIModelName(), runtime.Unknown{}.OpenAPIModelName(), auditv1.ObjectReference{}.OpenAPIModelName()},
+			authenticationv1.UserInfo{}.OpenAPIModelName(), metav1.MicroTime{}.OpenAPIModelName(), metav1.Status{}.OpenAPIModelName(), runtime.Unknown{}.OpenAPIModelName(), auditv1.AuthenticationMetadata{}.OpenAPIModelName(), auditv1.ObjectReference{}.OpenAPIModelName()},
 	}
 }
 

staging/src/k8s.io/apiserver/pkg/apis/audit/types.go

@@ -97,6 +97,10 @@ type Event struct {
 	// Impersonated user information.
 	// +optional
 	ImpersonatedUser *authnv1.UserInfo
+	// AuthenticationMetadata contains details about how the request was authenticated.
+	// +optional
+	AuthenticationMetadata *AuthenticationMetadata
+
 	// Source IPs, from where the request originated and intermediate proxies.
 	// The source IPs are listed from (in order):
 	// 1. X-Forwarded-For request header IPs
@@ -147,6 +151,13 @@ type Event struct {
 	Annotations map[string]string
 }
 
+type AuthenticationMetadata struct {
+	// ImpersonationConstraint is the verb associated with the constrained impersonation mode that was used to authorize
+	// the ImpersonatedUser associated with this audit event.  It is only set when constrained impersonation was used.
+	// +optional
+	ImpersonationConstraint string
+}
+
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
 
 // EventList is a list of audit Events.

staging/src/k8s.io/apiserver/pkg/apis/audit/v1/generated.pb.go

@@ -36,6 +36,8 @@ import (
 	k8s_io_apimachinery_pkg_types "k8s.io/apimachinery/pkg/types"
 )
 
+func (m *AuthenticationMetadata) Reset() { *m = AuthenticationMetadata{} }
+
 func (m *Event) Reset() { *m = Event{} }
 
 func (m *EventList) Reset() { *m = EventList{} }
@@ -50,6 +52,34 @@ func (m *PolicyList) Reset() { *m = PolicyList{} }
 
 func (m *PolicyRule) Reset() { *m = PolicyRule{} }
 
+func (m *AuthenticationMetadata) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *AuthenticationMetadata) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *AuthenticationMetadata) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	i -= len(m.ImpersonationConstraint)
+	copy(dAtA[i:], m.ImpersonationConstraint)
+	i = encodeVarintGenerated(dAtA, i, uint64(len(m.ImpersonationConstraint)))
+	i--
+	dAtA[i] = 0xa
+	return len(dAtA) - i, nil
+}
+
 func (m *Event) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
@@ -70,6 +100,20 @@ func (m *Event) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	_ = i
 	var l int
 	_ = l
+	if m.AuthenticationMetadata != nil {
+		{
+			size, err := m.AuthenticationMetadata.MarshalToSizedBuffer(dAtA[:i])
+			if err != nil {
+				return 0, err
+			}
+			i -= size
+			i = encodeVarintGenerated(dAtA, i, uint64(size))
+		}
+		i--
+		dAtA[i] = 0x1
+		i--
+		dAtA[i] = 0x8a
+	}
 	i -= len(m.UserAgent)
 	copy(dAtA[i:], m.UserAgent)
 	i = encodeVarintGenerated(dAtA, i, uint64(len(m.UserAgent)))
@@ -612,6 +656,17 @@ func encodeVarintGenerated(dAtA []byte, offset int, v uint64) int {
 	dAtA[offset] = uint8(v)
 	return base
 }
+func (m *AuthenticationMetadata) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	l = len(m.ImpersonationConstraint)
+	n += 1 + l + sovGenerated(uint64(l))
+	return n
+}
+
 func (m *Event) Size() (n int) {
 	if m == nil {
 		return 0
@@ -670,6 +725,10 @@ func (m *Event) Size() (n int) {
 	}
 	l = len(m.UserAgent)
 	n += 2 + l + sovGenerated(uint64(l))
+	if m.AuthenticationMetadata != nil {
+		l = m.AuthenticationMetadata.Size()
+		n += 2 + l + sovGenerated(uint64(l))
+	}
 	return n
 }
 
@@ -841,6 +900,16 @@ func sovGenerated(x uint64) (n int) {
 func sozGenerated(x uint64) (n int) {
 	return sovGenerated(uint64((x << 1) ^ uint64((int64(x) >> 63))))
 }
+func (this *AuthenticationMetadata) String() string {
+	if this == nil {
+		return "nil"
+	}
+	s := strings.Join([]string{`&AuthenticationMetadata{`,
+		`ImpersonationConstraint:` + fmt.Sprintf("%v", this.ImpersonationConstraint) + `,`,
+		`}`,
+	}, "")
+	return s
+}
 func (this *Event) String() string {
 	if this == nil {
 		return "nil"
@@ -872,6 +941,7 @@ func (this *Event) String() string {
 		`StageTimestamp:` + strings.Replace(strings.Replace(fmt.Sprintf("%v", this.StageTimestamp), "MicroTime", "v11.MicroTime", 1), `&`, ``, 1) + `,`,
 		`Annotations:` + mapStringForAnnotations + `,`,
 		`UserAgent:` + fmt.Sprintf("%v", this.UserAgent) + `,`,
+		`AuthenticationMetadata:` + strings.Replace(this.AuthenticationMetadata.String(), "AuthenticationMetadata", "AuthenticationMetadata", 1) + `,`,
 		`}`,
 	}, "")
 	return s
@@ -986,6 +1056,88 @@ func valueToStringGenerated(v interface{}) string {
 	pv := reflect.Indirect(rv).Interface()
 	return fmt.Sprintf("*%v", pv)
 }
+func (m *AuthenticationMetadata) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowGenerated
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: AuthenticationMetadata: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: AuthenticationMetadata: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field ImpersonationConstraint", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowGenerated
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				stringLen |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthGenerated
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return ErrInvalidLengthGenerated
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.ImpersonationConstraint = string(dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
+		default:
+			iNdEx = preIndex
+			skippy, err := skipGenerated(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
+				return ErrInvalidLengthGenerated
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
 func (m *Event) Unmarshal(dAtA []byte) error {
 	l := len(dAtA)
 	iNdEx := 0
@@ -1645,6 +1797,42 @@ func (m *Event) Unmarshal(dAtA []byte) error {
 			}
 			m.UserAgent = string(dAtA[iNdEx:postIndex])
 			iNdEx = postIndex
+		case 17:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field AuthenticationMetadata", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowGenerated
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthGenerated
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthGenerated
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if m.AuthenticationMetadata == nil {
+				m.AuthenticationMetadata = &AuthenticationMetadata{}
+			}
+			if err := m.AuthenticationMetadata.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
 		default:
 			iNdEx = preIndex
 			skippy, err := skipGenerated(dAtA[iNdEx:])

staging/src/k8s.io/apiserver/pkg/apis/audit/v1/generated.proto

@@ -29,6 +29,13 @@ import "k8s.io/apimachinery/pkg/runtime/schema/generated.proto";
 // Package-wide variables from generator "generated".
 option go_package = "k8s.io/apiserver/pkg/apis/audit/v1";
 
+message AuthenticationMetadata {
+  // ImpersonationConstraint is the verb associated with the constrained impersonation mode that was used to authorize
+  // the ImpersonatedUser associated with this audit event.  It is only set when constrained impersonation was used.
+  // +optional
+  optional string impersonationConstraint = 1;
+}
+
 // Event captures all the information that can be included in an API audit log.
 message Event {
   // AuditLevel at which event was generated
@@ -54,6 +61,10 @@ message Event {
   // +optional
   optional .k8s.io.api.authentication.v1.UserInfo impersonatedUser = 7;
 
+  // AuthenticationMetadata contains details about how the request was authenticated.
+  // +optional
+  optional AuthenticationMetadata authenticationMetadata = 17;
+
   // Source IPs, from where the request originated and intermediate proxies.
   // The source IPs are listed from (in order):
   // 1. X-Forwarded-For request header IPs

staging/src/k8s.io/apiserver/pkg/apis/audit/v1/generated.protomessage.pb.go

@@ -21,6 +21,8 @@ limitations under the License.
 
 package v1
 
+func (*AuthenticationMetadata) ProtoMessage() {}
+
 func (*Event) ProtoMessage() {}
 
 func (*EventList) ProtoMessage() {}

staging/src/k8s.io/apiserver/pkg/apis/audit/v1/types.go

@@ -90,6 +90,9 @@ type Event struct {
 	// Impersonated user information.
 	// +optional
 	ImpersonatedUser *authnv1.UserInfo `json:"impersonatedUser,omitempty" protobuf:"bytes,7,opt,name=impersonatedUser"`
+	// AuthenticationMetadata contains details about how the request was authenticated.
+	// +optional
+	AuthenticationMetadata *AuthenticationMetadata `json:"authenticationMetadata,omitempty" protobuf:"bytes,17,opt,name=authenticationMetadata"`
 	// Source IPs, from where the request originated and intermediate proxies.
 	// The source IPs are listed from (in order):
 	// 1. X-Forwarded-For request header IPs
@@ -142,6 +145,13 @@ type Event struct {
 	Annotations map[string]string `json:"annotations,omitempty" protobuf:"bytes,15,rep,name=annotations"`
 }
 
+type AuthenticationMetadata struct {
+	// ImpersonationConstraint is the verb associated with the constrained impersonation mode that was used to authorize
+	// the ImpersonatedUser associated with this audit event.  It is only set when constrained impersonation was used.
+	// +optional
+	ImpersonationConstraint string `json:"impersonationConstraint,omitempty" protobuf:"bytes,1,opt,name=impersonationConstraint"`
+}
+
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
 
 // EventList is a list of audit Events.

staging/src/k8s.io/apiserver/pkg/apis/audit/v1/zz_generated.conversion.go

@@ -39,6 +39,16 @@ func init() {
 // RegisterConversions adds conversion functions to the given scheme.
 // Public to allow building arbitrary schemes.
 func RegisterConversions(s *runtime.Scheme) error {
+	if err := s.AddGeneratedConversionFunc((*AuthenticationMetadata)(nil), (*audit.AuthenticationMetadata)(nil), func(a, b interface{}, scope conversion.Scope) error {
+		return Convert_v1_AuthenticationMetadata_To_audit_AuthenticationMetadata(a.(*AuthenticationMetadata), b.(*audit.AuthenticationMetadata), scope)
+	}); err != nil {
+		return err
+	}
+	if err := s.AddGeneratedConversionFunc((*audit.AuthenticationMetadata)(nil), (*AuthenticationMetadata)(nil), func(a, b interface{}, scope conversion.Scope) error {
+		return Convert_audit_AuthenticationMetadata_To_v1_AuthenticationMetadata(a.(*audit.AuthenticationMetadata), b.(*AuthenticationMetadata), scope)
+	}); err != nil {
+		return err
+	}
 	if err := s.AddGeneratedConversionFunc((*Event)(nil), (*audit.Event)(nil), func(a, b interface{}, scope conversion.Scope) error {
 		return Convert_v1_Event_To_audit_Event(a.(*Event), b.(*audit.Event), scope)
 	}); err != nil {
@@ -112,6 +122,26 @@ func RegisterConversions(s *runtime.Scheme) error {
 	return nil
 }
 
+func autoConvert_v1_AuthenticationMetadata_To_audit_AuthenticationMetadata(in *AuthenticationMetadata, out *audit.AuthenticationMetadata, s conversion.Scope) error {
+	out.ImpersonationConstraint = in.ImpersonationConstraint
+	return nil
+}
+
+// Convert_v1_AuthenticationMetadata_To_audit_AuthenticationMetadata is an autogenerated conversion function.
+func Convert_v1_AuthenticationMetadata_To_audit_AuthenticationMetadata(in *AuthenticationMetadata, out *audit.AuthenticationMetadata, s conversion.Scope) error {
+	return autoConvert_v1_AuthenticationMetadata_To_audit_AuthenticationMetadata(in, out, s)
+}
+
+func autoConvert_audit_AuthenticationMetadata_To_v1_AuthenticationMetadata(in *audit.AuthenticationMetadata, out *AuthenticationMetadata, s conversion.Scope) error {
+	out.ImpersonationConstraint = in.ImpersonationConstraint
+	return nil
+}
+
+// Convert_audit_AuthenticationMetadata_To_v1_AuthenticationMetadata is an autogenerated conversion function.
+func Convert_audit_AuthenticationMetadata_To_v1_AuthenticationMetadata(in *audit.AuthenticationMetadata, out *AuthenticationMetadata, s conversion.Scope) error {
+	return autoConvert_audit_AuthenticationMetadata_To_v1_AuthenticationMetadata(in, out, s)
+}
+
 func autoConvert_v1_Event_To_audit_Event(in *Event, out *audit.Event, s conversion.Scope) error {
 	out.Level = audit.Level(in.Level)
 	out.AuditID = types.UID(in.AuditID)
@@ -120,6 +150,7 @@ func autoConvert_v1_Event_To_audit_Event(in *Event, out *audit.Event, s conversi
 	out.Verb = in.Verb
 	out.User = in.User
 	out.ImpersonatedUser = (*authenticationv1.UserInfo)(unsafe.Pointer(in.ImpersonatedUser))
+	out.AuthenticationMetadata = (*audit.AuthenticationMetadata)(unsafe.Pointer(in.AuthenticationMetadata))
 	out.SourceIPs = *(*[]string)(unsafe.Pointer(&in.SourceIPs))
 	out.UserAgent = in.UserAgent
 	out.ObjectRef = (*audit.ObjectReference)(unsafe.Pointer(in.ObjectRef))
@@ -145,6 +176,7 @@ func autoConvert_audit_Event_To_v1_Event(in *audit.Event, out *Event, s conversi
 	out.Verb = in.Verb
 	out.User = in.User
 	out.ImpersonatedUser = (*authenticationv1.UserInfo)(unsafe.Pointer(in.ImpersonatedUser))
+	out.AuthenticationMetadata = (*AuthenticationMetadata)(unsafe.Pointer(in.AuthenticationMetadata))
 	out.SourceIPs = *(*[]string)(unsafe.Pointer(&in.SourceIPs))
 	out.UserAgent = in.UserAgent
 	out.ObjectRef = (*ObjectReference)(unsafe.Pointer(in.ObjectRef))

staging/src/k8s.io/apiserver/pkg/apis/audit/v1/zz_generated.deepcopy.go

@@ -27,6 +27,22 @@ import (
 	runtime "k8s.io/apimachinery/pkg/runtime"
 )
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *AuthenticationMetadata) DeepCopyInto(out *AuthenticationMetadata) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new AuthenticationMetadata.
+func (in *AuthenticationMetadata) DeepCopy() *AuthenticationMetadata {
+	if in == nil {
+		return nil
+	}
+	out := new(AuthenticationMetadata)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Event) DeepCopyInto(out *Event) {
 	*out = *in
@@ -37,6 +53,11 @@ func (in *Event) DeepCopyInto(out *Event) {
 		*out = new(authenticationv1.UserInfo)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.AuthenticationMetadata != nil {
+		in, out := &in.AuthenticationMetadata, &out.AuthenticationMetadata
+		*out = new(AuthenticationMetadata)
+		**out = **in
+	}
 	if in.SourceIPs != nil {
 		in, out := &in.SourceIPs, &out.SourceIPs
 		*out = make([]string, len(*in))

staging/src/k8s.io/apiserver/pkg/apis/audit/v1/zz_generated.model_name.go

@@ -21,6 +21,11 @@ limitations under the License.
 
 package v1
 
+// OpenAPIModelName returns the OpenAPI model name for this type.
+func (in AuthenticationMetadata) OpenAPIModelName() string {
+	return "io.k8s.apiserver.pkg.apis.audit.v1.AuthenticationMetadata"
+}
+
 // OpenAPIModelName returns the OpenAPI model name for this type.
 func (in Event) OpenAPIModelName() string {
 	return "io.k8s.apiserver.pkg.apis.audit.v1.Event"

staging/src/k8s.io/apiserver/pkg/apis/audit/zz_generated.deepcopy.go

@@ -27,6 +27,22 @@ import (
 	runtime "k8s.io/apimachinery/pkg/runtime"
 )
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *AuthenticationMetadata) DeepCopyInto(out *AuthenticationMetadata) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new AuthenticationMetadata.
+func (in *AuthenticationMetadata) DeepCopy() *AuthenticationMetadata {
+	if in == nil {
+		return nil
+	}
+	out := new(AuthenticationMetadata)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Event) DeepCopyInto(out *Event) {
 	*out = *in
@@ -37,6 +53,11 @@ func (in *Event) DeepCopyInto(out *Event) {
 		*out = new(v1.UserInfo)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.AuthenticationMetadata != nil {
+		in, out := &in.AuthenticationMetadata, &out.AuthenticationMetadata
+		*out = new(AuthenticationMetadata)
+		**out = **in
+	}
 	if in.SourceIPs != nil {
 		in, out := &in.SourceIPs, &out.SourceIPs
 		*out = make([]string, len(*in))

staging/src/k8s.io/apiserver/pkg/audit/context.go

@@ -132,7 +132,7 @@ func (ac *AuditContext) ProcessEventStage(ctx context.Context, stage auditintern
 	return processed
 }
 
-func (ac *AuditContext) LogImpersonatedUser(user user.Info) {
+func (ac *AuditContext) LogImpersonatedUser(user user.Info, constraint string) {
 	ac.visitEvent(func(ev *auditinternal.Event) {
 		if ev == nil || ev.Level.Less(auditinternal.LevelMetadata) {
 			return
@@ -146,6 +146,12 @@ func (ac *AuditContext) LogImpersonatedUser(user user.Info) {
 		for k, v := range user.GetExtra() {
 			ev.ImpersonatedUser.Extra[k] = authnv1.ExtraValue(v)
 		}
+		if len(constraint) > 0 {
+			if ev.AuthenticationMetadata == nil {
+				ev.AuthenticationMetadata = &auditinternal.AuthenticationMetadata{}
+			}
+			ev.AuthenticationMetadata.ImpersonationConstraint = constraint
+		}
 	})
 }
 

staging/src/k8s.io/apiserver/pkg/audit/request.go

@@ -82,12 +82,12 @@ func LogRequestMetadata(ctx context.Context, req *http.Request, requestReceivedT
 }
 
 // LogImpersonatedUser fills in the impersonated user attributes into an audit event.
-func LogImpersonatedUser(ctx context.Context, user user.Info) {
+func LogImpersonatedUser(ctx context.Context, user user.Info, constraint string) {
 	ac := AuditContextFrom(ctx)
 	if !ac.Enabled() {
 		return
 	}
-	ac.LogImpersonatedUser(user)
+	ac.LogImpersonatedUser(user, constraint)
 }
 
 // LogRequestObject fills in the request object into an audit event. The passed runtime.Object

staging/src/k8s.io/apiserver/pkg/authorization/authorizer/interfaces.go

@@ -18,6 +18,7 @@ package authorizer
 
 import (
 	"context"
+	"fmt"
 	"net/http"
 
 	"k8s.io/apimachinery/pkg/fields"
@@ -182,3 +183,16 @@ const (
 	// to allow or deny an action.
 	DecisionNoOpinion
 )
+
+func (d Decision) String() string {
+	switch d {
+	case DecisionDeny:
+		return "Deny"
+	case DecisionAllow:
+		return "Allow"
+	case DecisionNoOpinion:
+		return "NoOpinion"
+	default:
+		return fmt.Sprintf("Unknown (%d)", int(d))
+	}
+}

staging/src/k8s.io/apiserver/pkg/endpoints/filters/authorization.go

@@ -94,7 +94,7 @@ func withAuthorization(handler http.Handler, a authorizer.Authorizer, s runtime.
 		audit.AddAuditAnnotations(ctx,
 			decisionAnnotationKey, decisionForbid,
 			reasonAnnotationKey, reason)
-		responsewriters.Forbidden(ctx, attributes, w, req, reason, s)
+		responsewriters.Forbidden(attributes, w, req, reason, s)
 	})
 }
 

staging/src/k8s.io/apiserver/pkg/endpoints/filters/impersonation/OWNERS

@@ -0,0 +1,8 @@
+# See the OWNERS docs at https://go.k8s.io/owners
+
+approvers:
+  - sig-auth-authenticators-approvers
+reviewers:
+  - sig-auth-authenticators-reviewers
+labels:
+  - sig/auth

staging/src/k8s.io/apiserver/pkg/endpoints/filters/impersonation/cache.go

@@ -0,0 +1,320 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package impersonation
+
+import (
+	"crypto/sha256"
+	"fmt"
+	"hash/fnv"
+	"time"
+
+	"golang.org/x/crypto/cryptobyte"
+
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/util/cache"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/apiserver/pkg/authentication/user"
+	"k8s.io/apiserver/pkg/authorization/authorizer"
+	"k8s.io/utils/lru"
+)
+
+// modeIndexCache is a simple username -> impersonation mode cache that is based on the assumption
+// that a particular user is likely to use a single mode of impersonation for all impersonated requests
+// that they make.  it remembers which impersonation mode was last successful for a username, and tries
+// that mode first for future impersonation checks.  this makes it so that the amortized cost of legacy
+// impersonation remains the same, and the cost of constrained impersonation is one extra authorization
+// check in additional to the existing checks of regular impersonation.
+type modeIndexCache struct {
+	cache *lru.Cache
+}
+
+func (c *modeIndexCache) get(attributes authorizer.Attributes) (int, bool) {
+	idx, ok := c.cache.Get(modeIndexCacheKey(attributes))
+	if !ok {
+		return 0, false
+	}
+	return idx.(int), true
+}
+
+func (c *modeIndexCache) set(attributes authorizer.Attributes, idx int) {
+	c.cache.Add(modeIndexCacheKey(attributes), idx)
+}
+
+func modeIndexCacheKey(attributes authorizer.Attributes) string {
+	key := attributes.GetUser().GetName()
+	// hash the name so our cache size is predicable regardless of the size of usernames
+	// collisions do not matter for this logic as it simply changes the ordering of the modes used
+	hash := fnvSum128a([]byte(key))
+	return fmt.Sprintf("%x", hash)
+}
+
+func fnvSum128a(data []byte) []byte {
+	h := fnv.New128a()
+	h.Write(data)
+	var sum [16]byte
+	return h.Sum(sum[:0])
+}
+
+func newModeIndexCache() *modeIndexCache {
+	return &modeIndexCache{
+		// each entry is roughly ~24 bytes (16 bytes for the hashed key, 8 bytes for value)
+		// thus at even 10k entries, we should use less than 1 MB memory
+		// this hardcoded size allows us to remember many users without leaking memory
+		cache: lru.New(10_000),
+	}
+}
+
+// impersonationCache tracks successful impersonation attempts for a given mode with a short TTL.
+//
+// when skipAttributes is false, it maps [wantedUser, attributes] -> impersonatedUserInfo
+// when skipAttributes is true, it maps [wantedUser, requestor] -> impersonatedUserInfo
+//
+// thus each constrained impersonation mode needs two of these caches:
+// the outer cache sets skipAttributes to false and thus covers the overall impersonation attempt, see constrainedImpersonationModeState.check.
+// the inner cache sets skipAttributes to true and only covers the authorization checks that
+// are not dependent on the specific request being made, see impersonationModeState.check.
+type impersonationCache struct {
+	cache          *cache.Expiring
+	skipAttributes bool
+}
+
+func (c *impersonationCache) get(k *impersonationCacheKey) *impersonatedUserInfo {
+	key, err := k.key(c.skipAttributes)
+	if err != nil {
+		utilruntime.HandleError(fmt.Errorf("failed to build impersonation cache key: %w", err))
+		return nil
+	}
+	impersonatedUser, ok := c.cache.Get(key)
+	if !ok {
+		return nil
+	}
+	return impersonatedUser.(*impersonatedUserInfo)
+}
+
+func (c *impersonationCache) set(k *impersonationCacheKey, impersonatedUser *impersonatedUserInfo) {
+	key, err := k.key(c.skipAttributes)
+	if err != nil {
+		utilruntime.HandleError(fmt.Errorf("failed to build impersonation cache key: %w", err))
+		return
+	}
+	c.cache.Set(key, impersonatedUser, 10*time.Second) // hardcode the same short TTL as used by TokenSuccessCacheTTL
+}
+
+func newImpersonationCache(skipAttributes bool) *impersonationCache {
+	return &impersonationCache{
+		cache:          cache.NewExpiring(),
+		skipAttributes: skipAttributes,
+	}
+}
+
+// The attribute accessors known to cache key construction. If this fails to compile, the cache
+// implementation may need to be updated.
+var _ authorizer.Attributes = (interface {
+	GetUser() user.Info
+	GetVerb() string
+	IsReadOnly() bool
+	GetNamespace() string
+	GetResource() string
+	GetSubresource() string
+	GetName() string
+	GetAPIGroup() string
+	GetAPIVersion() string
+	IsResourceRequest() bool
+	GetPath() string
+	GetFieldSelector() (fields.Requirements, error)
+	GetLabelSelector() (labels.Requirements, error)
+})(nil)
+
+// The user info accessors known to cache key construction. If this fails to compile, the cache
+// implementation may need to be updated.
+var _ user.Info = (interface {
+	GetName() string
+	GetUID() string
+	GetGroups() []string
+	GetExtra() map[string][]string
+})(nil)
+
+// impersonationCacheKey allows for lazy building of string cache keys based on the inputs.
+// See impersonationCache details above for the semantics around skipAttributes.
+// Note that the same impersonationCacheKey can be used with any value for skipAttributes.
+type impersonationCacheKey struct {
+	wantedUser *user.DefaultInfo
+	attributes authorizer.Attributes
+
+	// lazily calculated values at point of use
+	keyAttr string
+	errAttr error
+	keyUser string
+	errUser error
+}
+
+func (k *impersonationCacheKey) key(skipAttributes bool) (string, error) {
+	if skipAttributes {
+		return k.keyWithoutAttributes()
+	}
+	return k.keyWithAttributes()
+}
+
+func (k *impersonationCacheKey) keyWithAttributes() (string, error) {
+	if len(k.keyAttr) != 0 || k.errAttr != nil {
+		return k.keyAttr, k.errAttr
+	}
+
+	k.keyAttr, k.errAttr = buildKey(k.wantedUser, k.attributes)
+	return k.keyAttr, k.errAttr
+}
+
+func (k *impersonationCacheKey) keyWithoutAttributes() (string, error) {
+	if len(k.keyUser) != 0 || k.errUser != nil {
+		return k.keyUser, k.errUser
+	}
+
+	// fake attributes that just contain the requestor to allow us to reuse buildKey
+	requestor := k.attributes.GetUser()
+	attributes := authorizer.AttributesRecord{User: requestor}
+
+	k.keyUser, k.errUser = buildKey(k.wantedUser, attributes)
+	return k.keyUser, k.errUser
+}
+
+// buildKey creates a hashed string key based on the inputs that is namespaced to the requestor.
+// A cryptographically secure hash is used to minimize the chance of collisions.
+func buildKey(wantedUser *user.DefaultInfo, attributes authorizer.Attributes) (string, error) {
+	fieldSelector, err := attributes.GetFieldSelector()
+	if err != nil {
+		return "", err // if we do not fully understand the attributes, just skip caching altogether
+	}
+
+	labelSelector, err := attributes.GetLabelSelector()
+	if err != nil {
+		return "", err // if we do not fully understand the attributes, just skip caching altogether
+	}
+
+	requestor := attributes.GetUser()
+
+	// the chance of a hash collision is impractically small, but the only way that would lead to a
+	// privilege escalation is if you could get the cache key of a different user.  if you somehow
+	// get a collision with your own username, you already have that permission since we only set
+	// values in the cache after a successful impersonation.  Thus, we include the requestor
+	// username in the cache key.  It is safe to assume that a user has no control over their own
+	// username since that is controlled by the authenticator.  Even though many of the other inputs
+	// are under the control of the requestor, they cannot explode the cache due to the hashing.
+	b := newCacheKeyBuilder(requestor.GetName())
+
+	addUser(b, wantedUser)
+	addUser(b, requestor)
+
+	b.addLengthPrefixed(func(b *cacheKeyBuilder) {
+		b.
+			addString(attributes.GetVerb()).
+			addBool(attributes.IsReadOnly()).
+			addString(attributes.GetNamespace()).
+			addString(attributes.GetResource()).
+			addString(attributes.GetSubresource()).
+			addString(attributes.GetName()).
+			addString(attributes.GetAPIGroup()).
+			addString(attributes.GetAPIVersion()).
+			addBool(attributes.IsResourceRequest()).
+			addString(attributes.GetPath())
+	})
+
+	b.addLengthPrefixed(func(b *cacheKeyBuilder) {
+		for _, req := range fieldSelector {
+			b.addStringSlice([]string{req.Field, string(req.Operator), req.Value})
+		}
+	})
+
+	b.addLengthPrefixed(func(b *cacheKeyBuilder) {
+		for _, req := range labelSelector {
+			b.addString(req.String())
+		}
+	})
+
+	return b.build()
+}
+
+func addUser(b *cacheKeyBuilder, u user.Info) {
+	b.addLengthPrefixed(func(b *cacheKeyBuilder) {
+		b.
+			addString(u.GetName()).
+			addString(u.GetUID()).
+			addStringSlice(u.GetGroups()).
+			addLengthPrefixed(func(b *cacheKeyBuilder) {
+				extra := u.GetExtra()
+				for _, key := range sets.StringKeySet(extra).List() {
+					b.addString(key)
+					b.addStringSlice(extra[key])
+				}
+			})
+	})
+}
+
+// cacheKeyBuilder adds syntactic sugar on top of cryptobyte.Builder to make it easier to use for complex inputs.
+type cacheKeyBuilder struct {
+	namespace string // in the programming sense, not the Kubernetes concept
+	builder   *cryptobyte.Builder
+}
+
+func newCacheKeyBuilder(namespace string) *cacheKeyBuilder {
+	// start with a reasonable size to avoid too many allocations
+	return &cacheKeyBuilder{namespace: namespace, builder: cryptobyte.NewBuilder(make([]byte, 0, 384))}
+}
+
+func (c *cacheKeyBuilder) addString(value string) *cacheKeyBuilder {
+	c.addLengthPrefixed(func(c *cacheKeyBuilder) {
+		c.builder.AddBytes([]byte(value))
+	})
+	return c
+}
+
+func (c *cacheKeyBuilder) addStringSlice(values []string) *cacheKeyBuilder {
+	c.addLengthPrefixed(func(c *cacheKeyBuilder) {
+		for _, v := range values {
+			c.addString(v)
+		}
+	})
+	return c
+}
+
+func (c *cacheKeyBuilder) addBool(value bool) *cacheKeyBuilder {
+	var b byte
+	if value {
+		b = 1
+	}
+	c.builder.AddUint8(b)
+	return c
+}
+
+type builderContinuation func(child *cacheKeyBuilder)
+
+func (c *cacheKeyBuilder) addLengthPrefixed(f builderContinuation) {
+	c.builder.AddUint32LengthPrefixed(func(b *cryptobyte.Builder) {
+		c := &cacheKeyBuilder{namespace: c.namespace, builder: b}
+		f(c)
+	})
+}
+
+func (c *cacheKeyBuilder) build() (string, error) {
+	key, err := c.builder.Bytes()
+	if err != nil {
+		return "", err
+	}
+	hash := sha256.Sum256(key) // reduce the size of the cache key to keep the overall cache size small
+	return fmt.Sprintf("%x/%s", hash[:], c.namespace), nil
+}

staging/src/k8s.io/apiserver/pkg/endpoints/filters/impersonation/constrained_impersonation.go

@@ -0,0 +1,251 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package impersonation
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"strings"
+
+	authenticationv1 "k8s.io/api/authentication/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apiserver/pkg/audit"
+	"k8s.io/apiserver/pkg/authentication/user"
+	"k8s.io/apiserver/pkg/authorization/authorizer"
+	"k8s.io/apiserver/pkg/endpoints/filters"
+	"k8s.io/apiserver/pkg/endpoints/handlers/responsewriters"
+	"k8s.io/apiserver/pkg/endpoints/request"
+	"k8s.io/apiserver/pkg/server/httplog"
+	"k8s.io/klog/v2"
+)
+
+// WithConstrainedImpersonation implements constrained impersonation as described in https://kep.k8s.io/5284
+// It also includes a complete reimplementation of legacy impersonation for backwards compatibility.
+// At a high level, constrained impersonation uses multiple authorization checks to allow for the granular
+// expression of impersonation access.  For example, a service account may be authorized to impersonate the
+// node that it is associated with but only when listing pods.  See the linked KEP for further details.
+func WithConstrainedImpersonation(handler http.Handler, a authorizer.Authorizer, s runtime.NegotiatedSerializer) http.Handler {
+	return &constrainedImpersonationHandler{
+		handler: handler,
+		tracker: newImpersonationModesTracker(a),
+		s:       s,
+	}
+}
+
+type constrainedImpersonationHandler struct {
+	handler http.Handler
+	tracker *impersonationModesTracker
+	s       runtime.NegotiatedSerializer
+}
+
+func (c *constrainedImpersonationHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
+	ctx := req.Context()
+
+	wantedUser, err := processImpersonationHeaders(req.Header)
+	if err != nil {
+		responsewriters.RespondWithError(w, req, err, c.s)
+		return
+	}
+	if wantedUser == nil { // impersonation was not attempted so skip to the next handler
+		c.handler.ServeHTTP(w, req)
+		return
+	}
+	attributes, err := filters.GetAuthorizerAttributes(ctx)
+	if err != nil {
+		responsewriters.InternalError(w, req, err)
+		return
+	}
+	requestor := attributes.GetUser()
+	if requestor == nil {
+		responsewriters.InternalError(w, req, errors.New("no User found in the context"))
+		return
+	}
+
+	impersonatedUser, err := c.tracker.getImpersonatedUser(ctx, wantedUser, attributes)
+	if err != nil {
+		klog.V(4).InfoS("Forbidden", "URI", req.RequestURI, "err", err)
+		responsewriters.RespondWithError(w, req, err, c.s)
+		return
+	}
+
+	req = req.WithContext(request.WithUser(ctx, impersonatedUser.user))
+	httplog.LogOf(req, w).Addf("%v is impersonating %v", userString(requestor), userString(impersonatedUser.user))
+	audit.LogImpersonatedUser(ctx, impersonatedUser.user, impersonatedUser.constraint)
+
+	c.handler.ServeHTTP(w, req)
+}
+
+// processImpersonationHeaders converts the impersonation headers in the given input headers
+// into the equivalent user.DefaultInfo.  The resulting user is a raw representation of the
+// input headers, that is, no defaulting or other mutations have been applied to it.
+func processImpersonationHeaders(headers http.Header) (*user.DefaultInfo, error) {
+	wantedUser := &user.DefaultInfo{}
+
+	wantedUser.Name = headers.Get(authenticationv1.ImpersonateUserHeader)
+	hasUser := len(wantedUser.Name) > 0
+
+	wantedUser.UID = headers.Get(authenticationv1.ImpersonateUIDHeader)
+	hasUID := len(wantedUser.UID) > 0
+
+	hasGroups := false
+	for _, group := range headers[authenticationv1.ImpersonateGroupHeader] {
+		hasGroups = true
+		wantedUser.Groups = append(wantedUser.Groups, group)
+	}
+
+	hasUserExtra := false
+	for headerName, values := range headers {
+		if !strings.HasPrefix(headerName, authenticationv1.ImpersonateUserExtraHeaderPrefix) {
+			continue
+		}
+
+		hasUserExtra = true
+
+		if len(values) == 0 {
+			// this looks a little strange but matches the behavior of buildImpersonationRequests from legacy impersonation
+			// http1 uses textproto.Reader#ReadMIMEHeader which does seem to allow an empty slice for values
+			// http2 uses http.Header#Add which will cause the values slice to always be non-empty
+			continue
+		}
+
+		extraKey := unescapeExtraKey(strings.ToLower(headerName[len(authenticationv1.ImpersonateUserExtraHeaderPrefix):]))
+
+		if wantedUser.Extra == nil {
+			wantedUser.Extra = map[string][]string{}
+		}
+		wantedUser.Extra[extraKey] = append(wantedUser.Extra[extraKey], values...)
+	}
+
+	if !hasUser && (hasUID || hasGroups || hasUserExtra) {
+		return nil, apierrors.NewBadRequest(fmt.Sprintf("requested %#v without impersonating a user name", wantedUser))
+	}
+
+	if !hasUser {
+		return nil, nil
+	}
+
+	// clear all the impersonation headers from the request to prevent downstream layers from knowing that impersonation was used
+	// we do not want anything outside of this package trying to behave differently based on if impersonation was used
+	headers.Del(authenticationv1.ImpersonateUserHeader)
+	headers.Del(authenticationv1.ImpersonateUIDHeader)
+	headers.Del(authenticationv1.ImpersonateGroupHeader)
+	for headerName := range headers {
+		if strings.HasPrefix(headerName, authenticationv1.ImpersonateUserExtraHeaderPrefix) {
+			headers.Del(headerName)
+		}
+	}
+
+	return wantedUser, nil
+}
+
+// impersonationModesTracker records which impersonation mode was last successful for a given requestor user.
+// this allows us to check for the more secure constrained impersonation modes first while keeping the overall
+// cost of legacy impersonation unchanged (as we will support legacy impersonation forever).
+type impersonationModesTracker struct {
+	modes    []impersonationMode
+	idxCache *modeIndexCache
+}
+
+func newImpersonationModesTracker(a authorizer.Authorizer) *impersonationModesTracker {
+	loggingAuthorizer := authorizer.AuthorizerFunc(func(ctx context.Context, attributes authorizer.Attributes) (authorizer.Decision, string, error) {
+		decision, reason, err := a.Authorize(ctx, attributes)
+		// build a detailed log of the authorization
+		// make the whole block conditional so we do not do a lot of string-building we will not use
+		if klogV := klog.V(5); klogV.Enabled() { // same log level that the RBAC authorizer uses for verbose logging
+			u := attributes.GetUser()
+			fieldSelector, _ := attributes.GetFieldSelector()
+			labelSelector, _ := attributes.GetLabelSelector()
+			klogV.InfoSDepth(3, "Impersonation authorization check",
+				// we cannot just pass attributes to the logger as that will not capture the actual result of calling these methods
+				// impersonation makes heavy use of wrapping these methods to add extra logic
+				"username", u.GetName(),
+				"uid", u.GetUID(),
+				"groups", u.GetGroups(),
+				"extra", u.GetExtra(),
+
+				"isResourceRequest", attributes.IsResourceRequest(),
+
+				"namespace", attributes.GetNamespace(),
+				"verb", attributes.GetVerb(),
+				"group", attributes.GetAPIGroup(),
+				"version", attributes.GetAPIVersion(),
+				"resource", attributes.GetResource(),
+				"subresource", attributes.GetSubresource(),
+				"name", attributes.GetName(),
+				"fieldSelector", fieldSelector,
+				"labelSelector", labelSelector,
+
+				"path", attributes.GetPath(),
+
+				"decision", decision,
+				"reason", reason,
+				"err", err,
+			)
+		}
+		return decision, reason, err
+	})
+	return &impersonationModesTracker{
+		modes:    allImpersonationModes(loggingAuthorizer),
+		idxCache: newModeIndexCache(),
+	}
+}
+
+func (t *impersonationModesTracker) getImpersonatedUser(ctx context.Context, wantedUser *user.DefaultInfo, attributes authorizer.Attributes) (*impersonatedUserInfo, error) {
+	// share a single cache key across all modes so that we only lazily build it once
+	key := &impersonationCacheKey{wantedUser: wantedUser, attributes: attributes}
+	var firstErr error
+
+	// try the last successful mode first to reduce the amortized cost of impersonation
+	// we attempt all modes unless we short-circuit due to a successful impersonation
+	modeIdx, modeIdxOk := t.idxCache.get(attributes)
+	if modeIdxOk {
+		impersonatedUser, err := t.modes[modeIdx].check(ctx, key, wantedUser, attributes)
+		if err == nil && impersonatedUser != nil {
+			return impersonatedUser, nil
+		}
+		firstErr = err
+	}
+
+	for i, mode := range t.modes {
+		if modeIdxOk && i == modeIdx {
+			continue // skip already attempted mode
+		}
+
+		impersonatedUser, err := mode.check(ctx, key, wantedUser, attributes)
+		if err != nil {
+			if firstErr == nil {
+				firstErr = err
+			}
+			continue
+		}
+		if impersonatedUser == nil {
+			continue
+		}
+		t.idxCache.set(attributes, i)
+		return impersonatedUser, nil
+	}
+
+	if firstErr != nil {
+		return nil, firstErr
+	}
+
+	// this should not happen, but make sure we fail closed when no impersonation mode succeeded
+	return nil, errors.New("all impersonation modes failed")
+}

staging/src/k8s.io/apiserver/pkg/endpoints/filters/impersonation/impersonation.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package filters
+package impersonation
 
 import (
 	"errors"
@@ -27,6 +27,7 @@ import (
 
 	authenticationv1 "k8s.io/api/authentication/v1"
 	"k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apiserver/pkg/audit"
 	"k8s.io/apiserver/pkg/authentication/serviceaccount"
@@ -43,7 +44,7 @@ func WithImpersonation(handler http.Handler, a authorizer.Authorizer, s runtime.
 		impersonationRequests, err := buildImpersonationRequests(req.Header)
 		if err != nil {
 			klog.V(4).Infof("%v", err)
-			responsewriters.InternalError(w, req, err)
+			responsewriters.RespondWithError(w, req, err, s)
 			return
 		}
 		if len(impersonationRequests) == 0 {
@@ -110,14 +111,14 @@ func WithImpersonation(handler http.Handler, a authorizer.Authorizer, s runtime.
 
 			default:
 				klog.V(4).InfoS("unknown impersonation request type", "request", impersonationRequest)
-				responsewriters.Forbidden(ctx, actingAsAttributes, w, req, fmt.Sprintf("unknown impersonation request type: %v", impersonationRequest), s)
+				responsewriters.Forbidden(actingAsAttributes, w, req, fmt.Sprintf("unknown impersonation request type: %v", impersonationRequest), s)
 				return
 			}
 
 			decision, reason, err := a.Authorize(ctx, actingAsAttributes)
 			if err != nil || decision != authorizer.DecisionAllow {
 				klog.V(4).InfoS("Forbidden", "URI", req.RequestURI, "reason", reason, "err", err)
-				responsewriters.Forbidden(ctx, actingAsAttributes, w, req, reason, s)
+				responsewriters.Forbidden(actingAsAttributes, w, req, reason, s)
 				return
 			}
 		}
@@ -166,7 +167,7 @@ func WithImpersonation(handler http.Handler, a authorizer.Authorizer, s runtime.
 		oldUser, _ := request.UserFrom(ctx)
 		httplog.LogOf(req, w).Addf("%v is impersonating %v", userString(oldUser), userString(newUser))
 
-		audit.LogImpersonatedUser(audit.WithAuditContext(ctx), newUser)
+		audit.LogImpersonatedUser(audit.WithAuditContext(ctx), newUser, "")
 
 		// clear all the impersonation headers from the request
 		req.Header.Del(authenticationv1.ImpersonateUserHeader)
@@ -266,7 +267,7 @@ func buildImpersonationRequests(headers http.Header) ([]v1.ObjectReference, erro
 	}
 
 	if (hasGroups || hasUserExtra || hasUID) && !hasUser {
-		return nil, fmt.Errorf("requested %v without impersonating a user", impersonationRequests)
+		return nil, apierrors.NewBadRequest(fmt.Sprintf("requested %v without impersonating a user", impersonationRequests))
 	}
 
 	return impersonationRequests, nil

staging/src/k8s.io/apiserver/pkg/endpoints/filters/impersonation/impersonation_test.go

@@ -14,19 +14,19 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package filters
+package impersonation
 
 import (
 	"context"
 	"fmt"
 	"net/http"
 	"net/http/httptest"
-	"reflect"
 	"strings"
 	"sync"
 	"testing"
 
 	authenticationapi "k8s.io/api/authentication/v1"
+	apiequality "k8s.io/apimachinery/pkg/api/equality"
 	"k8s.io/apimachinery/pkg/runtime"
 	serializer "k8s.io/apimachinery/pkg/runtime/serializer"
 	"k8s.io/apiserver/pkg/authentication/user"
@@ -143,7 +143,7 @@ func TestImpersonationFilter(t *testing.T) {
 			expectedUser: &user.DefaultInfo{
 				Name: "tester",
 			},
-			expectedCode: http.StatusInternalServerError,
+			expectedCode: http.StatusBadRequest,
 		},
 		{
 			name: "impersonating-extra-without-user",
@@ -154,7 +154,7 @@ func TestImpersonationFilter(t *testing.T) {
 			expectedUser: &user.DefaultInfo{
 				Name: "tester",
 			},
-			expectedCode: http.StatusInternalServerError,
+			expectedCode: http.StatusBadRequest,
 		},
 		{
 			name: "impersonating-uid-without-user",
@@ -165,7 +165,7 @@ func TestImpersonationFilter(t *testing.T) {
 			expectedUser: &user.DefaultInfo{
 				Name: "tester",
 			},
-			expectedCode: http.StatusInternalServerError,
+			expectedCode: http.StatusBadRequest,
 		},
 		{
 			name: "disallowed-group",
@@ -497,7 +497,7 @@ func TestImpersonationFilter(t *testing.T) {
 		}
 
 	})
-	handler := func(delegate http.Handler) http.Handler {
+	delegateHandler := func(delegate http.Handler) http.Handler {
 		return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
 			defer func() {
 				if r := recover(); r != nil {
@@ -519,53 +519,66 @@ func TestImpersonationFilter(t *testing.T) {
 
 			delegate.ServeHTTP(w, req)
 		})
-	}(WithImpersonation(doNothingHandler, impersonateAuthorizer{}, serializer.NewCodecFactory(runtime.NewScheme())))
+	}
 
-	server := httptest.NewServer(handler)
-	defer server.Close()
+	handlersToTest := map[string]http.Handler{
+		"impersonation":            delegateHandler(WithImpersonation(doNothingHandler, impersonateAuthorizer{}, serializer.NewCodecFactory(runtime.NewScheme()))),
+		"constrainedImpersonation": delegateHandler(WithConstrainedImpersonation(doNothingHandler, impersonateAuthorizer{}, serializer.NewCodecFactory(runtime.NewScheme()))),
+	}
 
-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			func() {
-				lock.Lock()
-				defer lock.Unlock()
-				ctx = request.WithUser(request.NewContext(), tc.user)
-			}()
+	for name, handler := range handlersToTest {
+		server := httptest.NewServer(handler)
 
-			req, err := http.NewRequestWithContext(ctx, http.MethodGet, server.URL, nil)
-			if err != nil {
-				t.Errorf("%s: unexpected error: %v", tc.name, err)
-				return
-			}
-			if len(tc.impersonationUser) > 0 {
-				req.Header.Add(authenticationapi.ImpersonateUserHeader, tc.impersonationUser)
-			}
-			for _, group := range tc.impersonationGroups {
-				req.Header.Add(authenticationapi.ImpersonateGroupHeader, group)
-			}
-			for extraKey, values := range tc.impersonationUserExtras {
-				for _, value := range values {
-					req.Header.Add(authenticationapi.ImpersonateUserExtraHeaderPrefix+extraKey, value)
+		for _, tc := range testCases {
+			t.Run(name+"/"+tc.name, func(t *testing.T) {
+				func() {
+					lock.Lock()
+					defer lock.Unlock()
+					ctx = request.WithUser(request.NewContext(), tc.user)
+					ctx = request.WithRequestInfo(ctx, &request.RequestInfo{
+						IsResourceRequest: true,
+						Verb:              "get",
+						APIVersion:        "v1",
+						Resource:          "pods",
+					})
+				}()
+
+				req, err := http.NewRequestWithContext(ctx, http.MethodGet, server.URL, nil)
+				if err != nil {
+					t.Errorf("%s: unexpected error: %v", tc.name, err)
+					return
+				}
+				if len(tc.impersonationUser) > 0 {
+					req.Header.Add(authenticationapi.ImpersonateUserHeader, tc.impersonationUser)
+				}
+				for _, group := range tc.impersonationGroups {
+					req.Header.Add(authenticationapi.ImpersonateGroupHeader, group)
+				}
+				for extraKey, values := range tc.impersonationUserExtras {
+					for _, value := range values {
+						req.Header.Add(authenticationapi.ImpersonateUserExtraHeaderPrefix+extraKey, value)
+					}
+				}
+				if len(tc.impersonationUid) > 0 {
+					req.Header.Add(authenticationapi.ImpersonateUIDHeader, tc.impersonationUid)
 				}
-			}
-			if len(tc.impersonationUid) > 0 {
-				req.Header.Add(authenticationapi.ImpersonateUIDHeader, tc.impersonationUid)
-			}
 
-			resp, err := http.DefaultClient.Do(req)
-			if err != nil {
-				t.Errorf("%s: unexpected error: %v", tc.name, err)
-				return
-			}
-			if resp.StatusCode != tc.expectedCode {
-				t.Errorf("%s: expected %v, actual %v", tc.name, tc.expectedCode, resp.StatusCode)
-				return
-			}
+				resp, err := http.DefaultClient.Do(req)
+				if err != nil {
+					t.Errorf("%s: unexpected error: %v", tc.name, err)
+					return
+				}
+				if resp.StatusCode != tc.expectedCode {
+					t.Errorf("%s: expected %v, actual %v", tc.name, tc.expectedCode, resp.StatusCode)
+					return
+				}
 
-			if !reflect.DeepEqual(actualUser, tc.expectedUser) {
-				t.Errorf("%s: expected %#v, actual %#v", tc.name, tc.expectedUser, actualUser)
-				return
-			}
-		})
+				if !apiequality.Semantic.DeepEqual(actualUser, tc.expectedUser) {
+					t.Errorf("%s: expected %#v, actual %#v", tc.name, tc.expectedUser, actualUser)
+					return
+				}
+			})
+		}
+		server.Close()
 	}
 }

staging/src/k8s.io/apiserver/pkg/endpoints/filters/impersonation/mode.go

@@ -0,0 +1,637 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package impersonation
+
+import (
+	"context"
+	"fmt"
+	"slices"
+	"strings"
+
+	authenticationv1 "k8s.io/api/authentication/v1"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/validation"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/sets"
+	utilvalidation "k8s.io/apimachinery/pkg/util/validation"
+	"k8s.io/apimachinery/pkg/util/validation/field"
+	"k8s.io/apiserver/pkg/authentication/serviceaccount"
+	"k8s.io/apiserver/pkg/authentication/user"
+	"k8s.io/apiserver/pkg/authorization/authorizer"
+	"k8s.io/apiserver/pkg/endpoints/handlers/responsewriters"
+)
+
+type impersonatedUserInfo struct {
+	user       user.Info
+	constraint string // the verb used in impersonationModeState.check that allowed this user to be impersonated
+}
+
+// impersonationMode is a type that represents a specific impersonation mode
+// it checks if a requester is allowed to make an API request (the attributes) while impersonating a user (the wantedUser)
+// a mode may return a cached result if it supports caching (using the input cache key if appropriate)
+// a nil impersonatedUserInfo is returned if the mode does not support impersonating the wantedUser
+type impersonationMode interface {
+	check(ctx context.Context, key *impersonationCacheKey, wantedUser *user.DefaultInfo, attributes authorizer.Attributes) (*impersonatedUserInfo, error)
+
+	// all methods below are only used in unit tests
+	verbForTests() string
+	cachesForTests() (outer, inner *impersonationCache)
+}
+
+// constrainedImpersonationModeFilter is a function that defines if a specific constrained impersonation mode
+// supports the requestor impersonating the wantedUser.  It serves as a sudo authorization check for the mode.
+type constrainedImpersonationModeFilter func(wantedUser *user.DefaultInfo, requestor user.Info) bool
+
+func allImpersonationModes(a authorizer.Authorizer) []impersonationMode {
+	return []impersonationMode{
+		associatedNodeImpersonationMode(a),
+		arbitraryNodeImpersonationMode(a),
+		serviceAccountImpersonationMode(a),
+		userInfoImpersonationMode(a),
+		legacyImpersonationMode(a),
+	}
+}
+
+// associatedNodeImpersonationMode allows a requestor service account to impersonate the node that it is
+// associated with.  this is by far the most complex impersonation mode because it caches successful
+// impersonation attempts in a way that results in a high cache hit ratio even when the same service account
+// is used across different pods running on different nodes (i.e. a node agent running as a daemonset).
+// only the username can be specified by the requester.  all other fields in user.Info are controlled by the API server.
+func associatedNodeImpersonationMode(a authorizer.Authorizer) impersonationMode {
+	// we wrap the authorizer so that we can override the requestor service account's extra values
+	// and the node name used in the authorization check.  this makes our authorization checks match
+	// the exact semantics of our cache key which prevents unexpected privilege escalation on a cache
+	// hit.  see the comment below for the cache key details.
+	wrappedAuthorizer := authorizer.AuthorizerFunc(func(ctx context.Context, attributes authorizer.Attributes) (authorizer.Decision, string, error) {
+		// we use checkAuthorization instead of directly calling the authorizer so we can
+		// make the error message line up with the actual attributes authorized against
+		if err := checkAuthorization(ctx, a, &associatedNodeImpersonationAttributes{Attributes: attributes}); err != nil {
+			return authorizer.DecisionDeny, "", err
+		}
+		return authorizer.DecisionAllow, "", nil
+	})
+	mode := newConstrainedImpersonationMode(wrappedAuthorizer, "associated-node",
+		func(wantedUser *user.DefaultInfo, requestor user.Info) bool {
+			wantedNodeName := wantedUser.Name
+			return onlyUsernameSet(wantedUser) && requesterAssociatedWithRequestedNodeUsername(requestor, wantedNodeName)
+		},
+	)
+	return &associatedNodeImpersonationCheck{mode: mode}
+}
+
+type associatedNodeImpersonationCheck struct {
+	mode impersonationMode
+}
+
+func (a *associatedNodeImpersonationCheck) check(ctx context.Context, _ *impersonationCacheKey, wantedUser *user.DefaultInfo, attributes authorizer.Attributes) (*impersonatedUserInfo, error) {
+	wantedNodeName := wantedUser.Name
+	// ignore the input cache key because the cache semantics for associated-node require custom logic.
+	// we know that by the time this key is used, the filter has already verified that the requestor is
+	// a service account with a node ref that matches the node it is trying to impersonate.
+	// the actual node being impersonated is not relevant to the cache key, so we just use a static wantedUser.
+	// we wrap the attributes so that we can drop the requestor service account's extra values.
+	// this results in the cache key being the same for the same service account across all nodes.
+	// this is only safe because of the aforementioned filter running before the cache lookup happens.
+	key := &impersonationCacheKey{wantedUser: &user.DefaultInfo{Name: "system:node:*"}, attributes: &associatedNodeImpersonationAttributes{Attributes: attributes}}
+	impersonatedNodeWithMaybeIncorrectUsername, err := a.mode.check(ctx, key, wantedUser, attributes)
+	if err != nil || impersonatedNodeWithMaybeIncorrectUsername == nil {
+		return nil, err
+	}
+	// at this point, we know that we have a successful associated-node impersonation.
+	// the value could have come from the cache and thus could be for any node, so we wrap the result
+	// here so that the username matches the node associated with the requestor service account.
+	return &impersonatedUserInfo{
+		user: &associatedNodeImpersonationWantedUserInfo{
+			Info: impersonatedNodeWithMaybeIncorrectUsername.user,
+			name: wantedNodeName,
+		},
+		constraint: impersonatedNodeWithMaybeIncorrectUsername.constraint,
+	}, nil
+}
+
+func (a *associatedNodeImpersonationCheck) verbForTests() string {
+	return a.mode.verbForTests()
+}
+
+func (a *associatedNodeImpersonationCheck) cachesForTests() (*impersonationCache, *impersonationCache) {
+	return a.mode.cachesForTests()
+}
+
+type associatedNodeImpersonationAttributes struct {
+	authorizer.Attributes
+}
+
+func (a *associatedNodeImpersonationAttributes) GetUser() user.Info {
+	return &associatedNodeImpersonationRequestorUserInfo{Info: a.Attributes.GetUser()}
+}
+
+func (a *associatedNodeImpersonationAttributes) GetName() string {
+	if a.GetVerb() == "impersonate:associated-node" {
+		return "*" // our cache key ignores the node name, so our authorization check needs to be valid for all node names
+	}
+	return a.Attributes.GetName()
+}
+
+type associatedNodeImpersonationRequestorUserInfo struct {
+	user.Info
+}
+
+func (a *associatedNodeImpersonationRequestorUserInfo) GetExtra() map[string][]string {
+	return map[string][]string{
+		// we know the requestor is a service account with a node ref that matches the node it is trying to impersonate
+		// basically all the extra values would cause cache misses (for example, the node name itself)
+		// so we drop all the extra values but keep the associated key names
+		// the authorizer can trust that we have performed the node association check correctly
+		// the audit log will still contain the full requestor extra fields
+		// different bound object ref types can result in different extra keys, and thus a different cache key
+		"authentication.kubernetes.io/associated-node-keys": sets.StringKeySet(a.Info.GetExtra()).List(),
+	}
+}
+
+type associatedNodeImpersonationWantedUserInfo struct {
+	user.Info
+	name string
+}
+
+func (a *associatedNodeImpersonationWantedUserInfo) GetName() string {
+	return a.name
+}
+
+// arbitraryNodeImpersonationMode implements constrained impersonation for nodes.
+// Only the username can be specified by the requester.  All other fields in user.Info are controlled by the API server.
+func arbitraryNodeImpersonationMode(a authorizer.Authorizer) impersonationMode {
+	return newConstrainedImpersonationMode(a, "arbitrary-node",
+		func(wantedUser *user.DefaultInfo, _ user.Info) bool {
+			if !onlyUsernameSet(wantedUser) {
+				return false
+			}
+			_, ok := isNodeUsername(wantedUser.Name)
+			return ok
+		},
+	)
+}
+
+// serviceAccountImpersonationMode implements constrained impersonation for service accounts.
+// Only the username can be specified by the requester.  All other fields in user.Info are controlled by the API server.
+func serviceAccountImpersonationMode(a authorizer.Authorizer) impersonationMode {
+	return newConstrainedImpersonationMode(a, "serviceaccount",
+		func(wantedUser *user.DefaultInfo, _ user.Info) bool {
+			if !onlyUsernameSet(wantedUser) {
+				return false
+			}
+			_, _, ok := isServiceAccountUsername(wantedUser.Name)
+			return ok
+		},
+	)
+}
+
+// userInfoImpersonationMode implements constrained impersonation for non-node and non-service account users.
+// Unlike the other constrained impersonation modes, it supports impersonating all fields of user.Info.
+func userInfoImpersonationMode(a authorizer.Authorizer) impersonationMode {
+	return newConstrainedImpersonationMode(a, "user-info",
+		func(wantedUser *user.DefaultInfo, _ user.Info) bool {
+			// nodes and service accounts cannot be impersonated in this mode.
+			// the user-info bucket is reserved for the "other" users, that is,
+			// users that do not have an explicit schema defined by Kube.
+			if _, ok := isNodeUsername(wantedUser.Name); ok {
+				return false
+			}
+			if _, _, ok := isServiceAccountUsername(wantedUser.Name); ok {
+				return false
+			}
+			return true
+		},
+	)
+}
+
+// legacyImpersonationMode is a complete reimplementation of the original impersonation mode that has
+// existed in kube since v1.3.  The behavior is expected to be identical to the original implementation.
+func legacyImpersonationMode(a authorizer.Authorizer) impersonationMode {
+	return &legacyImpersonationCheck{m: newImpersonationModeState(a, "impersonate", false)}
+}
+
+type legacyImpersonationCheck struct {
+	m *impersonationModeState
+}
+
+func (l *legacyImpersonationCheck) check(ctx context.Context, key *impersonationCacheKey, wantedUser *user.DefaultInfo, attributes authorizer.Attributes) (*impersonatedUserInfo, error) {
+	requestor := attributes.GetUser()
+	return l.m.check(ctx, key, wantedUser, requestor)
+}
+
+func (l *legacyImpersonationCheck) verbForTests() string {
+	return l.m.verb
+}
+
+func (l *legacyImpersonationCheck) cachesForTests() (*impersonationCache, *impersonationCache) {
+	// legacy impersonation has no outer layer so just return an empty cache
+	// though an inner cache is present, it is unused
+	return newImpersonationCache(false), l.m.cache
+}
+
+func newConstrainedImpersonationMode(a authorizer.Authorizer, mode string, filter constrainedImpersonationModeFilter) impersonationMode {
+	return &constrainedImpersonationModeState{
+		state:      newImpersonationModeState(a, "impersonate:"+mode, true),
+		cache:      newImpersonationCache(false),
+		authorizer: a,
+		mode:       mode,
+		filter:     filter,
+	}
+}
+
+// constrainedImpersonationModeState implements the secondary authorization check via impersonate-on:<mode>:<verb> to
+// determine if the requestor is authorized to perform the specific verb when impersonating the wantedUser via mode.
+// if this check succeeds, the primary authorization checks are run, see impersonationModeState for details.
+// if the mode's filter does not match the inputs, the impersonation automatically fails and returns a nil impersonatedUserInfo.
+type constrainedImpersonationModeState struct {
+	state *impersonationModeState
+	// this outer cache covers the overall impersonation for this mode, i.e. a cache hit here short-circuits all checks
+	// skipAttributes is false, i.e. this cache depends on the request being made, not just the user being impersonated by the requestor
+	// it is expected to have a low hit ratio because the requestor is unlikely to make the same request multiple times in a short period
+	cache      *impersonationCache
+	authorizer authorizer.Authorizer
+	mode       string
+	filter     constrainedImpersonationModeFilter
+}
+
+func (c *constrainedImpersonationModeState) check(ctx context.Context, key *impersonationCacheKey, wantedUser *user.DefaultInfo, attributes authorizer.Attributes) (*impersonatedUserInfo, error) {
+	requestor := attributes.GetUser()
+	// we must call the filter before doing anything because this serves as a sudo authorization check to say "does this mode even apply?"
+	// also the cache key is not always a direct match with wantedUser+attributes, so again, we must call the filter first
+	if !c.filter(wantedUser, requestor) {
+		return nil, nil
+	}
+
+	if impersonatedUser := c.cache.get(key); impersonatedUser != nil {
+		return impersonatedUser, nil
+	}
+
+	if err := checkAuthorization(ctx, c.authorizer, &impersonateOnAttributes{mode: c.mode, Attributes: attributes}); err != nil {
+		return nil, err
+	}
+
+	impersonatedUser, err := c.state.check(ctx, key, wantedUser, requestor)
+	if err != nil || impersonatedUser == nil {
+		return nil, err
+	}
+	c.cache.set(key, impersonatedUser)
+	return impersonatedUser, nil
+}
+
+func (c *constrainedImpersonationModeState) verbForTests() string {
+	return c.state.verb
+}
+
+func (c *constrainedImpersonationModeState) cachesForTests() (*impersonationCache, *impersonationCache) {
+	return c.cache, c.state.cache
+}
+
+// impersonationModeState implements the primary authorization checks via the impersonate:<mode> verb for constrained
+// impersonation and the impersonate verb for legacy impersonation.  each field that is set in the wantedUser
+// results in one or more authorization checks to determine if the requestor has access to impersonate that value.
+type impersonationModeState struct {
+	authorizer                 authorizer.Authorizer
+	verb                       string
+	isConstrainedImpersonation bool
+
+	usernameAndGroupGV schema.GroupVersion
+	constraint         string
+
+	// this inner cache covers the checks related to the specific fields set in wantedUser
+	// skipAttributes is true, i.e. this cache only depends on the user being impersonated by the requestor
+	// it is expected to have a high hit ratio because the requestor may impersonate the same user for many different requests
+	cache *impersonationCache
+}
+
+func newImpersonationModeState(a authorizer.Authorizer, verb string, isConstrainedImpersonation bool) *impersonationModeState {
+	usernameAndGroupGV := authenticationv1.SchemeGroupVersion
+	constraint := verb
+	if !isConstrainedImpersonation {
+		usernameAndGroupGV = corev1.SchemeGroupVersion
+		constraint = ""
+	}
+	return &impersonationModeState{
+		authorizer:                 a,
+		verb:                       verb,
+		isConstrainedImpersonation: isConstrainedImpersonation,
+
+		usernameAndGroupGV: usernameAndGroupGV,
+		constraint:         constraint,
+		cache:              newImpersonationCache(true),
+	}
+}
+
+func (m *impersonationModeState) check(ctx context.Context, key *impersonationCacheKey, wantedUser *user.DefaultInfo, requestor user.Info) (*impersonatedUserInfo, error) {
+	// we only use caching in constrained impersonation mode to avoid any behavioral changes with legacy impersonation
+	if m.isConstrainedImpersonation {
+		if impersonatedUser := m.cache.get(key); impersonatedUser != nil {
+			return impersonatedUser, nil
+		}
+	}
+
+	actualUser := *wantedUser
+
+	if err := m.authorizeUsername(ctx, requestor, wantedUser.Name, wantedUser.Groups, &actualUser); err != nil {
+		return nil, err
+	}
+
+	if err := m.authorizeUID(ctx, requestor, wantedUser.UID); err != nil {
+		return nil, err
+	}
+
+	if err := m.authorizeGroups(ctx, requestor, wantedUser.Groups); err != nil {
+		return nil, err
+	}
+
+	if err := m.authorizeExtra(ctx, requestor, wantedUser.Extra); err != nil {
+		return nil, err
+	}
+
+	if actualUser.Name == user.Anonymous {
+		ensureGroup(&actualUser, user.AllUnauthenticated)
+	} else if !slices.Contains(actualUser.Groups, user.AllUnauthenticated) {
+		ensureGroup(&actualUser, user.AllAuthenticated)
+	}
+
+	impersonatedUser := &impersonatedUserInfo{user: &actualUser, constraint: m.constraint}
+	if m.isConstrainedImpersonation {
+		m.cache.set(key, impersonatedUser)
+	}
+	return impersonatedUser, nil
+}
+
+func (m *impersonationModeState) authorizeUsername(ctx context.Context, requestor user.Info, username string, wantedUserGroups []string, actualUser *user.DefaultInfo) error {
+	usernameAttributes := impersonationAttributes(requestor, m.usernameAndGroupGV, m.verb, "users", username)
+
+	if m.isConstrainedImpersonation {
+		if name, ok := isNodeUsername(username); ok {
+			usernameAttributes.Resource = "nodes"
+			usernameAttributes.Name = name
+
+			// this should be impossible to reach but check just in case
+			if len(wantedUserGroups) != 0 {
+				return responsewriters.ForbiddenStatusError(usernameAttributes, fmt.Sprintf("when impersonating a node, cannot impersonate groups %q", wantedUserGroups))
+			}
+
+			actualUser.Groups = []string{user.NodesGroup} // all nodes have a fixed group list in constrained impersonation
+		}
+	}
+
+	if namespace, name, ok := isServiceAccountUsername(username); ok {
+		usernameAttributes.Resource = "serviceaccounts"
+		usernameAttributes.Namespace = namespace
+		usernameAttributes.Name = name
+
+		// this should be impossible to reach but check just in case
+		if m.isConstrainedImpersonation && len(wantedUserGroups) != 0 {
+			return responsewriters.ForbiddenStatusError(usernameAttributes, fmt.Sprintf("when impersonating a service account, cannot impersonate groups %q", wantedUserGroups))
+		}
+
+		if len(wantedUserGroups) == 0 {
+			// if groups are not specified for a service account, we know the groups because it is a fixed mapping.  Add them
+			actualUser.Groups = serviceaccount.MakeGroupNames(namespace)
+		}
+	}
+
+	return checkAuthorization(ctx, m.authorizer, usernameAttributes)
+}
+
+func (m *impersonationModeState) authorizeUID(ctx context.Context, requestor user.Info, uid string) error {
+	if len(uid) == 0 {
+		return nil
+	}
+	uidAttributes := impersonationAttributes(requestor, authenticationv1.SchemeGroupVersion, m.verb, "uids", uid)
+	return checkAuthorization(ctx, m.authorizer, uidAttributes)
+}
+
+// manyAuthorizationChecksInLoop is an arbitrary value used in constrained impersonation modes to decide if they
+// should try to perform a single wildcard authorization check before making many individual checks in a loop.
+const manyAuthorizationChecksInLoop = 4
+
+func (m *impersonationModeState) authorizeGroups(ctx context.Context, requestor user.Info, groups []string) error {
+	if len(groups) == 0 {
+		return nil
+	}
+
+	groupAttributes := impersonationAttributes(requestor, m.usernameAndGroupGV, m.verb, "groups", "")
+
+	// perform extra sanity checks that would be backwards incompatible with legacy impersonation
+	if m.isConstrainedImpersonation {
+		if slices.Contains(groups, "") {
+			return responsewriters.ForbiddenStatusError(groupAttributes, "impersonating the empty string group is not allowed")
+		}
+		if slices.Contains(groups, user.SystemPrivilegedGroup) {
+			groupAttributes.Name = user.SystemPrivilegedGroup
+			return responsewriters.ForbiddenStatusError(groupAttributes, "impersonating the system:masters group is not allowed")
+		}
+	}
+
+	// if the requestor is trying to impersonate many groups at once, see if they are authorized to impersonate all groups
+	// we only do this in constrained impersonation mode to avoid any behavioral changes with legacy impersonation
+	if m.isConstrainedImpersonation && len(groups) >= manyAuthorizationChecksInLoop {
+		groupAttributes.Name = "*"
+		if err := checkAuthorization(ctx, m.authorizer, groupAttributes); err == nil {
+			return nil
+		}
+	}
+
+	for _, group := range groups {
+		groupAttributes.Name = group
+		if err := checkAuthorization(ctx, m.authorizer, groupAttributes); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (m *impersonationModeState) authorizeExtra(ctx context.Context, requestor user.Info, extra map[string][]string) error {
+	if len(extra) == 0 {
+		return nil
+	}
+
+	extraAttributes := impersonationAttributes(requestor, authenticationv1.SchemeGroupVersion, m.verb, "userextras", "")
+
+	// perform extra sanity checks that would be backwards incompatible with legacy impersonation
+	if m.isConstrainedImpersonation {
+		if err := validateExtra(extra); err != nil {
+			return responsewriters.ForbiddenStatusError(extraAttributes, err.Error())
+		}
+	}
+
+	// if the requestor is trying to impersonate many extras at once, see if they are authorized to impersonate all extras
+	// we only do this in constrained impersonation mode to avoid any behavioral changes with legacy impersonation
+	if m.isConstrainedImpersonation && isLargeExtra(extra) {
+		extraAttributes.Subresource = "*"
+		extraAttributes.Name = "*"
+		if err := checkAuthorization(ctx, m.authorizer, extraAttributes); err == nil {
+			return nil
+		}
+	}
+
+	for key, values := range extra {
+		extraAttributes.Subresource = key
+		for _, value := range values {
+			extraAttributes.Name = value
+			if err := checkAuthorization(ctx, m.authorizer, extraAttributes); err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+
+func validateExtra(extra map[string][]string) error {
+	fp := field.NewPath("extra", "key")
+	for key, values := range extra {
+		if len(key) == 0 {
+			return fmt.Errorf("impersonating the empty string key in extra is not allowed")
+		}
+		if err := utilvalidation.IsDomainPrefixedPath(fp, key).ToAggregate(); err != nil {
+			return fmt.Errorf("impersonating an invalid key in extra is not allowed: %w", err)
+		}
+		if key != strings.ToLower(key) {
+			return fmt.Errorf("impersonating a non-lowercase key in extra is not allowed: %q", key)
+		}
+		if len(values) == 0 {
+			return fmt.Errorf("impersonating empty values in extra is not allowed")
+		}
+		if slices.Contains(values, "") {
+			return fmt.Errorf("impersonating the empty string value in extra is not allowed")
+		}
+	}
+	return nil
+}
+
+func isLargeExtra(extra map[string][]string) bool {
+	if len(extra) >= manyAuthorizationChecksInLoop {
+		return true
+	}
+	var count int
+	for _, values := range extra {
+		count += len(values)
+		if count >= manyAuthorizationChecksInLoop {
+			return true
+		}
+	}
+	return false
+}
+
+func impersonationAttributes(requestor user.Info, gv schema.GroupVersion, verb, resource, name string) authorizer.AttributesRecord {
+	return authorizer.AttributesRecord{
+		User:            requestor,
+		Verb:            verb,
+		APIGroup:        gv.Group,
+		APIVersion:      gv.Version,
+		Resource:        resource,
+		Name:            name,
+		ResourceRequest: true,
+	}
+}
+
+// impersonateOnAttributes is a simple wrapper that updates the verb of the attributes to impersonate-on:<mode>:<verb>
+// This allows the expression of "a subject can perform this verb while using this impersonation mode"
+type impersonateOnAttributes struct {
+	mode string
+	authorizer.Attributes
+}
+
+func (i *impersonateOnAttributes) GetVerb() string {
+	return "impersonate-on:" + i.mode + ":" + i.Attributes.GetVerb()
+}
+
+func checkAuthorization(ctx context.Context, a authorizer.Authorizer, attributes authorizer.Attributes) error {
+	authorized, reason, err := a.Authorize(ctx, attributes)
+
+	// an authorizer like RBAC could encounter evaluation errors and still allow the request, so authorizer decision is checked before error here.
+	if authorized == authorizer.DecisionAllow {
+		return nil
+	}
+
+	// if the authorizer gave us a forbidden error, do not wrap it again
+	if errors.IsForbidden(err) {
+		return err
+	}
+
+	msg := reason
+	switch {
+	case err != nil && len(reason) > 0:
+		msg = fmt.Sprintf("%v: %s", err, reason)
+	case err != nil:
+		msg = err.Error()
+	}
+
+	return responsewriters.ForbiddenStatusError(attributes, msg)
+}
+
+func ensureGroup(u *user.DefaultInfo, group string) {
+	if slices.Contains(u.Groups, group) {
+		return
+	}
+
+	// do not mutate a slice that we did not create
+	groups := make([]string, 0, len(u.Groups)+1)
+	groups = append(groups, u.Groups...)
+	groups = append(groups, group)
+	u.Groups = groups
+}
+
+func isServiceAccountUsername(username string) (namespace, name string, ok bool) {
+	namespace, name, err := serviceaccount.SplitUsername(username)
+	return namespace, name, err == nil
+}
+
+// matches the real ValidateNodeName from k8s.io/kubernetes/pkg/apis/core/validation
+// which we are not allowed to import here
+var validateNodeName = validation.NameIsDNSSubdomain
+
+func isNodeUsername(username string) (string, bool) {
+	const nodeUsernamePrefix = "system:node:"
+	if !strings.HasPrefix(username, nodeUsernamePrefix) {
+		return "", false
+	}
+	name := strings.TrimPrefix(username, nodeUsernamePrefix)
+	if len(validateNodeName(name, false)) != 0 {
+		return "", false
+	}
+	return name, true
+}
+
+func requesterAssociatedWithRequestedNodeUsername(requestor user.Info, username string) bool {
+	nodeName, ok := isNodeUsername(username)
+	if !ok {
+		return false
+	}
+	if _, _, ok := isServiceAccountUsername(requestor.GetName()); !ok {
+		return false
+	}
+	return getExtraValue(requestor, serviceaccount.NodeNameKey) == nodeName
+}
+
+func getExtraValue(u user.Info, key string) string {
+	values := u.GetExtra()[key]
+	if len(values) != 1 {
+		return ""
+	}
+	return values[0]
+}
+
+func onlyUsernameSet(u user.Info) bool {
+	return len(u.GetUID()) == 0 && len(u.GetGroups()) == 0 && len(u.GetExtra()) == 0
+}

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/responsewriters/errors.go

@@ -17,12 +17,12 @@ limitations under the License.
 package responsewriters
 
 import (
-	"context"
 	"fmt"
 	"net/http"
 	"strings"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
@@ -33,10 +33,13 @@ import (
 var sanitizer = strings.NewReplacer(`&`, "&amp;", `<`, "&lt;", `>`, "&gt;")
 
 // Forbidden renders a simple forbidden error
-func Forbidden(ctx context.Context, attributes authorizer.Attributes, w http.ResponseWriter, req *http.Request, reason string, s runtime.NegotiatedSerializer) {
+func Forbidden(attributes authorizer.Attributes, w http.ResponseWriter, req *http.Request, reason string, s runtime.NegotiatedSerializer) {
+	RespondWithError(w, req, ForbiddenStatusError(attributes, reason), s)
+}
+
+func RespondWithError(w http.ResponseWriter, req *http.Request, err error, s runtime.NegotiatedSerializer) {
 	w.Header().Set("X-Content-Type-Options", "nosniff")
-	gv := schema.GroupVersion{Group: attributes.GetAPIGroup(), Version: attributes.GetAPIVersion()}
-	ErrorNegotiated(ForbiddenStatusError(attributes, reason), s, gv, w, req)
+	ErrorNegotiated(err, s, metav1.Unversioned, w, req)
 }
 
 func ForbiddenStatusError(attributes authorizer.Attributes, reason string) *apierrors.StatusError {

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/responsewriters/errors_test.go

@@ -78,7 +78,7 @@ func TestForbidden(t *testing.T) {
 		observer := httptest.NewRecorder()
 		scheme := runtime.NewScheme()
 		negotiatedSerializer := serializer.NewCodecFactory(scheme).WithoutConversion()
-		Forbidden(request.NewDefaultContext(), test.attributes, observer, &http.Request{URL: &url.URL{Path: "/path"}}, test.reason, negotiatedSerializer)
+		Forbidden(test.attributes, observer, &http.Request{URL: &url.URL{Path: "/path"}}, test.reason, negotiatedSerializer)
 		result := observer.Body.String()
 		if result != test.expected {
 			t.Errorf("Forbidden response body(%#v...)\n expected: %v\ngot:       %v", test.attributes, test.expected, result)

staging/src/k8s.io/apiserver/pkg/features/kube_features.go

@@ -109,6 +109,12 @@ const (
 	// Allow the API server to serve consistent lists from cache
 	ConsistentListFromCache featuregate.Feature = "ConsistentListFromCache"
 
+	// owner: @enj @qiujian16
+	// kep: https://kep.k8s.io/5284
+	//
+	// Enables impersonation that is constrained to specific requests instead of being all or nothing.
+	ConstrainedImpersonation featuregate.Feature = "ConstrainedImpersonation"
+
 	// owner: @jefftree
 	// kep: https://kep.k8s.io/4355
 	//
@@ -370,6 +376,10 @@ var defaultVersionedKubernetesFeatureGates = map[featuregate.Feature]featuregate
 		{Version: version.MustParse("1.34"), Default: true, PreRelease: featuregate.GA, LockToDefault: true},
 	},
 
+	ConstrainedImpersonation: {
+		{Version: version.MustParse("1.35"), Default: false, PreRelease: featuregate.Alpha},
+	},
+
 	CoordinatedLeaderElection: {
 		{Version: version.MustParse("1.31"), Default: false, PreRelease: featuregate.Alpha},
 		{Version: version.MustParse("1.33"), Default: false, PreRelease: featuregate.Beta},

staging/src/k8s.io/apiserver/pkg/server/config.go

@@ -55,6 +55,7 @@ import (
 	discoveryendpoint "k8s.io/apiserver/pkg/endpoints/discovery/aggregated"
 	"k8s.io/apiserver/pkg/endpoints/filterlatency"
 	genericapifilters "k8s.io/apiserver/pkg/endpoints/filters"
+	"k8s.io/apiserver/pkg/endpoints/filters/impersonation"
 	apiopenapi "k8s.io/apiserver/pkg/endpoints/openapi"
 	apirequest "k8s.io/apiserver/pkg/endpoints/request"
 	genericfeatures "k8s.io/apiserver/pkg/features"
@@ -1030,8 +1031,13 @@ func DefaultBuildHandlerChain(apiHandler http.Handler, c *Config) http.Handler {
 	}
 
 	handler = filterlatency.TrackCompleted(handler)
-	handler = genericapifilters.WithImpersonation(handler, c.Authorization.Authorizer, c.Serializer)
-	handler = filterlatency.TrackStarted(handler, c.TracerProvider, "impersonation")
+	if c.FeatureGate.Enabled(genericfeatures.ConstrainedImpersonation) {
+		handler = impersonation.WithConstrainedImpersonation(handler, c.Authorization.Authorizer, c.Serializer)
+		handler = filterlatency.TrackStarted(handler, c.TracerProvider, "constrainedimpersonation")
+	} else {
+		handler = impersonation.WithImpersonation(handler, c.Authorization.Authorizer, c.Serializer)
+		handler = filterlatency.TrackStarted(handler, c.TracerProvider, "impersonation")
+	}
 
 	handler = filterlatency.TrackCompleted(handler)
 	handler = genericapifilters.WithAudit(handler, c.AuditBackend, c.AuditPolicyRuleEvaluator, c.LongRunningFunc)

test/compatibility_lifecycle/reference/versioned_feature_list.yaml

@@ -253,6 +253,12 @@
     lockToDefault: true
     preRelease: GA
     version: "1.34"
+- name: ConstrainedImpersonation
+  versionedSpecs:
+  - default: false
+    lockToDefault: false
+    preRelease: Alpha
+    version: "1.35"
 - name: ContainerCheckpoint
   versionedSpecs:
   - default: false

test/integration/auth/auth_test.go

@@ -56,8 +56,12 @@ import (
 	"k8s.io/apiserver/pkg/authentication/request/bearertoken"
 	"k8s.io/apiserver/pkg/authentication/serviceaccount"
 	"k8s.io/apiserver/pkg/authentication/token/cache"
+	"k8s.io/apiserver/pkg/authentication/token/tokenfile"
+	"k8s.io/apiserver/pkg/authentication/user"
 	"k8s.io/apiserver/pkg/authorization/authorizer"
 	unionauthz "k8s.io/apiserver/pkg/authorization/union"
+	"k8s.io/apiserver/pkg/features"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	webhookutil "k8s.io/apiserver/pkg/util/webhook"
 	"k8s.io/apiserver/plugin/pkg/authenticator/token/webhook"
 	clientset "k8s.io/client-go/kubernetes"
@@ -65,6 +69,7 @@ import (
 	v1 "k8s.io/client-go/tools/clientcmd/api/v1"
 	resttransport "k8s.io/client-go/transport"
 	utiltesting "k8s.io/client-go/util/testing"
+	featuregatetesting "k8s.io/component-base/featuregate/testing"
 	"k8s.io/kubernetes/cmd/kube-apiserver/app/options"
 	kubeapiservertesting "k8s.io/kubernetes/cmd/kube-apiserver/app/testing"
 	"k8s.io/kubernetes/pkg/apis/autoscaling"
@@ -1007,16 +1012,14 @@ func TestImpersonateWithUID(t *testing.T) {
 		client := clientset.NewForConfigOrDie(clientConfig)
 		_, err := client.CoreV1().Nodes().List(ctx, metav1.ListOptions{})
 
-		if !errors.IsInternalError(err) {
-			t.Fatalf("expected internal error, got %T %v", err, err)
+		if !errors.IsBadRequest(err) {
+			t.Fatalf("expected bad request, got %T %v", err, err)
 		}
 		if diff := cmp.Diff(
-			`an error on the server ("Internal Server Error: \"/api/v1/nodes\": `+
-				`requested [{UID  1234  authentication.k8s.io/v1  }] without impersonating a user") `+
-				`has prevented the request from succeeding (get nodes)`,
+			`requested [{UID  1234  authentication.k8s.io/v1  }] without impersonating a user`,
 			err.Error(),
 		); diff != "" {
-			t.Fatalf("internal error different than expected, -got, +want:\n %s", diff)
+			t.Fatalf("bad request different than expected, -got, +want:\n %s", diff)
 		}
 	})
 
@@ -1054,6 +1057,391 @@ func TestImpersonateWithUID(t *testing.T) {
 	})
 }
 
+// TestConstrainedImpersonation tests the constrained impersonation feature.
+// It ensures that users can only perform actions on behalf of other users
+// if they have the appropriate permissions.
+// This test covers:
+// - A user attempting to impersonate another user.
+// - A user attempting to impersonate a node.
+// - A service account attempting to impersonate a node it is scheduled on.
+// - The fallback to legacy impersonation when the feature is enabled.
+func TestConstrainedImpersonation(t *testing.T) {
+	superUser := "admin/system:masters"
+
+	authenticator := group.NewAuthenticatedGroupAdder(bearertoken.New(tokenfile.New(map[string]*user.DefaultInfo{
+		superUser: {Name: "admin", Groups: []string{"system:masters"}},
+		"bob":     {Name: "bob"},
+		"alice":   {Name: "alice"},
+		"node1":   {Name: "system:node:node1", Groups: []string{user.NodesGroup}},
+		"serviceaccount1": {Name: "system:serviceaccount:default:sa1", Extra: map[string][]string{
+			"authentication.kubernetes.io/node-name": {"node1"},
+		}},
+		"serviceaccount2": {Name: "system:serviceaccount:default:sa2", Extra: map[string][]string{
+			"authentication.kubernetes.io/node-name": {"node2"},
+		}},
+	})))
+
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
+	t.Cleanup(cancel)
+
+	featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ConstrainedImpersonation, true)
+	_, kubeConfig, tearDownFn := framework.StartTestServer(ctx, t, framework.TestServerSetup{
+		ModifyServerRunOptions: func(opts *options.ServerRunOptions) {
+			opts.Authorization.Modes = []string{"RBAC"}
+		},
+		ModifyServerConfig: func(config *controlplane.Config) {
+			config.ControlPlane.Generic.Authentication.Authenticator = authenticator
+		},
+	})
+	t.Cleanup(tearDownFn)
+
+	superuserClient, _ := clientsetForToken(superUser, kubeConfig)
+
+	// preset permissions for users to be impersonated
+	authutil.GrantUserAuthorization(t, ctx, superuserClient, "system:node:node1", rbacv1.PolicyRule{
+		Verbs:     []string{"list"},
+		APIGroups: []string{""},
+		Resources: []string{"pods"},
+	})
+	authutil.GrantUserAuthorization(t, ctx, superuserClient, "alice", rbacv1.PolicyRule{
+		Verbs:     []string{"list"},
+		APIGroups: []string{""},
+		Resources: []string{"pods"},
+	})
+
+	t.Run("bob impersonating alice", func(t *testing.T) {
+		impersonatorClientConfig := rest.CopyConfig(kubeConfig)
+		impersonatorClientConfig.BearerToken = "bob"
+		impersonatorClientConfig.Impersonate = rest.ImpersonationConfig{
+			UserName: "alice",
+		}
+
+		client := clientset.NewForConfigOrDie(impersonatorClientConfig)
+		_, err := client.CoreV1().Pods(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
+		if !errors.IsForbidden(err) {
+			t.Fatalf("expected forbidden error, got %T %v", err, err)
+		}
+
+		if diff := cmp.Diff(
+			`pods is forbidden: User "bob" cannot impersonate-on:user-info:list resource "pods" in API group "" at the cluster scope`,
+			err.Error(),
+		); diff != "" {
+			t.Fatalf("forbidden error different than expected, -got, +want:\n %s", diff)
+		}
+
+		// with impersonation:user-info permission added, bob still cannot list pods
+		authutil.GrantUserAuthorization(t, ctx, superuserClient, "bob", rbacv1.PolicyRule{
+			Verbs:     []string{"impersonate:user-info"},
+			APIGroups: []string{"authentication.k8s.io"},
+			Resources: []string{"users"},
+		})
+
+		_, err = client.CoreV1().Pods(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
+		if !errors.IsForbidden(err) {
+			t.Fatalf("expected forbidden error, got %T %v", err, err)
+		}
+
+		if diff := cmp.Diff(
+			`pods is forbidden: User "bob" cannot impersonate-on:user-info:list resource "pods" in API group "" at the cluster scope`,
+			err.Error(),
+		); diff != "" {
+			t.Fatalf("forbidden error different than expected, -got, +want:\n %s", diff)
+		}
+
+		// with impersonate-on:list permission added, bob can list but not watch pods.
+		authutil.GrantUserAuthorization(t, ctx, superuserClient, "bob", rbacv1.PolicyRule{
+			Verbs:     []string{"impersonate-on:user-info:list"},
+			APIGroups: []string{""},
+			Resources: []string{"pods"},
+		})
+
+		_, err = client.CoreV1().Pods(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
+		if err != nil {
+			t.Fatalf("expected no error, got %T %v", err, err)
+		}
+
+		_, err = client.CoreV1().Pods(metav1.NamespaceAll).Watch(ctx, metav1.ListOptions{})
+		if !errors.IsForbidden(err) {
+			t.Fatalf("expected forbidden error, got %T %v", err, err)
+		}
+	})
+
+	t.Run("bob impersonating a node", func(t *testing.T) {
+		impersonatorClientConfig := rest.CopyConfig(kubeConfig)
+		impersonatorClientConfig.BearerToken = "bob"
+		impersonatorClientConfig.Impersonate = rest.ImpersonationConfig{
+			UserName: "system:node:node1",
+		}
+
+		client := clientset.NewForConfigOrDie(impersonatorClientConfig)
+		_, err := client.CoreV1().Pods(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
+		if !errors.IsForbidden(err) {
+			t.Fatalf("expected forbidden error, got %T %v", err, err)
+		}
+
+		if diff := cmp.Diff(
+			`pods is forbidden: User "bob" cannot impersonate-on:arbitrary-node:list resource "pods" in API group "" at the cluster scope`,
+			err.Error(),
+		); diff != "" {
+			t.Fatalf("forbidden error different than expected, -got, +want:\n %s", diff)
+		}
+
+		// with permissions added, bob still cannot list pods since bob needs
+		// permission to impersonate node.
+		authutil.GrantUserAuthorization(t, ctx, superuserClient, "bob", rbacv1.PolicyRule{
+			Verbs:     []string{"impersonate:user-info"},
+			APIGroups: []string{"authentication.k8s.io"},
+			Resources: []string{"users"},
+		})
+		authutil.GrantUserAuthorization(t, ctx, superuserClient, "bob", rbacv1.PolicyRule{
+			Verbs:     []string{"impersonate-on:arbitrary-node:list"},
+			APIGroups: []string{""},
+			Resources: []string{"pods"},
+		})
+
+		_, err = client.CoreV1().Pods(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
+		if !errors.IsForbidden(err) {
+			t.Fatalf("expected forbidden error, got %T %v", err, err)
+		}
+
+		if diff := cmp.Diff(
+			`nodes.authentication.k8s.io "node1" is forbidden: User "bob" cannot impersonate:arbitrary-node resource "nodes" in API group "authentication.k8s.io" at the cluster scope`,
+			err.Error(),
+		); diff != "" {
+			t.Fatalf("forbidden error different than expected, -got, +want:\n %s", diff)
+		}
+
+		authutil.GrantUserAuthorization(t, ctx, superuserClient, "bob", rbacv1.PolicyRule{
+			Verbs:     []string{"impersonate:arbitrary-node"},
+			APIGroups: []string{"authentication.k8s.io"},
+			Resources: []string{"nodes"},
+		})
+
+		_, err = client.CoreV1().Pods(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
+		if err != nil {
+			t.Fatalf("expected no error, got %T %v", err, err)
+		}
+	})
+
+	t.Run("impersonating scheduled node", func(t *testing.T) {
+		impersonatorClientConfig := rest.CopyConfig(kubeConfig)
+		impersonatorClientConfig.BearerToken = "serviceaccount2"
+		impersonatorClientConfig.Impersonate = rest.ImpersonationConfig{
+			UserName: "system:node:node1",
+		}
+
+		client := clientset.NewForConfigOrDie(impersonatorClientConfig)
+
+		// with permissions added, it cannot list pods since sa on the node2 instead of node1.
+		authutil.GrantUserAuthorization(t, ctx, superuserClient, "system:serviceaccount:default:sa2", rbacv1.PolicyRule{
+			Verbs:     []string{"impersonate:associated-node"},
+			APIGroups: []string{"authentication.k8s.io"},
+			Resources: []string{"nodes"},
+		})
+		authutil.GrantUserAuthorization(t, ctx, superuserClient, "system:serviceaccount:default:sa2", rbacv1.PolicyRule{
+			Verbs:     []string{"impersonate-on:associated-node:list"},
+			APIGroups: []string{""},
+			Resources: []string{"pods"},
+		})
+
+		_, err := client.CoreV1().Pods(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
+		if !errors.IsForbidden(err) {
+			t.Fatalf("expected forbidden error, got %T %v", err, err)
+		}
+
+		if diff := cmp.Diff(
+			`pods is forbidden: User "system:serviceaccount:default:sa2" cannot impersonate-on:arbitrary-node:list resource "pods" in API group "" at the cluster scope`,
+			err.Error(),
+		); diff != "" {
+			t.Fatalf("forbidden error different than expected, -got, +want:\n %s", diff)
+		}
+
+		// change to service account1 which is at node1
+		impersonatorClientConfig.BearerToken = "serviceaccount1"
+
+		client = clientset.NewForConfigOrDie(impersonatorClientConfig)
+		authutil.GrantUserAuthorization(t, ctx, superuserClient, "system:serviceaccount:default:sa1", rbacv1.PolicyRule{
+			Verbs:     []string{"impersonate:associated-node"},
+			APIGroups: []string{"authentication.k8s.io"},
+			Resources: []string{"nodes"},
+		})
+		authutil.GrantUserAuthorization(t, ctx, superuserClient, "system:serviceaccount:default:sa1", rbacv1.PolicyRule{
+			Verbs:     []string{"impersonate-on:associated-node:list"},
+			APIGroups: []string{""},
+			Resources: []string{"pods"},
+		})
+
+		_, err = client.CoreV1().Pods(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
+		if err != nil {
+			t.Fatalf("expected no error, got %T %v", err, err)
+		}
+	})
+
+	t.Run("fallback to legacy impersonation", func(t *testing.T) {
+		impersonatorClientConfig := rest.CopyConfig(kubeConfig)
+		impersonatorClientConfig.BearerToken = "bob"
+		impersonatorClientConfig.Impersonate = rest.ImpersonationConfig{
+			UserName: "alice",
+		}
+
+		client := clientset.NewForConfigOrDie(impersonatorClientConfig)
+
+		// with legacy impersonation verb, bob can impersonate alice.
+		authutil.GrantUserAuthorization(t, ctx, superuserClient, "bob", rbacv1.PolicyRule{
+			Verbs:     []string{"impersonate"},
+			APIGroups: []string{""},
+			Resources: []string{"users"},
+		})
+
+		_, err := client.CoreV1().Pods(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
+		if err != nil {
+			t.Fatalf("expected no error, got %T %v", err, err)
+		}
+	})
+}
+
+// TestConstrainedImpersonationDisabled tests the impersonation behavior when the
+// ConstrainedImpersonation feature gate is disabled. In this mode, the legacy
+// impersonation behavior is expected, where a user only needs the "impersonate"
+// permission on the user, group, or service account they are trying to
+// impersonate.
+func TestConstrainedImpersonationDisabled(t *testing.T) {
+	superUser := "admin/system:masters"
+
+	authenticator := group.NewAuthenticatedGroupAdder(bearertoken.New(tokenfile.New(map[string]*user.DefaultInfo{
+		superUser: {Name: "admin", Groups: []string{"system:masters"}},
+		"bob":     {Name: "bob"},
+		"alice":   {Name: "alice"},
+		"node1":   {Name: "system:node:node1", Groups: []string{user.NodesGroup}},
+		"serviceaccount1": {Name: "system:serviceaccount:default:sa1", Extra: map[string][]string{
+			"authentication.kubernetes.io/node-name": {"node1"},
+		}},
+	})))
+
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
+	t.Cleanup(cancel)
+
+	featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ConstrainedImpersonation, false)
+	_, kubeConfig, tearDownFn := framework.StartTestServer(ctx, t, framework.TestServerSetup{
+		ModifyServerRunOptions: func(opts *options.ServerRunOptions) {
+			opts.Authorization.Modes = []string{"RBAC"}
+		},
+		ModifyServerConfig: func(config *controlplane.Config) {
+			config.ControlPlane.Generic.Authentication.Authenticator = authenticator
+		},
+	})
+	t.Cleanup(tearDownFn)
+
+	superuserClient, _ := clientsetForToken(superUser, kubeConfig)
+
+	// preset permissions for users to be impersonated
+	authutil.GrantUserAuthorization(t, ctx, superuserClient, "system:node:node1", rbacv1.PolicyRule{
+		Verbs:     []string{"list"},
+		APIGroups: []string{""},
+		Resources: []string{"pods"},
+	})
+	authutil.GrantUserAuthorization(t, ctx, superuserClient, "alice", rbacv1.PolicyRule{
+		Verbs:     []string{"list"},
+		APIGroups: []string{""},
+		Resources: []string{"pods"},
+	})
+
+	t.Run("bob impersonating alice", func(t *testing.T) {
+		authutil.GrantUserAuthorization(t, ctx, superuserClient, "bob", rbacv1.PolicyRule{
+			Verbs:     []string{"impersonate:user-info"},
+			APIGroups: []string{"authentication.k8s.io"},
+			Resources: []string{"users"},
+		})
+		authutil.GrantUserAuthorization(t, ctx, superuserClient, "bob", rbacv1.PolicyRule{
+			Verbs:     []string{"impersonate-on:user-info:list"},
+			APIGroups: []string{""},
+			Resources: []string{"pods"},
+		})
+
+		impersonatorClientConfig := rest.CopyConfig(kubeConfig)
+		impersonatorClientConfig.BearerToken = "bob"
+		impersonatorClientConfig.Impersonate = rest.ImpersonationConfig{
+			UserName: "alice",
+		}
+
+		client := clientset.NewForConfigOrDie(impersonatorClientConfig)
+		_, err := client.CoreV1().Pods(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
+		if !errors.IsForbidden(err) {
+			t.Fatalf("expected forbidden error, got %T %v", err, err)
+		}
+
+		if diff := cmp.Diff(
+			`users "alice" is forbidden: User "bob" cannot impersonate resource "users" in API group "" at the cluster scope`,
+			err.Error(),
+		); diff != "" {
+			t.Fatalf("forbidden error different than expected, -got, +want:\n %s", diff)
+		}
+
+		// with legacy impersonation permission added, bob can list pods
+		authutil.GrantUserAuthorization(t, ctx, superuserClient, "bob", rbacv1.PolicyRule{
+			Verbs:     []string{"impersonate"},
+			APIGroups: []string{""},
+			Resources: []string{"users"},
+		})
+
+		_, err = client.CoreV1().Pods(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
+		if err != nil {
+			t.Fatalf("expected no error, got %T %v", err, err)
+		}
+	})
+
+	t.Run("serviceaccount impersonating a node", func(t *testing.T) {
+		authutil.GrantUserAuthorization(t, ctx, superuserClient, "system:serviceaccount:default:sa1", rbacv1.PolicyRule{
+			Verbs:     []string{"impersonate:associated-node"},
+			APIGroups: []string{"authentication.k8s.io"},
+			Resources: []string{"nodes"},
+		})
+		authutil.GrantUserAuthorization(t, ctx, superuserClient, "system:serviceaccount:default:sa1", rbacv1.PolicyRule{
+			Verbs:     []string{"impersonate-on:associated-node:list"},
+			APIGroups: []string{""},
+			Resources: []string{"pods"},
+		})
+
+		impersonatorClientConfig := rest.CopyConfig(kubeConfig)
+		impersonatorClientConfig.BearerToken = "serviceaccount1"
+		impersonatorClientConfig.Impersonate = rest.ImpersonationConfig{
+			UserName: "system:node:node1",
+		}
+
+		client := clientset.NewForConfigOrDie(impersonatorClientConfig)
+		_, err := client.CoreV1().Pods(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
+		if !errors.IsForbidden(err) {
+			t.Fatalf("expected forbidden error, got %T %v", err, err)
+		}
+
+		if diff := cmp.Diff(
+			`users "system:node:node1" is forbidden: User "system:serviceaccount:default:sa1" cannot impersonate resource "users" in API group "" at the cluster scope`,
+			err.Error(),
+		); diff != "" {
+			t.Fatalf("forbidden error different than expected, -got, +want:\n %s", diff)
+		}
+
+		// with legacy impersonation permission added, sa can list pods
+		authutil.GrantUserAuthorization(t, ctx, superuserClient, "system:serviceaccount:default:sa1", rbacv1.PolicyRule{
+			Verbs:         []string{"impersonate"},
+			APIGroups:     []string{""},
+			Resources:     []string{"users"},
+			ResourceNames: []string{"system:node:node1"},
+		})
+		authutil.GrantUserAuthorization(t, ctx, superuserClient, "system:serviceaccount:default:sa1", rbacv1.PolicyRule{
+			Verbs:         []string{"impersonate"},
+			APIGroups:     []string{""},
+			Resources:     []string{"groups"},
+			ResourceNames: []string{"system:nodes"},
+		})
+
+		_, err = client.CoreV1().Pods(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
+		if err != nil {
+			t.Fatalf("expected no error, got %T %v", err, err)
+		}
+	})
+}
+
 func csrPEM(t *testing.T) []byte {
 	t.Helper()