diff --git a/deploy/spoke/clusterrole.yaml b/deploy/spoke/clusterrole.yaml
index a545f5918..90fda0fa5 100644
--- a/deploy/spoke/clusterrole.yaml
+++ b/deploy/spoke/clusterrole.yaml
@@ -15,7 +15,10 @@ rules:
 - apiGroups: ["work.open-cluster-management.io"]
   resources: ["appliedmanifestworks/finalizers"]
   verbs: ["update"]
-# Allow agent to create subjectaccessreviews
+# Allow agent to check executor permissions
 - apiGroups: ["authorization.k8s.io"]
   resources: ["subjectaccessreviews"]
   verbs: ["create"]
+- apiGroups: [""]
+  resources: ["serviceaccounts"]
+  verbs: ["impersonate"]
diff --git a/pkg/spoke/auth/auth.go b/pkg/spoke/auth/auth.go
index ce3476455..0d9adef98 100644
--- a/pkg/spoke/auth/auth.go
+++ b/pkg/spoke/auth/auth.go
@@ -7,9 +7,14 @@ import (
 	"time"
 
 	authorizationv1 "k8s.io/api/authorization/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+	"k8s.io/klog/v2"
 
 	workapiv1 "open-cluster-management.io/api/work/v1"
 )
@@ -29,8 +34,8 @@ const (
 type ExecutorValidator interface {
 	// Validate whether the work executor subject has permission to perform action on the specific manifest,
 	// if there is no permission will return a kubernetes forbidden error.
-	Validate(ctx context.Context, executor *workapiv1.ManifestWorkExecutor,
-		gvr schema.GroupVersionResource, namespace, name string, action ExecuteAction) error
+	Validate(ctx context.Context, executor *workapiv1.ManifestWorkExecutor, gvr schema.GroupVersionResource,
+		namespace, name string, obj *unstructured.Unstructured, action ExecuteAction) error
 }
 
 type NotAllowedError struct {
@@ -46,18 +51,33 @@ func (e *NotAllowedError) Error() string {
 	return err
 }
 
-func NewExecutorValidator(kubeClient kubernetes.Interface) ExecutorValidator {
+func NewExecutorValidator(config *rest.Config, kubeClient kubernetes.Interface) ExecutorValidator {
 	return &sarValidator{
-		kubeClient: kubeClient,
+		kubeClient:               kubeClient,
+		config:                   config,
+		newImpersonateClientFunc: defaultNewImpersonateClient,
 	}
 }
 
 type sarValidator struct {
-	kubeClient kubernetes.Interface
+	kubeClient               kubernetes.Interface
+	config                   *rest.Config
+	newImpersonateClientFunc newImpersonateClient
+}
+
+type newImpersonateClient func(config *rest.Config, username string) (dynamic.Interface, error)
+
+func defaultNewImpersonateClient(config *rest.Config, username string) (dynamic.Interface, error) {
+	if config == nil {
+		return nil, fmt.Errorf("kube config should not be nil")
+	}
+	impersonatedConfig := *config
+	impersonatedConfig.Impersonate.UserName = username
+	return dynamic.NewForConfig(&impersonatedConfig)
 }
 
 func (v *sarValidator) Validate(ctx context.Context, executor *workapiv1.ManifestWorkExecutor,
-	gvr schema.GroupVersionResource, namespace, name string, action ExecuteAction) error {
+	gvr schema.GroupVersionResource, namespace, name string, obj *unstructured.Unstructured, action ExecuteAction) error {
 	if executor == nil {
 		return nil
 	}
@@ -97,15 +117,63 @@ func (v *sarValidator) Validate(ctx context.Context, executor *workapiv1.Manifes
 
 	if !allowed {
 		return &NotAllowedError{
-			Err: fmt.Errorf("not allowed to %s the resource %s %s, name: %s",
-				strings.ToLower(string(action)), resource.Group, resource.Resource, resource.Name),
+			Err: fmt.Errorf("not allowed to %s the resource %s %s, %s %s",
+				strings.ToLower(string(action)), resource.Group, resource.Resource, resource.Namespace, resource.Name),
 			RequeueTime: 60 * time.Second,
 		}
 	}
 
+	switch {
+	case action != ApplyAction:
+		return nil
+	case gvr.Group != "rbac.authorization.k8s.io":
+		return nil
+	case gvr.Resource == "roles", gvr.Resource == "rolebindings",
+		gvr.Resource == "clusterroles", gvr.Resource == "clusterrolebindings":
+		// subjectaccessreview can not permission escalation, use an impersonation request to check again
+		return v.checkEscalation(ctx, sa, gvr, namespace, name, obj)
+	}
+
 	return nil
 }
 
+func (v *sarValidator) checkEscalation(ctx context.Context, sa *workapiv1.ManifestWorkSubjectServiceAccount,
+	gvr schema.GroupVersionResource, namespace, name string, obj *unstructured.Unstructured) error {
+
+	dynamicClient, err := v.newImpersonateClientFunc(v.config, username(sa.Namespace, sa.Name))
+	if err != nil {
+		return err
+	}
+
+	_, err = dynamicClient.Resource(gvr).Namespace(namespace).Create(ctx, obj, metav1.CreateOptions{
+		DryRun: []string{"All"},
+	})
+	if apierrors.IsForbidden(err) {
+		klog.Infof("not allowed to apply the resource %s %s, %s %s, error: %s",
+			gvr.Group, gvr.Resource, namespace, name, err.Error())
+		return &NotAllowedError{
+			Err: fmt.Errorf("not allowed to apply the resource %s %s, %s %s, error: permission escalation",
+				gvr.Group, gvr.Resource, namespace, name),
+			RequeueTime: 60 * time.Second,
+		}
+	}
+
+	if apierrors.IsAlreadyExists(err) {
+		// it is not necessary to further check the permission for update when the resource exists, because
+		// the API server checks the permission escalation before checking the existence.
+		return nil
+	}
+	return err
+}
+
+func username(saNamespace, saName string) string {
+	return fmt.Sprintf("system:serviceaccount:%s:%s", saNamespace, saName)
+}
+func groups(saNamespace string) []string {
+	return []string{"system:serviceaccounts", "system:authenticated",
+		fmt.Sprintf("system:serviceaccounts:%s", saNamespace)}
+}
+
 func buildSubjectAccessReviews(saNamespace string, saName string,
 	resource authorizationv1.ResourceAttributes,
 	verbs ...string) []authorizationv1.SubjectAccessReview {
@@ -123,9 +191,8 @@ func buildSubjectAccessReviews(saNamespace string, saName string,
 					Namespace:   resource.Namespace,
 					Verb:        verb,
 				},
-				User: fmt.Sprintf("system:serviceaccount:%s:%s", saNamespace, saName),
-				Groups: []string{"system:serviceaccounts", "system:authenticated",
-					fmt.Sprintf("system:serviceaccounts:%s", saNamespace)},
+				User:   username(saNamespace, saName),
+				Groups: groups(saNamespace),
 			},
 		})
 	}
diff --git a/pkg/spoke/auth/auth_test.go b/pkg/spoke/auth/auth_test.go
index 7c193db2a..1c109be06 100644
--- a/pkg/spoke/auth/auth_test.go
+++ b/pkg/spoke/auth/auth_test.go
@@ -1,4 +1,4 @@
-package auth_test
+package auth
 
 import (
 	"context"
@@ -6,13 +6,18 @@ import (
 	"testing"
 
 	v1 "k8s.io/api/authorization/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/dynamic"
+	fakedynamic "k8s.io/client-go/dynamic/fake"
 	fakekube "k8s.io/client-go/kubernetes/fake"
+	"k8s.io/client-go/rest"
 	clienttesting "k8s.io/client-go/testing"
 
 	workapiv1 "open-cluster-management.io/api/work/v1"
-	"open-cluster-management.io/work/pkg/spoke/auth"
+	"open-cluster-management.io/work/pkg/spoke/spoketesting"
 )
 
 func TestValidate(t *testing.T) {
@@ -21,7 +26,7 @@ func TestValidate(t *testing.T) {
 		executor  *workapiv1.ManifestWorkExecutor
 		namespace string
 		name      string
-		action    auth.ExecuteAction
+		action    ExecuteAction
 		expect    error
 	}{
 		"executor nil": {
@@ -69,8 +74,8 @@ func TestValidate(t *testing.T) {
 			},
 			namespace: "test-deny",
 			name:      "test",
-			action:    auth.ApplyAction,
-			expect:    fmt.Errorf("not allowed to apply the resource  secrets, name: test, will try again in 1m0s"),
+			action:    ApplyAction,
+			expect:    fmt.Errorf("not allowed to apply the resource  secrets, test-deny test, will try again in 1m0s"),
 		},
 		"allow": {
 			executor: &workapiv1.ManifestWorkExecutor{
@@ -84,7 +89,7 @@ func TestValidate(t *testing.T) {
 			},
 			namespace: "test-allow",
 			name:      "test",
-			action:    auth.ApplyAction,
+			action:    ApplyAction,
 			expect:    nil,
 		},
 	}
@@ -113,10 +118,101 @@ func TestValidate(t *testing.T) {
 			return false, nil, nil
 		},
 	)
-	validator := auth.NewExecutorValidator(kubeClient)
+	validator := NewExecutorValidator(nil, kubeClient)
 	for testName, test := range tests {
 		t.Run(testName, func(t *testing.T) {
-			err := validator.Validate(context.TODO(), test.executor, gvr, test.namespace, test.name, test.action)
+			err := validator.Validate(context.TODO(), test.executor, gvr, test.namespace, test.name, nil, test.action)
+			if test.expect == nil {
+				if err != nil {
+					t.Errorf("expect nil but got %s", err)
+				}
+			} else if err == nil || err.Error() != test.expect.Error() {
+				t.Errorf("expect %s but got %s", test.expect, err)
+			}
+		})
+	}
+}
+
+func TestValidateEscalation(t *testing.T) {
+
+	tests := map[string]struct {
+		executor  *workapiv1.ManifestWorkExecutor
+		namespace string
+		name      string
+		obj       *unstructured.Unstructured
+		expect    error
+	}{
+		"forbideen": {
+			executor: &workapiv1.ManifestWorkExecutor{
+				Subject: workapiv1.ManifestWorkExecutorSubject{
+					Type: workapiv1.ExecutorSubjectTypeServiceAccount,
+					ServiceAccount: &workapiv1.ManifestWorkSubjectServiceAccount{
+						Namespace: "test-ns",
+						Name:      "test-name",
+					},
+				},
+			},
+			namespace: "test-deny",
+			name:      "test",
+			obj:       spoketesting.NewUnstructured("v1", "ClusterRole", "", "test"),
+			expect:    fmt.Errorf("not allowed to apply the resource rbac.authorization.k8s.io roles, test-deny test, error: permission escalation, will try again in 1m0s"),
+		},
+		"allow": {
+			executor: &workapiv1.ManifestWorkExecutor{
+				Subject: workapiv1.ManifestWorkExecutorSubject{
+					Type: workapiv1.ExecutorSubjectTypeServiceAccount,
+					ServiceAccount: &workapiv1.ManifestWorkSubjectServiceAccount{
+						Namespace: "test-ns",
+						Name:      "test-name",
+					},
+				},
+			},
+			namespace: "test-allow",
+			name:      "test",
+			obj:       spoketesting.NewUnstructured("v1", "Role", "ns1", "test"),
+			expect:    nil,
+		},
+	}
+
+	gvr := schema.GroupVersionResource{Group: "rbac.authorization.k8s.io", Version: "v1", Resource: "roles"}
+	kubeClient := fakekube.NewSimpleClientset()
+	kubeClient.PrependReactor("create", "subjectaccessreviews",
+		func(action clienttesting.Action) (handled bool, ret runtime.Object, err error) {
+			return true, &v1.SubjectAccessReview{
+				Status: v1.SubjectAccessReviewStatus{
+					Allowed: true,
+				},
+			}, nil
+		},
+	)
+
+	scheme := runtime.NewScheme()
+	dynamicClient := fakedynamic.NewSimpleDynamicClient(scheme)
+	dynamicClient.PrependReactor("create", "roles",
+		func(action clienttesting.Action) (handled bool, ret runtime.Object, err error) {
+			obj := action.(clienttesting.CreateActionImpl).Object.(*unstructured.Unstructured)
+
+			if obj.GetKind() == "Role" {
+				return true, obj, nil
+			}
+			if obj.GetKind() == "ClusterRole" {
+				return true, obj, apierrors.NewForbidden(
+					schema.GroupResource{Group: "rbac.authorization.k8s.io", Resource: "clusterroles"},
+					obj.GetName(),
+					fmt.Errorf("escalation"))
+			}
+			return false, nil, nil
+		})
+	validator := &sarValidator{
+		kubeClient: kubeClient,
+		newImpersonateClientFunc: func(config *rest.Config, username string) (dynamic.Interface, error) {
+			return dynamicClient, nil
+		},
+	}
+
+	for testName, test := range tests {
+		t.Run(testName, func(t *testing.T) {
+			err := validator.Validate(context.TODO(), test.executor, gvr, test.namespace, test.name, test.obj, ApplyAction)
 			if test.expect == nil {
 				if err != nil {
 					t.Errorf("expect nil but got %s", err)
diff --git a/pkg/spoke/controllers/manifestcontroller/manifestwork_controller.go b/pkg/spoke/controllers/manifestcontroller/manifestwork_controller.go
index a9c6d45b8..917781a5e 100644
--- a/pkg/spoke/controllers/manifestcontroller/manifestwork_controller.go
+++ b/pkg/spoke/controllers/manifestcontroller/manifestwork_controller.go
@@ -73,7 +73,8 @@ func NewManifestWorkController(
 	appliedManifestWorkClient workv1client.AppliedManifestWorkInterface,
 	appliedManifestWorkInformer workinformer.AppliedManifestWorkInformer,
 	hubHash string,
-	restMapper meta.RESTMapper) factory.Controller {
+	restMapper meta.RESTMapper,
+	validator auth.ExecutorValidator) factory.Controller {
 
 	controller := &ManifestWorkController{
 		manifestWorkClient:        manifestWorkClient,
@@ -84,7 +85,7 @@ func NewManifestWorkController(
 		hubHash:                   hubHash,
 		restMapper:                restMapper,
 		appliers:                  apply.NewAppliers(spokeDynamicClient, spokeKubeClient, spokeAPIExtensionClient),
-		validator:                 auth.NewExecutorValidator(spokeKubeClient),
+		validator:                 validator,
 	}
 
 	return factory.New().
@@ -274,7 +275,7 @@ func (m *ManifestWorkController) applyOneManifest(
 	}
 
 	// check the Executor subject permission before applying
-	err = m.validator.Validate(ctx, workSpec.Executor, gvr, resMeta.Namespace, resMeta.Name, auth.ApplyAction)
+	err = m.validator.Validate(ctx, workSpec.Executor, gvr, resMeta.Namespace, resMeta.Name, required, auth.ApplyAction)
 	if err != nil {
 		result.Error = err
 		return result
diff --git a/pkg/spoke/controllers/manifestcontroller/manifestwork_controller_test.go b/pkg/spoke/controllers/manifestcontroller/manifestwork_controller_test.go
index 7a2e13b63..76a6e42eb 100644
--- a/pkg/spoke/controllers/manifestcontroller/manifestwork_controller_test.go
+++ b/pkg/spoke/controllers/manifestcontroller/manifestwork_controller_test.go
@@ -44,7 +44,7 @@ func newController(t *testing.T, work *workapiv1.ManifestWork, appliedWork *work
 		appliedManifestWorkClient: fakeWorkClient.WorkV1().AppliedManifestWorks(),
 		appliedManifestWorkLister: workInformerFactory.Work().V1().AppliedManifestWorks().Lister(),
 		restMapper:                mapper,
-		validator:                 auth.NewExecutorValidator(spokeKubeClient),
+		validator:                 auth.NewExecutorValidator(nil, spokeKubeClient),
 	}
 
 	if err := workInformerFactory.Work().V1().ManifestWorks().Informer().GetStore().Add(work); err != nil {
diff --git a/pkg/spoke/spokeagent.go b/pkg/spoke/spokeagent.go
index f0f0a9950..90d90818e 100644
--- a/pkg/spoke/spokeagent.go
+++ b/pkg/spoke/spokeagent.go
@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"open-cluster-management.io/work/pkg/helper"
+	"open-cluster-management.io/work/pkg/spoke/auth"
 	"open-cluster-management.io/work/pkg/spoke/controllers/appliedmanifestcontroller"
 	"open-cluster-management.io/work/pkg/spoke/controllers/finalizercontroller"
 	"open-cluster-management.io/work/pkg/spoke/controllers/manifestcontroller"
@@ -103,6 +104,7 @@ func (o *WorkloadAgentOptions) RunWorkloadAgent(ctx context.Context, controllerC
 		return err
 	}
 
+	validator := auth.NewExecutorValidator(spokeRestConfig, spokeKubeClient)
 	manifestWorkController := manifestcontroller.NewManifestWorkController(
 		ctx,
 		controllerContext.EventRecorder,
@@ -116,6 +118,7 @@ func (o *WorkloadAgentOptions) RunWorkloadAgent(ctx context.Context, controllerC
 		spokeWorkInformerFactory.Work().V1().AppliedManifestWorks(),
 		hubhash,
 		restMapper,
+		validator,
 	)
 	addFinalizerController := finalizercontroller.NewAddFinalizerController(
 		controllerContext.EventRecorder,
diff --git a/test/integration/executor_test.go b/test/integration/executor_test.go
index 5f90324d9..8a06d4582 100644
--- a/test/integration/executor_test.go
+++ b/test/integration/executor_test.go
@@ -209,4 +209,318 @@ var _ = ginkgo.Describe("ManifestWork Executor Subject", func() {
 			util.AssertExistenceOfConfigMaps(manifests, spokeKubeClient, eventuallyTimeout, eventuallyInterval)
 		})
 	})
+
+	ginkgo.Context("Apply the resource with executor escalation validating", func() {
+		executorName := "test-executor"
+		ginkgo.BeforeEach(func() {
+			manifests = []workapiv1.Manifest{
+				util.ToManifest(util.NewConfigmap(o.SpokeClusterName, "cm1", map[string]string{"a": "b"}, []string{})),
+				util.ToManifest(util.NewRoleForManifest(o.SpokeClusterName, "role-cm-creator", rbacv1.PolicyRule{
+					Verbs:     []string{"create", "update", "patch", "get", "list", "delete"},
+					APIGroups: []string{""},
+					Resources: []string{"configmaps"},
+				})),
+				util.ToManifest(util.NewRoleBindingForManifest(o.SpokeClusterName, "role-cm-creator-binding",
+					rbacv1.RoleRef{
+						Kind: "Role",
+						Name: "role-cm-creator",
+					},
+					rbacv1.Subject{
+						Kind:      "ServiceAccount",
+						Namespace: o.SpokeClusterName,
+						Name:      executorName,
+					})),
+			}
+			executor = &workapiv1.ManifestWorkExecutor{
+				Subject: workapiv1.ManifestWorkExecutorSubject{
+					Type: workapiv1.ExecutorSubjectTypeServiceAccount,
+					ServiceAccount: &workapiv1.ManifestWorkSubjectServiceAccount{
+						Namespace: o.SpokeClusterName,
+						Name:      executorName,
+					},
+				},
+			}
+		})
+
+		ginkgo.It("no permission", func() {
+			roleName := "role1"
+			_, err = spokeKubeClient.RbacV1().Roles(o.SpokeClusterName).Create(
+				context.TODO(), &rbacv1.Role{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      roleName,
+						Namespace: o.SpokeClusterName,
+					},
+					Rules: []rbacv1.PolicyRule{
+						{
+							// no "escalate" and "bind" verb
+							Verbs:     []string{"create", "update", "patch", "get", "list", "delete"},
+							APIGroups: []string{"rbac.authorization.k8s.io"},
+							Resources: []string{"roles", "rolebindings"},
+						},
+					},
+				}, metav1.CreateOptions{})
+			gomega.Expect(err).ToNot(gomega.HaveOccurred())
+			_, err = spokeKubeClient.RbacV1().RoleBindings(o.SpokeClusterName).Create(
+				context.TODO(), &rbacv1.RoleBinding{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      roleName,
+						Namespace: o.SpokeClusterName,
+					},
+					Subjects: []rbacv1.Subject{
+						{
+							Kind:      "ServiceAccount",
+							Namespace: o.SpokeClusterName,
+							Name:      executorName,
+						},
+					},
+					RoleRef: rbacv1.RoleRef{
+						APIGroup: "rbac.authorization.k8s.io",
+						Kind:     "Role",
+						Name:     roleName,
+					},
+				}, metav1.CreateOptions{})
+			gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+			work, err = hubWorkClient.WorkV1().ManifestWorks(o.SpokeClusterName).Create(
+				context.Background(), work, metav1.CreateOptions{})
+			gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+			util.AssertWorkCondition(work.Namespace, work.Name, hubWorkClient, string(workapiv1.WorkApplied),
+				metav1.ConditionFalse,
+				[]metav1.ConditionStatus{metav1.ConditionFalse, metav1.ConditionFalse, metav1.ConditionFalse},
+				eventuallyTimeout, eventuallyInterval)
+			util.AssertWorkCondition(work.Namespace, work.Name, hubWorkClient, string(workapiv1.WorkAvailable),
+				metav1.ConditionFalse,
+				[]metav1.ConditionStatus{metav1.ConditionFalse, metav1.ConditionFalse, metav1.ConditionFalse},
+				eventuallyTimeout, eventuallyInterval)
+
+			// ensure configmap not exist
+			util.AssertNonexistenceOfConfigMaps(
+				[]workapiv1.Manifest{
+					util.ToManifest(util.NewConfigmap(o.SpokeClusterName, "cm1", map[string]string{"a": "b"}, []string{})),
+				}, spokeKubeClient, eventuallyTimeout, eventuallyInterval)
+		})
+
+		ginkgo.It("no permission for already exist resource", func() {
+			roleName := "role1"
+			_, err = spokeKubeClient.RbacV1().Roles(o.SpokeClusterName).Create(
+				context.TODO(), &rbacv1.Role{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      roleName,
+						Namespace: o.SpokeClusterName,
+					},
+					Rules: []rbacv1.PolicyRule{
+						{
+							// no "escalate" and "bind" verb
+							Verbs:     []string{"create", "update", "patch", "get", "list", "delete"},
+							APIGroups: []string{"rbac.authorization.k8s.io"},
+							Resources: []string{"roles", "rolebindings"},
+						},
+					},
+				}, metav1.CreateOptions{})
+			gomega.Expect(err).ToNot(gomega.HaveOccurred())
+			_, err = spokeKubeClient.RbacV1().RoleBindings(o.SpokeClusterName).Create(
+				context.TODO(), &rbacv1.RoleBinding{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      roleName,
+						Namespace: o.SpokeClusterName,
+					},
+					Subjects: []rbacv1.Subject{
+						{
+							Kind:      "ServiceAccount",
+							Namespace: o.SpokeClusterName,
+							Name:      executorName,
+						},
+					},
+					RoleRef: rbacv1.RoleRef{
+						APIGroup: "rbac.authorization.k8s.io",
+						Kind:     "Role",
+						Name:     roleName,
+					},
+				}, metav1.CreateOptions{})
+			gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+			// make the role exist with lower permission
+			_, err = spokeKubeClient.RbacV1().Roles(o.SpokeClusterName).Create(
+				context.TODO(), &rbacv1.Role{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "role-cm-creator",
+						Namespace: o.SpokeClusterName,
+					},
+					Rules: []rbacv1.PolicyRule{
+						{
+							Verbs:     []string{"get", "list"},
+							APIGroups: []string{""},
+							Resources: []string{"configmaps"},
+						},
+					},
+				}, metav1.CreateOptions{})
+			gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+			work, err = hubWorkClient.WorkV1().ManifestWorks(o.SpokeClusterName).Create(
+				context.Background(), work, metav1.CreateOptions{})
+			gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+			util.AssertWorkCondition(work.Namespace, work.Name, hubWorkClient, string(workapiv1.WorkApplied),
+				metav1.ConditionFalse,
+				[]metav1.ConditionStatus{metav1.ConditionFalse, metav1.ConditionFalse, metav1.ConditionFalse},
+				eventuallyTimeout, eventuallyInterval)
+			util.AssertWorkCondition(work.Namespace, work.Name, hubWorkClient, string(workapiv1.WorkAvailable),
+				metav1.ConditionFalse,
+				// the cluster role already esists, so the ailable status is true enen if the applied status is false
+				[]metav1.ConditionStatus{metav1.ConditionFalse, metav1.ConditionTrue, metav1.ConditionFalse},
+				eventuallyTimeout, eventuallyInterval)
+
+			// ensure configmap not exist
+			util.AssertNonexistenceOfConfigMaps(
+				[]workapiv1.Manifest{
+					util.ToManifest(util.NewConfigmap(o.SpokeClusterName, "cm1", map[string]string{"a": "b"}, []string{})),
+				}, spokeKubeClient, eventuallyTimeout, eventuallyInterval)
+		})
+
+		ginkgo.It("with permission", func() {
+			roleName := "role1"
+			_, err = spokeKubeClient.RbacV1().Roles(o.SpokeClusterName).Create(
+				context.TODO(), &rbacv1.Role{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      roleName,
+						Namespace: o.SpokeClusterName,
+					},
+					Rules: []rbacv1.PolicyRule{
+						{
+							// with "escalate" and "bind" verb
+							Verbs:     []string{"create", "update", "patch", "get", "list", "delete", "escalate", "bind"},
+							APIGroups: []string{"rbac.authorization.k8s.io"},
+							Resources: []string{"roles"},
+						},
+						{
+							Verbs:     []string{"create", "update", "patch", "get", "list", "delete"},
+							APIGroups: []string{"rbac.authorization.k8s.io"},
+							Resources: []string{"rolebindings"},
+						},
+					},
+				}, metav1.CreateOptions{})
+			gomega.Expect(err).ToNot(gomega.HaveOccurred())
+			_, err = spokeKubeClient.RbacV1().RoleBindings(o.SpokeClusterName).Create(
+				context.TODO(), &rbacv1.RoleBinding{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      roleName,
+						Namespace: o.SpokeClusterName,
+					},
+					Subjects: []rbacv1.Subject{
+						{
+							Kind:      "ServiceAccount",
+							Namespace: o.SpokeClusterName,
+							Name:      executorName,
+						},
+					},
+					RoleRef: rbacv1.RoleRef{
+						APIGroup: "rbac.authorization.k8s.io",
+						Kind:     "Role",
+						Name:     roleName,
+					},
+				}, metav1.CreateOptions{})
+			gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+			work, err = hubWorkClient.WorkV1().ManifestWorks(o.SpokeClusterName).Create(
+				context.Background(), work, metav1.CreateOptions{})
+			gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+			util.AssertWorkCondition(work.Namespace, work.Name, hubWorkClient, string(workapiv1.WorkApplied),
+				metav1.ConditionTrue,
+				[]metav1.ConditionStatus{metav1.ConditionTrue, metav1.ConditionTrue, metav1.ConditionTrue},
+				eventuallyTimeout*3, eventuallyInterval)
+			util.AssertWorkCondition(work.Namespace, work.Name, hubWorkClient, string(workapiv1.WorkAvailable),
+				metav1.ConditionTrue,
+				[]metav1.ConditionStatus{metav1.ConditionTrue, metav1.ConditionTrue, metav1.ConditionTrue},
+				eventuallyTimeout, eventuallyInterval)
+
+			// ensure configmaps exist
+			util.AssertExistenceOfConfigMaps(
+				[]workapiv1.Manifest{
+					util.ToManifest(util.NewConfigmap(o.SpokeClusterName, "cm1", map[string]string{"a": "b"}, []string{})),
+				}, spokeKubeClient, eventuallyTimeout, eventuallyInterval)
+		})
+
+		ginkgo.It("with permission for already exist resource", func() {
+			roleName := "role1"
+			_, err = spokeKubeClient.RbacV1().Roles(o.SpokeClusterName).Create(
+				context.TODO(), &rbacv1.Role{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      roleName,
+						Namespace: o.SpokeClusterName,
+					},
+					Rules: []rbacv1.PolicyRule{
+						{
+							// with "escalate" and "bind" verb
+							Verbs:     []string{"create", "update", "patch", "get", "list", "delete", "escalate", "bind"},
+							APIGroups: []string{"rbac.authorization.k8s.io"},
+							Resources: []string{"roles"},
+						},
+						{
+							Verbs:     []string{"create", "update", "patch", "get", "list", "delete"},
+							APIGroups: []string{"rbac.authorization.k8s.io"},
+							Resources: []string{"rolebindings"},
+						},
+					},
+				}, metav1.CreateOptions{})
+			gomega.Expect(err).ToNot(gomega.HaveOccurred())
+			_, err = spokeKubeClient.RbacV1().RoleBindings(o.SpokeClusterName).Create(
+				context.TODO(), &rbacv1.RoleBinding{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      roleName,
+						Namespace: o.SpokeClusterName,
+					},
+					Subjects: []rbacv1.Subject{
+						{
+							Kind:      "ServiceAccount",
+							Namespace: o.SpokeClusterName,
+							Name:      executorName,
+						},
+					},
+					RoleRef: rbacv1.RoleRef{
+						APIGroup: "rbac.authorization.k8s.io",
+						Kind:     "Role",
+						Name:     roleName,
+					},
+				}, metav1.CreateOptions{})
+			gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+			// make the role exist with lower permission
+			_, err = spokeKubeClient.RbacV1().Roles(o.SpokeClusterName).Create(
+				context.TODO(), &rbacv1.Role{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "role-cm-creator",
+						Namespace: o.SpokeClusterName,
+					},
+					Rules: []rbacv1.PolicyRule{
+						{
+							Verbs:     []string{"get", "list"},
+							APIGroups: []string{""},
+							Resources: []string{"configmaps"},
+						},
+					},
+				}, metav1.CreateOptions{})
+			gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+			work, err = hubWorkClient.WorkV1().ManifestWorks(o.SpokeClusterName).Create(
+				context.Background(), work, metav1.CreateOptions{})
+			gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+			util.AssertWorkCondition(work.Namespace, work.Name, hubWorkClient, string(workapiv1.WorkApplied),
+				metav1.ConditionTrue,
+				[]metav1.ConditionStatus{metav1.ConditionTrue, metav1.ConditionTrue, metav1.ConditionTrue},
+				eventuallyTimeout*3, eventuallyInterval)
+			util.AssertWorkCondition(work.Namespace, work.Name, hubWorkClient, string(workapiv1.WorkAvailable),
+				metav1.ConditionTrue,
+				[]metav1.ConditionStatus{metav1.ConditionTrue, metav1.ConditionTrue, metav1.ConditionTrue},
+				eventuallyTimeout, eventuallyInterval)
+
+			// ensure configmaps exist
+			util.AssertExistenceOfConfigMaps(
+				[]workapiv1.Manifest{
+					util.ToManifest(util.NewConfigmap(o.SpokeClusterName, "cm1", map[string]string{"a": "b"}, []string{})),
+				}, spokeKubeClient, eventuallyTimeout, eventuallyInterval)
+		})
+	})
 })
