grafana-iam: Use parent folder to authorize ResourcePermissions (#115008)

* `grafana-iam`: Fetch target parent folder

* WIP add different ParentProviders

* Add version

* Move code to a different file

* Instantiate resourceParentProvider

* same import name

* imports

* Add tests

* Remove unecessary test

* forgot wire

* WIP integration tests

* Add test to cover list

* Fix caching problem in integration tests

* comments

* Logger and comments

* Add lazy creation and caching

* Instantiate clients only once

* Rerun wire gen

pkg/registry/apis/iam/authorizer/parent_provider.go

@@ -0,0 +1,164 @@
+package authorizer
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"sync"
+
+	"github.com/grafana/authlib/authn"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/rest"
+
+	dashboardv1 "github.com/grafana/grafana/apps/dashboard/pkg/apis/dashboard/v1beta1"
+	folderv1 "github.com/grafana/grafana/apps/folder/pkg/apis/folder/v1beta1"
+	"github.com/grafana/grafana/apps/provisioning/pkg/auth"
+	"github.com/grafana/grafana/pkg/apimachinery/utils"
+)
+
+var (
+	ErrNoConfigProvider = errors.New("no config provider for group resource")
+	ErrNoVersionInfo    = errors.New("no version info for group resource")
+
+	Versions = map[schema.GroupResource]string{
+		{Group: folderv1.GROUP, Resource: folderv1.RESOURCE}:                 folderv1.VERSION,
+		{Group: dashboardv1.GROUP, Resource: dashboardv1.DASHBOARD_RESOURCE}: dashboardv1.VERSION,
+	}
+)
+
+// ConfigProvider is a function that provides a rest.Config for a given context.
+type ConfigProvider func(ctx context.Context) (*rest.Config, error)
+
+// DynamicClientFactory is a function that creates a dynamic.Interface from a rest.Config.
+// This can be overridden in tests.
+type DynamicClientFactory func(config *rest.Config) (dynamic.Interface, error)
+
+// ParentProvider implementation that fetches the parent folder information from remote API servers.
+type ParentProviderImpl struct {
+	configProviders      map[schema.GroupResource]ConfigProvider
+	versions             map[schema.GroupResource]string
+	dynamicClientFactory DynamicClientFactory
+
+	// Cache of dynamic clients for each group resource
+	// This is used to avoid creating a new dynamic client for each request
+	// and to reuse the same client for the same group resource.
+	clients   map[schema.GroupResource]dynamic.Interface
+	clientsMu sync.Mutex
+}
+
+// DialConfig holds the configuration for dialing a remote API server.
+type DialConfig struct {
+	Host     string
+	Insecure bool
+	CAFile   string
+	Audience string
+}
+
+// NewLocalConfigProvider creates a map of ConfigProviders that return the same given config for local API servers.
+func NewLocalConfigProvider(
+	configProvider ConfigProvider,
+) map[schema.GroupResource]ConfigProvider {
+	return map[schema.GroupResource]ConfigProvider{
+		{Group: folderv1.GROUP, Resource: folderv1.RESOURCE}:                 configProvider,
+		{Group: dashboardv1.GROUP, Resource: dashboardv1.DASHBOARD_RESOURCE}: configProvider,
+	}
+}
+
+// NewRemoteConfigProvider creates a map of ConfigProviders for remote API servers based on the given DialConfig.
+func NewRemoteConfigProvider(cfg map[schema.GroupResource]DialConfig, exchangeClient authn.TokenExchanger) map[schema.GroupResource]ConfigProvider {
+	configProviders := make(map[schema.GroupResource]ConfigProvider, len(cfg))
+	for gr, dialConfig := range cfg {
+		configProviders[gr] = func(ctx context.Context) (*rest.Config, error) {
+			return &rest.Config{
+				Host: dialConfig.Host,
+				WrapTransport: func(rt http.RoundTripper) http.RoundTripper {
+					return auth.NewRoundTripper(exchangeClient, rt, dialConfig.Audience)
+				},
+				TLSClientConfig: rest.TLSClientConfig{
+					Insecure: dialConfig.Insecure,
+					CAFile:   dialConfig.CAFile,
+				},
+				QPS:   50,
+				Burst: 100,
+			}, nil
+		}
+	}
+	return configProviders
+}
+
+// NewApiParentProvider creates a new ParentProviderImpl with the given config providers and version info.
+func NewApiParentProvider(
+	configProviders map[schema.GroupResource]ConfigProvider,
+	version map[schema.GroupResource]string,
+) *ParentProviderImpl {
+	return &ParentProviderImpl{
+		configProviders: configProviders,
+		versions:        version,
+		dynamicClientFactory: func(config *rest.Config) (dynamic.Interface, error) {
+			return dynamic.NewForConfig(config)
+		},
+		clients: make(map[schema.GroupResource]dynamic.Interface),
+	}
+}
+
+func (p *ParentProviderImpl) HasParent(gr schema.GroupResource) bool {
+	_, ok := p.configProviders[gr]
+	return ok
+}
+
+func (p *ParentProviderImpl) getClient(ctx context.Context, gr schema.GroupResource) (dynamic.Interface, error) {
+	p.clientsMu.Lock()
+	client, ok := p.clients[gr]
+	p.clientsMu.Unlock()
+
+	if ok {
+		return client, nil
+	}
+
+	provider, ok := p.configProviders[gr]
+	if !ok {
+		return nil, fmt.Errorf("%w: %s", ErrNoConfigProvider, gr.String())
+	}
+	restConfig, err := provider(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	client, err = p.dynamicClientFactory(restConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	p.clientsMu.Lock()
+	p.clients[gr] = client
+	p.clientsMu.Unlock()
+
+	return client, nil
+}
+
+func (p *ParentProviderImpl) GetParent(ctx context.Context, gr schema.GroupResource, namespace, name string) (string, error) {
+	client, err := p.getClient(ctx, gr)
+	if err != nil {
+		return "", err
+	}
+
+	version, ok := p.versions[gr]
+	if !ok {
+		return "", fmt.Errorf("%w: %s", ErrNoVersionInfo, gr.String())
+	}
+	resourceClient := client.Resource(schema.GroupVersionResource{
+		Group:    gr.Group,
+		Resource: gr.Resource,
+		Version:  version,
+	}).Namespace(namespace)
+
+	unstructObj, err := resourceClient.Get(ctx, name, metav1.GetOptions{})
+	if err != nil {
+		return "", err
+	}
+
+	return unstructObj.GetAnnotations()[utils.AnnoKeyFolder], nil
+}

pkg/registry/apis/iam/authorizer/parent_provider_test.go

@@ -0,0 +1,198 @@
+package authorizer
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	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/rest"
+
+	folderv1 "github.com/grafana/grafana/apps/folder/pkg/apis/folder/v1beta1"
+	"github.com/grafana/grafana/pkg/apimachinery/utils"
+)
+
+var configProvider = func(ctx context.Context) (*rest.Config, error) {
+	return &rest.Config{}, nil
+}
+
+func TestParentProviderImpl_GetParent(t *testing.T) {
+	tests := []struct {
+		name            string
+		gr              schema.GroupResource
+		namespace       string
+		resourceName    string
+		parentFolder    string
+		setupFake       func(*fakeDynamicClient, *fakeResourceInterface)
+		configProviders map[schema.GroupResource]ConfigProvider
+		versions        map[schema.GroupResource]string
+		expectedError   string
+		expectedParent  string
+	}{
+		{
+			name:         "successfully get parent folder",
+			gr:           schema.GroupResource{Group: folderv1.GROUP, Resource: folderv1.RESOURCE},
+			namespace:    "org-1",
+			resourceName: "dash1",
+			parentFolder: "fold1",
+			setupFake: func(fakeClient *fakeDynamicClient, fakeResource *fakeResourceInterface) {
+				fakeClient.resourceInterface = fakeResource
+				fakeResource.getFunc = func(ctx context.Context, name string, opts metav1.GetOptions, subresources ...string) (*unstructured.Unstructured, error) {
+					obj := &unstructured.Unstructured{}
+					obj.SetAnnotations(map[string]string{utils.AnnoKeyFolder: "fold1"})
+					return obj, nil
+				}
+			},
+			configProviders: map[schema.GroupResource]ConfigProvider{
+				{Group: folderv1.GROUP, Resource: folderv1.RESOURCE}: configProvider,
+			},
+			versions:       Versions,
+			expectedParent: "fold1",
+		},
+		{
+			name:         "resource without parent annotation returns empty",
+			gr:           schema.GroupResource{Group: folderv1.GROUP, Resource: folderv1.RESOURCE},
+			namespace:    "org-1",
+			resourceName: "dash1",
+			setupFake: func(fakeClient *fakeDynamicClient, fakeResource *fakeResourceInterface) {
+				fakeClient.resourceInterface = fakeResource
+				fakeResource.getFunc = func(ctx context.Context, name string, opts metav1.GetOptions, subresources ...string) (*unstructured.Unstructured, error) {
+					obj := &unstructured.Unstructured{}
+					obj.SetAnnotations(map[string]string{})
+					return obj, nil
+				}
+			},
+			configProviders: map[schema.GroupResource]ConfigProvider{
+				{Group: folderv1.GROUP, Resource: folderv1.RESOURCE}: configProvider,
+			},
+			versions:       Versions,
+			expectedParent: "",
+		},
+		{
+			name:            "no config provider returns error",
+			gr:              schema.GroupResource{Group: "unknown.group", Resource: "unknown"},
+			namespace:       "org-1",
+			resourceName:    "resource-1",
+			configProviders: map[schema.GroupResource]ConfigProvider{},
+			versions:        Versions,
+			expectedError:   ErrNoConfigProvider.Error(),
+		},
+		{
+			name:         "config provider returns error",
+			gr:           schema.GroupResource{Group: folderv1.GROUP, Resource: folderv1.RESOURCE},
+			namespace:    "org-1",
+			resourceName: "resource-1",
+			configProviders: map[schema.GroupResource]ConfigProvider{
+				{Group: folderv1.GROUP, Resource: folderv1.RESOURCE}: func(ctx context.Context) (*rest.Config, error) {
+					return nil, errors.New("config provider error")
+				},
+			},
+			versions:      Versions,
+			expectedError: "config provider error",
+		},
+		{
+			name:         "no version info returns error",
+			gr:           schema.GroupResource{Group: folderv1.GROUP, Resource: folderv1.RESOURCE},
+			namespace:    "org-1",
+			resourceName: "resource-1",
+			configProviders: map[schema.GroupResource]ConfigProvider{
+				{Group: folderv1.GROUP, Resource: folderv1.RESOURCE}: func(ctx context.Context) (*rest.Config, error) {
+					return &rest.Config{}, nil
+				},
+			},
+			versions:      map[schema.GroupResource]string{},
+			expectedError: ErrNoVersionInfo.Error(),
+		},
+		{
+			name:         "resource get returns error",
+			gr:           schema.GroupResource{Group: folderv1.GROUP, Resource: folderv1.RESOURCE},
+			namespace:    "org-1",
+			resourceName: "resource-1",
+			setupFake: func(fakeClient *fakeDynamicClient, fakeResource *fakeResourceInterface) {
+				fakeClient.resourceInterface = fakeResource
+				fakeResource.getFunc = func(ctx context.Context, name string, opts metav1.GetOptions, subresources ...string) (*unstructured.Unstructured, error) {
+					return nil, errors.New("resource not found")
+				}
+			},
+			configProviders: map[schema.GroupResource]ConfigProvider{
+				{Group: folderv1.GROUP, Resource: folderv1.RESOURCE}: func(ctx context.Context) (*rest.Config, error) {
+					return &rest.Config{}, nil
+				},
+			},
+			versions:      Versions,
+			expectedError: "resource not found",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			fakeClient := &fakeDynamicClient{}
+			fakeResource := &fakeResourceInterface{}
+			if tt.setupFake != nil {
+				tt.setupFake(fakeClient, fakeResource)
+			}
+
+			provider := &ParentProviderImpl{
+				configProviders: tt.configProviders,
+				versions:        tt.versions,
+				dynamicClientFactory: func(config *rest.Config) (dynamic.Interface, error) {
+					return fakeClient, nil
+				},
+				clients: make(map[schema.GroupResource]dynamic.Interface),
+			}
+
+			parent, err := provider.GetParent(context.Background(), tt.gr, tt.namespace, tt.resourceName)
+
+			if tt.expectedError != "" {
+				require.Error(t, err)
+				assert.Contains(t, err.Error(), tt.expectedError)
+				assert.Empty(t, parent)
+			} else {
+				require.NoError(t, err)
+				assert.Equal(t, tt.expectedParent, parent)
+			}
+		})
+	}
+}
+
+// fakeDynamicClient is a fake implementation of dynamic.Interface
+type fakeDynamicClient struct {
+	resourceInterface dynamic.ResourceInterface
+}
+
+func (f *fakeDynamicClient) Resource(resource schema.GroupVersionResource) dynamic.NamespaceableResourceInterface {
+	return &fakeNamespaceableResourceInterface{
+		resourceInterface: f.resourceInterface,
+	}
+}
+
+// fakeNamespaceableResourceInterface is a fake implementation of dynamic.NamespaceableResourceInterface
+type fakeNamespaceableResourceInterface struct {
+	dynamic.NamespaceableResourceInterface
+	resourceInterface dynamic.ResourceInterface
+}
+
+func (f *fakeNamespaceableResourceInterface) Namespace(namespace string) dynamic.ResourceInterface {
+	if f.resourceInterface != nil {
+		return f.resourceInterface
+	}
+	return &fakeResourceInterface{}
+}
+
+// fakeResourceInterface is a fake implementation of dynamic.ResourceInterface
+type fakeResourceInterface struct {
+	dynamic.ResourceInterface
+	getFunc func(ctx context.Context, name string, opts metav1.GetOptions, subresources ...string) (*unstructured.Unstructured, error)
+}
+
+func (f *fakeResourceInterface) Get(ctx context.Context, name string, opts metav1.GetOptions, subresources ...string) (*unstructured.Unstructured, error) {
+	if f.getFunc != nil {
+		return f.getFunc(ctx, name, opts, subresources...)
+	}
+	return &unstructured.Unstructured{}, nil
+}

