FEATURE: Uppy direct S3 multipart uploads in composer (#14051)

This pull request introduces the endpoints required, and the JavaScript functionality in the `ComposerUppyUpload` mixin, for direct S3 multipart uploads. There are four new endpoints in the uploads controller:

* `create-multipart.json` - Creates the multipart upload in S3 along with an `ExternalUploadStub` record, storing information about the file in the same way as `generate-presigned-put.json` does for regular direct S3 uploads
* `batch-presign-multipart-parts.json` - Takes a list of part numbers and the unique identifier for an `ExternalUploadStub` record, and generates the presigned URLs for those parts if the multipart upload still exists and if the user has permission to access that upload
* `complete-multipart.json` - Completes the multipart upload in S3. Needs the full list of part numbers and their associated ETags which are returned when the part is uploaded to the presigned URL above. Only works if the user has permission to access the associated `ExternalUploadStub` record and the multipart upload still exists.

  After we confirm the upload is complete in S3, we go through the regular `UploadCreator` flow, the same as `complete-external-upload.json`, and promote the temporary upload S3 into a full `Upload` record, moving it to its final destination.
* `abort-multipart.json` - Aborts the multipart upload on S3 and destroys the `ExternalUploadStub` record if the user has permission to access that upload.

Also added are a few new columns to `ExternalUploadStub`:

* multipart - Whether or not this is a multipart upload
* external_upload_identifier - The "upload ID" for an S3 multipart upload
* filesize - The size of the file when the `create-multipart.json` or `generate-presigned-put.json` is called. This is used for validation.

When the user completes a direct S3 upload, either regular or multipart, we take the `filesize` that was captured when the `ExternalUploadStub` was first created and compare it with the final `Content-Length` size of the file where it is stored in S3. Then, if the two do not match, we throw an error, delete the file on S3, and ban the user from uploading files for N (default 5) minutes. This would only happen if the user uploads a different file than what they first specified, or in the case of multipart uploads uploaded larger chunks than needed. This is done to prevent abuse of S3 storage by bad actors.

Also included in this PR is an update to vendor/uppy.js. This has been built locally from the latest uppy source at d613b849a6. This must be done so that I can get my multipart upload changes into Discourse. When the Uppy team cuts a proper release, we can bump the package.json versions instead.

app/assets/javascripts/discourse-shims.js

@@ -32,7 +32,7 @@ define("@popperjs/core", ["exports"], function (__exports__) {
 
 define("@uppy/core", ["exports"], function (__exports__) {
   __exports__.default = window.Uppy.Core;
-  __exports__.Plugin = window.Uppy.Plugin;
+  __exports__.BasePlugin = window.Uppy.BasePlugin;
 });
 
 define("@uppy/aws-s3", ["exports"], function (__exports__) {

app/assets/javascripts/discourse/app/lib/uppy-checksum-plugin.js

@@ -1,8 +1,8 @@
-import { Plugin } from "@uppy/core";
+import { BasePlugin } from "@uppy/core";
 import { warn } from "@ember/debug";
 import { Promise } from "rsvp";
 
-export default class UppyChecksum extends Plugin {
+export default class UppyChecksum extends BasePlugin {
   constructor(uppy, opts) {
     super(uppy, opts);
     this.id = opts.id || "uppy-checksum";

app/assets/javascripts/discourse/app/lib/uppy-media-optimization-plugin.js

@@ -1,8 +1,8 @@
-import { Plugin } from "@uppy/core";
+import { BasePlugin } from "@uppy/core";
 import { warn } from "@ember/debug";
 import { Promise } from "rsvp";
 
-export default class UppyMediaOptimization extends Plugin {
+export default class UppyMediaOptimization extends BasePlugin {
   constructor(uppy, opts) {
     super(uppy, opts);
     this.id = opts.id || "uppy-media-optimization";
@@ -30,7 +30,10 @@ export default class UppyMediaOptimization extends Plugin {
             id: "discourse.uppy-media-optimization",
           });
         } else {
-          this.uppy.setFileState(fileId, { data: optimizedFile });
+          this.uppy.setFileState(fileId, {
+            data: optimizedFile,
+            size: optimizedFile.size,
+          });
         }
         this.uppy.emit("preprocess-complete", this.pluginClass, file);
       })

app/assets/javascripts/discourse/app/mixins/composer-upload-uppy.js

@@ -1,10 +1,12 @@
 import Mixin from "@ember/object/mixin";
+import { ajax } from "discourse/lib/ajax";
 import { deepMerge } from "discourse-common/lib/object";
 import UppyChecksum from "discourse/lib/uppy-checksum-plugin";
 import UppyMediaOptimization from "discourse/lib/uppy-media-optimization-plugin";
 import Uppy from "@uppy/core";
 import DropTarget from "@uppy/drop-target";
 import XHRUpload from "@uppy/xhr-upload";
+import AwsS3Multipart from "@uppy/aws-s3-multipart";
 import { warn } from "@ember/debug";
 import I18n from "I18n";
 import getURL from "discourse-common/lib/get-url";
@@ -70,6 +72,7 @@ export default Mixin.create({
 
   _bindUploadTarget() {
     this.placeholders = {};
+    this._inProgressUploads = 0;
     this._preProcessorStatus = {};
     this.fileInputEl = document.getElementById("file-uploader");
     const isPrivateMessage = this.get("composer.privateMessage");
@@ -140,9 +143,12 @@ export default Mixin.create({
     // name for the preprocess-X events.
     this._trackPreProcessorStatus(UppyChecksum);
 
-    // TODO (martin) support for direct S3 uploads will come later, for now
+    // hidden setting like enable_experimental_image_uploader
-    // we just want the regular /uploads.json endpoint to work well
+    if (this.siteSettings.enable_direct_s3_uploads) {
-    this._useXHRUploads();
+      this._useS3MultipartUploads();
+    } else {
+      this._useXHRUploads();
+    }
 
     // TODO (martin) develop upload handler guidance and an API to use; will
     // likely be using uppy plugins for this
@@ -171,6 +177,7 @@ export default Mixin.create({
       });
 
       files.forEach((file) => {
+        this._inProgressUploads++;
         const placeholder = this._uploadPlaceholder(file);
         this.placeholders[file.id] = {
           uploadPlaceholder: placeholder,
@@ -199,14 +206,7 @@ export default Mixin.create({
       this.appEvents.trigger("composer:upload-success", file.name, upload);
     });
 
-    this._uppyInstance.on("upload-error", (file, error, response) => {
+    this._uppyInstance.on("upload-error", this._handleUploadError.bind(this));
-      this._resetUpload(file, { removePlaceholder: true });
-
-      if (!this.userCancelled) {
-        displayErrorForUpload(response, this.siteSettings, file.name);
-        this.appEvents.trigger("composer:upload-error", file);
-      }
-    });
 
     this._uppyInstance.on("complete", () => {
       this.appEvents.trigger("composer:all-uploads-complete");
@@ -235,6 +235,20 @@ export default Mixin.create({
     this._setupPreprocessing();
   },
 
+  _handleUploadError(file, error, response) {
+    this._inProgressUploads--;
+    this._resetUpload(file, { removePlaceholder: true });
+
+    if (!this.userCancelled) {
+      displayErrorForUpload(response || error, this.siteSettings, file.name);
+      this.appEvents.trigger("composer:upload-error", file);
+    }
+
+    if (this._inProgressUploads === 0) {
+      this._reset();
+    }
+  },
+
   _setupPreprocessing() {
     Object.keys(this.uploadProcessorActions).forEach((action) => {
       switch (action) {
@@ -343,6 +357,99 @@ export default Mixin.create({
     });
   },
 
+  _useS3MultipartUploads() {
+    const self = this;
+
+    this._uppyInstance.use(AwsS3Multipart, {
+      // controls how many simultaneous _chunks_ are uploaded, not files,
+      // which in turn controls the minimum number of chunks presigned
+      // in each batch (limit / 2)
+      //
+      // the default, and minimum, chunk size is 5mb. we can control the
+      // chunk size via getChunkSize(file), so we may want to increase
+      // the chunk size for larger files
+      limit: 10,
+
+      createMultipartUpload(file) {
+        return ajax("/uploads/create-multipart.json", {
+          type: "POST",
+          data: {
+            file_name: file.name,
+            file_size: file.size,
+            upload_type: file.meta.upload_type,
+          },
+          // uppy is inconsistent, an error here fires the upload-error event
+        }).then((data) => {
+          file.meta.unique_identifier = data.unique_identifier;
+          return {
+            uploadId: data.external_upload_identifier,
+            key: data.key,
+          };
+        });
+      },
+
+      prepareUploadParts(file, partData) {
+        return (
+          ajax("/uploads/batch-presign-multipart-parts.json", {
+            type: "POST",
+            data: {
+              part_numbers: partData.partNumbers,
+              unique_identifier: file.meta.unique_identifier,
+            },
+          })
+            .then((data) => {
+              return { presignedUrls: data.presigned_urls };
+            })
+            // uppy is inconsistent, an error here does not fire the upload-error event
+            .catch((err) => {
+              self._handleUploadError(file, err);
+            })
+        );
+      },
+
+      completeMultipartUpload(file, data) {
+        const parts = data.parts.map((part) => {
+          return { part_number: part.PartNumber, etag: part.ETag };
+        });
+        return ajax("/uploads/complete-multipart.json", {
+          type: "POST",
+          contentType: "application/json",
+          data: JSON.stringify({
+            parts,
+            unique_identifier: file.meta.unique_identifier,
+          }),
+          // uppy is inconsistent, an error here fires the upload-error event
+        }).then((responseData) => {
+          return responseData;
+        });
+      },
+
+      abortMultipartUpload(file, { key, uploadId }) {
+        // if the user cancels the upload before the key and uploadId
+        // are stored from the createMultipartUpload response then they
+        // will not be set, and we don't have to abort the upload because
+        // it will not exist yet
+        if (!key || !uploadId) {
+          return;
+        }
+
+        return ajax("/uploads/abort-multipart.json", {
+          type: "POST",
+          data: {
+            external_upload_identifier: uploadId,
+          },
+          // uppy is inconsistent, an error here does not fire the upload-error event
+        }).catch((err) => {
+          self._handleUploadError(file, err);
+        });
+      },
+
+      // we will need a listParts function at some point when we want to
+      // resume multipart uploads; this is used by uppy to figure out
+      // what parts are uploaded and which still need to be
+    });
+  },
+
   _reset() {
     this._uppyInstance?.reset();
     this.setProperties({

app/assets/javascripts/discourse/app/mixins/uppy-upload.js

@@ -175,7 +175,11 @@ export default Mixin.create({
     this.set("usingS3Uploads", true);
     this._uppyInstance.use(AwsS3, {
       getUploadParameters: (file) => {
-        const data = { file_name: file.name, type: this.type };
+        const data = {
+          file_name: file.name,
+          file_size: file.size,
+          type: this.type,
+        };
 
         // the sha1 checksum is set by the UppyChecksum plugin, except
         // for in cases where the browser does not support the required

app/controllers/uploads_controller.rb

@@ -9,14 +9,30 @@ class UploadsController < ApplicationController
   protect_from_forgery except: :show
 
   before_action :is_asset_path, :apply_cdn_headers, only: [:show, :show_short, :show_secure]
-  before_action :external_store_check, only: [:show_secure, :generate_presigned_put, :complete_external_upload]
+  before_action :external_store_check, only: [
+    :show_secure,
+    :generate_presigned_put,
+    :complete_external_upload,
+    :create_multipart,
+    :batch_presign_multipart_parts,
+    :abort_multipart,
+    :complete_multipart
+  ]
+  before_action :direct_s3_uploads_check, only: [
+    :generate_presigned_put,
+    :complete_external_upload,
+    :create_multipart,
+    :batch_presign_multipart_parts,
+    :abort_multipart,
+    :complete_multipart
+  ]
+  before_action :can_upload_external?, only: [:create_multipart, :generate_presigned_put]
 
   SECURE_REDIRECT_GRACE_SECONDS = 5
-  PRESIGNED_PUT_RATE_LIMIT_PER_MINUTE = 5
+  PRESIGNED_PUT_RATE_LIMIT_PER_MINUTE = 10
-
+  CREATE_MULTIPART_RATE_LIMIT_PER_MINUTE = 10
-  def external_store_check
+  COMPLETE_MULTIPART_RATE_LIMIT_PER_MINUTE = 10
-    return render_404 if !Discourse.store.external?
+  BATCH_PRESIGN_RATE_LIMIT_PER_MINUTE = 10
-  end
 
   def create
     # capture current user for block later on
@@ -193,15 +209,21 @@ class UploadsController < ApplicationController
   end
 
   def generate_presigned_put
-    return render_404 if !SiteSetting.enable_direct_s3_uploads
-
     RateLimiter.new(
       current_user, "generate-presigned-put-upload-stub", PRESIGNED_PUT_RATE_LIMIT_PER_MINUTE, 1.minute
     ).performed!
 
     file_name = params.require(:file_name)
+    file_size = params.require(:file_size).to_i
     type = params.require(:type)
 
+    if file_size_too_big?(file_name, file_size)
+      return render_json_error(
+        I18n.t("upload.attachments.too_large", max_size_kb: SiteSetting.max_attachment_size_kb),
+        status: 422
+      )
+    end
+
     # don't want people posting arbitrary S3 metadata so we just take the
     # one we need. all of these will be converted to x-amz-meta- metadata
     # fields in S3 so it's best to use dashes in the names for consistency
@@ -225,33 +247,37 @@ class UploadsController < ApplicationController
       key: key,
       created_by: current_user,
       original_filename: file_name,
-      upload_type: type
+      upload_type: type,
+      filesize: file_size
     )
 
     render json: { url: url, key: key, unique_identifier: upload_stub.unique_identifier }
   end
 
   def complete_external_upload
-    return render_404 if !SiteSetting.enable_direct_s3_uploads
-
     unique_identifier = params.require(:unique_identifier)
     external_upload_stub = ExternalUploadStub.find_by(
       unique_identifier: unique_identifier, created_by: current_user
     )
     return render_404 if external_upload_stub.blank?
 
-    raise Discourse::InvalidAccess if external_upload_stub.created_by_id != current_user.id
+    complete_external_upload_via_manager(external_upload_stub)
-    external_upload_manager = ExternalUploadManager.new(external_upload_stub)
+  end
 
+  def complete_external_upload_via_manager(external_upload_stub)
+    external_upload_manager = ExternalUploadManager.new(external_upload_stub)
     hijack do
       begin
         upload = external_upload_manager.promote_to_upload!
         if upload.errors.empty?
-          external_upload_manager.destroy!
+          external_upload_stub.destroy!
           render json: UploadsController.serialize_upload(upload), status: 200
         else
           render_json_error(upload.errors.to_hash.values.flatten, status: 422)
         end
+      rescue ExternalUploadManager::SizeMismatchError => err
+        debug_upload_error(err, "upload.size_mismatch_failure", additional_detail: err.message)
+        render_json_error(I18n.t("upload.failed"), status: 422)
       rescue ExternalUploadManager::ChecksumMismatchError => err
         debug_upload_error(err, "upload.checksum_mismatch_failure")
         render_json_error(I18n.t("upload.failed"), status: 422)
@@ -270,6 +296,179 @@ class UploadsController < ApplicationController
     end
   end
 
+  def create_multipart
+    RateLimiter.new(
+      current_user, "create-multipart-upload", CREATE_MULTIPART_RATE_LIMIT_PER_MINUTE, 1.minute
+    ).performed!
+
+    file_name = params.require(:file_name)
+    file_size = params.require(:file_size).to_i
+    upload_type = params.require(:upload_type)
+    content_type = MiniMime.lookup_by_filename(file_name)&.content_type
+
+    if file_size_too_big?(file_name, file_size)
+      return render_json_error(
+        I18n.t("upload.attachments.too_large", max_size_kb: SiteSetting.max_attachment_size_kb),
+        status: 422
+      )
+    end
+
+    begin
+      multipart_upload = Discourse.store.create_multipart(
+        file_name, content_type
+      )
+    rescue Aws::S3::Errors::ServiceError => err
+      debug_upload_error(err, "upload.create_mutlipart_failure")
+      return render_json_error(I18n.t("upload.failed"), status: 422)
+    end
+
+    upload_stub = ExternalUploadStub.create!(
+      key: multipart_upload[:key],
+      created_by: current_user,
+      original_filename: file_name,
+      upload_type: upload_type,
+      external_upload_identifier: multipart_upload[:upload_id],
+      multipart: true,
+      filesize: file_size
+    )
+
+    render json: {
+      external_upload_identifier: upload_stub.external_upload_identifier,
+      key: upload_stub.key,
+      unique_identifier: upload_stub.unique_identifier
+    }
+  end
+
+  def batch_presign_multipart_parts
+    part_numbers = params.require(:part_numbers)
+    unique_identifier = params.require(:unique_identifier)
+
+    RateLimiter.new(
+      current_user, "batch-presign", BATCH_PRESIGN_RATE_LIMIT_PER_MINUTE, 1.minute
+    ).performed!
+
+    part_numbers = part_numbers.map do |part_number|
+      validate_part_number(part_number)
+    end
+
+    external_upload_stub = ExternalUploadStub.find_by(
+      unique_identifier: unique_identifier, created_by: current_user
+    )
+    return render_404 if external_upload_stub.blank?
+
+    if !multipart_upload_exists?(external_upload_stub)
+      return render_404
+    end
+
+    presigned_urls = {}
+    part_numbers.each do |part_number|
+      presigned_urls[part_number] = Discourse.store.presign_multipart_part(
+        upload_id: external_upload_stub.external_upload_identifier,
+        key: external_upload_stub.key,
+        part_number: part_number
+      )
+    end
+
+    render json: { presigned_urls: presigned_urls }
+  end
+
+  def validate_part_number(part_number)
+    part_number = part_number.to_i
+    if !part_number.between?(1, 10000)
+      raise Discourse::InvalidParameters.new(
+        "Each part number should be between 1 and 10000"
+      )
+    end
+    part_number
+  end
+
+  def multipart_upload_exists?(external_upload_stub)
+    begin
+      Discourse.store.list_multipart_parts(
+        upload_id: external_upload_stub.external_upload_identifier, key: external_upload_stub.key
+      )
+    rescue Aws::S3::Errors::NoSuchUpload => err
+      debug_upload_error(err, "upload.external_upload_not_found", { additional_detail: "path: #{external_upload_stub.key}" })
+      return false
+    end
+    true
+  end
+
+  def abort_multipart
+    external_upload_identifier = params.require(:external_upload_identifier)
+    external_upload_stub = ExternalUploadStub.find_by(
+      external_upload_identifier: external_upload_identifier
+    )
+
+    # The stub could have already been deleted by an earlier error via
+    # ExternalUploadManager, so we consider this a great success if the
+    # stub is already gone.
+    return render json: success_json if external_upload_stub.blank?
+
+    return render_404 if external_upload_stub.created_by_id != current_user.id
+
+    begin
+      Discourse.store.abort_multipart(
+        upload_id: external_upload_stub.external_upload_identifier,
+        key: external_upload_stub.key
+      )
+    rescue Aws::S3::Errors::ServiceError => err
+      debug_upload_error(err, "upload.abort_mutlipart_failure", additional_detail: "external upload stub id: #{external_upload_stub.id}")
+      return render_json_error(I18n.t("upload.failed"), status: 422)
+    end
+
+    external_upload_stub.destroy!
+
+    render json: success_json
+  end
+
+  def complete_multipart
+    unique_identifier = params.require(:unique_identifier)
+    parts = params.require(:parts)
+
+    RateLimiter.new(
+      current_user, "complete-multipart-upload", COMPLETE_MULTIPART_RATE_LIMIT_PER_MINUTE, 1.minute
+    ).performed!
+
+    external_upload_stub = ExternalUploadStub.find_by(
+      unique_identifier: unique_identifier, created_by: current_user
+    )
+    return render_404 if external_upload_stub.blank?
+
+    if !multipart_upload_exists?(external_upload_stub)
+      return render_404
+    end
+
+    parts = parts.map do |part|
+      part_number = part[:part_number]
+      etag = part[:etag]
+      part_number = validate_part_number(part_number)
+
+      if etag.blank?
+        raise Discourse::InvalidParameters.new("All parts must have an etag and a valid part number")
+      end
+
+      # this is done so it's an array of hashes rather than an array of
+      # ActionController::Parameters
+      { part_number: part_number, etag: etag }
+    end.sort_by do |part|
+      part[:part_number]
+    end
+
+    begin
+      complete_response = Discourse.store.complete_multipart(
+        upload_id: external_upload_stub.external_upload_identifier,
+        key: external_upload_stub.key,
+        parts: parts
+      )
+    rescue Aws::S3::Errors::ServiceError => err
+      debug_upload_error(err, "upload.complete_mutlipart_failure", additional_detail: "external upload stub id: #{external_upload_stub.id}")
+      return render_json_error(I18n.t("upload.failed"), status: 422)
+    end
+
+    complete_external_upload_via_manager(external_upload_stub)
+  end
+
   protected
 
   def force_download?
@@ -339,6 +538,25 @@ class UploadsController < ApplicationController
 
   private
 
+  def external_store_check
+    return render_404 if !Discourse.store.external?
+  end
+
+  def direct_s3_uploads_check
+    return render_404 if !SiteSetting.enable_direct_s3_uploads
+  end
+
+  def can_upload_external?
+    raise Discourse::InvalidAccess if !guardian.can_upload_external?
+  end
+
+  # We can pre-emptively check size for attachments, but not for images
+  # as they may be further reduced in size by UploadCreator (at this point
+  # they may have already been reduced in size by preprocessors)
+  def file_size_too_big?(file_name, file_size)
+    !FileHelper.is_supported_image?(file_name) && file_size >= SiteSetting.max_attachment_size_kb.kilobytes
+  end
+
   def send_file_local_upload(upload)
     opts = {
       filename: upload.original_filename,
@@ -357,8 +575,8 @@ class UploadsController < ApplicationController
     send_file(file_path, opts)
   end
 
-  def debug_upload_error(translation_key, err)
+  def debug_upload_error(err, translation_key, translation_params = {})
     return if !SiteSetting.enable_upload_debug_mode
-    Discourse.warn_exception(err, message: I18n.t(translation_key))
+    Discourse.warn_exception(err, message: I18n.t(translation_key, translation_params))
   end
 end

app/models/external_upload_stub.rb

@@ -5,9 +5,14 @@ require "digest/sha1"
 class ExternalUploadStub < ActiveRecord::Base
   CREATED_EXPIRY_HOURS = 1
   UPLOADED_EXPIRY_HOURS = 24
+  FAILED_EXPIRY_HOURS = 48
 
   belongs_to :created_by, class_name: 'User'
 
+  validates :filesize, numericality: {
+    allow_nil: false, only_integer: true, greater_than_or_equal_to: 1
+  }
+
   scope :expired_created, -> {
     where(
       "status = ? AND created_at <= ?",
@@ -33,7 +38,6 @@ class ExternalUploadStub < ActiveRecord::Base
     @statuses ||= Enum.new(
       created: 1,
       uploaded: 2,
-      failed: 3
     )
   end
 
@@ -50,19 +54,23 @@ end
 #
 # Table name: external_upload_stubs
 #
-#  id                :bigint           not null, primary key
+#  id                         :bigint           not null, primary key
-#  key               :string           not null
+#  key                        :string           not null
-#  original_filename :string           not null
+#  original_filename          :string           not null
-#  status            :integer          default(1), not null
+#  status                     :integer          default(1), not null
-#  unique_identifier :uuid             not null
+#  unique_identifier          :uuid             not null
-#  created_by_id     :integer          not null
+#  created_by_id              :integer          not null
-#  upload_type       :string           not null
+#  upload_type                :string           not null
-#  created_at        :datetime         not null
+#  created_at                 :datetime         not null
-#  updated_at        :datetime         not null
+#  updated_at                 :datetime         not null
+#  multipart                  :boolean          default(FALSE), not null
+#  external_upload_identifier :string
+#  filesize                   :bigint           not null
 #
 # Indexes
 #
-#  index_external_upload_stubs_on_created_by_id  (created_by_id)
+#  index_external_upload_stubs_on_created_by_id               (created_by_id)
-#  index_external_upload_stubs_on_key            (key) UNIQUE
+#  index_external_upload_stubs_on_external_upload_identifier  (external_upload_identifier)
-#  index_external_upload_stubs_on_status         (status)
+#  index_external_upload_stubs_on_key                         (key) UNIQUE
+#  index_external_upload_stubs_on_status                      (status)
 #

app/services/external_upload_manager.rb

@@ -2,13 +2,24 @@
 
 class ExternalUploadManager
   DOWNLOAD_LIMIT = 100.megabytes
+  SIZE_MISMATCH_BAN_MINUTES = 5
+  BAN_USER_REDIS_PREFIX = "ban_user_from_external_uploads_"
 
   class ChecksumMismatchError < StandardError; end
   class DownloadFailedError < StandardError; end
   class CannotPromoteError < StandardError; end
+  class SizeMismatchError < StandardError; end
 
   attr_reader :external_upload_stub
 
+  def self.ban_user_from_external_uploads!(user:, ban_minutes: 5)
+    Discourse.redis.setex("#{BAN_USER_REDIS_PREFIX}#{user.id}", ban_minutes.minutes.to_i, "1")
+  end
+
+  def self.user_banned?(user)
+    Discourse.redis.get("#{BAN_USER_REDIS_PREFIX}#{user.id}") == "1"
+  end
+
   def initialize(external_upload_stub)
     @external_upload_stub = external_upload_stub
   end
@@ -31,6 +42,19 @@ class ExternalUploadManager
     # variable as well to check.
     tempfile = nil
     should_download = external_size < DOWNLOAD_LIMIT
+
+    # We require that the file size is specified ahead of time, and compare
+    # it here to make sure that people are not uploading excessively large
+    # files to the external provider. If this happens, the user will be banned
+    # from uploading to the external provider for N minutes.
+    if external_size != external_upload_stub.filesize
+      ExternalUploadManager.ban_user_from_external_uploads!(
+        user: external_upload_stub.created_by,
+        ban_minutes: SIZE_MISMATCH_BAN_MINUTES
+      )
+      raise SizeMismatchError.new("expected: #{external_upload_stub.filesize}, actual: #{external_size}")
+    end
+
     if should_download
       tempfile = download(external_upload_stub.key, external_upload_stub.upload_type)
 
@@ -60,16 +84,17 @@ class ExternalUploadManager
       external_upload_stub.created_by_id
     )
   rescue
-    external_upload_stub.update!(status: ExternalUploadStub.statuses[:failed])
+    # We don't need to do anything special to abort multipart uploads here,
+    # because at this point (calling promote_to_upload!), the multipart
+    # upload would already be complete.
+    Discourse.store.delete_file(external_upload_stub.key)
+    external_upload_stub.destroy!
+
     raise
   ensure
     tempfile&.close!
   end
 
-  def destroy!
-    external_upload_stub.destroy!
-  end
-
   private
 
   def download(key, type)

config/locales/server.en.yml

@@ -4009,6 +4009,11 @@ en:
     png_to_jpg_conversion_failure_message: "An error happened when converting from PNG to JPG."
     optimize_failure_message: "An error occurred while optimizing the uploaded image."
     download_failure: "Downloading the file from the external provider failed."
+    size_mismatch_failure: "The size of the file uploaded to S3 did not match the external upload stub's intended size. %{additional_detail}"
+    create_mutlipart_failure: "Failed to create multipart upload in the external store."
+    abort_mutlipart_failure: "Failed to abort multipart upload in the external store."
+    complete_mutlipart_failure: "Failed to complete multipart upload in the external store."
+    external_upload_not_found: "The upload was not found in the external store. %{additional_detail}"
     checksum_mismatch_failure: "The checksum of the file you uploaded does not match. The file contents may have changed on upload. Please try again."
     cannot_promote_failure: "The upload cannot be completed, it may have already completed or previously failed."
     attachments:

config/routes.rb

@@ -541,8 +541,15 @@ Discourse::Application.routes.draw do
     post "uploads" => "uploads#create"
     post "uploads/lookup-urls" => "uploads#lookup_urls"
 
-    post "uploads/generate-presigned-put" => "uploads#generate_presigned_put"
+    # direct to s3 uploads
-    post "uploads/complete-external-upload" => "uploads#complete_external_upload"
+    post "uploads/generate-presigned-put" => "uploads#generate_presigned_put", format: :json
+    post "uploads/complete-external-upload" => "uploads#complete_external_upload", format: :json
+
+    # multipart uploads
+    post "uploads/create-multipart" => "uploads#create_multipart", format: :json
+    post "uploads/complete-multipart" => "uploads#complete_multipart", format: :json
+    post "uploads/abort-multipart" => "uploads#abort_multipart", format: :json
+    post "uploads/batch-presign-multipart-parts" => "uploads#batch_presign_multipart_parts", format: :json
 
     # used to download original images
     get "uploads/:site/:sha(.:extension)" => "uploads#show", constraints: { site: /\w+/, sha: /\h{40}/, extension: /[a-z0-9\._]+/i }

db/migrate/20210812033033_add_multipart_and_size_columns_to_external_upload_stubs.rb

@@ -0,0 +1,23 @@
+# frozen_string_literal: true
+
+class AddMultipartAndSizeColumnsToExternalUploadStubs < ActiveRecord::Migration[6.1]
+  def up
+    add_column :external_upload_stubs, :multipart, :boolean, default: false, null: false
+    add_column :external_upload_stubs, :external_upload_identifier, :string, null: true
+    add_column :external_upload_stubs, :filesize, :bigint
+
+    add_index :external_upload_stubs, :external_upload_identifier
+
+    # this feature is not actively used yet so this will be safe, also the rows in this
+    # table are regularly deleted
+    DB.exec("UPDATE external_upload_stubs SET filesize = 0 WHERE filesize IS NULL")
+
+    change_column_null :external_upload_stubs, :filesize, false
+  end
+
+  def down
+    remove_column :external_upload_stubs, :multipart
+    remove_column :external_upload_stubs, :external_upload_identifier
+    remove_column :external_upload_stubs, :filesize
+  end
+end

lib/file_store/s3_store.rb

@@ -97,12 +97,13 @@ module FileStore
 
       # if this fails, it will throw an exception
       if opts[:move_existing] && opts[:existing_external_upload_key]
+        original_path = opts[:existing_external_upload_key]
         path, etag = s3_helper.copy(
-          opts[:existing_external_upload_key],
+          original_path,
           path,
           options: options
         )
-        s3_helper.delete_object(opts[:existing_external_upload_key])
+        delete_file(original_path)
       else
         path, etag = s3_helper.upload(file, path, options)
       end
@@ -111,6 +112,12 @@ module FileStore
       [File.join(absolute_base_url, path), etag]
     end
 
+    def delete_file(path)
+      # delete the object outright without moving to tombstone,
+      # not recommended for most use cases
+      s3_helper.delete_object(path)
+    end
+
     def remove_file(url, path)
       return unless has_been_uploaded?(url)
       # copy the removed file to tombstone
@@ -217,7 +224,15 @@ module FileStore
 
     def signed_url_for_temporary_upload(file_name, expires_in: S3Helper::UPLOAD_URL_EXPIRES_AFTER_SECONDS, metadata: {})
       key = temporary_upload_path(file_name)
-      presigned_put_url(key, expires_in: expires_in, metadata: metadata)
+      presigned_url(
+        key,
+        method: :put_object,
+        expires_in: expires_in,
+        opts: {
+          metadata: metadata,
+          acl: "private"
+        }
+      )
     end
 
     def temporary_upload_path(file_name)
@@ -297,17 +312,72 @@ module FileStore
       FileUtils.mv(old_upload_path, public_upload_path) if old_upload_path
     end
 
-    private
+    def abort_multipart(key:, upload_id:)
-
+      s3_helper.s3_client.abort_multipart_upload(
-    def presigned_put_url(key, expires_in: S3Helper::UPLOAD_URL_EXPIRES_AFTER_SECONDS, metadata: {})
-      signer = Aws::S3::Presigner.new(client: s3_helper.s3_client)
-      signer.presigned_url(
-        :put_object,
         bucket: s3_bucket_name,
         key: key,
+        upload_id: upload_id
+      )
+    end
+
+    def create_multipart(file_name, content_type)
+      key = temporary_upload_path(file_name)
+      response = s3_helper.s3_client.create_multipart_upload(
         acl: "private",
-        expires_in: expires_in,
+        bucket: s3_bucket_name,
-        metadata: metadata
+        key: key,
+        content_type: content_type
+      )
+      { upload_id: response.upload_id, key: key }
+    end
+
+    def presign_multipart_part(upload_id:, key:, part_number:)
+      presigned_url(
+        key,
+        method: :upload_part,
+        expires_in: S3Helper::UPLOAD_URL_EXPIRES_AFTER_SECONDS,
+        opts: {
+          part_number: part_number,
+          upload_id: upload_id
+        }
+      )
+    end
+
+    def list_multipart_parts(upload_id:, key:)
+      s3_helper.s3_client.list_parts(
+        bucket: s3_bucket_name,
+        key: key,
+        upload_id: upload_id
+      )
+    end
+
+    def complete_multipart(upload_id:, key:, parts:)
+      s3_helper.s3_client.complete_multipart_upload(
+        bucket: s3_bucket_name,
+        key: key,
+        upload_id: upload_id,
+        multipart_upload: {
+          parts: parts
+        }
+      )
+    end
+
+    private
+
+    def presigned_url(
+      key,
+      method:,
+      expires_in: S3Helper::UPLOAD_URL_EXPIRES_AFTER_SECONDS,
+      opts: {}
+    )
+      signer = Aws::S3::Presigner.new(client: s3_helper.s3_client)
+      signer.presigned_url(
+        method,
+        {
+          bucket: s3_bucket_name,
+          key: key,
+          expires_in: expires_in,
+        }.merge(opts)
       )
     end
 

lib/guardian/user_guardian.rb

@@ -176,6 +176,10 @@ module UserGuardian
     (is_me?(user) && user.has_trust_level?(SiteSetting.min_trust_level_to_allow_user_card_background.to_i)) || is_staff?
   end
 
+  def can_upload_external?
+    !ExternalUploadManager.user_banned?(user)
+  end
+
   def can_delete_sso_record?(user)
     SiteSetting.enable_discourse_connect && user && is_admin?
   end

lib/upload_creator.rb

@@ -32,6 +32,9 @@ class UploadCreator
     @opts = opts
     @filesize = @opts[:filesize] if @opts[:external_upload_too_big]
     @opts[:validate] = opts[:skip_validations].present? ? !ActiveRecord::Type::Boolean.new.cast(opts[:skip_validations]) : true
+
+    # TODO (martin) Validate @opts[:type] to make sure only blessed types are passed
+    # in, since the clientside can pass any type it wants.
   end
 
   def create_for(user_id)
@@ -50,6 +53,11 @@ class UploadCreator
     # so we have not downloaded it to a tempfile. no modifications can be made to the
     # file in this case because it does not exist; we simply move it to its new location
     # in S3
+    #
+    # TODO (martin) I've added a bunch of external_upload_too_big checks littered
+    # throughout the UploadCreator code. It would be better to have two seperate
+    # classes with shared methods, rather than doing all these checks all over the
+    # place. Needs a refactor.
     external_upload_too_big = @opts[:external_upload_too_big]
     sha1_before_changes = Upload.generate_digest(@file) if @file
 

spec/components/guardian/user_guardian_spec.rb

@@ -492,4 +492,17 @@ describe UserGuardian do
       end
     end
   end
+
+  describe "#can_upload_external?" do
+    after { Discourse.redis.flushdb }
+
+    it "is true by default" do
+      expect(Guardian.new(user).can_upload_external?).to eq(true)
+    end
+
+    it "is false if the user has been banned from external uploads for a time period" do
+      ExternalUploadManager.ban_user_from_external_uploads!(user: user)
+      expect(Guardian.new(user).can_upload_external?).to eq(false)
+    end
+  end
 end

spec/fabricators/external_upload_stub_fabricator.rb

@@ -5,15 +5,18 @@ Fabricator(:external_upload_stub) do
   original_filename "test.txt"
   key { Discourse.store.temporary_upload_path("test.txt") }
   upload_type "card_background"
+  filesize 1024
   status 1
 end
 
 Fabricator(:image_external_upload_stub, from: :external_upload_stub) do
   original_filename "logo.png"
+  filesize 1024
   key { Discourse.store.temporary_upload_path("logo.png") }
 end
 
 Fabricator(:attachment_external_upload_stub, from: :external_upload_stub) do
   original_filename "file.pdf"
+  filesize 1024
   key { Discourse.store.temporary_upload_path("file.pdf") }
 end

spec/requests/uploads_controller_spec.rb

@@ -721,7 +721,9 @@ describe UploadsController do
       end
 
       it "generates a presigned URL and creates an external upload stub" do
-        post "/uploads/generate-presigned-put.json", params: { file_name: "test.png", type: "card_background" }
+        post "/uploads/generate-presigned-put.json", params: {
+          file_name: "test.png", type: "card_background", file_size: 1024
+        }
         expect(response.status).to eq(200)
 
         result = response.parsed_body
@@ -730,7 +732,8 @@ describe UploadsController do
           unique_identifier: result["unique_identifier"],
           original_filename: "test.png",
           created_by: user,
-          upload_type: "card_background"
+          upload_type: "card_background",
+          filesize: 1024
         )
         expect(external_upload_stub.exists?).to eq(true)
         expect(result["key"]).to include(FileStore::S3Store::TEMPORARY_UPLOAD_PREFIX)
@@ -742,6 +745,7 @@ describe UploadsController do
         post "/uploads/generate-presigned-put.json", {
           params: {
             file_name: "test.png",
+            file_size: 1024,
             type: "card_background",
             metadata: {
               "sha1-checksum" => "testing",
@@ -761,8 +765,8 @@ describe UploadsController do
         RateLimiter.clear_all!
 
         stub_const(UploadsController, "PRESIGNED_PUT_RATE_LIMIT_PER_MINUTE", 1) do
-          post "/uploads/generate-presigned-put.json", params: { file_name: "test.png", type: "card_background" }
+          post "/uploads/generate-presigned-put.json", params: { file_name: "test.png", type: "card_background", file_size: 1024 }
-          post "/uploads/generate-presigned-put.json", params: { file_name: "test.png", type: "card_background" }
+          post "/uploads/generate-presigned-put.json", params: { file_name: "test.png", type: "card_background", file_size: 1024 }
         end
         expect(response.status).to eq(429)
       end
@@ -774,7 +778,566 @@ describe UploadsController do
       end
 
       it "returns 404" do
-        post "/uploads/generate-presigned-put.json", params: { file_name: "test.png", type: "card_background" }
+        post "/uploads/generate-presigned-put.json", params: { file_name: "test.png", type: "card_background", file_size: 1024 }
+        expect(response.status).to eq(404)
+      end
+    end
+  end
+
+  describe "#create_multipart" do
+    context "when the store is external" do
+      let(:mock_multipart_upload_id) { "ibZBv_75gd9r8lH_gqXatLdxMVpAlj6CFTR.OwyF3953YdwbcQnMA2BLGn8Lx12fQNICtMw5KyteFeHw.Sjng--" }
+
+      before do
+        sign_in(user)
+        SiteSetting.enable_direct_s3_uploads = true
+        setup_s3
+        FileStore::S3Store.any_instance.stubs(:temporary_upload_path).returns(
+          "uploads/default/test_0/temp/28fccf8259bbe75b873a2bd2564b778c/test.png"
+        )
+      end
+
+      it "errors if the correct params are not provided" do
+        post "/uploads/create-multipart.json", params: { file_name: "test.png" }
+        expect(response.status).to eq(400)
+        post "/uploads/create-multipart.json", params: { upload_type: "composer" }
+        expect(response.status).to eq(400)
+        post "/uploads/create-multipart.json", params: { content_type: "image/jpeg" }
+        expect(response.status).to eq(400)
+      end
+
+      it "returns 422 when the create request errors" do
+        FileStore::S3Store.any_instance.stubs(:create_multipart).raises(Aws::S3::Errors::ServiceError.new({}, "test"))
+        post "/uploads/create-multipart.json", {
+          params: {
+            file_name: "test.png",
+            file_size: 1024,
+            upload_type: "composer",
+            content_type: "image/png"
+          }
+        }
+        expect(response.status).to eq(422)
+      end
+
+      it "returns 422 when the file is an attachment and it's too big" do
+        SiteSetting.max_attachment_size_kb = 1000
+        post "/uploads/create-multipart.json", {
+          params: {
+            file_name: "test.zip",
+            file_size: 9999999,
+            upload_type: "composer",
+            content_type: "application/zip"
+          }
+        }
+        expect(response.status).to eq(422)
+        expect(response.body).to include(I18n.t("upload.attachments.too_large", max_size_kb: SiteSetting.max_attachment_size_kb))
+      end
+
+      def stub_create_multipart_request
+        create_multipart_result = <<~BODY
+        <?xml version=\"1.0\" encoding=\"UTF-8\"?>\n
+        <InitiateMultipartUploadResult>
+           <Bucket>s3-upload-bucket</Bucket>
+           <Key>uploads/default/test_0/temp/28fccf8259bbe75b873a2bd2564b778c/test.png</Key>
+           <UploadId>#{mock_multipart_upload_id}</UploadId>
+        </InitiateMultipartUploadResult>
+        BODY
+        stub_request(
+          :post,
+          "https://s3-upload-bucket.s3.us-west-1.amazonaws.com/uploads/default/test_0/temp/28fccf8259bbe75b873a2bd2564b778c/test.png?uploads"
+        ).to_return({ status: 200, body: create_multipart_result })
+      end
+
+      it "creates a multipart upload and creates an external upload stub that is marked as multipart" do
+        stub_create_multipart_request
+        post "/uploads/create-multipart.json", {
+          params: {
+            file_name: "test.png",
+            file_size: 1024,
+            upload_type: "composer",
+            content_type: "image/png"
+          }
+        }
+
+        expect(response.status).to eq(200)
+        result = response.parsed_body
+
+        external_upload_stub = ExternalUploadStub.where(
+          unique_identifier: result["unique_identifier"],
+          original_filename: "test.png",
+          created_by: user,
+          upload_type: "composer",
+          key: result["key"],
+          external_upload_identifier: mock_multipart_upload_id,
+          multipart: true,
+          filesize: 1024
+        )
+        expect(external_upload_stub.exists?).to eq(true)
+        expect(result["key"]).to include(FileStore::S3Store::TEMPORARY_UPLOAD_PREFIX)
+        expect(result["external_upload_identifier"]).to eq(mock_multipart_upload_id)
+        expect(result["key"]).to eq(external_upload_stub.last.key)
+      end
+
+      it "rate limits" do
+        RateLimiter.enable
+        RateLimiter.clear_all!
+
+        stub_create_multipart_request
+        stub_const(UploadsController, "CREATE_MULTIPART_RATE_LIMIT_PER_MINUTE", 1) do
+          post "/uploads/create-multipart.json", params: {
+            file_name: "test.png",
+            upload_type: "composer",
+            content_type: "image/png",
+            file_size: 1024
+          }
+          expect(response.status).to eq(200)
+
+          post "/uploads/create-multipart.json", params: {
+            file_name: "test.png",
+            upload_type: "composer",
+            content_type: "image/png",
+            file_size: 1024
+          }
+          expect(response.status).to eq(429)
+        end
+      end
+    end
+
+    context "when the store is not external" do
+      before do
+        sign_in(user)
+      end
+
+      it "returns 404" do
+        post "/uploads/create-multipart.json", params: {
+          file_name: "test.png",
+          upload_type: "composer",
+          content_type: "image/png",
+          file_size: 1024
+        }
+        expect(response.status).to eq(404)
+      end
+    end
+  end
+
+  describe "#batch_presign_multipart_parts" do
+    fab!(:mock_multipart_upload_id) { "ibZBv_75gd9r8lH_gqXatLdxMVpAlj6CFTR.OwyF3953YdwbcQnMA2BLGn8Lx12fQNICtMw5KyteFeHw.Sjng--" }
+    fab!(:external_upload_stub) do
+      Fabricate(:image_external_upload_stub, created_by: user, multipart: true, external_upload_identifier: mock_multipart_upload_id)
+    end
+
+    context "when the store is external" do
+      before do
+        sign_in(user)
+        SiteSetting.enable_direct_s3_uploads = true
+        setup_s3
+      end
+
+      def stub_list_multipart_request
+        list_multipart_result = <<~BODY
+        <?xml version=\"1.0\" encoding=\"UTF-8\"?>\n
+        <ListPartsResult>
+           <Bucket>s3-upload-bucket</Bucket>
+           <Key>#{external_upload_stub.key}</Key>
+           <UploadId>#{mock_multipart_upload_id}</UploadId>
+           <PartNumberMarker>0</PartNumberMarker>
+           <NextPartNumberMarker>0</NextPartNumberMarker>
+           <MaxParts>1</MaxParts>
+           <IsTruncated>false</IsTruncated>
+           <Part>
+              <ETag>test</ETag>
+              <LastModified>#{Time.zone.now}</LastModified>
+              <PartNumber>1</PartNumber>
+              <Size>#{5.megabytes}</Size>
+           </Part>
+           <Initiator>
+              <DisplayName>test-upload-user</DisplayName>
+              <ID>arn:aws:iam::123:user/test-upload-user</ID>
+           </Initiator>
+           <Owner>
+              <DisplayName></DisplayName>
+              <ID>12345</ID>
+           </Owner>
+           <StorageClass>STANDARD</StorageClass>
+        </ListPartsResult>
+        BODY
+        stub_request(:get, "https://s3-upload-bucket.s3.us-west-1.amazonaws.com/#{external_upload_stub.key}?uploadId=#{mock_multipart_upload_id}").to_return({ status: 200, body: list_multipart_result })
+      end
+
+      it "errors if the correct params are not provided" do
+        post "/uploads/batch-presign-multipart-parts.json", params: {}
+        expect(response.status).to eq(400)
+      end
+
+      it "errors if the part_numbers do not contain numbers between 1 and 10000" do
+        post "/uploads/batch-presign-multipart-parts.json", params: {
+          unique_identifier: external_upload_stub.unique_identifier,
+          part_numbers: [-1, 0, 1, 2, 3, 4]
+        }
+        expect(response.status).to eq(400)
+        expect(response.body).to include("You supplied invalid parameters to the request: Each part number should be between 1 and 10000")
+        post "/uploads/batch-presign-multipart-parts.json", params: {
+          unique_identifier: external_upload_stub.unique_identifier,
+          part_numbers: [3, 4, "blah"]
+        }
+        expect(response.status).to eq(400)
+        expect(response.body).to include("You supplied invalid parameters to the request: Each part number should be between 1 and 10000")
+      end
+
+      it "returns 404 when the upload stub does not exist" do
+        post "/uploads/batch-presign-multipart-parts.json", params: {
+          unique_identifier: "unknown",
+          part_numbers: [1, 2, 3]
+        }
+        expect(response.status).to eq(404)
+      end
+
+      it "returns 404 when the upload stub does not belong to the user" do
+        external_upload_stub.update!(created_by: Fabricate(:user))
+        post "/uploads/batch-presign-multipart-parts.json", params: {
+          unique_identifier: external_upload_stub.unique_identifier,
+          part_numbers: [1, 2, 3]
+        }
+        expect(response.status).to eq(404)
+      end
+
+      it "returns 404 when the multipart upload does not exist" do
+        FileStore::S3Store.any_instance.stubs(:list_multipart_parts).raises(Aws::S3::Errors::NoSuchUpload.new("test", "test"))
+        post "/uploads/batch-presign-multipart-parts.json", params: {
+          unique_identifier: external_upload_stub.unique_identifier,
+          part_numbers: [1, 2, 3]
+        }
+        expect(response.status).to eq(404)
+      end
+
+      it "returns an object with the presigned URLs with the part numbers as keys" do
+        stub_list_multipart_request
+        post "/uploads/batch-presign-multipart-parts.json", params: {
+          unique_identifier: external_upload_stub.unique_identifier,
+          part_numbers: [2, 3, 4]
+        }
+
+        expect(response.status).to eq(200)
+        result = response.parsed_body
+        expect(result["presigned_urls"].keys).to eq(["2", "3", "4"])
+        expect(result["presigned_urls"]["2"]).to include("?partNumber=2&uploadId=#{mock_multipart_upload_id}")
+        expect(result["presigned_urls"]["3"]).to include("?partNumber=3&uploadId=#{mock_multipart_upload_id}")
+        expect(result["presigned_urls"]["4"]).to include("?partNumber=4&uploadId=#{mock_multipart_upload_id}")
+      end
+
+      it "rate limits" do
+        RateLimiter.enable
+        RateLimiter.clear_all!
+
+        stub_const(UploadsController, "BATCH_PRESIGN_RATE_LIMIT_PER_MINUTE", 1) do
+          stub_list_multipart_request
+          post "/uploads/batch-presign-multipart-parts.json", params: {
+            unique_identifier: external_upload_stub.unique_identifier,
+            part_numbers: [1, 2, 3]
+          }
+
+          expect(response.status).to eq(200)
+
+          post "/uploads/batch-presign-multipart-parts.json", params: {
+            unique_identifier: external_upload_stub.unique_identifier,
+            part_numbers: [1, 2, 3]
+          }
+
+          expect(response.status).to eq(429)
+        end
+      end
+    end
+
+    context "when the store is not external" do
+      before do
+        sign_in(user)
+      end
+
+      it "returns 404" do
+        post "/uploads/batch-presign-multipart-parts.json", params: {
+          unique_identifier: external_upload_stub.unique_identifier,
+          part_numbers: [1, 2, 3]
+        }
+        expect(response.status).to eq(404)
+      end
+    end
+  end
+
+  describe "#complete_multipart" do
+    let(:upload_base_url) { "https://#{SiteSetting.s3_upload_bucket}.s3.#{SiteSetting.s3_region}.amazonaws.com" }
+    let(:mock_multipart_upload_id) { "ibZBv_75gd9r8lH_gqXatLdxMVpAlj6CFTR.OwyF3953YdwbcQnMA2BLGn8Lx12fQNICtMw5KyteFeHw.Sjng--" }
+    let!(:external_upload_stub) do
+      Fabricate(:image_external_upload_stub, created_by: user, multipart: true, external_upload_identifier: mock_multipart_upload_id)
+    end
+
+    context "when the store is external" do
+      before do
+        sign_in(user)
+        SiteSetting.enable_direct_s3_uploads = true
+        setup_s3
+      end
+
+      def stub_list_multipart_request
+        list_multipart_result = <<~BODY
+        <?xml version=\"1.0\" encoding=\"UTF-8\"?>\n
+        <ListPartsResult>
+           <Bucket>s3-upload-bucket</Bucket>
+           <Key>#{external_upload_stub.key}</Key>
+           <UploadId>#{mock_multipart_upload_id}</UploadId>
+           <PartNumberMarker>0</PartNumberMarker>
+           <NextPartNumberMarker>0</NextPartNumberMarker>
+           <MaxParts>1</MaxParts>
+           <IsTruncated>false</IsTruncated>
+           <Part>
+              <ETag>test</ETag>
+              <LastModified>#{Time.zone.now}</LastModified>
+              <PartNumber>1</PartNumber>
+              <Size>#{5.megabytes}</Size>
+           </Part>
+           <Initiator>
+              <DisplayName>test-upload-user</DisplayName>
+              <ID>arn:aws:iam::123:user/test-upload-user</ID>
+           </Initiator>
+           <Owner>
+              <DisplayName></DisplayName>
+              <ID>12345</ID>
+           </Owner>
+           <StorageClass>STANDARD</StorageClass>
+        </ListPartsResult>
+        BODY
+        stub_request(:get, "#{upload_base_url}/#{external_upload_stub.key}?uploadId=#{mock_multipart_upload_id}").to_return({ status: 200, body: list_multipart_result })
+      end
+
+      it "errors if the correct params are not provided" do
+        post "/uploads/complete-multipart.json", params: {}
+        expect(response.status).to eq(400)
+      end
+
+      it "errors if the part_numbers do not contain numbers between 1 and 10000" do
+        stub_list_multipart_request
+        post "/uploads/complete-multipart.json", params: {
+          unique_identifier: external_upload_stub.unique_identifier,
+          parts: [{ part_number: -1, etag: "test1" }]
+        }
+        expect(response.status).to eq(400)
+        expect(response.body).to include("You supplied invalid parameters to the request: Each part number should be between 1 and 10000")
+        post "/uploads/complete-multipart.json", params: {
+          unique_identifier: external_upload_stub.unique_identifier,
+          parts: [{ part_number: 20001, etag: "test1" }]
+        }
+        expect(response.status).to eq(400)
+        expect(response.body).to include("You supplied invalid parameters to the request: Each part number should be between 1 and 10000")
+        post "/uploads/complete-multipart.json", params: {
+          unique_identifier: external_upload_stub.unique_identifier,
+          parts: [{ part_number: "blah", etag: "test1" }]
+        }
+        expect(response.status).to eq(400)
+        expect(response.body).to include("You supplied invalid parameters to the request: Each part number should be between 1 and 10000")
+      end
+
+      it "errors if any of the parts objects have missing values" do
+        stub_list_multipart_request
+        post "/uploads/complete-multipart.json", params: {
+          unique_identifier: external_upload_stub.unique_identifier,
+          parts: [{ part_number: 1 }]
+        }
+        expect(response.status).to eq(400)
+        expect(response.body).to include("All parts must have an etag")
+      end
+
+      it "returns 404 when the upload stub does not exist" do
+        post "/uploads/complete-multipart.json", params: {
+          unique_identifier: "unknown",
+          parts: [{ part_number: 1, etag: "test1" }]
+        }
+        expect(response.status).to eq(404)
+      end
+
+      it "returns 422 when the complete request errors" do
+        FileStore::S3Store.any_instance.stubs(:complete_multipart).raises(Aws::S3::Errors::ServiceError.new({}, "test"))
+        stub_list_multipart_request
+        post "/uploads/complete-multipart.json", params: {
+          unique_identifier: external_upload_stub.unique_identifier,
+          parts: [{ part_number: 1, etag: "test1" }]
+        }
+        expect(response.status).to eq(422)
+      end
+
+      it "returns 404 when the upload stub does not belong to the user" do
+        external_upload_stub.update!(created_by: Fabricate(:user))
+        post "/uploads/complete-multipart.json", params: {
+          unique_identifier: external_upload_stub.unique_identifier,
+          parts: [{ part_number: 1, etag: "test1" }]
+        }
+        expect(response.status).to eq(404)
+      end
+
+      it "returns 404 when the multipart upload does not exist" do
+        FileStore::S3Store.any_instance.stubs(:list_multipart_parts).raises(Aws::S3::Errors::NoSuchUpload.new("test", "test"))
+        post "/uploads/complete-multipart.json", params: {
+          unique_identifier: external_upload_stub.unique_identifier,
+          parts: [{ part_number: 1, etag: "test1" }]
+        }
+        expect(response.status).to eq(404)
+      end
+
+      it "completes the multipart upload, creates the Upload record, and returns a serialized Upload record" do
+        temp_location = "#{upload_base_url}/#{external_upload_stub.key}"
+        stub_list_multipart_request
+        stub_request(
+          :post,
+          "#{temp_location}?uploadId=#{external_upload_stub.external_upload_identifier}"
+        ).with(
+          body: "<CompleteMultipartUpload xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\">\n  <Part>\n    <ETag>test1</ETag>\n    <PartNumber>1</PartNumber>\n  </Part>\n  <Part>\n    <ETag>test2</ETag>\n    <PartNumber>2</PartNumber>\n  </Part>\n</CompleteMultipartUpload>\n"
+        ).to_return(status: 200, body: <<~XML)
+          <?xml version="1.0" encoding="UTF-8"?>
+          <CompleteMultipartUploadResult>
+             <Location>#{temp_location}</Location>
+             <Bucket>s3-upload-bucket</Bucket>
+             <Key>#{external_upload_stub.key}</Key>
+             <ETag>testfinal</ETag>
+          </CompleteMultipartUploadResult>
+        XML
+
+        # all the functionality for ExternalUploadManager is already tested along
+        # with stubs to S3 in its own test, we can just stub the response here
+        upload = Fabricate(:upload)
+        ExternalUploadManager.any_instance.stubs(:promote_to_upload!).returns(upload)
+
+        post "/uploads/complete-multipart.json", params: {
+          unique_identifier: external_upload_stub.unique_identifier,
+          parts: [{ part_number: 1, etag: "test1" }, { part_number: 2, etag: "test2" }]
+        }
+
+        expect(response.status).to eq(200)
+        result = response.parsed_body
+        expect(result[:upload]).to eq(JSON.parse(UploadSerializer.new(upload).to_json)[:upload])
+      end
+
+      it "rate limits" do
+        RateLimiter.enable
+        RateLimiter.clear_all!
+
+        stub_const(UploadsController, "COMPLETE_MULTIPART_RATE_LIMIT_PER_MINUTE", 1) do
+          post "/uploads/complete-multipart.json", params: {
+            unique_identifier: "blah",
+            parts: [{ part_number: 1, etag: "test1" }, { part_number: 2, etag: "test2" }]
+          }
+          post "/uploads/complete-multipart.json", params: {
+            unique_identifier: "blah",
+            parts: [{ part_number: 1, etag: "test1" }, { part_number: 2, etag: "test2" }]
+          }
+        end
+        expect(response.status).to eq(429)
+      end
+    end
+
+    context "when the store is not external" do
+      before do
+        sign_in(user)
+      end
+
+      it "returns 404" do
+        post "/uploads/complete-multipart.json", params: {
+          unique_identifier: external_upload_stub.external_upload_identifier,
+          parts: [
+            {
+              part_number: 1,
+              etag: "test1"
+            },
+            {
+              part_number: 2,
+              etag: "test2"
+            }
+          ]
+        }
+        expect(response.status).to eq(404)
+      end
+    end
+  end
+
+  describe "#abort_multipart" do
+    let(:upload_base_url) { "https://#{SiteSetting.s3_upload_bucket}.s3.#{SiteSetting.s3_region}.amazonaws.com" }
+    let(:mock_multipart_upload_id) { "ibZBv_75gd9r8lH_gqXatLdxMVpAlj6CFTR.OwyF3953YdwbcQnMA2BLGn8Lx12fQNICtMw5KyteFeHw.Sjng--" }
+    let!(:external_upload_stub) do
+      Fabricate(:image_external_upload_stub, created_by: user, multipart: true, external_upload_identifier: mock_multipart_upload_id)
+    end
+
+    context "when the store is external" do
+      before do
+        sign_in(user)
+        SiteSetting.enable_direct_s3_uploads = true
+        setup_s3
+      end
+
+      def stub_abort_request
+        temp_location = "#{upload_base_url}/#{external_upload_stub.key}"
+        stub_request(
+          :delete,
+          "#{temp_location}?uploadId=#{external_upload_stub.external_upload_identifier}"
+        ).to_return(status: 200, body: "")
+      end
+
+      it "errors if the correct params are not provided" do
+        post "/uploads/abort-multipart.json", params: {}
+        expect(response.status).to eq(400)
+      end
+
+      it "returns 200 when the stub does not exist, assumes it has already been deleted" do
+        FileStore::S3Store.any_instance.expects(:abort_multipart).never
+        post "/uploads/abort-multipart.json", params: {
+          external_upload_identifier: "unknown",
+        }
+        expect(response.status).to eq(200)
+      end
+
+      it "returns 404 when the upload stub does not belong to the user" do
+        external_upload_stub.update!(created_by: Fabricate(:user))
+        post "/uploads/abort-multipart.json", params: {
+          external_upload_identifier: external_upload_stub.external_upload_identifier
+        }
+        expect(response.status).to eq(404)
+      end
+
+      it "aborts the multipart upload and deletes the stub" do
+        stub_abort_request
+
+        post "/uploads/abort-multipart.json", params: {
+          external_upload_identifier: external_upload_stub.external_upload_identifier
+        }
+
+        expect(response.status).to eq(200)
+        expect(ExternalUploadStub.exists?(id: external_upload_stub.id)).to eq(false)
+      end
+
+      it "returns 422 when the abort request errors" do
+        FileStore::S3Store.any_instance.stubs(:abort_multipart).raises(Aws::S3::Errors::ServiceError.new({}, "test"))
+        post "/uploads/abort-multipart.json", params: {
+          external_upload_identifier: external_upload_stub.external_upload_identifier
+        }
+        expect(response.status).to eq(422)
+      end
+    end
+
+    context "when the store is not external" do
+      before do
+        sign_in(user)
+      end
+
+      it "returns 404" do
+        post "/uploads/complete-multipart.json", params: {
+          unique_identifier: external_upload_stub.external_upload_identifier,
+          parts: [
+            {
+              part_number: 1,
+              etag: "test1"
+            },
+            {
+              part_number: 2,
+              etag: "test2"
+            }
+          ]
+        }
         expect(response.status).to eq(404)
       end
     end
@@ -786,7 +1349,7 @@ describe UploadsController do
     end
 
     context "when the store is external" do
-      fab!(:external_upload_stub) { Fabricate(:external_upload_stub, created_by: user) }
+      fab!(:external_upload_stub) { Fabricate(:image_external_upload_stub, created_by: user) }
       let(:upload) { Fabricate(:upload) }
 
       before do
@@ -813,6 +1376,13 @@ describe UploadsController do
         expect(response.parsed_body["errors"].first).to eq(I18n.t("upload.failed"))
       end
 
+      it "handles SizeMismatchError" do
+        ExternalUploadManager.any_instance.stubs(:promote_to_upload!).raises(ExternalUploadManager::SizeMismatchError.new("expected: 10, actual: 1000"))
+        post "/uploads/complete-external-upload.json", params: { unique_identifier: external_upload_stub.unique_identifier }
+        expect(response.status).to eq(422)
+        expect(response.parsed_body["errors"].first).to eq(I18n.t("upload.failed"))
+      end
+
       it "handles CannotPromoteError" do
         ExternalUploadManager.any_instance.stubs(:promote_to_upload!).raises(ExternalUploadManager::CannotPromoteError)
         post "/uploads/complete-external-upload.json", params: { unique_identifier: external_upload_stub.unique_identifier }

spec/services/external_upload_manager_spec.rb

@@ -31,6 +31,15 @@ RSpec.describe ExternalUploadManager do
     stub_delete_object
   end
 
+  describe "#ban_user_from_external_uploads!" do
+    after { Discourse.redis.flushdb }
+
+    it "bans the user from external uploads using a redis key" do
+      ExternalUploadManager.ban_user_from_external_uploads!(user: user)
+      expect(ExternalUploadManager.user_banned?(user)).to eq(true)
+    end
+  end
+
   describe "#can_promote?" do
     it "returns false if the external stub status is not created" do
       external_upload_stub.update!(status: ExternalUploadStub.statuses[:uploaded])
@@ -40,7 +49,7 @@ RSpec.describe ExternalUploadManager do
 
   describe "#promote_to_upload!" do
     context "when stubbed upload is < DOWNLOAD_LIMIT (small enough to download + generate sha)" do
-      let!(:external_upload_stub) { Fabricate(:image_external_upload_stub, created_by: user) }
+      let!(:external_upload_stub) { Fabricate(:image_external_upload_stub, created_by: user, filesize: object_size) }
       let(:object_size) { 1.megabyte }
       let(:object_file) { logo_file }
 
@@ -114,18 +123,36 @@ RSpec.describe ExternalUploadManager do
         context "when the downloaded file sha1 does not match the client sha1" do
           let(:client_sha1) { "blahblah" }
 
-          it "raises an error and marks upload as failed" do
+          it "raises an error, deletes the stub" do
             expect { subject.promote_to_upload! }.to raise_error(ExternalUploadManager::ChecksumMismatchError)
-            expect(external_upload_stub.reload.status).to eq(ExternalUploadStub.statuses[:failed])
+            expect(ExternalUploadStub.exists?(id: external_upload_stub.id)).to eq(false)
           end
         end
       end
+
+      context "when the downloaded file size does not match the expected file size for the upload stub" do
+        before do
+          external_upload_stub.update!(filesize: 10)
+        end
+
+        after { Discourse.redis.flushdb }
+
+        it "raises an error, deletes the file immediately, and prevents the user from uploading external files for a few minutes" do
+          expect { subject.promote_to_upload! }.to raise_error(ExternalUploadManager::SizeMismatchError)
+          expect(ExternalUploadStub.exists?(id: external_upload_stub.id)).to eq(false)
+          expect(Discourse.redis.get("#{ExternalUploadManager::BAN_USER_REDIS_PREFIX}#{external_upload_stub.created_by_id}")).to eq("1")
+          expect(WebMock).to have_requested(
+            :delete,
+            "#{upload_base_url}/#{external_upload_stub.key}"
+          )
+        end
+      end
     end
 
     context "when stubbed upload is > DOWNLOAD_LIMIT (too big to download, generate a fake sha)" do
       let(:object_size) { 200.megabytes }
       let(:object_file) { pdf_file }
-      let!(:external_upload_stub) { Fabricate(:attachment_external_upload_stub, created_by: user) }
+      let!(:external_upload_stub) { Fabricate(:attachment_external_upload_stub, created_by: user, filesize: object_size) }
 
       before do
         UploadCreator.any_instance.stubs(:generate_fake_sha1_hash).returns("testbc60eb18e8f974cbfae8bb0f069c3a311024")