[MM-27277] Move away from storing uploaded files into memory (#15616)

* Move away from storing uploaded files into memory * Improve function readability * Fix translations * Revert version bump * Log error * Fix possible race condition and goroutine leak * Improve default case
2020-10-09 10:14:19 +02:00
--- a/app/file.go
+++ b/app/file.go
@@ -623,33 +623,41 @@ func (a *App) UploadFileX(channelId, name string, input io.Reader,
 		}
 	}

-	aerr = t.readAll()
-	if aerr != nil {
-		return t.fileinfo, aerr
-	}
-
-	aerr = t.runPlugins()
-	if aerr != nil {
-		return t.fileinfo, aerr
-	}
-
-	// Concurrently post-process the image.
-	wg := sync.WaitGroup{}
-
-	if !t.Raw && t.fileinfo.IsImage() {
-		wg.Add(1)
-		go func() {
-			t.postprocessImage()
-			wg.Done()
-		}()
-	}
-
-	_, aerr = t.writeFile(t.newReader(), t.fileinfo.Path)
+	written, aerr := t.writeFile(io.MultiReader(t.buf, t.limitedInput), t.fileinfo.Path)
 	if aerr != nil {
 		return nil, aerr
 	}

-	wg.Wait()
+	if written > t.maxFileSize {
+		if fileErr := a.RemoveFile(t.fileinfo.Path); fileErr != nil {
+			mlog.Error("Failed to remove file", mlog.Err(fileErr))
+		}
+		return nil, t.newAppError("api.file.upload_file.too_large_detailed.app_error",
+			"", http.StatusRequestEntityTooLarge, "Length", t.ContentLength, "Limit", t.maxFileSize)
+	}
+
+	t.fileinfo.Size = written
+
+	file, aerr := a.FileReader(t.fileinfo.Path)
+	if aerr != nil {
+		return nil, aerr
+	}
+	defer file.Close()
+
+	aerr = a.runPluginsHook(t.fileinfo, file)
+	if aerr != nil {
+		return nil, aerr
+	}
+
+	if !t.Raw && t.fileinfo.IsImage() {
+		file, aerr = a.FileReader(t.fileinfo.Path)
+		if aerr != nil {
+			return nil, aerr
+		}
+		defer file.Close()
+		t.postprocessImage(file)
+	}
+
 	if _, err := t.saveToDatabase(t.fileinfo); err != nil {
 		var appErr *model.AppError
 		switch {
@@ -663,69 +671,10 @@ func (a *App) UploadFileX(channelId, name string, input io.Reader,
 	return t.fileinfo, nil
 }

-func (t *UploadFileTask) readAll() *model.AppError {
-	_, err := t.buf.ReadFrom(t.limitedInput)
-	if err != nil {
-		// Ugly hack: the error is not exported from net/http.
-		if err.Error() == "http: request body too large" {
-			return t.newAppError("api.file.upload_file.too_large_detailed.app_error",
-				"", http.StatusRequestEntityTooLarge, "Length", t.buf.Len(), "Limit", t.limit)
-		}
-		return t.newAppError("api.file.upload_file.read_request.app_error",
-			err.Error(), http.StatusBadRequest)
-	}
-	if int64(t.buf.Len()) > t.limit {
-		return t.newAppError("api.file.upload_file.too_large_detailed.app_error",
-			"", http.StatusRequestEntityTooLarge, "Length", t.buf.Len(), "Limit", t.limit)
-	}
-	t.fileinfo.Size = int64(t.buf.Len())
-
-	t.limitedInput = nil
-	t.teeInput = nil
-	return nil
-}
-
-func (t *UploadFileTask) runPlugins() *model.AppError {
-	if t.pluginsEnvironment == nil {
-		return nil
-	}
-
-	pluginContext := &plugin.Context{}
-	var rejectionError *model.AppError
-
-	t.pluginsEnvironment.RunMultiPluginHook(func(hooks plugin.Hooks) bool {
-		buf := &bytes.Buffer{}
-		replacementInfo, rejectionReason := hooks.FileWillBeUploaded(pluginContext,
-			t.fileinfo, t.newReader(), buf)
-		if rejectionReason != "" {
-			rejectionError = t.newAppError("api.file.upload_file.rejected_by_plugin.app_error",
-				rejectionReason, http.StatusForbidden, "Reason", rejectionReason)
-			return false
-		}
-		if replacementInfo != nil {
-			t.fileinfo = replacementInfo
-		}
-		if buf.Len() != 0 {
-			t.buf = buf
-			t.teeInput = nil
-			t.limitedInput = nil
-			t.fileinfo.Size = int64(buf.Len())
-		}
-
-		return true
-	}, plugin.FileWillBeUploadedId)
-
-	if rejectionError != nil {
-		return rejectionError
-	}
-
-	return nil
-}
-
 func (t *UploadFileTask) preprocessImage() *model.AppError {
 	// If SVG, attempt to extract dimensions and then return
 	if t.fileinfo.MimeType == "image/svg+xml" {
-		svgInfo, err := parseSVG(t.newReader())
+		svgInfo, err := parseSVG(t.teeInput)
 		if err != nil {
 			mlog.Error("Failed to parse SVG", mlog.Err(err))
 		}
@@ -738,7 +687,7 @@ func (t *UploadFileTask) preprocessImage() *model.AppError {
 	}

 	// If we fail to decode, return "as is".
-	config, _, err := image.DecodeConfig(t.newReader())
+	config, _, err := image.DecodeConfig(t.teeInput)
 	if err != nil {
 		return nil
 	}
@@ -762,7 +711,7 @@ func (t *UploadFileTask) preprocessImage() *model.AppError {
 	// check the image orientation with goexif; consume the bytes we
 	// already have first, then keep Tee-ing from input.
 	// TODO: try to reuse exif's .Raw buffer rather than Tee-ing
-	if t.imageOrientation, err = getImageOrientation(t.newReader()); err == nil &&
+	if t.imageOrientation, err = getImageOrientation(io.MultiReader(bytes.NewReader(t.buf.Bytes()), t.teeInput)); err == nil &&
 		(t.imageOrientation == RotatedCWMirrored ||
 			t.imageOrientation == RotatedCCW ||
 			t.imageOrientation == RotatedCCWMirrored ||
@@ -773,13 +722,10 @@ func (t *UploadFileTask) preprocessImage() *model.AppError {
 	// For animated GIFs disable the preview; since we have to Decode gifs
 	// anyway, cache the decoded image for later.
 	if t.fileinfo.MimeType == "image/gif" {
-		gifConfig, err := gif.DecodeAll(t.newReader())
+		gifConfig, err := gif.DecodeAll(io.MultiReader(bytes.NewReader(t.buf.Bytes()), t.teeInput))
 		if err == nil {
-			if len(gifConfig.Image) >= 1 {
-				t.fileinfo.HasPreviewImage = false
-
-			}
 			if len(gifConfig.Image) > 0 {
+				t.fileinfo.HasPreviewImage = false
 				t.decoded = gifConfig.Image[0]
 				t.imageType = "gif"
 			}
@@ -789,7 +735,7 @@ func (t *UploadFileTask) preprocessImage() *model.AppError {
 	return nil
 }

-func (t *UploadFileTask) postprocessImage() {
+func (t *UploadFileTask) postprocessImage(file io.Reader) {
 	// don't try to process SVG files
 	if t.fileinfo.MimeType == "image/svg+xml" {
 		return
@@ -798,7 +744,7 @@ func (t *UploadFileTask) postprocessImage() {
 	decoded, typ := t.decoded, t.imageType
 	if decoded == nil {
 		var err error
-		decoded, typ, err = image.Decode(t.newReader())
+		decoded, typ, err = image.Decode(file)
 		if err != nil {
 			mlog.Error("Unable to decode image", mlog.Err(err))
 			return
@@ -839,17 +785,19 @@ func (t *UploadFileTask) postprocessImage() {
 	}

 	var wg sync.WaitGroup
-	wg.Add(3)
-	go func() {
-		defer wg.Done()
-		writeJPEG(genThumbnail(decoded), t.fileinfo.ThumbnailPath)
-	}()
-
-	go func() {
-		defer wg.Done()
-		writeJPEG(genPreview(decoded), t.fileinfo.PreviewPath)
-	}()
+	wg.Add(1)
+	if t.fileinfo.HasPreviewImage {
+		wg.Add(2)
+		go func() {
+			defer wg.Done()
+			writeJPEG(genThumbnail(decoded), t.fileinfo.ThumbnailPath)
+		}()

+		go func() {
+			defer wg.Done()
+			writeJPEG(genPreview(decoded), t.fileinfo.PreviewPath)
+		}()
+	}
 	go func() {
 		defer wg.Done()
 		if t.fileinfo.MiniPreview == nil {
@@ -859,14 +807,6 @@ func (t *UploadFileTask) postprocessImage() {
 	wg.Wait()
 }

-func (t UploadFileTask) newReader() io.Reader {
-	if t.teeInput != nil {
-		return io.MultiReader(bytes.NewReader(t.buf.Bytes()), t.teeInput)
-	} else {
-		return bytes.NewReader(t.buf.Bytes())
-	}
-}
-
 func (t UploadFileTask) pathPrefix() string {
 	return t.Timestamp.Format("20060102") +
 		"/teams/" + t.TeamId +
--- a/app/upload.go
+++ b/app/upload.go
@@ -9,6 +9,7 @@ import (
 	"net/http"
 	"path/filepath"
 	"strings"
+	"sync"
 	"time"

 	"github.com/mattermost/mattermost-server/v5/mlog"
@@ -19,6 +20,85 @@ import (

 const minFirstPartSize = 5 * 1024 * 1024 // 5MB

+func (a *App) runPluginsHook(info *model.FileInfo, file io.Reader) *model.AppError {
+	pluginsEnvironment := a.GetPluginsEnvironment()
+	if pluginsEnvironment == nil {
+		return nil
+	}
+
+	filePath := info.Path
+	// using a pipe to avoid loading the whole file content in memory.
+	r, w := io.Pipe()
+	errChan := make(chan *model.AppError, 1)
+	hookHasRunCh := make(chan struct{})
+
+	go func() {
+		defer w.Close()
+		defer close(hookHasRunCh)
+		defer close(errChan)
+		var rejErr *model.AppError
+		var once sync.Once
+		pluginContext := a.PluginContext()
+		pluginsEnvironment.RunMultiPluginHook(func(hooks plugin.Hooks) bool {
+			once.Do(func() {
+				hookHasRunCh <- struct{}{}
+			})
+			newInfo, rejStr := hooks.FileWillBeUploaded(pluginContext, info, file, w)
+			if rejStr != "" {
+				rejErr = model.NewAppError("runPluginsHook", "app.upload.run_plugins_hook.rejected",
+					map[string]interface{}{"Filename": info.Name, "Reason": rejStr}, "", http.StatusBadRequest)
+				return false
+			}
+			if newInfo != nil {
+				info = newInfo
+			}
+			return true
+		}, plugin.FileWillBeUploadedId)
+		if rejErr != nil {
+			errChan <- rejErr
+		}
+	}()
+
+	// If the plugin hook has not run we can return early.
+	if _, ok := <-hookHasRunCh; !ok {
+		return nil
+	}
+
+	tmpPath := filePath + ".tmp"
+	written, err := a.WriteFile(r, tmpPath)
+	if err != nil {
+		if fileErr := a.RemoveFile(tmpPath); fileErr != nil {
+			mlog.Error("Failed to remove file", mlog.Err(fileErr))
+		}
+		return err
+	}
+
+	if err = <-errChan; err != nil {
+		if fileErr := a.RemoveFile(info.Path); fileErr != nil {
+			mlog.Error("Failed to remove file", mlog.Err(fileErr))
+		}
+		if fileErr := a.RemoveFile(tmpPath); fileErr != nil {
+			mlog.Error("Failed to remove file", mlog.Err(fileErr))
+		}
+		return err
+	}
+
+	if written > 0 {
+		info.Size = written
+		if fileErr := a.MoveFile(tmpPath, info.Path); fileErr != nil {
+			mlog.Error("Failed to move file", mlog.Err(fileErr))
+			return model.NewAppError("runPluginsHook", "app.upload.run_plugins_hook.move_fail",
+				nil, fileErr.Error(), http.StatusInternalServerError)
+		}
+	} else {
+		if fileErr := a.RemoveFile(tmpPath); fileErr != nil {
+			mlog.Error("Failed to remove file", mlog.Err(fileErr))
+		}
+	}
+
+	return nil
+}
+
 func (a *App) CreateUploadSession(us *model.UploadSession) (*model.UploadSession, *model.AppError) {
 	if us.FileSize > *a.Config().FileSettings.MaxFileSize {
 		return nil, model.NewAppError("CreateUploadSession", "app.upload.create.upload_too_large.app_error",
@@ -162,58 +242,9 @@ func (a *App) UploadData(us *model.UploadSession, rd io.Reader) (*model.FileInfo
 	info.CreatorId = us.UserId
 	info.Path = us.Path

-	// call plugins upload hook
-	if pluginsEnvironment := a.GetPluginsEnvironment(); pluginsEnvironment != nil {
-		// using a pipe to avoid loading the whole file content in memory.
-		r, w := io.Pipe()
-		errChan := make(chan *model.AppError, 1)
-		go func() {
-			defer w.Close()
-			defer close(errChan)
-			pluginContext := a.PluginContext()
-			pluginsEnvironment.RunMultiPluginHook(func(hooks plugin.Hooks) bool {
-				newInfo, rejStr := hooks.FileWillBeUploaded(pluginContext, info, file, w)
-				if rejStr != "" {
-					errChan <- model.NewAppError("UploadData", "File rejected by plugin. "+rejStr, nil, "", http.StatusBadRequest)
-					return false
-				}
-				if newInfo != nil {
-					info = newInfo
-				}
-				return true
-			}, plugin.FileWillBeUploadedId)
-		}()
-
-		var written int64
-		tmpPath := us.Path + ".tmp"
-		written, err = a.WriteFile(r, tmpPath)
-		if err != nil {
-			if fileErr := a.RemoveFile(tmpPath); fileErr != nil {
-				mlog.Error("Failed to remove file", mlog.Err(fileErr))
-			}
-			return nil, err
-		}
-
-		if err = <-errChan; err != nil {
-			if fileErr := a.RemoveFile(us.Path); fileErr != nil {
-				mlog.Error("Failed to remove file", mlog.Err(fileErr))
-			}
-			if fileErr := a.RemoveFile(tmpPath); fileErr != nil {
-				mlog.Error("Failed to remove file", mlog.Err(fileErr))
-			}
-			return nil, err
-		}
-
-		if written > 0 {
-			info.Size = written
-			if fileErr := a.MoveFile(tmpPath, us.Path); fileErr != nil {
-				mlog.Error("Failed to move file", mlog.Err(fileErr))
-			}
-		} else {
-			if fileErr := a.RemoveFile(tmpPath); fileErr != nil {
-				mlog.Error("Failed to remove file", mlog.Err(fileErr))
-			}
-		}
+	// run plugins upload hook
+	if err := a.runPluginsHook(info, file); err != nil {
+		return nil, err
 	}

 	// image post-processing
--- a/go.tools.sum
+++ b/go.tools.sum
@@ -68,6 +68,8 @@ github.com/mattermost/mattermost-utilities/mmgotool v0.0.0-20200828152206-519b99
 github.com/mattermost/mattermost-utilities/mmgotool v0.0.0-20200828152206-519b99a4e51e/go.mod h1:3gKozJI8n2Y/vW37GfnFWAdehGXe5yZlt+HykK6Y3DM=
 github.com/mattermost/mattermost-utilities/mmgotool v0.0.0-20200915114419-f4421bc07461 h1:dn2/HZjzUY5PQmKDmH95vgwVqpbR84FiU/t060g7rqg=
 github.com/mattermost/mattermost-utilities/mmgotool v0.0.0-20200915114419-f4421bc07461/go.mod h1:3gKozJI8n2Y/vW37GfnFWAdehGXe5yZlt+HykK6Y3DM=
+github.com/mattermost/mattermost-utilities/mmgotool v0.0.0-20200926180007-fd1b679200e5 h1:1fVtMi+1XPAtxffRZiMSiy7Zm4Od3fyHWnC2BydfYHY=
+github.com/mattermost/mattermost-utilities/mmgotool v0.0.0-20200926180007-fd1b679200e5/go.mod h1:3gKozJI8n2Y/vW37GfnFWAdehGXe5yZlt+HykK6Y3DM=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
 github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
--- a/i18n/en.json
+++ b/i18n/en.json
@@ -1456,10 +1456,6 @@
    "id": "api.file.upload_file.read_request.app_error",
    "translation": "Unable to upload file(s). Error reading or parsing request data."
  },
-  {
-    "id": "api.file.upload_file.rejected_by_plugin.app_error",
-    "translation": "Unable to upload file {{.Filename}}. Rejected by plugin: {{.Reason}}"
-  },
  {
    "id": "api.file.upload_file.storage.app_error",
    "translation": "Unable to upload file. Image storage is not configured."
@@ -5302,6 +5298,14 @@
    "id": "app.upload.get_for_user.app_error",
    "translation": "Failed to get uploads for user."
  },
+  {
+    "id": "app.upload.run_plugins_hook.move_fail",
+    "translation": "Failed to move file."
+  },
+  {
+    "id": "app.upload.run_plugins_hook.rejected",
+    "translation": "Unable to upload file {{.Filename}}. Rejected by plugin: {{.Reason}}"
+  },
  {
    "id": "app.upload.upload_data.concurrent.app_error",
    "translation": "Unable to upload data from more than one request."