Search v1: Add support for inherited folder permissions if nested folders are enabled (#63275)

* Add features dependency to SQLBuilder

* Add features dependency to AccessControlDashboardPermissionFilter

* Add test for folder inheritance

* Dashboard permissions: Return recursive query

* Recursive query for inherited folders

* Modify search builder

* Adjust db.SQLBuilder

* Pass flag to SQLbuilder if CTEs are supported

* Add support for mysql < 8.0

* Add benchmarking for search with nested folders

* Set features to AlertStore

* Update pkg/infra/db/sqlbuilder.go

Co-authored-by: Ieva <ieva.vasiljeva@grafana.com>

* Set features to LibraryElementService

* SQLBuilder tests with nested folder flag set

* Apply suggestion from code review

Co-authored-by: IevaVasiljeva <ieva.vasiljeva@grafana.com>
Co-authored-by: Emil Tullstedt <emil.tullstedt@grafana.com>

pkg/services/sqlstore/permissions/dashboard.go

@@ -1,16 +1,23 @@
 package permissions
 
 import (
+	"bytes"
+	"fmt"
 	"strings"
 
 	"github.com/grafana/grafana/pkg/services/accesscontrol"
 	"github.com/grafana/grafana/pkg/services/dashboards"
+	"github.com/grafana/grafana/pkg/services/featuremgmt"
+	"github.com/grafana/grafana/pkg/services/folder"
 	"github.com/grafana/grafana/pkg/services/org"
 	"github.com/grafana/grafana/pkg/services/sqlstore/migrator"
 	"github.com/grafana/grafana/pkg/services/sqlstore/searchstore"
 	"github.com/grafana/grafana/pkg/services/user"
 )
 
+// maximum possible capacity for recursive queries array: one query for folder and one for dashboard actions
+const maximumRecursiveQueries = 2
+
 type DashboardPermissionFilter struct {
 	OrgRole         org.RoleType
 	Dialect         migrator.Dialect
@@ -78,14 +85,25 @@ func (d DashboardPermissionFilter) Where() (string, []interface{}) {
 	return sql, params
 }
 
-type AccessControlDashboardPermissionFilter struct {
+type clause struct {
+	string
+	params []interface{}
+}
+
+type accessControlDashboardPermissionFilter struct {
 	user             *user.SignedInUser
 	dashboardActions []string
 	folderActions    []string
+	features         featuremgmt.FeatureToggles
+
+	where clause
+	// any recursive CTE queries (if supported)
+	recQueries                   []clause
+	recursiveQueriesAreSupported bool
 }
 
 // NewAccessControlDashboardPermissionFilter creates a new AccessControlDashboardPermissionFilter that is configured with specific actions calculated based on the dashboards.PermissionType and query type
-func NewAccessControlDashboardPermissionFilter(user *user.SignedInUser, permissionLevel dashboards.PermissionType, queryType string) AccessControlDashboardPermissionFilter {
+func NewAccessControlDashboardPermissionFilter(user *user.SignedInUser, permissionLevel dashboards.PermissionType, queryType string, features featuremgmt.FeatureToggles, recursiveQueriesAreSupported bool) *accessControlDashboardPermissionFilter {
 	needEdit := permissionLevel > dashboards.PERMISSION_VIEW
 
 	var folderActions []string
@@ -121,12 +139,26 @@ func NewAccessControlDashboardPermissionFilter(user *user.SignedInUser, permissi
 		}
 	}
 
-	return AccessControlDashboardPermissionFilter{user: user, folderActions: folderActions, dashboardActions: dashboardActions}
+	f := accessControlDashboardPermissionFilter{user: user, folderActions: folderActions, dashboardActions: dashboardActions, features: features,
+		recursiveQueriesAreSupported: recursiveQueriesAreSupported,
+	}
+
+	f.buildClauses()
+
+	return &f
 }
 
-func (f AccessControlDashboardPermissionFilter) Where() (string, []interface{}) {
+// Where returns:
+// - a where clause for filtering dashboards with expected permissions
+// - an array with the query parameters
+func (f *accessControlDashboardPermissionFilter) Where() (string, []interface{}) {
+	return f.where.string, f.where.params
+}
+
+func (f *accessControlDashboardPermissionFilter) buildClauses() {
 	if f.user == nil || f.user.Permissions == nil || f.user.Permissions[f.user.OrgID] == nil {
-		return "(1 = 0)", nil
+		f.where = clause{string: "(1 = 0)"}
+		return
 	}
 	dashWildcards := accesscontrol.WildcardsFromPrefix(dashboards.ScopeDashboardsPrefix)
 	folderWildcards := accesscontrol.WildcardsFromPrefix(dashboards.ScopeFoldersPrefix)
@@ -136,6 +168,10 @@ func (f AccessControlDashboardPermissionFilter) Where() (string, []interface{})
 	var args []interface{}
 	builder := strings.Builder{}
 	builder.WriteRune('(')
+
+	permSelector := strings.Builder{}
+	var permSelectorArgs []interface{}
+
 	if len(f.dashboardActions) > 0 {
 		toCheck := actionsToCheck(f.dashboardActions, f.user.Permissions[f.user.OrgID], dashWildcards, folderWildcards)
 
@@ -155,24 +191,50 @@ func (f AccessControlDashboardPermissionFilter) Where() (string, []interface{})
 			builder.WriteString(") AND NOT dashboard.is_folder)")
 
 			builder.WriteString(" OR ")
-			builder.WriteString("(dashboard.folder_id IN (SELECT id FROM dashboard as d WHERE d.uid IN (SELECT substr(scope, 13) FROM permission WHERE scope LIKE 'folders:uid:%' ")
-			builder.WriteString(rolesFilter)
-			args = append(args, params...)
+			permSelector.WriteString("(SELECT substr(scope, 13) FROM permission WHERE scope LIKE 'folders:uid:%' ")
+			permSelector.WriteString(rolesFilter)
+			permSelectorArgs = append(permSelectorArgs, params...)
 
 			if len(toCheck) == 1 {
-				builder.WriteString(" AND action = ?")
-				args = append(args, toCheck[0])
+				permSelector.WriteString(" AND action = ?")
+				permSelectorArgs = append(permSelectorArgs, toCheck[0])
 			} else {
-				builder.WriteString(" AND action IN (?" + strings.Repeat(", ?", len(toCheck)-1) + ") GROUP BY role_id, scope HAVING COUNT(action) = ?")
-				args = append(args, toCheck...)
-				args = append(args, len(toCheck))
+				permSelector.WriteString(" AND action IN (?" + strings.Repeat(", ?", len(toCheck)-1) + ") GROUP BY role_id, scope HAVING COUNT(action) = ?")
+				permSelectorArgs = append(permSelectorArgs, toCheck...)
+				permSelectorArgs = append(permSelectorArgs, len(toCheck))
 			}
-			builder.WriteString(")) AND NOT dashboard.is_folder)")
+			permSelector.WriteRune(')')
+
+			switch f.features.IsEnabled(featuremgmt.FlagNestedFolders) {
+			case true:
+				switch f.recursiveQueriesAreSupported {
+				case true:
+					recQueryName := fmt.Sprintf("RecQry%d", len(f.recQueries))
+					f.addRecQry(recQueryName, permSelector.String(), permSelectorArgs)
+					builder.WriteString("(dashboard.folder_id IN (SELECT d.id FROM dashboard as d ")
+					builder.WriteString(fmt.Sprintf("WHERE d.uid IN (SELECT uid FROM %s)", recQueryName))
+				default:
+					nestedFoldersSelectors, nestedFoldersArgs := nestedFoldersSelectors(permSelector.String(), permSelectorArgs, "folder_id", "id")
+					builder.WriteRune('(')
+					builder.WriteString(nestedFoldersSelectors)
+					args = append(args, nestedFoldersArgs...)
+				}
+			default:
+				builder.WriteString("(dashboard.folder_id IN (SELECT d.id FROM dashboard as d ")
+				builder.WriteString("WHERE d.uid IN ")
+				builder.WriteString(permSelector.String())
+				args = append(args, permSelectorArgs...)
+			}
+			builder.WriteString(") AND NOT dashboard.is_folder)")
 		} else {
 			builder.WriteString("NOT dashboard.is_folder")
 		}
 	}
 
+	// recycle and reuse
+	permSelector.Reset()
+	permSelectorArgs = permSelectorArgs[:0]
+
 	if len(f.folderActions) > 0 {
 		if len(f.dashboardActions) > 0 {
 			builder.WriteString(" OR ")
@@ -180,24 +242,80 @@ func (f AccessControlDashboardPermissionFilter) Where() (string, []interface{})
 
 		toCheck := actionsToCheck(f.folderActions, f.user.Permissions[f.user.OrgID], folderWildcards)
 		if len(toCheck) > 0 {
-			builder.WriteString("(dashboard.uid IN (SELECT substr(scope, 13) FROM permission WHERE scope LIKE 'folders:uid:%'")
-			builder.WriteString(rolesFilter)
-			args = append(args, params...)
+			permSelector.WriteString("(SELECT substr(scope, 13) FROM permission WHERE scope LIKE 'folders:uid:%'")
+			permSelector.WriteString(rolesFilter)
+			permSelectorArgs = append(permSelectorArgs, params...)
 			if len(toCheck) == 1 {
-				builder.WriteString(" AND action = ?")
-				args = append(args, toCheck[0])
+				permSelector.WriteString(" AND action = ?")
+				permSelectorArgs = append(permSelectorArgs, toCheck[0])
 			} else {
-				builder.WriteString(" AND action IN (?" + strings.Repeat(", ?", len(toCheck)-1) + ") GROUP BY role_id, scope HAVING COUNT(action) = ?")
-				args = append(args, toCheck...)
-				args = append(args, len(toCheck))
+				permSelector.WriteString(" AND action IN (?" + strings.Repeat(", ?", len(toCheck)-1) + ") GROUP BY role_id, scope HAVING COUNT(action) = ?")
+				permSelectorArgs = append(permSelectorArgs, toCheck...)
+				permSelectorArgs = append(permSelectorArgs, len(toCheck))
 			}
-			builder.WriteString(") AND dashboard.is_folder)")
+			permSelector.WriteRune(')')
+
+			switch f.features.IsEnabled(featuremgmt.FlagNestedFolders) {
+			case true:
+				switch f.recursiveQueriesAreSupported {
+				case true:
+					recQueryName := fmt.Sprintf("RecQry%d", len(f.recQueries))
+					f.addRecQry(recQueryName, permSelector.String(), permSelectorArgs)
+					builder.WriteString("(dashboard.uid IN ")
+					builder.WriteString(fmt.Sprintf("(SELECT uid FROM %s)", recQueryName))
+				default:
+					nestedFoldersSelectors, nestedFoldersArgs := nestedFoldersSelectors(permSelector.String(), permSelectorArgs, "uid", "uid")
+					builder.WriteRune('(')
+					builder.WriteString(nestedFoldersSelectors)
+					builder.WriteRune(')')
+					args = append(args, nestedFoldersArgs...)
+				}
+			default:
+				builder.WriteString("(dashboard.uid IN ")
+				builder.WriteString(permSelector.String())
+				args = append(args, permSelectorArgs...)
+			}
+			builder.WriteString(" AND dashboard.is_folder)")
 		} else {
 			builder.WriteString("dashboard.is_folder")
 		}
 	}
 	builder.WriteRune(')')
-	return builder.String(), args
+
+	f.where = clause{string: builder.String(), params: args}
+}
+
+// With returns:
+// - a with clause for fetching folders with inherited permissions if nested folders are enabled or an empty string
+func (f *accessControlDashboardPermissionFilter) With() (string, []interface{}) {
+	var sb bytes.Buffer
+	var params []interface{}
+	if len(f.recQueries) > 0 {
+		sb.WriteString("WITH RECURSIVE ")
+		sb.WriteString(f.recQueries[0].string)
+		params = append(params, f.recQueries[0].params...)
+		for _, r := range f.recQueries[1:] {
+			sb.WriteRune(',')
+			sb.WriteString(r.string)
+			params = append(params, r.params...)
+		}
+	}
+	return sb.String(), params
+}
+
+func (f *accessControlDashboardPermissionFilter) addRecQry(queryName string, whereUIDSelect string, whereParams []interface{}) {
+	if f.recQueries == nil {
+		f.recQueries = make([]clause, 0, maximumRecursiveQueries)
+	}
+	c := make([]interface{}, len(whereParams))
+	copy(c, whereParams)
+	f.recQueries = append(f.recQueries, clause{
+		string: fmt.Sprintf(`%s AS (
+			SELECT uid, parent_uid, org_id FROM folder WHERE uid IN %s
+			UNION ALL SELECT f.uid, f.parent_uid, f.org_id FROM folder f INNER JOIN %s r ON f.parent_uid = r.uid and f.org_id = r.org_id
+		)`, queryName, whereUIDSelect, queryName),
+		params: c,
+	})
 }
 
 func actionsToCheck(actions []string, permissions map[string][]string, wildcards ...accesscontrol.Wildcards) []interface{} {
@@ -222,3 +340,28 @@ func actionsToCheck(actions []string, permissions map[string][]string, wildcards
 	}
 	return toCheck
 }
+
+func nestedFoldersSelectors(permSelector string, permSelectorArgs []interface{}, leftTableCol string, rightTableCol string) (string, []interface{}) {
+	wheres := make([]string, 0, folder.MaxNestedFolderDepth+1)
+	args := make([]interface{}, 0, len(permSelectorArgs)*(folder.MaxNestedFolderDepth+1))
+
+	joins := make([]string, 0, folder.MaxNestedFolderDepth+2)
+
+	tmpl := "INNER JOIN folder %s ON %s.%s = %s.uid AND %s.org_id = %s.org_id "
+
+	prev := "d"
+	onCol := "uid"
+	for i := 1; i <= folder.MaxNestedFolderDepth+2; i++ {
+		t := fmt.Sprintf("f%d", i)
+		s := fmt.Sprintf(tmpl, t, prev, onCol, t, prev, t)
+		joins = append(joins, s)
+
+		wheres = append(wheres, fmt.Sprintf("(dashboard.%s IN (SELECT d.%s FROM dashboard d %s WHERE %s.uid IN %s)", leftTableCol, rightTableCol, strings.Join(joins, " "), t, permSelector))
+		args = append(args, permSelectorArgs...)
+
+		prev = t
+		onCol = "parent_uid"
+	}
+
+	return strings.Join(wheres, ") OR "), args
+}

pkg/services/sqlstore/permissions/dashboard_test.go

@@ -9,14 +9,24 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/grafana/grafana/pkg/bus"
 	"github.com/grafana/grafana/pkg/components/simplejson"
 	"github.com/grafana/grafana/pkg/infra/db"
+	"github.com/grafana/grafana/pkg/infra/tracing"
 	"github.com/grafana/grafana/pkg/services/accesscontrol"
+	"github.com/grafana/grafana/pkg/services/accesscontrol/mock"
 	"github.com/grafana/grafana/pkg/services/dashboards"
+	"github.com/grafana/grafana/pkg/services/dashboards/database"
+	"github.com/grafana/grafana/pkg/services/featuremgmt"
+	"github.com/grafana/grafana/pkg/services/folder"
+	"github.com/grafana/grafana/pkg/services/folder/folderimpl"
+	"github.com/grafana/grafana/pkg/services/guardian"
 	"github.com/grafana/grafana/pkg/services/org"
+	"github.com/grafana/grafana/pkg/services/quota/quotatest"
 	"github.com/grafana/grafana/pkg/services/sqlstore"
 	"github.com/grafana/grafana/pkg/services/sqlstore/permissions"
 	"github.com/grafana/grafana/pkg/services/sqlstore/searchstore"
+	"github.com/grafana/grafana/pkg/services/tag/tagimpl"
 	"github.com/grafana/grafana/pkg/services/user"
 )
 
@@ -129,13 +139,18 @@ func TestIntegration_DashboardPermissionFilter(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.desc, func(t *testing.T) {
 			store := setupTest(t, 10, 100, tt.permissions)
+			recursiveQueriesAreSupported, err := store.RecursiveQueriesAreSupported()
+			require.NoError(t, err)
+
 			usr := &user.SignedInUser{OrgID: 1, OrgRole: org.RoleViewer, Permissions: map[int64]map[string][]string{1: accesscontrol.GroupScopesByAction(tt.permissions)}}
-			filter := permissions.NewAccessControlDashboardPermissionFilter(usr, tt.permission, tt.queryType)
+			filter := permissions.NewAccessControlDashboardPermissionFilter(usr, tt.permission, tt.queryType, featuremgmt.WithFeatures(), recursiveQueriesAreSupported)
 
 			var result int
-			err := store.WithDbSession(context.Background(), func(sess *sqlstore.DBSession) error {
+			err = store.WithDbSession(context.Background(), func(sess *sqlstore.DBSession) error {
 				q, params := filter.Where()
-				_, err := sess.SQL("SELECT COUNT(*) FROM dashboard WHERE "+q, params...).Get(&result)
+				recQry, recQryParams := filter.With()
+				params = append(recQryParams, params...)
+				_, err := sess.SQL(recQry+"\nSELECT COUNT(*) FROM dashboard WHERE "+q, params...).Get(&result)
 				return err
 			})
 			require.NoError(t, err)
@@ -145,7 +160,115 @@ func TestIntegration_DashboardPermissionFilter(t *testing.T) {
 	}
 }
 
+func TestIntegration_DashboardNestedPermissionFilter(t *testing.T) {
+	testCases := []struct {
+		desc           string
+		queryType      string
+		permission     dashboards.PermissionType
+		permissions    []accesscontrol.Permission
+		expectedResult []string
+		features       featuremgmt.FeatureToggles
+	}{
+		{
+			desc:       "Should be able to view dashboards under inherited folders if nested folders are enabled",
+			queryType:  searchstore.TypeDashboard,
+			permission: dashboards.PERMISSION_VIEW,
+			permissions: []accesscontrol.Permission{
+				{Action: dashboards.ActionDashboardsRead, Scope: "folders:uid:parent"},
+			},
+			features:       featuremgmt.WithFeatures(featuremgmt.FlagNestedFolders),
+			expectedResult: []string{"dashboard under parent folder", "dashboard under subfolder"},
+		},
+		{
+			desc:       "Should not be able to view dashboards under inherited folders if nested folders are not enabled",
+			queryType:  searchstore.TypeDashboard,
+			permission: dashboards.PERMISSION_VIEW,
+			permissions: []accesscontrol.Permission{
+				{Action: dashboards.ActionDashboardsRead, Scope: "folders:uid:parent"},
+			},
+			features:       featuremgmt.WithFeatures(),
+			expectedResult: []string{"dashboard under parent folder"},
+		},
+		{
+			desc:       "Should be able to view inherited folders if nested folders are enabled",
+			queryType:  searchstore.TypeFolder,
+			permission: dashboards.PERMISSION_VIEW,
+			permissions: []accesscontrol.Permission{
+				{Action: dashboards.ActionFoldersRead, Scope: "folders:uid:parent"},
+			},
+			features:       featuremgmt.WithFeatures(featuremgmt.FlagNestedFolders),
+			expectedResult: []string{"parent", "subfolder"},
+		},
+		{
+			desc:       "Should not be able to view inherited folders if nested folders are not enabled",
+			queryType:  searchstore.TypeFolder,
+			permission: dashboards.PERMISSION_VIEW,
+			permissions: []accesscontrol.Permission{
+				{Action: dashboards.ActionFoldersRead, Scope: "folders:uid:parent"},
+			},
+			features:       featuremgmt.WithFeatures(),
+			expectedResult: []string{"parent"},
+		},
+		{
+			desc:       "Should be able to view inherited dashboards and folders if nested folders are enabled",
+			permission: dashboards.PERMISSION_VIEW,
+			permissions: []accesscontrol.Permission{
+				{Action: dashboards.ActionFoldersRead, Scope: "folders:uid:parent"},
+				{Action: dashboards.ActionDashboardsRead, Scope: "folders:uid:parent"},
+			},
+			features:       featuremgmt.WithFeatures(featuremgmt.FlagNestedFolders),
+			expectedResult: []string{"parent", "subfolder", "dashboard under parent folder", "dashboard under subfolder"},
+		},
+		{
+			desc:       "Should not be able to view inherited dashboards and folders if nested folders are not enabled",
+			permission: dashboards.PERMISSION_VIEW,
+			permissions: []accesscontrol.Permission{
+				{Action: dashboards.ActionFoldersRead, Scope: "folders:uid:parent"},
+				{Action: dashboards.ActionDashboardsRead, Scope: "folders:uid:parent"},
+			},
+			features:       featuremgmt.WithFeatures(),
+			expectedResult: []string{"parent", "dashboard under parent folder"},
+		},
+	}
+
+	origNewGuardian := guardian.New
+	guardian.MockDashboardGuardian(&guardian.FakeDashboardGuardian{CanViewValue: true, CanSaveValue: true})
+	t.Cleanup(func() {
+		guardian.New = origNewGuardian
+	})
+
+	var orgID int64 = 1
+
+	for _, tc := range testCases {
+		t.Run(tc.desc, func(t *testing.T) {
+			tc.permissions = append(tc.permissions, accesscontrol.Permission{
+				Action: dashboards.ActionFoldersCreate,
+			}, accesscontrol.Permission{
+				Action: dashboards.ActionFoldersWrite,
+				Scope:  dashboards.ScopeFoldersAll,
+			})
+			usr := &user.SignedInUser{OrgID: orgID, OrgRole: org.RoleViewer, Permissions: map[int64]map[string][]string{orgID: accesscontrol.GroupScopesByAction(tc.permissions)}}
+			db := setupNestedTest(t, usr, tc.permissions, orgID, tc.features)
+			recursiveQueriesAreSupported, err := db.RecursiveQueriesAreSupported()
+			require.NoError(t, err)
+			filter := permissions.NewAccessControlDashboardPermissionFilter(usr, tc.permission, tc.queryType, tc.features, recursiveQueriesAreSupported)
+			var result []string
+			err = db.WithDbSession(context.Background(), func(sess *sqlstore.DBSession) error {
+				q, params := filter.Where()
+				recQry, recQryParams := filter.With()
+				params = append(recQryParams, params...)
+				err := sess.SQL(recQry+"\nSELECT title FROM dashboard WHERE "+q, params...).Find(&result)
+				return err
+			})
+			require.NoError(t, err)
+			assert.Equal(t, tc.expectedResult, result)
+		})
+	}
+}
+
 func setupTest(t *testing.T, numFolders, numDashboards int, permissions []accesscontrol.Permission) db.DB {
+	t.Helper()
+
 	store := db.InitTestDB(t)
 	err := store.WithDbSession(context.Background(), func(sess *sqlstore.DBSession) error {
 		dashes := make([]dashboards.Dashboard, 0, numFolders+numDashboards)
@@ -227,3 +350,95 @@ func setupTest(t *testing.T, numFolders, numDashboards int, permissions []access
 	require.NoError(t, err)
 	return store
 }
+
+func setupNestedTest(t *testing.T, usr *user.SignedInUser, perms []accesscontrol.Permission, orgID int64, features featuremgmt.FeatureToggles) db.DB {
+	t.Helper()
+
+	db := sqlstore.InitTestDB(t)
+
+	// dashboard store commands that should be called.
+	dashStore, err := database.ProvideDashboardStore(db, db.Cfg, features, tagimpl.ProvideService(db, db.Cfg), quotatest.New(false, nil))
+	require.NoError(t, err)
+
+	folderSvc := folderimpl.ProvideService(mock.New(), bus.ProvideBus(tracing.InitializeTracerForTest()), db.Cfg, dashStore, folderimpl.ProvideDashboardFolderStore(db), db, features)
+
+	// create parent folder
+	parent, err := folderSvc.Create(context.Background(), &folder.CreateFolderCommand{
+		UID:          "parent",
+		OrgID:        orgID,
+		Title:        "parent",
+		SignedInUser: usr,
+	})
+	require.NoError(t, err)
+
+	// create subfolder
+	subfolder, err := folderSvc.Create(context.Background(), &folder.CreateFolderCommand{
+		UID:          "subfolder",
+		ParentUID:    "parent",
+		OrgID:        orgID,
+		Title:        "subfolder",
+		SignedInUser: usr,
+	})
+	require.NoError(t, err)
+
+	// create dashboard under parent folder
+	_, err = dashStore.SaveDashboard(context.Background(), dashboards.SaveDashboardCommand{
+		OrgID:    orgID,
+		FolderID: parent.ID,
+		Dashboard: simplejson.NewFromAny(map[string]interface{}{
+			"title": "dashboard under parent folder",
+		}),
+	})
+	require.NoError(t, err)
+
+	// create dashboard under subfolder
+	_, err = dashStore.SaveDashboard(context.Background(), dashboards.SaveDashboardCommand{
+		OrgID:    orgID,
+		FolderID: subfolder.ID,
+		Dashboard: simplejson.NewFromAny(map[string]interface{}{
+			"title": "dashboard under subfolder",
+		}),
+	})
+	require.NoError(t, err)
+
+	err = db.WithDbSession(context.Background(), func(sess *sqlstore.DBSession) error {
+		role := &accesscontrol.Role{
+			OrgID:   0,
+			UID:     "basic_viewer",
+			Name:    "basic:viewer",
+			Updated: time.Now(),
+			Created: time.Now(),
+		}
+		_, err = sess.Insert(role)
+		if err != nil {
+			return err
+		}
+		_, err = sess.Insert(accesscontrol.BuiltinRole{
+			OrgID:   0,
+			RoleID:  role.ID,
+			Role:    "Viewer",
+			Created: time.Now(),
+			Updated: time.Now(),
+		})
+		if err != nil {
+			return err
+		}
+
+		for i := range perms {
+			perms[i].RoleID = role.ID
+			perms[i].Created = time.Now()
+			perms[i].Updated = time.Now()
+		}
+		if len(perms) > 0 {
+			_, err = sess.InsertMulti(&perms)
+			if err != nil {
+				return err
+			}
+		}
+
+		return nil
+	})
+	require.NoError(t, err)
+
+	return db
+}

pkg/services/sqlstore/permissions/dashboards_bench_test.go

@@ -10,26 +10,59 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/grafana/grafana/pkg/bus"
 	"github.com/grafana/grafana/pkg/components/simplejson"
 	"github.com/grafana/grafana/pkg/infra/db"
+	"github.com/grafana/grafana/pkg/infra/tracing"
 	"github.com/grafana/grafana/pkg/services/accesscontrol"
+	"github.com/grafana/grafana/pkg/services/accesscontrol/mock"
 	"github.com/grafana/grafana/pkg/services/dashboards"
+	"github.com/grafana/grafana/pkg/services/dashboards/database"
+	"github.com/grafana/grafana/pkg/services/featuremgmt"
+	"github.com/grafana/grafana/pkg/services/folder"
+	"github.com/grafana/grafana/pkg/services/folder/folderimpl"
+	"github.com/grafana/grafana/pkg/services/guardian"
 	"github.com/grafana/grafana/pkg/services/org"
+	"github.com/grafana/grafana/pkg/services/quota/quotatest"
 	"github.com/grafana/grafana/pkg/services/sqlstore"
 	"github.com/grafana/grafana/pkg/services/sqlstore/permissions"
+	"github.com/grafana/grafana/pkg/services/tag/tagimpl"
 	"github.com/grafana/grafana/pkg/services/user"
 )
 
-func benchmarkDashboardPermissionFilter(b *testing.B, numUsers, numDashboards int) {
-	store := setupBenchMark(b, numUsers, numDashboards)
+func benchmarkDashboardPermissionFilter(b *testing.B, numUsers, numDashboards, numFolders, nestingLevel int) {
+	usr := user.SignedInUser{UserID: 1, OrgID: 1, OrgRole: org.RoleViewer, Permissions: map[int64]map[string][]string{
+		1: accesscontrol.GroupScopesByAction([]accesscontrol.Permission{
+			{
+				Action: dashboards.ActionFoldersCreate,
+			},
+			{
+				Action: dashboards.ActionFoldersWrite,
+				Scope:  dashboards.ScopeFoldersAll,
+			},
+		}),
+	}}
+
+	features := featuremgmt.WithFeatures()
+	// if nestingLevel > 0 enable nested folders
+	if nestingLevel > 0 {
+		features = featuremgmt.WithFeatures(featuremgmt.FlagNestedFolders)
+	}
+
+	store := setupBenchMark(b, usr, features, numUsers, numDashboards, numFolders, nestingLevel)
+
+	recursiveQueriesAreSupported, err := store.RecursiveQueriesAreSupported()
+	require.NoError(b, err)
+
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
-		usr := &user.SignedInUser{UserID: 1, OrgID: 1, OrgRole: org.RoleViewer, Permissions: map[int64]map[string][]string{1: {}}}
-		filter := permissions.NewAccessControlDashboardPermissionFilter(usr, dashboards.PERMISSION_VIEW, "")
+		filter := permissions.NewAccessControlDashboardPermissionFilter(&usr, dashboards.PERMISSION_VIEW, "", features, recursiveQueriesAreSupported)
 		var result int
 		err := store.WithDbSession(context.Background(), func(sess *sqlstore.DBSession) error {
 			q, params := filter.Where()
-			_, err := sess.SQL("SELECT COUNT(*) FROM dashboard WHERE "+q, params...).Get(&result)
+			recQry, recQryParams := filter.With()
+			params = append(recQryParams, params...)
+			_, err := sess.SQL(recQry+"SELECT COUNT(*) FROM dashboard WHERE "+q, params...).Get(&result)
 			return err
 		})
 		require.NoError(b, err)
@@ -37,15 +70,78 @@ func benchmarkDashboardPermissionFilter(b *testing.B, numUsers, numDashboards in
 	}
 }
 
-func setupBenchMark(b *testing.B, numUsers, numDashboards int) db.DB {
+func setupBenchMark(b *testing.B, usr user.SignedInUser, features featuremgmt.FeatureToggles, numUsers, numDashboards, numFolders, nestingLevel int) db.DB {
+	if nestingLevel > folder.MaxNestedFolderDepth {
+		nestingLevel = folder.MaxNestedFolderDepth
+	}
+
 	store := db.InitTestDB(b)
-	now := time.Now()
-	err := store.WithDbSession(context.Background(), func(sess *sqlstore.DBSession) error {
-		dashes := make([]dashboards.Dashboard, 0, numDashboards)
-		for i := 1; i <= numDashboards; i++ {
+
+	quotaService := quotatest.New(false, nil)
+
+	dashboardWriteStore, err := database.ProvideDashboardStore(store, store.Cfg, features, tagimpl.ProvideService(store, store.Cfg), quotaService)
+	require.NoError(b, err)
+
+	folderSvc := folderimpl.ProvideService(mock.New(), bus.ProvideBus(tracing.InitializeTracerForTest()), store.Cfg, dashboardWriteStore, folderimpl.ProvideDashboardFolderStore(store), store, features)
+
+	origNewGuardian := guardian.New
+	guardian.MockDashboardGuardian(&guardian.FakeDashboardGuardian{CanViewValue: true, CanSaveValue: true})
+	b.Cleanup(func() {
+		guardian.New = origNewGuardian
+	})
+
+	rootFolders := make([]*folder.Folder, 0, numFolders)
+	dashes := make([]dashboards.Dashboard, 0, numDashboards)
+	parentUID := ""
+	for i := 0; i < numFolders; i++ {
+		uid := fmt.Sprintf("f%d", i)
+		f, err := folderSvc.Create(context.Background(), &folder.CreateFolderCommand{
+			UID:          uid,
+			OrgID:        usr.OrgID,
+			Title:        uid,
+			SignedInUser: &usr,
+			ParentUID:    parentUID,
+		})
+		require.NoError(b, err)
+		rootFolders = append(rootFolders, f)
+
+		parentUID := f.UID
+		var leaf *folder.Folder
+		for j := 1; j <= nestingLevel; j++ {
+			uid := fmt.Sprintf("f%d_%d", i, j)
+			sf, err := folderSvc.Create(context.Background(), &folder.CreateFolderCommand{
+				UID:          uid,
+				OrgID:        usr.OrgID,
+				Title:        uid,
+				SignedInUser: &usr,
+				ParentUID:    parentUID,
+			})
+			require.NoError(b, err)
+			parentUID = sf.UID
+			leaf = sf
+		}
+
+		str := fmt.Sprintf("dashboard under folder %s", leaf.Title)
+		now := time.Now()
+		dashes = append(dashes, dashboards.Dashboard{
+			OrgID:    usr.OrgID,
+			IsFolder: false,
+			UID:      str,
+			Slug:     str,
+			Title:    str,
+			Data:     simplejson.New(),
+			Created:  now,
+			Updated:  now,
+			FolderID: leaf.ID,
+		})
+	}
+
+	err = store.WithDbSession(context.Background(), func(sess *sqlstore.DBSession) error {
+		now := time.Now()
+		for i := len(dashes); i < numDashboards; i++ {
 			str := strconv.Itoa(i)
 			dashes = append(dashes, dashboards.Dashboard{
-				OrgID:    1,
+				OrgID:    usr.OrgID,
 				IsFolder: false,
 				UID:      str,
 				Slug:     str,
@@ -79,10 +175,24 @@ func setupBenchMark(b *testing.B, numUsers, numDashboards int) db.DB {
 				Created: now,
 			})
 			for _, dash := range dashes {
+				// add permission to read dashboards under the general
+				if dash.FolderID == 0 {
+					permissions = append(permissions, accesscontrol.Permission{
+						RoleID:  int64(i),
+						Action:  dashboards.ActionDashboardsRead,
+						Scope:   dashboards.ScopeDashboardsProvider.GetResourceScopeUID(dash.UID),
+						Updated: now,
+						Created: now,
+					})
+				}
+			}
+
+			for _, f := range rootFolders {
+				// add permission to read folders under specific folders
 				permissions = append(permissions, accesscontrol.Permission{
 					RoleID:  int64(i),
 					Action:  dashboards.ActionDashboardsRead,
-					Scope:   dashboards.ScopeDashboardsProvider.GetResourceScopeUID(dash.UID),
+					Scope:   dashboards.ScopeFoldersProvider.GetResourceScopeUID(f.UID),
 					Updated: now,
 					Created: now,
 				})
@@ -113,16 +223,52 @@ func setupBenchMark(b *testing.B, numUsers, numDashboards int) db.DB {
 	return store
 }
 
-func BenchmarkDashboardPermissionFilter_100_100(b *testing.B) {
-	benchmarkDashboardPermissionFilter(b, 100, 100)
+func BenchmarkDashboardPermissionFilter_100_100_0_0(b *testing.B) {
+	benchmarkDashboardPermissionFilter(b, 100, 100, 0, 0)
 }
 
-func BenchmarkDashboardPermissionFilter_100_1000(b *testing.B) {
-	benchmarkDashboardPermissionFilter(b, 100, 1000)
+func BenchmarkDashboardPermissionFilter_100_100_10_2(b *testing.B) {
+	benchmarkDashboardPermissionFilter(b, 100, 100, 10, 2)
 }
 
-func BenchmarkDashboardPermissionFilter_300_10000(b *testing.B) {
-	benchmarkDashboardPermissionFilter(b, 300, 10000)
+func BenchmarkDashboardPermissionFilter_100_100_10_4(b *testing.B) {
+	benchmarkDashboardPermissionFilter(b, 100, 100, 10, 4)
+}
+
+func BenchmarkDashboardPermissionFilter_100_100_10_8(b *testing.B) {
+	benchmarkDashboardPermissionFilter(b, 100, 100, 10, 8)
+}
+
+func BenchmarkDashboardPermissionFilter_100_1000_0_0(b *testing.B) {
+	benchmarkDashboardPermissionFilter(b, 100, 1000, 0, 0)
+}
+
+func BenchmarkDashboardPermissionFilter_100_1000_10_2(b *testing.B) {
+	benchmarkDashboardPermissionFilter(b, 100, 1000, 10, 2)
+}
+
+func BenchmarkDashboardPermissionFilter_100_1000_10_4(b *testing.B) {
+	benchmarkDashboardPermissionFilter(b, 100, 1000, 10, 4)
+}
+
+func BenchmarkDashboardPermissionFilter_100_1000_10_8(b *testing.B) {
+	benchmarkDashboardPermissionFilter(b, 100, 1000, 10, 8)
+}
+
+func BenchmarkDashboardPermissionFilter_300_10000_0_0(b *testing.B) {
+	benchmarkDashboardPermissionFilter(b, 300, 10000, 0, 0)
+}
+
+func BenchmarkDashboardPermissionFilter_300_10000_10_2(b *testing.B) {
+	benchmarkDashboardPermissionFilter(b, 300, 10000, 10, 2)
+}
+
+func BenchmarkDashboardPermissionFilter_300_10000_10_4(b *testing.B) {
+	benchmarkDashboardPermissionFilter(b, 300, 10000, 10, 4)
+}
+
+func BenchmarkDashboardPermissionFilter_300_10000_10_8(b *testing.B) {
+	benchmarkDashboardPermissionFilter(b, 300, 10000, 10, 8)
 }
 
 func batch(count, batchSize int, eachFn func(start, end int) error) error {

pkg/services/sqlstore/searchstore/builder.go

@@ -44,6 +44,9 @@ func (b *Builder) ToSQL(limit, page int64) (string, []interface{}) {
 }
 
 func (b *Builder) buildSelect() {
+	var recQuery string
+	var recQueryParams []interface{}
+
 	b.sql.WriteString(
 		`SELECT
 			dashboard.id,
@@ -61,9 +64,25 @@ func (b *Builder) buildSelect() {
 		if f, ok := f.(FilterSelect); ok {
 			b.sql.WriteString(fmt.Sprintf(", %s", f.Select()))
 		}
+
+		if f, ok := f.(FilterWith); ok {
+			recQuery, recQueryParams = f.With()
+		}
 	}
 
 	b.sql.WriteString(` FROM `)
+
+	if recQuery == "" {
+		return
+	}
+
+	// prepend recursive queries
+	var bf bytes.Buffer
+	bf.WriteString(recQuery)
+	bf.WriteString(b.sql.String())
+
+	b.sql = bf
+	b.params = append(recQueryParams, b.params...)
 }
 
 func (b *Builder) applyFilters() (ordering string) {

pkg/services/sqlstore/searchstore/filters.go

@@ -14,6 +14,12 @@ type FilterWhere interface {
 	Where() (string, []interface{})
 }
 
+// FilterWith returns any recursive CTE queries (if supported)
+// and their parameters
+type FilterWith interface {
+	With() (string, []interface{})
+}
+
 // FilterGroupBy should be used after performing an outer join on the
 // search result to ensure there is only one of each ID in the results.
 // The id column must be present in the result.

pkg/services/sqlstore/searchstore/search_test.go

@@ -10,7 +10,9 @@ import (
 
 	"github.com/grafana/grafana/pkg/components/simplejson"
 	"github.com/grafana/grafana/pkg/infra/db"
+	"github.com/grafana/grafana/pkg/services/accesscontrol"
 	"github.com/grafana/grafana/pkg/services/dashboards"
+	"github.com/grafana/grafana/pkg/services/featuremgmt"
 	"github.com/grafana/grafana/pkg/services/org"
 	"github.com/grafana/grafana/pkg/services/sqlstore/permissions"
 	"github.com/grafana/grafana/pkg/services/sqlstore/searchstore"
@@ -148,6 +150,145 @@ func TestBuilder_Permissions(t *testing.T) {
 	assert.Len(t, res, 0)
 }
 
+func TestBuilder_RBAC(t *testing.T) {
+	testsCases := []struct {
+		desc            string
+		userPermissions []accesscontrol.Permission
+		features        featuremgmt.FeatureToggles
+		expectedParams  []interface{}
+	}{
+		{
+			desc:     "no user permissions",
+			features: featuremgmt.WithFeatures(),
+			expectedParams: []interface{}{
+				int64(1),
+			},
+		},
+		{
+			desc: "user with view permission",
+			userPermissions: []accesscontrol.Permission{
+				{Action: dashboards.ActionDashboardsRead, Scope: "dashboards:uid:1"},
+			},
+			features: featuremgmt.WithFeatures(),
+			expectedParams: []interface{}{
+				int64(1),
+				int64(1),
+				int64(1),
+				0,
+				"Viewer",
+				int64(1),
+				0,
+				"dashboards:read",
+				"dashboards:write",
+				2,
+				int64(1),
+				int64(1),
+				0,
+				"Viewer",
+				int64(1),
+				0,
+				"dashboards:read",
+				"dashboards:write",
+				2,
+				int64(1),
+				int64(1),
+				0,
+				"Viewer",
+				int64(1),
+				0,
+				"folders:read",
+				"dashboards:create",
+				2,
+			},
+		},
+		{
+			desc: "user with view permission with nesting",
+			userPermissions: []accesscontrol.Permission{
+				{Action: dashboards.ActionDashboardsRead, Scope: "dashboards:uid:1"},
+			},
+			features: featuremgmt.WithFeatures(featuremgmt.FlagNestedFolders),
+			expectedParams: []interface{}{
+				int64(1),
+				int64(1),
+				0,
+				"Viewer",
+				int64(1),
+				0,
+				"dashboards:read",
+				"dashboards:write",
+				2,
+				int64(1),
+				int64(1),
+				0,
+				"Viewer",
+				int64(1),
+				0,
+				"folders:read",
+				"dashboards:create",
+				2,
+				int64(1),
+				int64(1),
+				int64(1),
+				0,
+				"Viewer",
+				int64(1),
+				0,
+				"dashboards:read",
+				"dashboards:write",
+				2,
+			},
+		},
+	}
+
+	user := &user.SignedInUser{
+		UserID:  1,
+		OrgID:   1,
+		OrgRole: org.RoleViewer,
+	}
+
+	store := setupTestEnvironment(t)
+	createDashboards(t, store, 0, 1, user.OrgID)
+
+	recursiveQueriesAreSupported, err := store.RecursiveQueriesAreSupported()
+	require.NoError(t, err)
+
+	for _, tc := range testsCases {
+		t.Run(tc.desc, func(t *testing.T) {
+			if len(tc.userPermissions) > 0 {
+				user.Permissions = map[int64]map[string][]string{1: accesscontrol.GroupScopesByAction(tc.userPermissions)}
+			}
+
+			level := dashboards.PERMISSION_EDIT
+
+			builder := &searchstore.Builder{
+				Filters: []interface{}{
+					searchstore.OrgFilter{OrgId: user.OrgID},
+					searchstore.TitleSorter{},
+					permissions.NewAccessControlDashboardPermissionFilter(
+						user,
+						level,
+						"",
+						tc.features,
+						recursiveQueriesAreSupported,
+					),
+				},
+				Dialect: store.GetDialect(),
+			}
+
+			res := []dashboards.DashboardSearchProjection{}
+			err := store.WithDbSession(context.Background(), func(sess *db.Session) error {
+				sql, params := builder.ToSQL(limit, page)
+				// TODO: replace with a proper test
+				assert.Equal(t, tc.expectedParams, params)
+				return sess.SQL(sql, params...).Find(&res)
+			})
+			require.NoError(t, err)
+
+			assert.Len(t, res, 0)
+		})
+	}
+}
+
 func setupTestEnvironment(t *testing.T) db.DB {
 	t.Helper()
 	store := db.InitTestDB(t)