diff --git a/test/integration/util/util.go b/test/integration/util/util.go
index bb1e4d179..ed67e148f 100644
--- a/test/integration/util/util.go
+++ b/test/integration/util/util.go
@@ -8,6 +8,7 @@ import (
 
 	"github.com/openshift/library-go/pkg/operator/events"
 	corev1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -75,6 +76,36 @@ func NewConfigmap(namespace, name string, data map[string]string, finalizers []s
 	return cm
 }
 
+func NewRoleForManifest(namespace, name string, rules ...rbacv1.PolicyRule) *rbacv1.Role {
+	return &rbacv1.Role{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Role",
+			APIVersion: "rbac.authorization.k8s.io/v1",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+		},
+		Rules: rules,
+	}
+}
+
+func NewRoleBindingForManifest(namespace, name string, rule rbacv1.RoleRef,
+	subjects ...rbacv1.Subject) *rbacv1.RoleBinding {
+	return &rbacv1.RoleBinding{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "RoleBinding",
+			APIVersion: "rbac.authorization.k8s.io/v1",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+		},
+		Subjects: subjects,
+		RoleRef:  rule,
+	}
+}
+
 func ToManifest(object runtime.Object) workapiv1.Manifest {
 	manifest := workapiv1.Manifest{}
 	manifest.Object = object