pkg/registry/apis/iam/authorizer/resource_permissions.go

@@ -10,24 +10,44 @@ import (
 
 	iamv0 "github.com/grafana/grafana/apps/iam/pkg/apis/iam/v0alpha1"
 	"github.com/grafana/grafana/pkg/apimachinery/utils"
+	"github.com/grafana/grafana/pkg/infra/log"
 	"github.com/grafana/grafana/pkg/services/apiserver/auth/authorizer/storewrapper"
 )
 
 // TODO: Logs, Metrics, Traces?
 
+// ParentProvider interface for fetching parent information of resources
+type ParentProvider interface {
+	// HasParent checks if the given GroupResource has a parent folder
+	HasParent(gr schema.GroupResource) bool
+	// GetParent fetches the parent folder name for the given resource
+	GetParent(ctx context.Context, gr schema.GroupResource, namespace, name string) (string, error)
+}
+
 // ResourcePermissionsAuthorizer
 type ResourcePermissionsAuthorizer struct {
-	accessClient types.AccessClient
+	accessClient   types.AccessClient
+	parentProvider ParentProvider
+	logger         log.Logger
 }
 
 var _ storewrapper.ResourceStorageAuthorizer = (*ResourcePermissionsAuthorizer)(nil)
 
-func NewResourcePermissionsAuthorizer(accessClient types.AccessClient) *ResourcePermissionsAuthorizer {
+func NewResourcePermissionsAuthorizer(
+	accessClient types.AccessClient,
+	parentProvider ParentProvider,
+) *ResourcePermissionsAuthorizer {
 	return &ResourcePermissionsAuthorizer{
-		accessClient: accessClient,
+		accessClient:   accessClient,
+		parentProvider: parentProvider,
+		logger:         log.New("iam.resource-permissions-authorizer"),
 	}
 }
 
+func isAccessPolicy(authInfo types.AuthInfo) bool {
+	return types.IsIdentityType(authInfo.GetIdentityType(), types.TypeAccessPolicy)
+}
+
 // AfterGet implements ResourceStorageAuthorizer.
 func (r *ResourcePermissionsAuthorizer) AfterGet(ctx context.Context, obj runtime.Object) error {
 	authInfo, ok := types.AuthInfoFrom(ctx)
@@ -37,9 +57,24 @@ func (r *ResourcePermissionsAuthorizer) AfterGet(ctx context.Context, obj runtim
 	switch o := obj.(type) {
 	case *iamv0.ResourcePermission:
 		target := o.Spec.Resource
+		targetGR := schema.GroupResource{Group: target.ApiGroup, Resource: target.Resource}
 
-		// TODO: Fetch the resource to retrieve its parent folder.
 		parent := ""
+		// Fetch the parent of the resource
+		// Access Policies have global scope, so no parent check needed
+		if !isAccessPolicy(authInfo) && r.parentProvider.HasParent(targetGR) {
+			p, err := r.parentProvider.GetParent(ctx, targetGR, o.Namespace, target.Name)
+			if err != nil {
+				r.logger.Error("after get: error fetching parent", "error", err.Error(),
+					"namespace", o.Namespace,
+					"group", target.ApiGroup,
+					"resource", target.Resource,
+					"name", target.Name,
+				)
+				return err
+			}
+			parent = p
+		}
 
 		checkReq := types.CheckRequest{
 			Namespace: o.Namespace,
@@ -72,9 +107,24 @@ func (r *ResourcePermissionsAuthorizer) beforeWrite(ctx context.Context, obj run
 	switch o := obj.(type) {
 	case *iamv0.ResourcePermission:
 		target := o.Spec.Resource
+		targetGR := schema.GroupResource{Group: target.ApiGroup, Resource: target.Resource}
 
-		// TODO: Fetch the resource to retrieve its parent folder.
 		parent := ""
+		// Fetch the parent of the resource
+		// Access Policies have global scope, so no parent check needed
+		if !isAccessPolicy(authInfo) && r.parentProvider.HasParent(targetGR) {
+			p, err := r.parentProvider.GetParent(ctx, targetGR, o.Namespace, target.Name)
+			if err != nil {
+				r.logger.Error("before write: error fetching parent", "error", err.Error(),
+					"namespace", o.Namespace,
+					"group", target.ApiGroup,
+					"resource", target.Resource,
+					"name", target.Name,
+				)
+				return err
+			}
+			parent = p
+		}
 
 		checkReq := types.CheckRequest{
 			Namespace: o.Namespace,
@@ -153,8 +203,29 @@ func (r *ResourcePermissionsAuthorizer) FilterList(ctx context.Context, list run
 				canViewFuncs[gr] = canView
 			}
 
-			// TODO : Fetch the resource to retrieve its parent folder.
+			target := item.Spec.Resource
+			targetGR := schema.GroupResource{Group: target.ApiGroup, Resource: target.Resource}
+
 			parent := ""
+			// Fetch the parent of the resource
+			// It's not efficient to do for every item in the list, but it's a good starting point.
+			// Access Policies have global scope, so no parent check needed
+			if !isAccessPolicy(authInfo) && r.parentProvider.HasParent(targetGR) {
+				p, err := r.parentProvider.GetParent(ctx, targetGR, item.Namespace, target.Name)
+				if err != nil {
+					// Skip item on error fetching parent
+					r.logger.Warn("filter list: error fetching parent, skipping item",
+						"error", err.Error(),
+						"namespace",
+						item.Namespace,
+						"group", target.ApiGroup,
+						"resource", target.Resource,
+						"name", target.Name,
+					)
+					continue
+				}
+				parent = p
+			}
 
 			allowed := canView(item.Spec.Resource.Name, parent)
 			if allowed {

pkg/registry/apis/iam/authorizer/resource_permissions_test.go

@@ -5,13 +5,15 @@ import (
 	"testing"
 
 	"github.com/go-jose/go-jose/v4/jwt"
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+
 	"github.com/grafana/authlib/authn"
 	"github.com/grafana/authlib/types"
 	iamv0 "github.com/grafana/grafana/apps/iam/pkg/apis/iam/v0alpha1"
 	"github.com/grafana/grafana/pkg/apimachinery/identity"
 	"github.com/grafana/grafana/pkg/apimachinery/utils"
-	"github.com/stretchr/testify/require"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 var (
@@ -63,6 +65,7 @@ func TestResourcePermissions_AfterGet(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
+			parent := "fold-1"
 			checkFunc := func(id types.AuthInfo, req *types.CheckRequest, folder string) (types.CheckResponse, error) {
 				require.NotNil(t, id)
 				// Check is called with the user's identity
@@ -74,12 +77,18 @@ func TestResourcePermissions_AfterGet(t *testing.T) {
 				require.Equal(t, fold1.Spec.Resource.Resource, req.Resource)
 				require.Equal(t, fold1.Spec.Resource.Name, req.Name)
 				require.Equal(t, utils.VerbGetPermissions, req.Verb)
+				require.Equal(t, parent, folder)
 
 				return types.CheckResponse{Allowed: tt.shouldAllow}, nil
 			}
+			getParentFunc := func(ctx context.Context, gr schema.GroupResource, namespace, name string) (string, error) {
+				// For this test, we can return a fixed parent folder ID
+				return parent, nil
+			}
 
 			accessClient := &fakeAccessClient{checkFunc: checkFunc}
-			resPermAuthz := NewResourcePermissionsAuthorizer(accessClient)
+			fakeParentProvider := &fakeParentProvider{hasParent: true, getParentFunc: getParentFunc}
+			resPermAuthz := NewResourcePermissionsAuthorizer(accessClient, fakeParentProvider)
 			ctx := types.WithAuthInfo(context.Background(), user)
 
 			err := resPermAuthz.AfterGet(ctx, fold1)
@@ -89,6 +98,7 @@ func TestResourcePermissions_AfterGet(t *testing.T) {
 				require.Error(t, err, "expected error for denied access")
 			}
 			require.True(t, accessClient.checkCalled, "accessClient.Check should be called")
+			require.True(t, fakeParentProvider.getParentCalled, "parentProvider.GetParent should be called")
 		})
 	}
 }
@@ -121,23 +131,32 @@ func TestResourcePermissions_FilterList(t *testing.T) {
 			require.Equal(t, "dashboards", req.Resource)
 		}
 
-		// Return a checker that allows only specific resources: fold-1 and dash-2
+		// Return a checker that allows access to fold-1 and its content
 		return func(name, folder string) bool {
-			if name == "fold-1" || name == "dash-2" {
+			if name == "fold-1" || folder == "fold-1" {
 				return true
 			}
 			return false
 		}, &types.NoopZookie{}, nil
 	}
 
+	getParentFunc := func(ctx context.Context, gr schema.GroupResource, namespace, name string) (string, error) {
+		if name == "dash-2" {
+			return "fold-1", nil
+		}
+		return "", nil
+	}
+
 	accessClient := &fakeAccessClient{compileFunc: compileFunc}
-	resPermAuthz := NewResourcePermissionsAuthorizer(accessClient)
+	fakeParentProvider := &fakeParentProvider{hasParent: true, getParentFunc: getParentFunc}
+	resPermAuthz := NewResourcePermissionsAuthorizer(accessClient, fakeParentProvider)
 	ctx := types.WithAuthInfo(context.Background(), user)
 
 	obj, err := resPermAuthz.FilterList(ctx, list)
 	require.NoError(t, err)
 	require.NotNil(t, list)
 	require.True(t, accessClient.compileCalled, "accessClient.Compile should be called")
+	require.True(t, fakeParentProvider.getParentCalled, "parentProvider.GetParent should be called")
 
 	filtered, ok := obj.(*iamv0.ResourcePermissionList)
 	require.True(t, ok, "response should be of type ResourcePermissionList")
@@ -165,6 +184,7 @@ func TestResourcePermissions_beforeWrite(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
+			parent := "fold-1"
 			checkFunc := func(id types.AuthInfo, req *types.CheckRequest, folder string) (types.CheckResponse, error) {
 				require.NotNil(t, id)
 				// Check is called with the user's identity
@@ -176,12 +196,18 @@ func TestResourcePermissions_beforeWrite(t *testing.T) {
 				require.Equal(t, fold1.Spec.Resource.Resource, req.Resource)
 				require.Equal(t, fold1.Spec.Resource.Name, req.Name)
 				require.Equal(t, utils.VerbSetPermissions, req.Verb)
+				require.Equal(t, parent, folder)
 
 				return types.CheckResponse{Allowed: tt.shouldAllow}, nil
 			}
 
+			getParentFunc := func(ctx context.Context, gr schema.GroupResource, namespace, name string) (string, error) {
+				return parent, nil
+			}
+
 			accessClient := &fakeAccessClient{checkFunc: checkFunc}
-			resPermAuthz := NewResourcePermissionsAuthorizer(accessClient)
+			fakeParentProvider := &fakeParentProvider{hasParent: true, getParentFunc: getParentFunc}
+			resPermAuthz := NewResourcePermissionsAuthorizer(accessClient, fakeParentProvider)
 			ctx := types.WithAuthInfo(context.Background(), user)
 
 			err := resPermAuthz.beforeWrite(ctx, fold1)
@@ -191,6 +217,7 @@ func TestResourcePermissions_beforeWrite(t *testing.T) {
 				require.Error(t, err, "expected error for denied delete")
 			}
 			require.True(t, accessClient.checkCalled, "accessClient.Check should be called")
+			require.True(t, fakeParentProvider.getParentCalled, "parentProvider.GetParent should be called")
 		})
 	}
 }
@@ -214,3 +241,18 @@ func (m *fakeAccessClient) Compile(ctx context.Context, id types.AuthInfo, req t
 }
 
 var _ types.AccessClient = (*fakeAccessClient)(nil)
+
+type fakeParentProvider struct {
+	hasParent       bool
+	getParentCalled bool
+	getParentFunc   func(ctx context.Context, gr schema.GroupResource, namespace, name string) (string, error)
+}
+
+func (f *fakeParentProvider) HasParent(gr schema.GroupResource) bool {
+	return f.hasParent
+}
+
+func (f *fakeParentProvider) GetParent(ctx context.Context, gr schema.GroupResource, namespace, name string) (string, error) {
+	f.getParentCalled = true
+	return f.getParentFunc(ctx, gr, namespace, name)
+}

pkg/registry/apis/iam/models.go

@@ -7,6 +7,7 @@ import (
 	"github.com/grafana/authlib/types"
 
 	"github.com/grafana/grafana/pkg/infra/log"
+	iamauthorizer "github.com/grafana/grafana/pkg/registry/apis/iam/authorizer"
 	"github.com/grafana/grafana/pkg/registry/apis/iam/externalgroupmapping"
 	"github.com/grafana/grafana/pkg/registry/apis/iam/legacy"
 	"github.com/grafana/grafana/pkg/registry/apis/iam/serviceaccount"
@@ -60,6 +61,10 @@ type IdentityAccessManagementAPIBuilder struct {
 	roleBindingsStorage         RoleBindingStorageBackend
 	externalGroupMappingStorage ExternalGroupMappingStorageBackend
 
+	// Required for resource permissions authorization
+	// fetches resources parent folders
+	resourceParentProvider iamauthorizer.ParentProvider
+
 	// Access Control
 	authorizer authorizer.Authorizer
 	// legacyAccessClient is used for the identity apis, we need to migrate to the access client

pkg/registry/apis/iam/register.go

@@ -41,6 +41,7 @@ import (
 	"github.com/grafana/grafana/pkg/registry/apis/iam/teambinding"
 	"github.com/grafana/grafana/pkg/registry/apis/iam/user"
 	"github.com/grafana/grafana/pkg/services/accesscontrol"
+	"github.com/grafana/grafana/pkg/services/apiserver"
 	gfauthorizer "github.com/grafana/grafana/pkg/services/apiserver/auth/authorizer"
 	"github.com/grafana/grafana/pkg/services/apiserver/auth/authorizer/storewrapper"
 	"github.com/grafana/grafana/pkg/services/apiserver/builder"
@@ -80,6 +81,7 @@ func RegisterAPIService(
 	orgService org.Service,
 	userService legacyuser.Service,
 	teamService teamservice.Service,
+	restConfig apiserver.RestConfigProvider,
 ) (*IdentityAccessManagementAPIBuilder, error) {
 	dbProvider := legacysql.NewDatabaseProvider(sql)
 	store := legacy.NewLegacySQLStores(dbProvider)
@@ -90,6 +92,11 @@ func RegisterAPIService(
 	//nolint:staticcheck // not yet migrated to OpenFeature
 	enableAuthnMutation := features.IsEnabledGlobally(featuremgmt.FlagKubernetesAuthnMutation)
 
+	resourceParentProvider := iamauthorizer.NewApiParentProvider(
+		iamauthorizer.NewLocalConfigProvider(restConfig.GetRestConfig),
+		iamauthorizer.Versions,
+	)
+
 	builder := &IdentityAccessManagementAPIBuilder{
 		store:                       store,
 		userLegacyStore:             user.NewLegacyStore(store, accessClient, enableAuthnMutation, tracing),
@@ -104,6 +111,7 @@ func RegisterAPIService(
 		externalGroupMappingStorage: externalGroupMappingStorageBackend,
 		teamGroupsHandler:           teamGroupsHandlerImpl,
 		sso:                         ssoService,
+		resourceParentProvider:      resourceParentProvider,
 		authorizer:                  authorizer,
 		legacyAccessClient:          legacyAccessClient,
 		accessClient:                accessClient,
@@ -142,6 +150,12 @@ func NewAPIService(
 	resourceAuthorizer := gfauthorizer.NewResourceAuthorizer(accessClient)
 	coreRoleAuthorizer := iamauthorizer.NewCoreRoleAuthorizer(accessClient)
 
+	// TODO: in a follow up PR, make this configurable
+	resourceParentProvider := iamauthorizer.NewApiParentProvider(
+		iamauthorizer.NewRemoteConfigProvider(map[schema.GroupResource]iamauthorizer.DialConfig{}, nil),
+		iamauthorizer.Versions,
+	)
+
 	return &IdentityAccessManagementAPIBuilder{
 		store:                      store,
 		display:                    user.NewLegacyDisplayREST(store),
@@ -152,6 +166,7 @@ func NewAPIService(
 		logger:                     log.New("iam.apis"),
 		features:                   features,
 		accessClient:               accessClient,
+		resourceParentProvider:     resourceParentProvider,
 		zClient:                    zClient,
 		zTickets:                   make(chan bool, MaxConcurrentZanzanaWrites),
 		reg:                        reg,
@@ -444,7 +459,7 @@ func (b *IdentityAccessManagementAPIBuilder) UpdateResourcePermissionsAPIGroup(
 		return fmt.Errorf("expected RegistryStoreDualWrite, got %T", dw)
 	}
 
-	authzWrapper := storewrapper.New(regStoreDW, iamauthorizer.NewResourcePermissionsAuthorizer(b.accessClient))
+	authzWrapper := storewrapper.New(regStoreDW, iamauthorizer.NewResourcePermissionsAuthorizer(b.accessClient, b.resourceParentProvider))
 
 	storage[iamv0.ResourcePermissionInfo.StoragePath()] = authzWrapper
 	return nil

pkg/server/wire_gen.go

@@ -879,7 +879,7 @@ func Initialize(ctx context.Context, cfg *setting.Cfg, opts Options, apiOpts api
 	folderAPIBuilder := folders.RegisterAPIService(cfg, featureToggles, apiserverService, folderimplService, folderPermissionsService, accessControl, acimplService, accessClient, registerer, resourceClient, zanzanaClient)
 	storageBackendImpl := noopstorage.ProvideStorageBackend()
 	noopTeamGroupsREST := externalgroupmapping.ProvideNoopTeamGroupsREST()
-	identityAccessManagementAPIBuilder, err := iam.RegisterAPIService(cfg, featureToggles, apiserverService, ssosettingsimplService, sqlStore, accessControl, accessClient, zanzanaClient, registerer, storageBackendImpl, storageBackendImpl, tracingService, storageBackendImpl, storageBackendImpl, noopTeamGroupsREST, dualwriteService, resourceClient, orgService, userService, teamService)
+	identityAccessManagementAPIBuilder, err := iam.RegisterAPIService(cfg, featureToggles, apiserverService, ssosettingsimplService, sqlStore, accessControl, accessClient, zanzanaClient, registerer, storageBackendImpl, storageBackendImpl, tracingService, storageBackendImpl, storageBackendImpl, noopTeamGroupsREST, dualwriteService, resourceClient, orgService, userService, teamService, eventualRestConfigProvider)
 	if err != nil {
 		return nil, err
 	}
@@ -1537,7 +1537,7 @@ func InitializeForTest(ctx context.Context, t sqlutil.ITestDB, testingT interfac
 	folderAPIBuilder := folders.RegisterAPIService(cfg, featureToggles, apiserverService, folderimplService, folderPermissionsService, accessControl, acimplService, accessClient, registerer, resourceClient, zanzanaClient)
 	storageBackendImpl := noopstorage.ProvideStorageBackend()
 	noopTeamGroupsREST := externalgroupmapping.ProvideNoopTeamGroupsREST()
-	identityAccessManagementAPIBuilder, err := iam.RegisterAPIService(cfg, featureToggles, apiserverService, ssosettingsimplService, sqlStore, accessControl, accessClient, zanzanaClient, registerer, storageBackendImpl, storageBackendImpl, tracingService, storageBackendImpl, storageBackendImpl, noopTeamGroupsREST, dualwriteService, resourceClient, orgService, userService, teamService)
+	identityAccessManagementAPIBuilder, err := iam.RegisterAPIService(cfg, featureToggles, apiserverService, ssosettingsimplService, sqlStore, accessControl, accessClient, zanzanaClient, registerer, storageBackendImpl, storageBackendImpl, tracingService, storageBackendImpl, storageBackendImpl, noopTeamGroupsREST, dualwriteService, resourceClient, orgService, userService, teamService, eventualRestConfigProvider)
 	if err != nil {
 		return nil, err
 	}

pkg/tests/apis/iam/resource_permissions_integration_test.go

@@ -0,0 +1,625 @@
+package identity
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"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"
+
+	iamv0 "github.com/grafana/grafana/apps/iam/pkg/apis/iam/v0alpha1"
+	"github.com/grafana/grafana/pkg/apimachinery/utils"
+	"github.com/grafana/grafana/pkg/apiserver/rest"
+	"github.com/grafana/grafana/pkg/services/featuremgmt"
+	"github.com/grafana/grafana/pkg/setting"
+	"github.com/grafana/grafana/pkg/tests/apis"
+	"github.com/grafana/grafana/pkg/tests/testinfra"
+	"github.com/grafana/grafana/pkg/util/testutil"
+)
+
+var gvrResourcePermissions = schema.GroupVersionResource{
+	Group:    "iam.grafana.app",
+	Version:  "v0alpha1",
+	Resource: "resourcepermissions",
+}
+
+var gvrFolders = schema.GroupVersionResource{
+	Group:    "folder.grafana.app",
+	Version:  "v1beta1",
+	Resource: "folders",
+}
+
+var gvrDashboards = schema.GroupVersionResource{
+	Group:    "dashboard.grafana.app",
+	Version:  "v1beta1",
+	Resource: "dashboards",
+}
+
+type permission struct {
+	kind string
+	name string
+	verb string
+}
+
+func newPermission(kind, name, verb string) permission {
+	return permission{
+		kind: kind,
+		name: name,
+		verb: verb,
+	}
+}
+
+func (p permission) ToMap() map[string]interface{} {
+	return map[string]interface{}{
+		"kind": p.kind,
+		"name": p.name,
+		"verb": p.verb,
+	}
+}
+
+func newPermissionMaps(permissions ...permission) []map[string]interface{} {
+	permissionsMaps := make([]map[string]interface{}, len(permissions))
+	for i, permission := range permissions {
+		permissionsMaps[i] = permission.ToMap()
+	}
+	return permissionsMaps
+}
+
+type k8sTestClients struct {
+	rpAdmin  *apis.K8sResourceClient
+	rpEditor *apis.K8sResourceClient
+	rpViewer *apis.K8sResourceClient
+}
+
+func newk8sTestHelperClients(helper *apis.K8sTestHelper) *k8sTestClients {
+	return &k8sTestClients{
+		rpAdmin: helper.GetResourceClient(apis.ResourceClientArgs{
+			User:      helper.Org1.Admin,
+			Namespace: helper.Namespacer(helper.Org1.OrgID),
+			GVR:       gvrResourcePermissions,
+		}),
+		rpEditor: helper.GetResourceClient(apis.ResourceClientArgs{
+			User:      helper.Org1.Editor,
+			Namespace: helper.Namespacer(helper.Org1.OrgID),
+			GVR:       gvrResourcePermissions,
+		}),
+		rpViewer: helper.GetResourceClient(apis.ResourceClientArgs{
+			User:      helper.Org1.Viewer,
+			Namespace: helper.Namespacer(helper.Org1.OrgID),
+			GVR:       gvrResourcePermissions,
+		}),
+	}
+}
+
+func TestIntegrationResourcePermissions(t *testing.T) {
+	testutil.SkipIntegrationTestInShortMode(t)
+
+	modes := []rest.DualWriterMode{rest.Mode0, rest.Mode1, rest.Mode2, rest.Mode3}
+	for _, mode := range modes {
+		if mode >= rest.Mode3 {
+			t.Skip("Skipping ResourcePermission tests for Mode3+ because default permissions are not written through the new APIs")
+			continue
+		}
+		t.Run(fmt.Sprintf("ResourcePermission CRUD with dual writer mode %d", mode), func(t *testing.T) {
+			// Turn off authorization cache so permission changes apply right away in tests
+			t.Setenv("GF_AUTHORIZATION_CACHE_TTL", "0s")
+
+			helper := apis.NewK8sTestHelper(t, testinfra.GrafanaOpts{
+				AppModeProduction:    false,
+				DisableAnonymous:     true,
+				APIServerStorageType: "unified",
+				UnifiedStorageConfig: map[string]setting.UnifiedStorageConfig{
+					"resourcepermissions.iam.grafana.app": {
+						DualWriterMode: mode,
+					},
+					"folders.folder.grafana.app": {
+						DualWriterMode: mode,
+					},
+					"dashboards.dashboard.grafana.app": {
+						DualWriterMode: mode,
+					},
+				},
+				EnableFeatureToggles: []string{
+					featuremgmt.FlagKubernetesAuthzResourcePermissionApis,
+					// Prevents nested folders from having default permissions
+					featuremgmt.FlagKubernetesDashboards,
+				},
+			})
+
+			// Work around the default permissions applied on root folders
+			// so we can test the ResourcePermission APIs without the default permissions interfering
+			parentFolder := createRootFolderWithoutDefaultPermissions(t, helper)
+			parentUID := parentFolder.GetName()
+
+			clients := newk8sTestHelperClients(helper)
+			doResourcePermissionCRUDTests(t, helper, clients, parentUID)
+			doResourcePermissionAuthzTests(t, helper, clients, parentUID)
+			doResourcePermissionHierarchyTests(t, helper, clients, parentUID)
+			doResourcePermissionListFilteringTests(t, helper, clients, parentUID)
+			// TODO: Add tests for External JWT authentication
+			// doResourcePermissionAccessPolicyTests(t, helper)
+		})
+	}
+}
+
+func doResourcePermissionCRUDTests(t *testing.T, helper *apis.K8sTestHelper, clients *k8sTestClients, parentUID string) {
+	t.Run("should create/get/update/delete ResourcePermission using the new APIs", func(t *testing.T) {
+		ctx := context.Background()
+
+		// Create ResourcePermission for the folder
+		permission := newPermission("ServiceAccount", helper.Org1.ViewerServiceAccount.UID, "view")
+		toCreate := createResourcePermissionObject(parentUID, gvrFolders.Group, gvrFolders.Resource, permission)
+
+		created, err := clients.rpAdmin.Resource.Create(ctx, toCreate, metav1.CreateOptions{})
+		require.NoError(t, err)
+		require.NotNil(t, created)
+
+		createdName := created.GetName()
+		require.NotEmpty(t, createdName)
+
+		// Verify spec
+		spec := created.Object["spec"].(map[string]interface{})
+		resource := spec["resource"].(map[string]interface{})
+		require.Equal(t, gvrFolders.Group, resource["apiGroup"])
+		require.Equal(t, gvrFolders.Resource, resource["resource"])
+		require.Equal(t, parentUID, resource["name"])
+
+		// Get the ResourcePermission
+		fetched, err := clients.rpAdmin.Resource.Get(ctx, createdName, metav1.GetOptions{})
+		require.NoError(t, err)
+		require.NotNil(t, fetched)
+		require.Equal(t, createdName, fetched.GetName())
+
+		// Update the ResourcePermission
+		fetched.Object["spec"].(map[string]interface{})["permissions"] = newPermissionMaps(
+			newPermission("User", helper.Org1.Viewer.Identity.GetIdentifier(), "edit"),
+		)
+		updated, err := clients.rpAdmin.Resource.Update(ctx, fetched, metav1.UpdateOptions{})
+		require.NoError(t, err)
+		require.NotNil(t, updated)
+
+		updatedSpec := updated.Object["spec"].(map[string]interface{})
+		permissions := updatedSpec["permissions"].([]interface{})
+		require.Len(t, permissions, 1)
+		perm := permissions[0].(map[string]interface{})
+		require.Equal(t, helper.Org1.Viewer.Identity.GetIdentifier(), perm["name"])
+		require.Equal(t, "edit", perm["verb"])
+
+		// Delete should work
+		err = clients.rpAdmin.Resource.Delete(ctx, createdName, metav1.DeleteOptions{})
+		require.NoError(t, err)
+
+		_, err = clients.rpAdmin.Resource.Get(ctx, createdName, metav1.GetOptions{})
+		require.Error(t, err)
+		var statusErr *errors.StatusError
+		require.ErrorAs(t, err, &statusErr)
+		require.Equal(t, int32(404), statusErr.ErrStatus.Code)
+	})
+
+	t.Run("should return 404 for non-existent ResourcePermission", func(t *testing.T) {
+		ctx := context.Background()
+
+		_, err := clients.rpAdmin.Resource.Get(ctx, "folder.grafana.app-folders-unknown", metav1.GetOptions{})
+		require.Error(t, err)
+		var statusErr *errors.StatusError
+		require.ErrorAs(t, err, &statusErr)
+		require.Equal(t, int32(404), statusErr.ErrStatus.Code)
+	})
+}
+
+func doResourcePermissionAuthzTests(t *testing.T, helper *apis.K8sTestHelper, clients *k8sTestClients, parentUID string) {
+	t.Run("admin can create/update/delete ResourcePermission", func(t *testing.T) {
+		ctx := context.Background()
+
+		folder := createTestFolder(t, helper, helper.Org1.Admin, "test-folder-admin", parentUID)
+		folderUID := folder.GetName()
+
+		permission := newPermission("User", helper.Org1.Admin.Identity.GetIdentifier(), "admin")
+
+		toCreate := createResourcePermissionObject(folderUID, gvrFolders.Group, gvrFolders.Resource, permission)
+		created, err := clients.rpAdmin.Resource.Create(ctx, toCreate, metav1.CreateOptions{})
+		require.NoError(t, err)
+		require.NotNil(t, created)
+
+		createdName := created.GetName()
+
+		// Get the created object
+		fetched, err := clients.rpAdmin.Resource.Get(ctx, createdName, metav1.GetOptions{})
+		require.NoError(t, err)
+		require.NotNil(t, fetched)
+
+		// Update should work
+		permission = newPermission("Team", helper.Org1.Staff.UID, "edit")
+		fetched.Object["spec"].(map[string]interface{})["permissions"] = []interface{}{permission.ToMap()}
+
+		_, err = clients.rpAdmin.Resource.Update(ctx, fetched, metav1.UpdateOptions{})
+		require.NoError(t, err)
+
+		// Delete should work
+		err = clients.rpAdmin.Resource.Delete(ctx, createdName, metav1.DeleteOptions{})
+		require.NoError(t, err)
+	})
+
+	t.Run("editor cannot create ResourcePermission (insufficient permissions)", func(t *testing.T) {
+		ctx := context.Background()
+
+		folder := createTestFolder(t, helper, helper.Org1.Admin, "test-folder-editor-deny", parentUID)
+		folderUID := folder.GetName()
+
+		permission := newPermission("User", helper.Org1.Editor.Identity.GetIdentifier(), "admin")
+		toCreate := createResourcePermissionObject(folderUID, gvrFolders.Group, gvrFolders.Resource, permission)
+		_, err := clients.rpEditor.Resource.Create(ctx, toCreate, metav1.CreateOptions{})
+		require.Error(t, err)
+		var statusErr *errors.StatusError
+		require.ErrorAs(t, err, &statusErr)
+		require.Equal(t, int32(403), statusErr.ErrStatus.Code)
+	})
+
+	t.Run("viewer cannot create ResourcePermission (insufficient permissions)", func(t *testing.T) {
+		ctx := context.Background()
+
+		folder := createTestFolder(t, helper, helper.Org1.Admin, "test-folder-viewer-deny", parentUID)
+		folderUID := folder.GetName()
+
+		permission := newPermission("User", helper.Org1.Viewer.Identity.GetIdentifier(), "admin")
+		toCreate := createResourcePermissionObject(folderUID, gvrFolders.Group, gvrFolders.Resource, permission)
+		_, err := clients.rpViewer.Resource.Create(ctx, toCreate, metav1.CreateOptions{})
+		require.Error(t, err)
+		var statusErr *errors.StatusError
+		require.ErrorAs(t, err, &statusErr)
+		require.Equal(t, int32(403), statusErr.ErrStatus.Code)
+	})
+
+	t.Run("viewer can update ResourcePermission of folder they can admin", func(t *testing.T) {
+		ctx := context.Background()
+
+		folder := createTestFolder(t, helper, helper.Org1.Admin, "test-folder-viewer-admin", parentUID)
+		folderUID := folder.GetName()
+
+		// Grant admin permissions to the viewer
+		permission := newPermission("User", helper.Org1.Viewer.Identity.GetIdentifier(), "admin")
+		toCreate := createResourcePermissionObject(folderUID, gvrFolders.Group, gvrFolders.Resource, permission)
+		_, err := clients.rpAdmin.Resource.Create(ctx, toCreate, metav1.CreateOptions{})
+		require.NoError(t, err)
+
+		// As a Viewer we should now be able to get and update the ResourcePermission of the folder
+		fetched, err := clients.rpViewer.Resource.Get(ctx, "folder.grafana.app-folders-"+folderUID, metav1.GetOptions{})
+		require.NoError(t, err)
+		require.NotNil(t, fetched)
+
+		// Update the ResourcePermission to grant editor permissions
+		fetched.Object["spec"].(map[string]interface{})["permissions"] = newPermissionMaps(
+			newPermission("BasicRole", "Editor", "edit"),
+			newPermission("User", helper.Org1.Viewer.Identity.GetIdentifier(), "admin"),
+		)
+
+		_, err = clients.rpViewer.Resource.Update(ctx, fetched, metav1.UpdateOptions{})
+		require.NoError(t, err)
+	})
+}
+
+func doResourcePermissionHierarchyTests(t *testing.T, helper *apis.K8sTestHelper, clients *k8sTestClients, parentUID string) {
+	permission := newPermission("BasicRole", "Editor", "admin")
+
+	t.Run("should respect folder hierarchy for folder permissions", func(t *testing.T) {
+		ctx := context.Background()
+
+		sub1 := createTestFolder(t, helper, helper.Org1.Admin, "sub1-folder-hierarchy", parentUID)
+		sub1UID := sub1.GetName()
+
+		sub2 := createTestFolder(t, helper, helper.Org1.Admin, "sub2-folder-hierarchy", sub1UID)
+		sub2UID := sub2.GetName()
+
+		toCreate := createResourcePermissionObject(sub2UID, gvrFolders.Group, gvrFolders.Resource, permission)
+		_, err := clients.rpAdmin.Resource.Create(ctx, toCreate, metav1.CreateOptions{})
+		require.NoError(t, err)
+
+		t.Run("editor can update ResourcePermission of sub2", func(t *testing.T) {
+			fetched, err := clients.rpEditor.Resource.Get(ctx, "folder.grafana.app-folders-"+sub2UID, metav1.GetOptions{})
+			require.NoError(t, err)
+
+			fetched.Object["spec"].(map[string]interface{})["permissions"] = newPermissionMaps(
+				newPermission("BasicRole", "Editor", "admin"),
+				newPermission("User", helper.Org1.Viewer.Identity.GetIdentifier(), "edit"),
+			)
+			_, err = clients.rpEditor.Resource.Update(ctx, fetched, metav1.UpdateOptions{})
+			require.NoError(t, err)
+		})
+		t.Run("editor cannot create ResourcePermission of sub1", func(t *testing.T) {
+			toCreate := createResourcePermissionObject(sub1UID, gvrFolders.Group, gvrFolders.Resource, permission)
+			_, err := clients.rpEditor.Resource.Create(ctx, toCreate, metav1.CreateOptions{})
+			require.Error(t, err)
+			var statusErr *errors.StatusError
+			require.ErrorAs(t, err, &statusErr)
+			require.Equal(t, int32(403), statusErr.ErrStatus.Code)
+		})
+
+		// Delete the ResourcePermission of sub2
+		err = clients.rpAdmin.Resource.Delete(ctx, "folder.grafana.app-folders-"+sub2UID, metav1.DeleteOptions{})
+		require.NoError(t, err)
+
+		// Create a new ResourcePermission for sub2
+		toCreate = createResourcePermissionObject(sub1UID, gvrFolders.Group, gvrFolders.Resource, permission)
+		_, err = clients.rpAdmin.Resource.Create(ctx, toCreate, metav1.CreateOptions{})
+		require.NoError(t, err)
+
+		t.Run("editor can create ResourcePermission of sub2 with parent folder permission", func(t *testing.T) {
+			toCreate := createResourcePermissionObject(sub2UID, gvrFolders.Group, gvrFolders.Resource, permission)
+
+			_, err = clients.rpEditor.Resource.Create(ctx, toCreate, metav1.CreateOptions{})
+			require.NoError(t, err)
+
+			fetched, err := clients.rpEditor.Resource.Get(ctx, "folder.grafana.app-folders-"+sub2UID, metav1.GetOptions{})
+			require.NoError(t, err)
+			require.NotNil(t, fetched)
+
+			permissions := fetched.Object["spec"].(map[string]interface{})["permissions"]
+			require.Len(t, permissions, 1)
+		})
+	})
+
+	t.Run("should respect folder hierarchy for dashboard permissions", func(t *testing.T) {
+		ctx := context.Background()
+
+		// Create folder and a nested dashboard
+		folder := createTestFolder(t, helper, helper.Org1.Admin, "sub1-dashboard-hierarchy", parentUID)
+		folderUID := folder.GetName()
+
+		dashboard := createTestDashboard(t, helper, helper.Org1.Admin, "sub2-dashboard", folderUID)
+		dashboardUID := dashboard.GetName()
+
+		// Verify dashboard has parent folder annotation
+		annotations := dashboard.GetAnnotations()
+		require.Equal(t, folderUID, annotations[utils.AnnoKeyFolder])
+
+		t.Run("editor cannot create ResourcePermission of dashboard without parent folder permission", func(t *testing.T) {
+			toCreate := createResourcePermissionObject(dashboardUID, gvrDashboards.Group, gvrDashboards.Resource, permission)
+			_, err := clients.rpEditor.Resource.Create(ctx, toCreate, metav1.CreateOptions{})
+			require.Error(t, err)
+			var statusErr *errors.StatusError
+			require.ErrorAs(t, err, &statusErr)
+			require.Equal(t, int32(403), statusErr.ErrStatus.Code)
+		})
+
+		// Admin creates ResourcePermission for parent folder
+		toCreate := createResourcePermissionObject(folderUID, gvrFolders.Group, gvrFolders.Resource, permission)
+		created, err := clients.rpAdmin.Resource.Create(ctx, toCreate, metav1.CreateOptions{})
+		require.NoError(t, err)
+		require.NotNil(t, created)
+
+		t.Run("editor can create ResourcePermission of dashboard with parent folder permission", func(t *testing.T) {
+			toCreate := createResourcePermissionObject(dashboardUID, gvrDashboards.Group, gvrDashboards.Resource, permission)
+			_, err := clients.rpEditor.Resource.Create(ctx, toCreate, metav1.CreateOptions{})
+			require.NoError(t, err)
+
+			fetched, err := clients.rpEditor.Resource.Get(ctx, "dashboard.grafana.app-dashboards-"+dashboardUID, metav1.GetOptions{})
+			require.NoError(t, err)
+			require.NotNil(t, fetched)
+
+			permissions := fetched.Object["spec"].(map[string]interface{})["permissions"]
+			require.Len(t, permissions, 1)
+		})
+	})
+}
+
+func doResourcePermissionListFilteringTests(t *testing.T, helper *apis.K8sTestHelper, clients *k8sTestClients, parentUID string) {
+	viewerCanAdmin := newPermission("BasicRole", "Viewer", "admin")
+	viewerCanView := newPermission("BasicRole", "Viewer", "view")
+	editorCanAdmin := newPermission("BasicRole", "Editor", "admin")
+
+	ctx := context.Background()
+
+	// Create two folders
+	editorFolder := createTestFolder(t, helper, helper.Org1.Admin, "editor-only-folder", parentUID)
+	editorFolderUID := editorFolder.GetName()
+
+	viewerFolder := createTestFolder(t, helper, helper.Org1.Admin, "viewer-only-folder", parentUID)
+	viewerFolderUID := viewerFolder.GetName()
+
+	dashboardViewerCanAdmin := createTestDashboard(t, helper, helper.Org1.Admin, "dashboard-in-editor-folder-viewer-can-admin", editorFolderUID)
+	dashboardViewerCanAdminUID := dashboardViewerCanAdmin.GetName()
+	dashboardViewerCanView := createTestDashboard(t, helper, helper.Org1.Admin, "dashboard-in-editor-folder-viewer-can-view", editorFolderUID)
+	dashboardViewerCanViewUID := dashboardViewerCanView.GetName()
+
+	// Grant admin permissions to the viewer on folder2 and dashboard1
+	rp1 := createResourcePermissionObject(viewerFolderUID, gvrFolders.Group, gvrFolders.Resource, viewerCanAdmin)
+	_, err := clients.rpAdmin.Resource.Create(ctx, rp1, metav1.CreateOptions{})
+	require.NoError(t, err)
+	rp2 := createResourcePermissionObject(dashboardViewerCanAdminUID, gvrDashboards.Group, gvrDashboards.Resource, viewerCanAdmin)
+	_, err = clients.rpAdmin.Resource.Create(ctx, rp2, metav1.CreateOptions{})
+	require.NoError(t, err)
+	// Grant admin permissions to the editor on folder1
+	rp3 := createResourcePermissionObject(editorFolderUID, gvrFolders.Group, gvrFolders.Resource, editorCanAdmin)
+	_, err = clients.rpAdmin.Resource.Create(ctx, rp3, metav1.CreateOptions{})
+	require.NoError(t, err)
+	// Grant view permissions to the viewer on dashboard2
+	rp4 := createResourcePermissionObject(dashboardViewerCanViewUID, gvrDashboards.Group, gvrDashboards.Resource, viewerCanView)
+	_, err = clients.rpAdmin.Resource.Create(ctx, rp4, metav1.CreateOptions{})
+	require.NoError(t, err)
+
+	t.Run("Admin can list all ResourcePermissions", func(t *testing.T) {
+		// Admin can list all ResourcePermissions
+		list, err := clients.rpAdmin.Resource.List(ctx, metav1.ListOptions{})
+		require.NoError(t, err)
+		require.NotNil(t, list)
+
+		// Check that all expected items are present (there may be more from other tests)
+		itemNames := getNamesFromList(list)
+		require.Contains(t, itemNames, rp1.GetName(), "Admin should see viewer folder permission")
+		require.Contains(t, itemNames, rp2.GetName(), "Admin should see dashboard viewer can admin permission")
+		require.Contains(t, itemNames, rp3.GetName(), "Admin should see editor folder permission")
+		require.Contains(t, itemNames, rp4.GetName(), "Admin should see dashboard viewer can view permission")
+	})
+
+	t.Run("Viewer can list ResourcePermissions of folder2 and dashboard1", func(t *testing.T) {
+		list, err := clients.rpViewer.Resource.List(ctx, metav1.ListOptions{})
+		require.NoError(t, err)
+		require.NotNil(t, list)
+
+		itemNames := getNamesFromList(list)
+		// Viewer should see permissions for resources they can admin
+		require.Contains(t, itemNames, rp1.GetName(), "Viewer should see viewer folder permission")
+		require.Contains(t, itemNames, rp2.GetName(), "Viewer should see dashboard viewer can admin permission")
+
+		// Viewer should NOT see permissions for resources they cannot admin
+		require.NotContains(t, itemNames, rp3.GetName(), "Viewer should NOT see editor folder permission")
+		require.NotContains(t, itemNames, rp4.GetName(), "Viewer should NOT see dashboard viewer can view permission")
+	})
+	t.Run("Editor can list ResourcePermissions of folder1 and its nested dashboards", func(t *testing.T) {
+		list, err := clients.rpEditor.Resource.List(ctx, metav1.ListOptions{})
+		require.NoError(t, err)
+		require.NotNil(t, list)
+
+		itemNames := getNamesFromList(list)
+		// Editor has admin on editorFolder, so they should see:
+		// - rp3 (editorFolder permission)
+		// - rp2 (dashboard in editorFolder with admin permission)
+		// - rp4 (dashboard in editorFolder with view permission)
+		require.Contains(t, itemNames, rp2.GetName(), "Editor should see dashboard admin permission in their folder")
+		require.Contains(t, itemNames, rp3.GetName(), "Editor should see their folder permission")
+		require.Contains(t, itemNames, rp4.GetName(), "Editor should see dashboard view permission in their folder")
+
+		// Editor should NOT see permissions for viewerFolder
+		require.NotContains(t, itemNames, rp1.GetName(), "Editor should NOT see viewer-only folder permission")
+	})
+}
+
+// Helper functions
+
+func createTestFolder(t *testing.T, helper *apis.K8sTestHelper, user apis.User, title string, parentUID string) *unstructured.Unstructured {
+	t.Helper()
+	ctx := context.Background()
+
+	folderClient := helper.GetResourceClient(apis.ResourceClientArgs{
+		User:      user,
+		Namespace: helper.Namespacer(user.Identity.GetOrgID()),
+		GVR:       gvrFolders,
+	})
+	metadata := map[string]interface{}{
+		"generateName": "test-folder-",
+		"namespace":    helper.Namespacer(user.Identity.GetOrgID()),
+	}
+
+	if parentUID != "" {
+		metadata["annotations"] = map[string]interface{}{
+			utils.AnnoKeyFolder: parentUID,
+		}
+	}
+
+	folder := &unstructured.Unstructured{
+		Object: map[string]interface{}{
+			"apiVersion": "folder.grafana.app/v1beta1",
+			"kind":       "Folder",
+			"metadata":   metadata,
+			"spec": map[string]interface{}{
+				"title": title,
+			},
+		},
+	}
+
+	created, err := folderClient.Resource.Create(ctx, folder, metav1.CreateOptions{})
+	require.NoError(t, err)
+	require.NotNil(t, created)
+	return created
+}
+
+// Helper function to delete default permissions
+func deleteDefaultPermissions(t *testing.T, client *apis.K8sResourceClient, resourceName string) {
+	ctx := context.Background()
+	// Delete the resource permission
+	err := client.Resource.Delete(ctx, resourceName, metav1.DeleteOptions{})
+	require.NoError(t, err)
+	// Check if the resource permission is deleted
+	_, err = client.Resource.Get(ctx, resourceName, metav1.GetOptions{})
+	require.Error(t, err)
+	var statusErr *errors.StatusError
+	require.ErrorAs(t, err, &statusErr)
+	require.Equal(t, int32(404), statusErr.ErrStatus.Code)
+}
+
+// Helper function to create a root folder without default permissions
+func createRootFolderWithoutDefaultPermissions(t *testing.T, helper *apis.K8sTestHelper) *unstructured.Unstructured {
+	t.Helper()
+
+	// Create folder as admin
+	folder := createTestFolder(t, helper, helper.Org1.Admin, "root-without-permissions", "")
+	folderUID := folder.GetName()
+
+	// Delete default permissions
+	rpClient := helper.GetResourceClient(apis.ResourceClientArgs{
+		User:      helper.Org1.Admin,
+		Namespace: helper.Namespacer(helper.Org1.OrgID),
+		GVR:       gvrResourcePermissions,
+	})
+
+	// It would be better to create the folder without default permissions, but this is a workaround for the time being
+	deleteDefaultPermissions(t, rpClient, "folder.grafana.app-folders-"+folderUID)
+
+	return folder
+}
+
+func createTestDashboard(t *testing.T, helper *apis.K8sTestHelper, user apis.User, title, folderUID string) *unstructured.Unstructured {
+	t.Helper()
+	ctx := context.Background()
+
+	dashboardClient := helper.GetResourceClient(apis.ResourceClientArgs{
+		User:      user,
+		Namespace: helper.Namespacer(user.Identity.GetOrgID()),
+		GVR:       gvrDashboards,
+	})
+
+	dashboard := &unstructured.Unstructured{
+		Object: map[string]interface{}{
+			"apiVersion": "dashboard.grafana.app/v1beta1",
+			"kind":       "Dashboard",
+			"metadata": map[string]interface{}{
+				"generateName": "test-dashboard-",
+				"namespace":    helper.Namespacer(user.Identity.GetOrgID()),
+				"annotations": map[string]interface{}{
+					utils.AnnoKeyFolder: folderUID,
+				},
+			},
+			"spec": map[string]interface{}{
+				"title": title,
+			},
+		},
+	}
+
+	created, err := dashboardClient.Resource.Create(ctx, dashboard, metav1.CreateOptions{})
+	require.NoError(t, err)
+	require.NotNil(t, created)
+	return created
+}
+
+func createResourcePermissionObject(resourceName, apiGroup, resource string, permissions ...permission) *unstructured.Unstructured {
+	permissionMaps := newPermissionMaps(permissions...)
+	return &unstructured.Unstructured{
+		Object: map[string]interface{}{
+			"apiVersion": iamv0.GROUP + "/" + iamv0.VERSION,
+			"kind":       "ResourcePermission",
+			"metadata": map[string]interface{}{
+				"name": apiGroup + "-" + resource + "-" + resourceName,
+			},
+			"spec": map[string]interface{}{
+				"resource": map[string]interface{}{
+					"apiGroup": apiGroup,
+					"resource": resource,
+					"name":     resourceName,
+				},
+				"permissions": permissionMaps,
+			},
+		},
+	}
+}
+
+func getNamesFromList(list *unstructured.UnstructuredList) []string {
+	names := make([]string, len(list.Items))
+	for i, item := range list.Items {
+		names[i] = item.GetName()
+	}
+	return names
+}