Storage: add delete / deleteFolder / createFolder (#51887)

* delete / delete folder / create folder

* add backend tests

* implement force delete

* fix merge

* lint fix

* fix delete root folder

* fix folder name validation

* fix mysql path_hash issue

* Fix returning error

pkg/infra/filestorage/api.go

@@ -157,6 +157,16 @@ type ListOptions struct {
 	Filter       PathFilter
 }
 
+type DeleteFolderOptions struct {
+	// Force if set to true, the `deleteFolder` operation will delete the selected folder together with all the nested files & folders
+	Force bool
+
+	// AccessFilter must match all the nested files & folders in order for the `deleteFolder` operation to succeed
+	// The access check is not performed if `AccessFilter` is nil
+	AccessFilter PathFilter
+}
+
+//go:generate mockery --name FileStorage --structname MockFileStorage --inpackage --filename file_storage_mock.go
 type FileStorage interface {
 	Get(ctx context.Context, path string) (*File, error)
 	Delete(ctx context.Context, path string) error
@@ -166,7 +176,7 @@ type FileStorage interface {
 	List(ctx context.Context, folderPath string, paging *Paging, options *ListOptions) (*ListResponse, error)
 
 	CreateFolder(ctx context.Context, path string) error
-	DeleteFolder(ctx context.Context, path string) error
+	DeleteFolder(ctx context.Context, path string, options *DeleteFolderOptions) error
 
 	close() error
 }

pkg/infra/filestorage/cdk_blob_filestorage.go

@@ -9,10 +9,9 @@ import (
 
 	"github.com/grafana/grafana/pkg/infra/log"
 	"gocloud.dev/blob"
-	"gocloud.dev/gcerrors"
-
 	_ "gocloud.dev/blob/fileblob"
 	_ "gocloud.dev/blob/memblob"
+	"gocloud.dev/gcerrors"
 )
 
 const (
@@ -215,20 +214,62 @@ func (c cdkBlobStorage) CreateFolder(ctx context.Context, path string) error {
 	return nil
 }
 
-func (c cdkBlobStorage) DeleteFolder(ctx context.Context, folderPath string) error {
-	directoryMarkerPath := fmt.Sprintf("%s%s%s", folderPath, Delimiter, directoryMarker)
-	exists, err := c.bucket.Exists(ctx, strings.ToLower(directoryMarkerPath))
-
-	if err != nil {
-		return err
+func (c cdkBlobStorage) DeleteFolder(ctx context.Context, folderPath string, options *DeleteFolderOptions) error {
+	folderPrefix := strings.ToLower(c.convertFolderPathToPrefix(folderPath))
+	directoryMarkerPath := folderPrefix + directoryMarker
+	if !options.Force {
+		return c.bucket.Delete(ctx, directoryMarkerPath)
 	}
 
-	if !exists {
-		return nil
+	iterators := []*blob.ListIterator{c.bucket.List(&blob.ListOptions{
+		Prefix:    folderPrefix,
+		Delimiter: Delimiter,
+	})}
+
+	var pathsToDelete []string
+
+	for len(iterators) > 0 {
+		obj, err := iterators[0].Next(ctx)
+		if errors.Is(err, io.EOF) {
+			iterators = iterators[1:]
+			continue
+		}
+
+		if err != nil {
+			c.log.Error("force folder delete: failed to retrieve next object", "err", err)
+			return err
+		}
+
+		path := obj.Key
+		lowerPath := strings.ToLower(path)
+		if obj.IsDir {
+			iterators = append([]*blob.ListIterator{c.bucket.List(&blob.ListOptions{
+				Prefix:    lowerPath,
+				Delimiter: Delimiter,
+			})}, iterators...)
+			continue
+		}
+
+		pathsToDelete = append(pathsToDelete, lowerPath)
 	}
 
-	err = c.bucket.Delete(ctx, strings.ToLower(directoryMarkerPath))
-	return err
+	for _, path := range pathsToDelete {
+		if !options.AccessFilter.IsAllowed(path) {
+			c.log.Error("force folder delete: unauthorized access", "path", path)
+			return fmt.Errorf("force folder delete error, unauthorized access to %s", path)
+		}
+	}
+
+	var lastErr error
+	for _, path := range pathsToDelete {
+		if err := c.bucket.Delete(ctx, path); err != nil {
+			c.log.Error("force folder delete: failed while deleting a file", "err", err, "path", path)
+			lastErr = err
+			// keep going and delete remaining files
+		}
+	}
+
+	return lastErr
 }
 
 //nolint: gocyclo

pkg/infra/filestorage/db_filestorage.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/md5"
 	"encoding/hex"
+	"reflect"
 
 	// can ignore because we don't need a cryptographically secure hash function
 	// sha1 low chance of collisions and better performance than sha256
@@ -135,30 +136,23 @@ func (s dbFileStorage) Delete(ctx context.Context, filePath string) error {
 	if err != nil {
 		return err
 	}
-	err = s.db.WithDbSession(ctx, func(sess *sqlstore.DBSession) error {
-		table := &file{}
-		exists, innerErr := sess.Table("file").Where("path_hash = ?", pathHash).Get(table)
-		if innerErr != nil {
-			return innerErr
+	err = s.db.WithTransactionalDbSession(ctx, func(sess *sqlstore.DBSession) error {
+		deletedFilesCount, err := sess.Table("file").Where("path_hash = ?", pathHash).Delete(&file{})
+		if err != nil {
+			return err
 		}
 
-		if !exists {
-			return nil
+		deletedMetaCount, err := sess.Table("file_meta").Where("path_hash = ?", pathHash).Delete(&fileMeta{})
+		if err != nil {
+			if rollErr := sess.Rollback(); rollErr != nil {
+				return fmt.Errorf("failed to roll back transaction due to error: %s: %w", rollErr, err)
+			}
+
+			return err
 		}
 
-		number, innerErr := sess.Table("file").Where("path_hash = ?", pathHash).Delete(table)
-		if innerErr != nil {
-			return innerErr
-		}
-		s.log.Info("Deleted file", "path", filePath, "affectedRecords", number)
-
-		metaTable := &fileMeta{}
-		number, innerErr = sess.Table("file_meta").Where("path_hash = ?", pathHash).Delete(metaTable)
-		if innerErr != nil {
-			return innerErr
-		}
-		s.log.Info("Deleted metadata", "path", filePath, "affectedRecords", number)
-		return innerErr
+		s.log.Info("Deleted file", "path", filePath, "deletedMetaCount", deletedMetaCount, "deletedFilesCount", deletedFilesCount)
+		return err
 	})
 
 	return err
@@ -490,24 +484,87 @@ func (s dbFileStorage) CreateFolder(ctx context.Context, path string) error {
 	return err
 }
 
-func (s dbFileStorage) DeleteFolder(ctx context.Context, folderPath string) error {
-	err := s.db.WithDbSession(ctx, func(sess *sqlstore.DBSession) error {
-		existing := &file{}
-		internalFolderPathHash, err := createPathHash(folderPath + Delimiter)
-		if err != nil {
-			return err
-		}
-		exists, err := sess.Table("file").Where("path_hash = ?", internalFolderPathHash).Get(existing)
+func (s dbFileStorage) DeleteFolder(ctx context.Context, folderPath string, options *DeleteFolderOptions) error {
+	lowerFolderPath := strings.ToLower(folderPath)
+	if lowerFolderPath == "" || lowerFolderPath == Delimiter {
+		lowerFolderPath = Delimiter
+	} else if !strings.HasSuffix(lowerFolderPath, Delimiter) {
+		lowerFolderPath = lowerFolderPath + Delimiter
+	}
+
+	if !options.Force {
+		return s.Delete(ctx, lowerFolderPath)
+	}
+
+	err := s.db.WithTransactionalDbSession(ctx, func(sess *sqlstore.DBSession) error {
+		var rawHashes []interface{}
+
+		// xorm does not support `.Delete()` with `.Join()`, so we first have to retrieve all path_hashes and then use them to filter `file_meta` table
+		err := sess.Table("file").
+			Cols("path_hash").
+			Where("LOWER(path) LIKE ?", lowerFolderPath+"%").
+			Find(&rawHashes)
 		if err != nil {
 			return err
 		}
 
-		if !exists {
+		if len(rawHashes) == 0 {
+			s.log.Info("Force deleted folder", "path", lowerFolderPath, "deletedFilesCount", 0, "deletedMetaCount", 0)
 			return nil
 		}
 
-		_, err = sess.Table("file").Where("path_hash = ?", internalFolderPathHash).Delete(existing)
-		return err
+		accessFilter := options.AccessFilter.asSQLFilter()
+		accessibleFilesCount, err := sess.Table("file").
+			Cols("path_hash").
+			Where("LOWER(path) LIKE ?", lowerFolderPath+"%").
+			Where(accessFilter.Where, accessFilter.Args...).
+			Count(&file{})
+		if err != nil {
+			return err
+		}
+
+		if int64(len(rawHashes)) != accessibleFilesCount {
+			s.log.Error("force folder delete: unauthorized access", "path", lowerFolderPath, "expectedAccessibleFilesCount", int64(len(rawHashes)), "actualAccessibleFilesCount", accessibleFilesCount)
+			return fmt.Errorf("force folder delete: unauthorized access for path %s", lowerFolderPath)
+		}
+
+		var hashes []interface{}
+		for _, hash := range rawHashes {
+			if hashString, ok := hash.(string); ok {
+				hashes = append(hashes, hashString)
+
+				// MySQL returns the `path_hash` field as []uint8
+			} else if hashUint, ok := hash.([]uint8); ok {
+				hashes = append(hashes, string(hashUint))
+			} else {
+				return fmt.Errorf("invalid hash type: %s", reflect.TypeOf(hash))
+			}
+		}
+
+		deletedFilesCount, err := sess.
+			Table("file").
+			In("path_hash", hashes...).
+			Delete(&file{})
+
+		if err != nil {
+			return err
+		}
+
+		deletedMetaCount, err := sess.
+			Table("file_meta").
+			In("path_hash", hashes...).
+			Delete(&fileMeta{})
+
+		if err != nil {
+			if rollErr := sess.Rollback(); rollErr != nil {
+				return fmt.Errorf("failed to roll back transaction due to error: %s: %w", rollErr, err)
+			}
+
+			return err
+		}
+
+		s.log.Info("Force deleted folder", "path", folderPath, "deletedFilesCount", deletedFilesCount, "deletedMetaCount", deletedMetaCount)
+		return nil
 	})
 
 	return err

pkg/infra/filestorage/file_storage_mock.go

@@ -0,0 +1,130 @@
+// Code generated by mockery v2.10.6. DO NOT EDIT.
+
+package filestorage
+
+import (
+	context "context"
+
+	mock "github.com/stretchr/testify/mock"
+)
+
+// MockFileStorage is an autogenerated mock type for the FileStorage type
+type MockFileStorage struct {
+	mock.Mock
+}
+
+// CreateFolder provides a mock function with given fields: ctx, path
+func (_m *MockFileStorage) CreateFolder(ctx context.Context, path string) error {
+	ret := _m.Called(ctx, path)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, string) error); ok {
+		r0 = rf(ctx, path)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// Delete provides a mock function with given fields: ctx, path
+func (_m *MockFileStorage) Delete(ctx context.Context, path string) error {
+	ret := _m.Called(ctx, path)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, string) error); ok {
+		r0 = rf(ctx, path)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// DeleteFolder provides a mock function with given fields: ctx, path, options
+func (_m *MockFileStorage) DeleteFolder(ctx context.Context, path string, options *DeleteFolderOptions) error {
+	ret := _m.Called(ctx, path, options)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, string, *DeleteFolderOptions) error); ok {
+		r0 = rf(ctx, path, options)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// Get provides a mock function with given fields: ctx, path
+func (_m *MockFileStorage) Get(ctx context.Context, path string) (*File, error) {
+	ret := _m.Called(ctx, path)
+
+	var r0 *File
+	if rf, ok := ret.Get(0).(func(context.Context, string) *File); ok {
+		r0 = rf(ctx, path)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*File)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, string) error); ok {
+		r1 = rf(ctx, path)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// List provides a mock function with given fields: ctx, folderPath, paging, options
+func (_m *MockFileStorage) List(ctx context.Context, folderPath string, paging *Paging, options *ListOptions) (*ListResponse, error) {
+	ret := _m.Called(ctx, folderPath, paging, options)
+
+	var r0 *ListResponse
+	if rf, ok := ret.Get(0).(func(context.Context, string, *Paging, *ListOptions) *ListResponse); ok {
+		r0 = rf(ctx, folderPath, paging, options)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*ListResponse)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, string, *Paging, *ListOptions) error); ok {
+		r1 = rf(ctx, folderPath, paging, options)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Upsert provides a mock function with given fields: ctx, command
+func (_m *MockFileStorage) Upsert(ctx context.Context, command *UpsertFileCommand) error {
+	ret := _m.Called(ctx, command)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, *UpsertFileCommand) error); ok {
+		r0 = rf(ctx, command)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// close provides a mock function with given fields:
+func (_m *MockFileStorage) close() error {
+	ret := _m.Called()
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func() error); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}

pkg/infra/filestorage/fs_integration_test.go

@@ -1183,6 +1183,123 @@ func TestIntegrationFsStorage(t *testing.T) {
 					},
 				},
 			},
+			{
+				name: "should be able to delete folders with files if using force",
+				steps: []interface{}{
+					cmdCreateFolder{
+						path: "/folder/dashboards/myNewFolder",
+					},
+					cmdUpsert{
+						cmd: UpsertFileCommand{
+							Path:     "/folder/dashboards/myNewFolder/file.jpg",
+							Contents: emptyContents,
+						},
+					},
+					cmdDeleteFolder{
+						path: "/folder/dashboards/myNewFolder",
+						options: &DeleteFolderOptions{
+							Force: true,
+						},
+					},
+					queryListFolders{
+						input: queryListFoldersInput{path: "/", options: &ListOptions{Recursive: true}},
+						checks: [][]interface{}{
+							checks(fPath("/folder")),
+							checks(fPath("/folder/dashboards")),
+						},
+					},
+					queryGet{
+						input: queryGetInput{
+							path: "/folder/dashboards/myNewFolder/file.jpg",
+						},
+					},
+				},
+			},
+			{
+				name: "should be able to delete root folder with force",
+				steps: []interface{}{
+					cmdCreateFolder{
+						path: "/folder/dashboards/myNewFolder",
+					},
+					cmdUpsert{
+						cmd: UpsertFileCommand{
+							Path:     "/folder/dashboards/myNewFolder/file.jpg",
+							Contents: emptyContents,
+						},
+					},
+					cmdDeleteFolder{
+						path: "/",
+						options: &DeleteFolderOptions{
+							Force: true,
+						},
+					},
+					queryListFolders{
+						input:  queryListFoldersInput{path: "/", options: &ListOptions{Recursive: true}},
+						checks: [][]interface{}{},
+					},
+					queryGet{
+						input: queryGetInput{
+							path: "/folder/dashboards/myNewFolder/file.jpg",
+						},
+					},
+				},
+			},
+			{
+				name: "should not be able to delete a folder unless have access to all nested files",
+				steps: []interface{}{
+					cmdCreateFolder{
+						path: "/folder/dashboards/myNewFolder",
+					},
+					cmdUpsert{
+						cmd: UpsertFileCommand{
+							Path:     "/folder/dashboards/myNewFolder/file.jpg",
+							Contents: emptyContents,
+						},
+					},
+					cmdUpsert{
+						cmd: UpsertFileCommand{
+							Path:     "/folder/dashboards/abc/file.jpg",
+							Contents: emptyContents,
+						},
+					},
+					cmdDeleteFolder{
+						path: "/",
+						options: &DeleteFolderOptions{
+							Force:        true,
+							AccessFilter: NewPathFilter([]string{"/"}, nil, nil, []string{"/folder/dashboards/abc/file.jpg"}),
+						},
+						error: &cmdErrorOutput{
+							message: "force folder delete: unauthorized access for path %s",
+							args:    []interface{}{"/"},
+						},
+					},
+					queryListFolders{
+						input: queryListFoldersInput{path: "/", options: &ListOptions{Recursive: true}},
+						checks: [][]interface{}{
+							checks(fPath("/folder")),
+							checks(fPath("/folder/dashboards")),
+							checks(fPath("/folder/dashboards/abc")),
+							checks(fPath("/folder/dashboards/myNewFolder")),
+						},
+					},
+					queryGet{
+						input: queryGetInput{
+							path: "/folder/dashboards/myNewFolder/file.jpg",
+						},
+						checks: checks(
+							fName("file.jpg"),
+						),
+					},
+					queryGet{
+						input: queryGetInput{
+							path: "/folder/dashboards/myNewFolder/file.jpg",
+						},
+						checks: checks(
+							fName("file.jpg"),
+						),
+					},
+				},
+			},
 		}
 	}
 

pkg/infra/filestorage/test_utils.go

@@ -32,8 +32,9 @@ type cmdCreateFolder struct {
 }
 
 type cmdDeleteFolder struct {
-	path  string
-	error *cmdErrorOutput
+	path    string
+	error   *cmdErrorOutput
+	options *DeleteFolderOptions
 }
 
 type queryGetInput struct {
@@ -175,7 +176,7 @@ func handleCommand(t *testing.T, ctx context.Context, cmd interface{}, cmdName s
 		}
 		expectedErr = c.error
 	case cmdDeleteFolder:
-		err = fs.DeleteFolder(ctx, c.path)
+		err = fs.DeleteFolder(ctx, c.path, c.options)
 		if c.error == nil {
 			require.NoError(t, err, "%s: should be able to delete %s", cmdName, c.path)
 		}

pkg/infra/filestorage/wrapper.go

@@ -256,26 +256,58 @@ func (b wrapper) CreateFolder(ctx context.Context, path string) error {
 	return b.wrapped.CreateFolder(ctx, rootedPath)
 }
 
-func (b wrapper) DeleteFolder(ctx context.Context, path string) error {
+func (b wrapper) deleteFolderOptionsWithDefaults(options *DeleteFolderOptions) *DeleteFolderOptions {
+	if options == nil {
+		return &DeleteFolderOptions{
+			Force:        false,
+			AccessFilter: b.filter,
+		}
+	}
+
+	if options.AccessFilter == nil {
+		return &DeleteFolderOptions{
+			Force:        options.Force,
+			AccessFilter: b.filter,
+		}
+	}
+
+	var filter PathFilter
+	if options.AccessFilter != nil {
+		filter = newAndPathFilter(b.filter, wrapPathFilter(options.AccessFilter, b.rootFolder))
+	} else {
+		filter = b.filter
+	}
+
+	return &DeleteFolderOptions{
+		Force:        options.Force,
+		AccessFilter: filter,
+	}
+}
+
+func (b wrapper) DeleteFolder(ctx context.Context, path string, options *DeleteFolderOptions) error {
 	if err := b.validatePath(path); err != nil {
 		return err
 	}
 
 	rootedPath := b.addRoot(path)
-	if !b.filter.IsAllowed(rootedPath) {
-		return nil
+
+	optionsWithDefaults := b.deleteFolderOptionsWithDefaults(options)
+	if !optionsWithDefaults.AccessFilter.IsAllowed(rootedPath) {
+		return fmt.Errorf("delete folder unauthorized - no access to %s", rootedPath)
 	}
 
-	isEmpty, err := b.isFolderEmpty(ctx, path)
-	if err != nil {
-		return err
+	if !optionsWithDefaults.Force {
+		isEmpty, err := b.isFolderEmpty(ctx, path)
+		if err != nil {
+			return err
+		}
+
+		if !isEmpty {
+			return fmt.Errorf("folder %s is not empty - cant remove it", path)
+		}
 	}
 
-	if !isEmpty {
-		return fmt.Errorf("folder %s is not empty - cant remove it", path)
-	}
-
-	return b.wrapped.DeleteFolder(ctx, rootedPath)
+	return b.wrapped.DeleteFolder(ctx, rootedPath, optionsWithDefaults)
 }
 
 func (b wrapper) List(ctx context.Context, folderPath string, paging *Paging, options *ListOptions) (*ListResponse, error) {