[MM-28422] Enable processing of import files through API (#16062)

* Implement unzip function

* Implement FileSize method

* Implement path rewriting for bulk import

* Small improvements

* Add ImportSettings to config

* Implement ListImports API endpoint

* Enable uploading import files

* Implement import process job

* Add missing license headers

* Address reviews

* Make path sanitization a bit smarter

* Clean path before calculating Dir

* [MM-30008] Add mmctl support for file imports (#16301)

* Add mmctl support for import files

* Improve test

* Remove unnecessary handlers

* Use th.TestForSystemAdminAndLocal

* Make nouser id a constant

app/app.go

@@ -115,6 +115,9 @@ func (a *App) initJobs() {
 	if productNoticesJobInterface != nil {
 		a.srv.Jobs.ProductNotices = productNoticesJobInterface(a)
 	}
+	if jobsImportProcessInterface != nil {
+		a.srv.Jobs.ImportProcess = jobsImportProcessInterface(a)
+	}
 
 	if jobsActiveUsersInterface != nil {
 		a.srv.Jobs.ActiveUsers = jobsActiveUsersInterface(a)

app/app_iface.go

@@ -385,6 +385,7 @@ type AppIface interface {
 	BuildSamlMetadataObject(idpMetadata []byte) (*model.SamlMetadataResponse, *model.AppError)
 	BulkExport(writer io.Writer, file string, pathToEmojiDir string, dirNameToExportEmoji string) *model.AppError
 	BulkImport(fileReader io.Reader, dryRun bool, workers int) (*model.AppError, int)
+	BulkImportWithPath(fileReader io.Reader, dryRun bool, workers int, importPath string) (*model.AppError, int)
 	CancelJob(jobId string) *model.AppError
 	ChannelMembersToAdd(since int64, channelID *string) ([]*model.UserChannelIDPair, *model.AppError)
 	ChannelMembersToRemove(teamID *string) ([]*model.ChannelMember, *model.AppError)
@@ -494,6 +495,7 @@ type AppIface interface {
 	FetchSamlMetadataFromIdp(url string) ([]byte, *model.AppError)
 	FileBackend() (filesstore.FileBackend, *model.AppError)
 	FileExists(path string) (bool, *model.AppError)
+	FileSize(path string) (int64, *model.AppError)
 	FillInChannelProps(channel *model.Channel) *model.AppError
 	FillInChannelsProps(channelList *model.ChannelList) *model.AppError
 	FilterUsersByVisible(viewer *model.User, otherUsers []*model.User) ([]*model.User, *model.AppError)
@@ -770,6 +772,7 @@ type AppIface interface {
 	LimitedClientConfig() map[string]string
 	ListAllCommands(teamId string, T goi18n.TranslateFunc) ([]*model.Command, *model.AppError)
 	ListDirectory(path string) ([]string, *model.AppError)
+	ListImports() ([]string, *model.AppError)
 	ListPluginKeys(pluginId string, page, perPage int) ([]string, *model.AppError)
 	ListTeamCommands(teamId string) ([]*model.Command, *model.AppError)
 	Log() *mlog.Logger

app/enterprise.go

@@ -108,6 +108,12 @@ func RegisterJobsExpiryNotifyJobInterface(f func(*App) tjobs.ExpiryNotifyJobInte
 	jobsExpiryNotifyInterface = f
 }
 
+var jobsImportProcessInterface func(*App) tjobs.ImportProcessInterface
+
+func RegisterJobsImportProcessInterface(f func(*App) tjobs.ImportProcessInterface) {
+	jobsImportProcessInterface = f
+}
+
 var productNoticesJobInterface func(*App) tjobs.ProductNoticesJobInterface
 
 func RegisterProductNoticesJobInterface(f func(*App) tjobs.ProductNoticesJobInterface) {

app/file.go

@@ -101,6 +101,14 @@ func (a *App) FileExists(path string) (bool, *model.AppError) {
 	return backend.FileExists(path)
 }
 
+func (a *App) FileSize(path string) (int64, *model.AppError) {
+	backend, err := a.FileBackend()
+	if err != nil {
+		return 0, err
+	}
+	return backend.FileSize(path)
+}
+
 func (a *App) MoveFile(oldPath, newPath string) *model.AppError {
 	backend, err := a.FileBackend()
 	if err != nil {

app/import.go

@@ -5,9 +5,11 @@ package app
 
 import (
 	"bufio"
+	"bytes"
 	"encoding/json"
 	"io"
 	"net/http"
+	"path/filepath"
 	"strings"
 	"sync"
 
@@ -29,6 +31,46 @@ func stopOnError(err LineImportWorkerError) bool {
 	return true
 }
 
+func rewriteAttachmentPaths(files *[]AttachmentImportData, basePath string) {
+	if files == nil {
+		return
+	}
+	for _, f := range *files {
+		if f.Path != nil {
+			*f.Path = filepath.Join(basePath, *f.Path)
+		}
+	}
+}
+
+func rewriteFilePaths(line *LineImportData, basePath string) {
+	switch line.Type {
+	case "post", "direct_post":
+		var replies []ReplyImportData
+		if line.Type == "direct_post" {
+			rewriteAttachmentPaths(line.DirectPost.Attachments, basePath)
+			if line.DirectPost.Replies != nil {
+				replies = *line.DirectPost.Replies
+			}
+		} else {
+			rewriteAttachmentPaths(line.Post.Attachments, basePath)
+			if line.Post.Replies != nil {
+				replies = *line.Post.Replies
+			}
+		}
+		for _, reply := range replies {
+			rewriteAttachmentPaths(reply.Attachments, basePath)
+		}
+	case "user":
+		if line.User.ProfileImage != nil {
+			*line.User.ProfileImage = filepath.Join(basePath, *line.User.ProfileImage)
+		}
+	case "emoji":
+		if line.Emoji.Image != nil {
+			*line.Emoji.Image = filepath.Join(basePath, *line.Emoji.Image)
+		}
+	}
+}
+
 func (a *App) bulkImportWorker(dryRun bool, wg *sync.WaitGroup, lines <-chan LineImportWorkerData, errors chan<- LineImportWorkerError) {
 	postLines := []LineImportWorkerData{}
 	directPostLines := []LineImportWorkerData{}
@@ -77,6 +119,14 @@ func (a *App) bulkImportWorker(dryRun bool, wg *sync.WaitGroup, lines <-chan Lin
 }
 
 func (a *App) BulkImport(fileReader io.Reader, dryRun bool, workers int) (*model.AppError, int) {
+	return a.bulkImport(fileReader, dryRun, workers, "")
+}
+
+func (a *App) BulkImportWithPath(fileReader io.Reader, dryRun bool, workers int, importPath string) (*model.AppError, int) {
+	return a.bulkImport(fileReader, dryRun, workers, importPath)
+}
+
+func (a *App) bulkImport(fileReader io.Reader, dryRun bool, workers int, importPath string) (*model.AppError, int) {
 	scanner := bufio.NewScanner(fileReader)
 	buf := make([]byte, 0, 64*1024)
 	scanner.Buffer(buf, maxScanTokenSize)
@@ -92,7 +142,7 @@ func (a *App) BulkImport(fileReader io.Reader, dryRun bool, workers int) (*model
 	lastLineType := ""
 
 	for scanner.Scan() {
-		decoder := json.NewDecoder(strings.NewReader(scanner.Text()))
+		decoder := json.NewDecoder(bytes.NewReader(scanner.Bytes()))
 		lineNumber++
 
 		var line LineImportData
@@ -100,6 +150,10 @@ func (a *App) BulkImport(fileReader io.Reader, dryRun bool, workers int) (*model
 			return model.NewAppError("BulkImport", "app.import.bulk_import.json_decode.error", nil, err.Error(), http.StatusBadRequest), lineNumber
 		}
 
+		if importPath != "" {
+			rewriteFilePaths(&line, importPath)
+		}
+
 		if lineNumber == 1 {
 			importDataFileVersion, appErr := processImportDataFileVersionLine(line)
 			if appErr != nil {
@@ -214,3 +268,20 @@ func (a *App) importLine(line LineImportData, dryRun bool) *model.AppError {
 		return model.NewAppError("BulkImport", "app.import.import_line.unknown_line_type.error", map[string]interface{}{"Type": line.Type}, "", http.StatusBadRequest)
 	}
 }
+
+func (a *App) ListImports() ([]string, *model.AppError) {
+	imports, appErr := a.ListDirectory(*a.Config().ImportSettings.Directory)
+	if appErr != nil {
+		return nil, appErr
+	}
+
+	results := make([]string, 0, len(imports))
+	for i := 0; i < len(imports); i++ {
+		filename := filepath.Base(imports[i])
+		if !strings.HasSuffix(filename, incompleteUploadSuffix) {
+			results = append(results, filename)
+		}
+	}
+
+	return results, nil
+}

app/import_functions.go

@@ -532,6 +532,7 @@ func (a *App) importUser(data *UserImportData, dryRun bool) *model.AppError {
 		if err != nil {
 			mlog.Error("Unable to open the profile image.", mlog.Any("err", err))
 		}
+		defer file.Close()
 		if err := a.SetProfileImageFromMultiPartFile(savedUser.Id, file); err != nil {
 			mlog.Error("Unable to set the profile image from a file.", mlog.Any("err", err))
 		}
@@ -1740,6 +1741,7 @@ func (a *App) importEmoji(data *EmojiImportData, dryRun bool) *model.AppError {
 	if err != nil {
 		return model.NewAppError("BulkImport", "app.import.emoji.bad_file.error", map[string]interface{}{"EmojiName": *data.Name}, "", http.StatusBadRequest)
 	}
+	defer file.Close()
 
 	if _, err := a.WriteFile(file, getEmojiImagePath(emoji.Id)); err != nil {
 		return err

app/import_test.go

@@ -4,8 +4,11 @@
 package app
 
 import (
+	"io/ioutil"
 	"net/http"
+	"os"
 	"path/filepath"
+	"runtime"
 	"strings"
 	"testing"
 
@@ -13,6 +16,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/mattermost/mattermost-server/v5/model"
+	"github.com/mattermost/mattermost-server/v5/utils"
 	"github.com/mattermost/mattermost-server/v5/utils/fileutils"
 )
 
@@ -192,7 +196,7 @@ func TestImportBulkImport(t *testing.T) {
 
 	// Run bulk import using a valid and large input and a \r\n line break.
 	t.Run("", func(t *testing.T) {
-		posts := `{"type": "post"` + strings.Repeat(`, "post": {"team": "`+teamName+`", "channel": "`+channelName+`", "user": "`+username+`", "message": "Repeat after me", "create_at": 193456789012}`, 1E4) + "}"
+		posts := `{"type": "post"` + strings.Repeat(`, "post": {"team": "`+teamName+`", "channel": "`+channelName+`", "user": "`+username+`", "message": "Repeat after me", "create_at": 193456789012}`, 1e4) + "}"
 		data4 := `{"type": "version", "version": 1}
 {"type": "team", "team": {"type": "O", "display_name": "lskmw2d7a5ao7ppwqh5ljchvr4", "name": "` + teamName + `"}}
 {"type": "channel", "channel": {"type": "O", "display_name": "xr6m6udffngark2uekvr3hoeny", "team": "` + teamName + `", "name": "` + channelName + `"}}
@@ -263,3 +267,124 @@ func AssertFileIdsInPost(files []*model.FileInfo, th *TestHelper, t *testing.T)
 		assert.Contains(t, posts[0].FileIds, file.Id)
 	}
 }
+
+func TestRewriteFilePaths(t *testing.T) {
+	genAttachments := func() *[]AttachmentImportData {
+		return &[]AttachmentImportData{
+			{
+				Path: model.NewString("file.jpg"),
+			},
+			{
+				Path: model.NewString("somedir/file.jpg"),
+			},
+		}
+	}
+
+	line := LineImportData{
+		Type: "post",
+		Post: &PostImportData{
+			Attachments: genAttachments(),
+		},
+	}
+
+	line2 := LineImportData{
+		Type: "direct_post",
+		DirectPost: &DirectPostImportData{
+			Attachments: genAttachments(),
+		},
+	}
+
+	userLine := LineImportData{
+		Type: "user",
+		User: &UserImportData{
+			ProfileImage: model.NewString("profile.jpg"),
+		},
+	}
+
+	emojiLine := LineImportData{
+		Type: "emoji",
+		Emoji: &EmojiImportData{
+			Image: model.NewString("emoji.png"),
+		},
+	}
+
+	t.Run("empty path", func(t *testing.T) {
+		expected := &[]AttachmentImportData{
+			{
+				Path: model.NewString("file.jpg"),
+			},
+			{
+				Path: model.NewString("somedir/file.jpg"),
+			},
+		}
+		rewriteFilePaths(&line, "")
+		require.Equal(t, expected, line.Post.Attachments)
+		rewriteFilePaths(&line2, "")
+		require.Equal(t, expected, line2.DirectPost.Attachments)
+	})
+
+	t.Run("valid path", func(t *testing.T) {
+		expected := &[]AttachmentImportData{
+			{
+				Path: model.NewString("/tmp/file.jpg"),
+			},
+			{
+				Path: model.NewString("/tmp/somedir/file.jpg"),
+			},
+		}
+
+		t.Run("post attachments", func(t *testing.T) {
+			rewriteFilePaths(&line, "/tmp")
+			require.Equal(t, expected, line.Post.Attachments)
+		})
+
+		t.Run("direct post attachments", func(t *testing.T) {
+			rewriteFilePaths(&line2, "/tmp")
+			require.Equal(t, expected, line2.DirectPost.Attachments)
+		})
+
+		t.Run("profile image", func(t *testing.T) {
+			expected := "/tmp/profile.jpg"
+			rewriteFilePaths(&userLine, "/tmp")
+			require.Equal(t, expected, *userLine.User.ProfileImage)
+		})
+
+		t.Run("emoji", func(t *testing.T) {
+			expected := "/tmp/emoji.png"
+			rewriteFilePaths(&emojiLine, "/tmp")
+			require.Equal(t, expected, *emojiLine.Emoji.Image)
+		})
+	})
+}
+
+func BenchmarkBulkImport(b *testing.B) {
+	th := Setup(b)
+	defer th.TearDown()
+
+	testsDir, _ := fileutils.FindDir("tests")
+
+	importFile, err := os.Open(testsDir + "/import_test.zip")
+	require.Nil(b, err)
+	defer importFile.Close()
+
+	info, err := importFile.Stat()
+	require.Nil(b, err)
+
+	dir, err := ioutil.TempDir("", "testimport")
+	require.Nil(b, err)
+	defer os.RemoveAll(dir)
+
+	_, err = utils.UnzipToPath(importFile, info.Size(), dir)
+	require.Nil(b, err)
+
+	jsonFile, err := os.Open(dir + "/import.jsonl")
+	require.Nil(b, err)
+	defer jsonFile.Close()
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		err, _ := th.App.BulkImportWithPath(jsonFile, false, runtime.NumCPU(), dir)
+		require.Nil(b, err)
+	}
+	b.StopTimer()
+}

app/opentracing/opentracing_layer.go

@@ -922,6 +922,28 @@ func (a *OpenTracingAppLayer) BulkImport(fileReader io.Reader, dryRun bool, work
 	return resultVar0, resultVar1
 }
 
+func (a *OpenTracingAppLayer) BulkImportWithPath(fileReader io.Reader, dryRun bool, workers int, importPath string) (*model.AppError, int) {
+	origCtx := a.ctx
+	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.BulkImportWithPath")
+
+	a.ctx = newCtx
+	a.app.Srv().Store.SetContext(newCtx)
+	defer func() {
+		a.app.Srv().Store.SetContext(origCtx)
+		a.ctx = origCtx
+	}()
+
+	defer span.Finish()
+	resultVar0, resultVar1 := a.app.BulkImportWithPath(fileReader, dryRun, workers, importPath)
+
+	if resultVar0 != nil {
+		span.LogFields(spanlog.Error(resultVar0))
+		ext.Error.Set(span, true)
+	}
+
+	return resultVar0, resultVar1
+}
+
 func (a *OpenTracingAppLayer) CancelJob(jobId string) *model.AppError {
 	origCtx := a.ctx
 	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.CancelJob")
@@ -3667,6 +3689,28 @@ func (a *OpenTracingAppLayer) FileReader(path string) (filesstore.ReadCloseSeeke
 	return resultVar0, resultVar1
 }
 
+func (a *OpenTracingAppLayer) FileSize(path string) (int64, *model.AppError) {
+	origCtx := a.ctx
+	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.FileSize")
+
+	a.ctx = newCtx
+	a.app.Srv().Store.SetContext(newCtx)
+	defer func() {
+		a.app.Srv().Store.SetContext(origCtx)
+		a.ctx = origCtx
+	}()
+
+	defer span.Finish()
+	resultVar0, resultVar1 := a.app.FileSize(path)
+
+	if resultVar1 != nil {
+		span.LogFields(spanlog.Error(resultVar1))
+		ext.Error.Set(span, true)
+	}
+
+	return resultVar0, resultVar1
+}
+
 func (a *OpenTracingAppLayer) FillInChannelProps(channel *model.Channel) *model.AppError {
 	origCtx := a.ctx
 	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.FillInChannelProps")
@@ -10252,6 +10296,28 @@ func (a *OpenTracingAppLayer) ListDirectory(path string) ([]string, *model.AppEr
 	return resultVar0, resultVar1
 }
 
+func (a *OpenTracingAppLayer) ListImports() ([]string, *model.AppError) {
+	origCtx := a.ctx
+	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.ListImports")
+
+	a.ctx = newCtx
+	a.app.Srv().Store.SetContext(newCtx)
+	defer func() {
+		a.app.Srv().Store.SetContext(origCtx)
+		a.ctx = origCtx
+	}()
+
+	defer span.Finish()
+	resultVar0, resultVar1 := a.app.ListImports()
+
+	if resultVar1 != nil {
+		span.LogFields(spanlog.Error(resultVar1))
+		ext.Error.Set(span, true)
+	}
+
+	return resultVar0, resultVar1
+}
+
 func (a *OpenTracingAppLayer) ListPluginKeys(pluginId string, page int, perPage int) ([]string, *model.AppError) {
 	origCtx := a.ctx
 	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.ListPluginKeys")

app/upload.go

@@ -19,6 +19,7 @@ import (
 )
 
 const minFirstPartSize = 5 * 1024 * 1024 // 5MB
+const incompleteUploadSuffix = ".tmp"
 
 func (a *App) runPluginsHook(info *model.FileInfo, file io.Reader) *model.AppError {
 	pluginsEnvironment := a.GetPluginsEnvironment()
@@ -108,19 +109,25 @@ func (a *App) CreateUploadSession(us *model.UploadSession) (*model.UploadSession
 	us.FileOffset = 0
 	now := time.Now()
 	us.CreateAt = model.GetMillisForTime(now)
-	us.Path = now.Format("20060102") + "/teams/noteam/channels/" + us.ChannelId + "/users/" + us.UserId + "/" + us.Id + "/" + filepath.Base(us.Filename)
+	if us.Type == model.UploadTypeAttachment {
+		us.Path = now.Format("20060102") + "/teams/noteam/channels/" + us.ChannelId + "/users/" + us.UserId + "/" + us.Id + "/" + filepath.Base(us.Filename)
+	} else if us.Type == model.UploadTypeImport {
+		us.Path = *a.Config().ImportSettings.Directory + "/" + us.Id + "_" + filepath.Base(us.Filename)
+	}
 	if err := us.IsValid(); err != nil {
 		return nil, err
 	}
 
-	channel, err := a.GetChannel(us.ChannelId)
-	if err != nil {
-		return nil, model.NewAppError("CreateUploadSession", "app.upload.create.incorrect_channel_id.app_error",
-			map[string]interface{}{"channelId": us.ChannelId}, "", http.StatusBadRequest)
-	}
-	if channel.DeleteAt != 0 {
-		return nil, model.NewAppError("CreateUploadSession", "app.upload.create.cannot_upload_to_deleted_channel.app_error",
-			map[string]interface{}{"channelId": us.ChannelId}, "", http.StatusBadRequest)
+	if us.Type == model.UploadTypeAttachment {
+		channel, err := a.GetChannel(us.ChannelId)
+		if err != nil {
+			return nil, model.NewAppError("CreateUploadSession", "app.upload.create.incorrect_channel_id.app_error",
+				map[string]interface{}{"channelId": us.ChannelId}, "", http.StatusBadRequest)
+		}
+		if channel.DeleteAt != 0 {
+			return nil, model.NewAppError("CreateUploadSession", "app.upload.create.cannot_upload_to_deleted_channel.app_error",
+				map[string]interface{}{"channelId": us.ChannelId}, "", http.StatusBadRequest)
+		}
 	}
 
 	us, storeErr := a.Srv().Store.UploadSession().Save(us)
@@ -186,6 +193,11 @@ func (a *App) UploadData(us *model.UploadSession, rd io.Reader) (*model.FileInfo
 			nil, "FileOffset mismatch", http.StatusBadRequest)
 	}
 
+	uploadPath := us.Path
+	if us.Type == model.UploadTypeImport {
+		uploadPath += incompleteUploadSuffix
+	}
+
 	// make sure it's not possible to upload more data than what is expected.
 	lr := &io.LimitedReader{
 		R: rd,
@@ -195,12 +207,12 @@ func (a *App) UploadData(us *model.UploadSession, rd io.Reader) (*model.FileInfo
 	var written int64
 	if us.FileOffset == 0 {
 		// new upload
-		written, err = a.WriteFile(lr, us.Path)
+		written, err = a.WriteFile(lr, uploadPath)
 		if err != nil && written == 0 {
 			return nil, err
 		}
 		if written < minFirstPartSize && written != us.FileSize {
-			a.RemoveFile(us.Path)
+			a.RemoveFile(uploadPath)
 			var errStr string
 			if err != nil {
 				errStr = err.Error()
@@ -210,7 +222,7 @@ func (a *App) UploadData(us *model.UploadSession, rd io.Reader) (*model.FileInfo
 		}
 	} else if us.FileOffset < us.FileSize {
 		// resume upload
-		written, err = a.AppendFile(lr, us.Path)
+		written, err = a.AppendFile(lr, uploadPath)
 	}
 	if written > 0 {
 		us.FileOffset += written
@@ -228,7 +240,7 @@ func (a *App) UploadData(us *model.UploadSession, rd io.Reader) (*model.FileInfo
 	}
 
 	// upload is done, create FileInfo
-	file, err := a.FileReader(us.Path)
+	file, err := a.FileReader(uploadPath)
 	if err != nil {
 		return nil, model.NewAppError("UploadData", "app.upload.upload_data.read_file.app_error", nil, err.Error(), http.StatusInternalServerError)
 	}
@@ -260,13 +272,19 @@ func (a *App) UploadData(us *model.UploadSession, rd io.Reader) (*model.FileInfo
 		nameWithoutExtension := info.Name[:strings.LastIndex(info.Name, ".")]
 		info.PreviewPath = filepath.Dir(info.Path) + "/" + nameWithoutExtension + "_preview.jpg"
 		info.ThumbnailPath = filepath.Dir(info.Path) + "/" + nameWithoutExtension + "_thumb.jpg"
-		imgData, fileErr := a.ReadFile(us.Path)
+		imgData, fileErr := a.ReadFile(uploadPath)
 		if fileErr != nil {
 			return nil, fileErr
 		}
 		a.HandleImages([]string{info.PreviewPath}, []string{info.ThumbnailPath}, [][]byte{imgData})
 	}
 
+	if us.Type == model.UploadTypeImport {
+		if err := a.MoveFile(uploadPath, us.Path); err != nil {
+			return nil, model.NewAppError("UploadData", "app.upload.upload_data.move_file.app_error", nil, err.Error(), http.StatusInternalServerError)
+		}
+	}
+
 	var storeErr error
 	if info, storeErr = a.Srv().Store.FileInfo().Save(info); storeErr != nil {
 		var appErr *model.AppError