add access control check for unsafe delete

add access control check to ensure that the user has permission to do
'unsafe-delete-ignore-read-error' on the resource being deleted

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/delete.go

@@ -30,11 +30,14 @@ import (
 	metainternalversionvalidation "k8s.io/apimachinery/pkg/apis/meta/internalversion/validation"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/validation"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/validation/field"
 	"k8s.io/apiserver/pkg/admission"
 	"k8s.io/apiserver/pkg/audit"
+	"k8s.io/apiserver/pkg/authorization/authorizer"
 	"k8s.io/apiserver/pkg/endpoints/handlers/finisher"
 	requestmetrics "k8s.io/apiserver/pkg/endpoints/handlers/metrics"
 	"k8s.io/apiserver/pkg/endpoints/handlers/negotiation"
@@ -45,6 +48,8 @@ import (
 	"k8s.io/apiserver/pkg/util/dryrun"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/component-base/tracing"
+
+	"k8s.io/klog/v2"
 	"k8s.io/utils/ptr"
 )
 
@@ -128,6 +133,9 @@ func DeleteResource(r rest.GracefulDeleter, allowsOptions bool, scope *RequestSc
 		}
 		options.TypeMeta.SetGroupVersionKind(metav1.SchemeGroupVersion.WithKind("DeleteOptions"))
 
+		userInfo, _ := request.UserFrom(ctx)
+		staticAdmissionAttrs := admission.NewAttributesRecord(nil, nil, scope.Kind, namespace, name, scope.Resource, scope.Subresource, admission.Delete, options, dryrun.IsDryRun(options.DryRun), userInfo)
+
 		if utilfeature.DefaultFeatureGate.Enabled(features.AllowUnsafeMalformedObjectDeletion) {
 			if options != nil && ptr.Deref(options.IgnoreStoreReadErrorWithClusterBreakingPotential, false) {
 				// let's make sure that the audit will reflect that this delete request
@@ -140,14 +148,21 @@ func DeleteResource(r rest.GracefulDeleter, allowsOptions bool, scope *RequestSc
 					scope.err(errors.NewInternalError(fmt.Errorf("no unsafe deleter provided, can not honor ignoreStoreReadErrorWithClusterBreakingPotential")), w, req)
 					return
 				}
+				if scope.Authorizer == nil {
+					scope.err(errors.NewInternalError(fmt.Errorf("no authorizer provided, unable to authorize unsafe delete")), w, req)
+					return
+				}
+				if err := authorizeUnsafeDelete(ctx, staticAdmissionAttrs, scope.Authorizer); err != nil {
+					scope.err(err, w, req)
+					return
+				}
+
 				r = p.GetCorruptObjDeleter()
 			}
 		}
 
 		span.AddEvent("About to delete object from database")
 		wasDeleted := true
-		userInfo, _ := request.UserFrom(ctx)
-		staticAdmissionAttrs := admission.NewAttributesRecord(nil, nil, scope.Kind, namespace, name, scope.Resource, scope.Subresource, admission.Delete, options, dryrun.IsDryRun(options.DryRun), userInfo)
 		result, err := finisher.FinishRequest(ctx, func() (runtime.Object, error) {
 			obj, deleted, err := r.Delete(ctx, name, rest.AdmissionToValidateObjectDeleteFunc(admit, staticAdmissionAttrs, scope), options)
 			wasDeleted = deleted
@@ -331,3 +346,77 @@ func DeleteCollection(r rest.CollectionDeleter, checkBody bool, scope *RequestSc
 		transformResponseObject(ctx, scope, req, w, http.StatusOK, outputMediaType, result)
 	}
 }
+
+// authorizeUnsafeDelete ensures that the user has permission to do
+// 'unsafe-delete-ignore-read-errors' on the resource being deleted when
+// ignoreStoreReadErrorWithClusterBreakingPotential is enabled
+func authorizeUnsafeDelete(ctx context.Context, attr admission.Attributes, authz authorizer.Authorizer) (err error) {
+	if attr.GetOperation() != admission.Delete || attr.GetOperationOptions() == nil {
+		return nil
+	}
+	options, ok := attr.GetOperationOptions().(*metav1.DeleteOptions)
+	if !ok {
+		return errors.NewInternalError(fmt.Errorf("expected an option of type: %T, but got: %T", &metav1.DeleteOptions{}, attr.GetOperationOptions()))
+	}
+	if !ptr.Deref(options.IgnoreStoreReadErrorWithClusterBreakingPotential, false) {
+		return nil
+	}
+
+	requestInfo, found := request.RequestInfoFrom(ctx)
+	if !found {
+		return admission.NewForbidden(attr, fmt.Errorf("no RequestInfo found in the context"))
+	}
+	if !requestInfo.IsResourceRequest || len(attr.GetSubresource()) > 0 {
+		return admission.NewForbidden(attr, fmt.Errorf("ignoreStoreReadErrorWithClusterBreakingPotential delete option is not allowed on a subresource or non-resource request"))
+	}
+
+	// if we are here, IgnoreStoreReadErrorWithClusterBreakingPotential
+	// is set to true in the delete options, the user must have permission
+	// to do 'unsafe-delete-ignore-read-errors' on the given resource.
+	record := authorizer.AttributesRecord{
+		User:            attr.GetUserInfo(),
+		Verb:            "unsafe-delete-ignore-read-errors",
+		Namespace:       attr.GetNamespace(),
+		Name:            attr.GetName(),
+		APIGroup:        attr.GetResource().Group,
+		APIVersion:      attr.GetResource().Version,
+		Resource:        attr.GetResource().Resource,
+		ResourceRequest: true,
+	}
+	// TODO: can't use ResourceAttributesFrom from k8s.io/kubernetes/pkg/registry/authorization/util
+	// due to prevent staging --> k8s.io/kubernetes dep issue
+	if utilfeature.DefaultFeatureGate.Enabled(features.AuthorizeWithSelectors) {
+		if len(requestInfo.FieldSelector) > 0 {
+			fieldSelector, err := fields.ParseSelector(requestInfo.FieldSelector)
+			if err != nil {
+				record.FieldSelectorRequirements, record.FieldSelectorParsingErr = nil, err
+			} else {
+				if requirements := fieldSelector.Requirements(); len(requirements) > 0 {
+					record.FieldSelectorRequirements, record.FieldSelectorParsingErr = fieldSelector.Requirements(), nil
+				}
+			}
+		}
+		if len(requestInfo.LabelSelector) > 0 {
+			labelSelector, err := labels.Parse(requestInfo.LabelSelector)
+			if err != nil {
+				record.LabelSelectorRequirements, record.LabelSelectorParsingErr = nil, err
+			} else {
+				if requirements, _ /*selectable*/ := labelSelector.Requirements(); len(requirements) > 0 {
+					record.LabelSelectorRequirements, record.LabelSelectorParsingErr = requirements, nil
+				}
+			}
+		}
+	}
+
+	decision, reason, err := authz.Authorize(ctx, record)
+	if err != nil {
+		err = fmt.Errorf("error while checking permission for %q, %w", record.Verb, err)
+		klog.FromContext(ctx).V(1).Error(err, "failed to authorize")
+		return admission.NewForbidden(attr, err)
+	}
+	if decision == authorizer.DecisionAllow {
+		return nil
+	}
+
+	return admission.NewForbidden(attr, fmt.Errorf("not permitted to do %q, reason: %s", record.Verb, reason))
+}

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/delete_test.go

@@ -18,6 +18,7 @@ package handlers
 
 import (
 	"context"
+	"fmt"
 	"io"
 	"net/http"
 	"net/http/httptest"
@@ -25,17 +26,24 @@ import (
 	"sync/atomic"
 	"testing"
 
+	"k8s.io/apimachinery/pkg/api/errors"
 	metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion"
 	metainternalversionscheme "k8s.io/apimachinery/pkg/apis/meta/internalversion/scheme"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/runtime/serializer"
+	"k8s.io/apiserver/pkg/admission"
 	auditapis "k8s.io/apiserver/pkg/apis/audit"
 	"k8s.io/apiserver/pkg/audit"
+	"k8s.io/apiserver/pkg/authentication/user"
+	"k8s.io/apiserver/pkg/authorization/authorizer"
 	"k8s.io/apiserver/pkg/endpoints/handlers/negotiation"
+	"k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/apiserver/pkg/registry/rest"
+
 	"k8s.io/utils/pointer"
+	"k8s.io/utils/ptr"
 )
 
 type mockCodecs struct {
@@ -273,3 +281,248 @@ func (n *fakeSerializer) EncoderForVersion(serializer runtime.Encoder, gv runtim
 func (n *fakeSerializer) DecoderToVersion(serializer runtime.Decoder, gv runtime.GroupVersioner) runtime.Decoder {
 	return n.serializer
 }
+
+func TestAuthorizeUnsafeDelete(t *testing.T) {
+	const verbWant = "unsafe-delete-ignore-read-errors"
+	tests := []struct {
+		name    string
+		reqInfo *request.RequestInfo
+		attr    admission.Attributes
+		authz   authorizer.Authorizer
+		err     func(admission.Attributes) error
+	}{
+		{
+			name:  "operation is not delete, admit",
+			attr:  newAttributes(attributes{operation: admission.Update}),
+			authz: nil, // Authorize should not be invoked
+		},
+		{
+			name: "feature enabled, delete, operation option is nil, admit",
+			attr: newAttributes(attributes{
+				operation:        admission.Delete,
+				operationOptions: nil,
+			}),
+			authz: nil, // Authorize should not be invoked
+		},
+		{
+			name: "delete, operation option is not a match, forbid",
+			attr: newAttributes(attributes{
+				operation:        admission.Delete,
+				operationOptions: &metav1.PatchOptions{},
+			}),
+			authz: nil, // Authorize should not be invoked
+			err: func(admission.Attributes) error {
+				return errors.NewInternalError(fmt.Errorf("expected an option of type: %T, but got: %T", &metav1.DeleteOptions{}, &metav1.PatchOptions{}))
+			},
+		},
+		{
+			name: "delete, IgnoreStoreReadErrorWithClusterBreakingPotential is nil, admit",
+			attr: newAttributes(attributes{
+				operation: admission.Delete,
+				operationOptions: &metav1.DeleteOptions{
+					IgnoreStoreReadErrorWithClusterBreakingPotential: nil,
+				},
+			}),
+			authz: nil, // Authorize should not be invoked
+		},
+		{
+			name: "delete, IgnoreStoreReadErrorWithClusterBreakingPotential is false, admit",
+			attr: newAttributes(attributes{
+				operation: admission.Delete,
+				operationOptions: &metav1.DeleteOptions{
+					IgnoreStoreReadErrorWithClusterBreakingPotential: ptr.To[bool](false),
+				},
+			}),
+			authz: nil, // Authorize should not be invoked
+		},
+		{
+			name:    "feature enabled, delete, IgnoreStoreReadErrorWithClusterBreakingPotential is true, no RequestInfo in request context, forbid",
+			reqInfo: nil,
+			attr: newAttributes(attributes{
+				operation: admission.Delete,
+				operationOptions: &metav1.DeleteOptions{
+					IgnoreStoreReadErrorWithClusterBreakingPotential: ptr.To[bool](true),
+				},
+			}),
+			authz: nil,
+			err: func(attr admission.Attributes) error {
+				return admission.NewForbidden(attr, fmt.Errorf("no RequestInfo found in the context"))
+			},
+		},
+		{
+			name:    "delete, IgnoreStoreReadErrorWithClusterBreakingPotential is true, subresource request, forbid",
+			reqInfo: &request.RequestInfo{IsResourceRequest: true},
+			attr: newAttributes(attributes{
+				operation:   admission.Delete,
+				subresource: "foo",
+				operationOptions: &metav1.DeleteOptions{
+					IgnoreStoreReadErrorWithClusterBreakingPotential: ptr.To[bool](true),
+				},
+			}),
+			authz: nil,
+			err: func(attr admission.Attributes) error {
+				return admission.NewForbidden(attr, fmt.Errorf("ignoreStoreReadErrorWithClusterBreakingPotential delete option is not allowed on a subresource or non-resource request"))
+			},
+		},
+		{
+			name:    "delete, IgnoreStoreReadErrorWithClusterBreakingPotential is true, subresource request, forbid",
+			reqInfo: &request.RequestInfo{IsResourceRequest: false},
+			attr: newAttributes(attributes{
+				operation:   admission.Delete,
+				subresource: "",
+				operationOptions: &metav1.DeleteOptions{
+					IgnoreStoreReadErrorWithClusterBreakingPotential: ptr.To[bool](true),
+				},
+			}),
+			authz: nil,
+			err: func(attr admission.Attributes) error {
+				return admission.NewForbidden(attr, fmt.Errorf("ignoreStoreReadErrorWithClusterBreakingPotential delete option is not allowed on a subresource or non-resource request"))
+			},
+		},
+		{
+			name:    "delete, IgnoreStoreReadErrorWithClusterBreakingPotential is true, authorizer returns error, forbid",
+			reqInfo: &request.RequestInfo{IsResourceRequest: true},
+			attr: newAttributes(attributes{
+				subresource: "",
+				operation:   admission.Delete,
+				operationOptions: &metav1.DeleteOptions{
+					IgnoreStoreReadErrorWithClusterBreakingPotential: ptr.To[bool](true),
+				},
+			}),
+			authz: &fakeAuthorizer{err: fmt.Errorf("unexpected error")},
+			err: func(attr admission.Attributes) error {
+				return admission.NewForbidden(attr, fmt.Errorf("error while checking permission for %q, %w", verbWant, fmt.Errorf("unexpected error")))
+			},
+		},
+		{
+			name:    "delete, IgnoreStoreReadErrorWithClusterBreakingPotential is true, user does not have permission, forbid",
+			reqInfo: &request.RequestInfo{IsResourceRequest: true},
+			attr: newAttributes(attributes{
+				operation:   admission.Delete,
+				subresource: "",
+				operationOptions: &metav1.DeleteOptions{
+					IgnoreStoreReadErrorWithClusterBreakingPotential: ptr.To[bool](true),
+				},
+			}),
+			authz: &fakeAuthorizer{
+				decision: authorizer.DecisionDeny,
+				reason:   "does not have permission",
+			},
+			err: func(attr admission.Attributes) error {
+				return admission.NewForbidden(attr, fmt.Errorf("not permitted to do %q, reason: %s", verbWant, "does not have permission"))
+			},
+		},
+		{
+			name:    "delete, IgnoreStoreReadErrorWithClusterBreakingPotential is true, authorizer gives no opinion, forbid",
+			reqInfo: &request.RequestInfo{IsResourceRequest: true},
+			attr: newAttributes(attributes{
+				operation:   admission.Delete,
+				subresource: "",
+				operationOptions: &metav1.DeleteOptions{
+					IgnoreStoreReadErrorWithClusterBreakingPotential: ptr.To[bool](true),
+				},
+			}),
+			authz: &fakeAuthorizer{
+				decision: authorizer.DecisionNoOpinion,
+				reason:   "no opinion",
+			},
+			err: func(attr admission.Attributes) error {
+				return admission.NewForbidden(attr, fmt.Errorf("not permitted to do %q, reason: %s", verbWant, "no opinion"))
+			},
+		},
+		{
+			name:    "delete, IgnoreStoreReadErrorWithClusterBreakingPotential is true, user has permission, admit",
+			reqInfo: &request.RequestInfo{IsResourceRequest: true},
+			attr: newAttributes(attributes{
+				operation:   admission.Delete,
+				subresource: "",
+				operationOptions: &metav1.DeleteOptions{
+					IgnoreStoreReadErrorWithClusterBreakingPotential: ptr.To[bool](true),
+				},
+				userInfo: &user.DefaultInfo{Name: "foo"},
+			}),
+			authz: &fakeAuthorizer{
+				decision: authorizer.DecisionAllow,
+				reason:   "permitted",
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			var want error
+			if test.err != nil {
+				want = test.err(test.attr)
+			}
+
+			ctx := context.Background()
+			if test.reqInfo != nil {
+				ctx = request.WithRequestInfo(ctx, test.reqInfo)
+			}
+
+			// wrap the attributes so we can access the annotations set during admission
+			attrs := &fakeAttributes{Attributes: test.attr}
+			got := authorizeUnsafeDelete(ctx, attrs, test.authz)
+			switch {
+			case want != nil:
+				if got == nil || want.Error() != got.Error() {
+					t.Errorf("expected error: %v, but got: %v", want, got)
+				}
+			default:
+				if got != nil {
+					t.Errorf("expected no error, but got: %v", got)
+				}
+			}
+		})
+	}
+}
+
+// attributes of interest for this test
+type attributes struct {
+	operation        admission.Operation
+	operationOptions runtime.Object
+	userInfo         user.Info
+	subresource      string
+}
+
+func newAttributes(attr attributes) admission.Attributes {
+	return admission.NewAttributesRecord(
+		nil,                           // this plugin should never inspect the object
+		nil,                           // old object, this plugin should never inspect it
+		schema.GroupVersionKind{},     // this plugin should never inspect kind
+		"",                            // namespace, leave it empty, this plugin only passes it along to the authorizer
+		"",                            // name, leave it empty, this plugin only passes it along to the authorizer
+		schema.GroupVersionResource{}, // resource, leave it empty, this plugin only passes it along to the authorizer
+		attr.subresource,
+		attr.operation,
+		attr.operationOptions,
+		false, // dryRun, this plugin should never inspect this attribute
+		attr.userInfo)
+}
+
+type fakeAttributes struct {
+	admission.Attributes
+	annotations map[string]string
+}
+
+func (f *fakeAttributes) AddAnnotation(key, value string) error {
+	if err := f.Attributes.AddAnnotation(key, value); err != nil {
+		return err
+	}
+
+	if len(f.annotations) == 0 {
+		f.annotations = map[string]string{}
+	}
+	f.annotations[key] = value
+	return nil
+}
+
+type fakeAuthorizer struct {
+	decision authorizer.Decision
+	reason   string
+	err      error
+}
+
+func (authorizer fakeAuthorizer) Authorize(ctx context.Context, a authorizer.Attributes) (authorized authorizer.Decision, reason string, err error) {
+	return authorizer.decision, authorizer.reason, authorizer.err
+}

test/integration/controlplane/transformation/secrets_transformation_test.go

@@ -30,15 +30,20 @@ import (
 	"testing"
 	"time"
 
+	rbacv1 "k8s.io/api/rbac/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/wait"
 	apiserverv1 "k8s.io/apiserver/pkg/apis/apiserver/v1"
 	genericfeatures "k8s.io/apiserver/pkg/features"
 	"k8s.io/apiserver/pkg/storage/value"
 	aestransformer "k8s.io/apiserver/pkg/storage/value/encrypt/aes"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	clientset "k8s.io/client-go/kubernetes"
+	restclient "k8s.io/client-go/rest"
 	featuregatetesting "k8s.io/component-base/featuregate/testing"
+	"k8s.io/kubernetes/test/integration/authutil"
 	"k8s.io/utils/ptr"
 )
 
@@ -139,6 +144,10 @@ func TestAllowUnsafeMalformedObjectDeletionFeature(t *testing.T) {
 		// what we expect for DELETE on the corrupt object, after encryption has
 		// broken, with the option to ignore store read error enabled
 		corrupObjDeleteWithOption verifier
+		// what we expect for DELETE on the corrupt object, after encryption has
+		// broken, with the option to ignore store read error enabled, and
+		// the user has the permission to do unsafe delete
+		corrupObjDeleteWithOptionAndPrivilege verifier
 		// what we expect for GET on the corrupt object (post deletion)
 		corrupObjGetPostDelete verifier
 	}{
@@ -151,8 +160,12 @@ func TestAllowUnsafeMalformedObjectDeletionFeature(t *testing.T) {
 			},
 			corruptObjGetPreDelete:      wantAPIStatusError{reason: metav1.StatusReasonInternalError},
 			corrupObjDeletWithoutOption: wantAPIStatusError{reason: metav1.StatusReasonInternalError},
-			corrupObjDeleteWithOption:   wantNoError{},
+			corrupObjDeleteWithOption: wantAPIStatusError{
-			corrupObjGetPostDelete:      wantAPIStatusError{reason: metav1.StatusReasonNotFound},
+				reason:          metav1.StatusReasonForbidden,
+				messageContains: `not permitted to do "unsafe-delete-ignore-read-errors"`,
+			},
+			corrupObjDeleteWithOptionAndPrivilege: wantNoError{},
+			corrupObjGetPostDelete:                wantAPIStatusError{reason: metav1.StatusReasonNotFound},
 		},
 		{
 			featureEnabled: false,
@@ -160,10 +173,11 @@ func TestAllowUnsafeMalformedObjectDeletionFeature(t *testing.T) {
 				return got.Status().Reason == metav1.StatusReasonInternalError &&
 					strings.Contains(got.Status().Message, "Internal error occurred: no matching prefix found")
 			},
-			corruptObjGetPreDelete:      wantAPIStatusError{reason: metav1.StatusReasonInternalError},
+			corruptObjGetPreDelete:                wantAPIStatusError{reason: metav1.StatusReasonInternalError},
-			corrupObjDeletWithoutOption: wantAPIStatusError{reason: metav1.StatusReasonInternalError},
+			corrupObjDeletWithoutOption:           wantAPIStatusError{reason: metav1.StatusReasonInternalError},
-			corrupObjDeleteWithOption:   wantAPIStatusError{reason: metav1.StatusReasonInternalError},
+			corrupObjDeleteWithOption:             wantAPIStatusError{reason: metav1.StatusReasonInternalError},
-			corrupObjGetPostDelete:      wantAPIStatusError{reason: metav1.StatusReasonInternalError},
+			corrupObjDeleteWithOptionAndPrivilege: wantAPIStatusError{reason: metav1.StatusReasonInternalError},
+			corrupObjGetPostDelete:                wantAPIStatusError{reason: metav1.StatusReasonInternalError},
 		},
 	}
 	for _, tc := range tests {
@@ -176,8 +190,27 @@ func TestAllowUnsafeMalformedObjectDeletionFeature(t *testing.T) {
 			}
 			defer test.cleanUp()
 
+			// a) set up a distinct client for the test user with the least
+			// privileges, we will grant permission as we progress through the test
+			testUser := "croc"
+			testUserConfig := restclient.CopyConfig(test.kubeAPIServer.ClientConfig)
+			testUserConfig.Impersonate.UserName = testUser
+			testUserClient := clientset.NewForConfigOrDie(testUserConfig)
+			adminClient := test.restClient
+
+			// b) use the admin client to grant the the test user initial permissions,
+			// we are not going to grant 'delete-ignore-read-errors' just yet
+			permitUserToDoVerbOnSecret(t, adminClient, testUser, testNamespace, []string{"create", "get", "delete", "update"})
+
+			// the test should not use the admin client going forward
+			test.restClient = testUserClient
+			defer func() {
+				// for any cleanup that requires admin privileges
+				test.restClient = adminClient
+			}()
+
 			secretCorrupt := "foo-with-unsafe-delete"
-			// a) create and delete the secret, we don't expect any error
+			// c) create and delete the secret, we don't expect any error
 			_, err = test.createSecret(secretCorrupt, testNamespace)
 			if err != nil {
 				t.Fatalf("'%s/%s' failed to create, got error: %v", err, testNamespace, secretCorrupt)
@@ -187,13 +220,13 @@ func TestAllowUnsafeMalformedObjectDeletionFeature(t *testing.T) {
 				t.Fatalf("'%s/%s' failed to delete, got error: %v", err, testNamespace, secretCorrupt)
 			}
 
-			// b) re-create the secret
+			// d) re-create the secret
 			test.secret, err = test.createSecret(secretCorrupt, testNamespace)
 			if err != nil {
 				t.Fatalf("Failed to create test secret, error: %v", err)
 			}
 
-			// c) update the secret with a finalizer
+			// e) update the secret with a finalizer
 			withFinalizer := test.secret.DeepCopy()
 			withFinalizer.Finalizers = append(withFinalizer.Finalizers, "tes.k8s.io/fake")
 			test.secret, err = test.restClient.CoreV1().Secrets(testNamespace).Update(context.Background(), withFinalizer, metav1.UpdateOptions{})
@@ -203,7 +236,7 @@ func TestAllowUnsafeMalformedObjectDeletionFeature(t *testing.T) {
 
 			test.runResource(test.TContext, unSealWithGCMTransformer, aesGCMPrefix, "", "v1", "secrets", test.secret.Name, test.secret.Namespace)
 
-			// d) override the config and break decryption of the old resources,
+			// f) override the config and break decryption of the old resources,
 			// the secret created in step b will be undecryptable
 			now := time.Now()
 			encryptionConf := filepath.Join(test.configDir, encryptionConfigFileName)
@@ -216,7 +249,7 @@ func TestAllowUnsafeMalformedObjectDeletionFeature(t *testing.T) {
 			body, _ = ioutil.ReadFile(encryptionConf)
 			t.Logf("file after write: %s", body)
 
-			// e) wait for the breaking changes to take effect
+			// g) wait for the breaking changes to take effect
 			testCtx, cancel := context.WithCancel(context.Background())
 			defer cancel()
 			// TODO: dynamic encryption config reload takes about 1m, so can't use
@@ -237,7 +270,7 @@ func TestAllowUnsafeMalformedObjectDeletionFeature(t *testing.T) {
 			}
 			t.Logf("it took %s for the apiserver to reload the encryption config", time.Since(now))
 
-			// f) create a new secret, and then delete it, it should work
+			// h) create a new secret, and then delete it, it should work
 			secretNormal := "bar-with-normal-delete"
 			_, err = test.createSecret(secretNormal, testNamespace)
 			if err != nil {
@@ -248,22 +281,31 @@ func TestAllowUnsafeMalformedObjectDeletionFeature(t *testing.T) {
 				t.Fatalf("'%s/%s' failed to create, got error: %v", err, testNamespace, secretNormal)
 			}
 
-			// g) let's try to get the broken secret created in step b, we expect it
+			// i) let's try to get the broken secret created in step b, we expect it
 			// to fail, the error will vary depending on whether the feature is enabled
 			_, err = test.restClient.CoreV1().Secrets(testNamespace).Get(context.Background(), secretCorrupt, metav1.GetOptions{})
 			tc.corruptObjGetPreDelete.verify(t, err)
 
-			// h) let's try the normal deletion flow, we expect an error
+			// j) let's try the normal deletion flow, we expect an error
 			err = test.restClient.CoreV1().Secrets(testNamespace).Delete(context.Background(), secretCorrupt, metav1.DeleteOptions{})
 			tc.corrupObjDeletWithoutOption.verify(t, err)
 
-			// i) make an attempt to delete the corrupt object by enabling the option
+			// k) make an attempt to delete the corrupt object by enabling the option,
+			// on the other hand, we have not granted the 'delete-ignore-read-errors'
+			// verb to the user yet, so we expect admission to deny the delete request
 			options := metav1.DeleteOptions{
 				IgnoreStoreReadErrorWithClusterBreakingPotential: ptr.To[bool](true),
 			}
 			err = test.restClient.CoreV1().Secrets(testNamespace).Delete(context.Background(), secretCorrupt, options)
 			tc.corrupObjDeleteWithOption.verify(t, err)
 
+			// l) grant the test user to do 'unsafe-delete-ignore-read-errors' on secrets
+			permitUserToDoVerbOnSecret(t, adminClient, testUser, testNamespace, []string{"unsafe-delete-ignore-read-errors"})
+
+			// m) let's try to do unsafe delete again
+			err = test.restClient.CoreV1().Secrets(testNamespace).Delete(context.Background(), secretCorrupt, options)
+			tc.corrupObjDeleteWithOptionAndPrivilege.verify(t, err)
+
 			// j) final get should return a NotFound error after the secret has been deleted
 			_, err = test.restClient.CoreV1().Secrets(testNamespace).Get(context.Background(), secretCorrupt, metav1.GetOptions{})
 			tc.corrupObjGetPostDelete.verify(t, err)
@@ -481,6 +523,52 @@ func TestListCorruptObjects(t *testing.T) {
 	}
 }
 
+func permitUserToDoVerbOnSecret(t *testing.T, client *clientset.Clientset, user, namespace string, verbs []string) {
+	t.Helper()
+
+	name := fmt.Sprintf("%s-can-do-%s-on-secrets-in-%s", user, strings.Join(verbs, "-"), namespace)
+	_, err := client.RbacV1().Roles(namespace).Create(context.TODO(), &rbacv1.Role{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+		},
+		Rules: []rbacv1.PolicyRule{
+			{
+				Verbs:     verbs,
+				APIGroups: []string{""},
+				Resources: []string{"secrets"},
+			},
+		},
+	}, metav1.CreateOptions{})
+	if err != nil {
+		t.Fatalf("error while creating role: %s, err: %v", name, err)
+	}
+
+	_, err = client.RbacV1().RoleBindings(namespace).Create(context.TODO(), &rbacv1.RoleBinding{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+		},
+		Subjects: []rbacv1.Subject{
+			{
+				Kind: rbacv1.UserKind,
+				Name: user,
+			},
+		},
+		RoleRef: rbacv1.RoleRef{
+			APIGroup: rbacv1.GroupName,
+			Kind:     "Role",
+			Name:     name,
+		},
+	}, metav1.CreateOptions{})
+	if err != nil {
+		t.Fatalf("error while creating role binding: %s, err: %v", name, err)
+	}
+
+	authutil.WaitForNamedAuthorizationUpdate(t, context.TODO(), client.AuthorizationV1(),
+		user, namespace, verbs[0], "", schema.GroupResource{Resource: "secrets"}, true)
+}
+
 // Baseline (no enveloping) - use to contrast with enveloping benchmarks.
 func BenchmarkBase(b *testing.B) {
 	runBenchmark(b, "")