fix: enforce 5gb copy source object size threshold.

Fixes #1896

Enforces the S3 `5 GiB` copy source size limit across the posix and azure
backends for `CopyObject` and `UploadPartCopy`, returning `InvalidRequest` when
the source object exceeds the threshold.

The limit is now configurable via `--copy-object-threshold`
(`VGW_COPY_OBJECT_THRESHOLD`, default 5 GiB).
A new `--mp-max-parts flag` (`VGW_MP_MAX_PARTS`, default `10000`) has been added to make multipart upload parts number limit configurable.

No integration test has been added, as GitHub Actions cannot reliably
handle large objects.

backend/azure/azure.go

@@ -95,16 +95,17 @@ func (key) Table() map[string]struct{} {
 type Azure struct {
 	backend.BackendUnsupported
 
-	client         *azblob.Client
-	sharedkeyCreds *azblob.SharedKeyCredential
-	defaultCreds   *azidentity.DefaultAzureCredential
-	serviceURL     string
-	sasToken       string
+	client              *azblob.Client
+	sharedkeyCreds      *azblob.SharedKeyCredential
+	defaultCreds        *azidentity.DefaultAzureCredential
+	serviceURL          string
+	sasToken            string
+	copyObjectThreshold int64
 }
 
 var _ backend.Backend = &Azure{}
 
-func New(accountName, accountKey, serviceURL, sasToken string) (*Azure, error) {
+func New(accountName, accountKey, serviceURL, sasToken string, copyObjectThreshold int64) (*Azure, error) {
 	url := serviceURL
 	if serviceURL == "" && accountName != "" {
 		// if not otherwise specified, use the typical form:
@@ -117,7 +118,12 @@ func New(accountName, accountKey, serviceURL, sasToken string) (*Azure, error) {
 		if err != nil {
 			return nil, fmt.Errorf("init client: %w", err)
 		}
-		return &Azure{client: client, serviceURL: serviceURL, sasToken: sasToken}, nil
+		return &Azure{
+			client:              client,
+			serviceURL:          serviceURL,
+			sasToken:            sasToken,
+			copyObjectThreshold: copyObjectThreshold,
+		}, nil
 	}
 
 	if accountName == "" {
@@ -134,7 +140,12 @@ func New(accountName, accountKey, serviceURL, sasToken string) (*Azure, error) {
 		if err != nil {
 			return nil, fmt.Errorf("init client: %w", err)
 		}
-		return &Azure{client: client, serviceURL: url, defaultCreds: cred}, nil
+		return &Azure{
+			client:              client,
+			serviceURL:          url,
+			defaultCreds:        cred,
+			copyObjectThreshold: copyObjectThreshold,
+		}, nil
 	}
 
 	cred, err := azblob.NewSharedKeyCredential(accountName, accountKey)
@@ -147,7 +158,12 @@ func New(accountName, accountKey, serviceURL, sasToken string) (*Azure, error) {
 		return nil, fmt.Errorf("init client: %w", err)
 	}
 
-	return &Azure{client: client, serviceURL: url, sharedkeyCreds: cred}, nil
+	return &Azure{
+		client:              client,
+		serviceURL:          url,
+		sharedkeyCreds:      cred,
+		copyObjectThreshold: copyObjectThreshold,
+	}, nil
 }
 
 func (az *Azure) Shutdown() {}
@@ -959,6 +975,15 @@ func (az *Azure) CopyObject(ctx context.Context, input s3response.CopyObjectInpu
 			return s3response.CopyObjectOutput{}, s3err.GetAPIError(s3err.ErrInvalidCopyDest)
 		}
 
+		resp, err := dstClient.GetProperties(ctx, nil)
+		if err != nil {
+			return s3response.CopyObjectOutput{}, azureErrToS3Err(err)
+		}
+
+		if resp.ContentLength != nil && *resp.ContentLength > az.copyObjectThreshold {
+			return s3response.CopyObjectOutput{}, s3err.GetCopySourceObjectTooLargeErr(az.copyObjectThreshold)
+		}
+
 		// Set object meta http headers
 		res, err := dstClient.SetHTTPHeaders(ctx, blob.HTTPHeaders{
 			BlobCacheControl:       input.CacheControl,
@@ -1045,6 +1070,10 @@ func (az *Azure) CopyObject(ctx context.Context, input s3response.CopyObjectInpu
 	}
 	defer downloadResp.Body.Close()
 
+	if downloadResp.ContentLength != nil && *downloadResp.ContentLength > az.copyObjectThreshold {
+		return s3response.CopyObjectOutput{}, s3err.GetCopySourceObjectTooLargeErr(az.copyObjectThreshold)
+	}
+
 	pInput := s3response.PutObjectInput{
 		Body:                      downloadResp.Body,
 		Bucket:                    input.Bucket,

backend/posix/posix.go

@@ -98,6 +98,12 @@ type Posix struct {
 	// execute blocking syscalls (stat, readdir, xattr, open, etc.), this limiter
 	// constrains parallelism to prevent excessive thread creation under load.
 	actionLimiter *semaphore.Weighted
+
+	// copyObjectThreshold is the maximum allowed size (in bytes) for a copy
+	// source object. Requests to copy objects larger than this value are
+	// rejected with an 'InvalidRequest' to comply with the S3 limit
+	// of 5 GiB.
+	copyObjectThreshold int64
 }
 
 var _ backend.Backend = &Posix{}
@@ -172,6 +178,11 @@ type PosixOpts struct {
 	// queue depth grows under sustained load, request latency increases and
 	// upstream timeouts may occur.
 	Concurrency int
+	// CopyObjectThreshold sets the maximum allowed source object size (in bytes)
+	// for CopyObject and UploadPartCopy operations. Requests exceeding this
+	// threshold are rejected with an 'InvalidRequest' error. Defaults to the
+	// S3 specification limit of 5 GiB.
+	CopyObjectThreshold int64
 }
 
 func New(rootdir string, meta meta.MetadataStorer, opts PosixOpts) (*Posix, error) {
@@ -235,6 +246,7 @@ func New(rootdir string, meta meta.MetadataStorer, opts PosixOpts) (*Posix, erro
 		forceNoCopyFileRange: opts.ForceNoCopyFileRange,
 		validateBucketName:   opts.ValidateBucketNames,
 		actionLimiter:        semaphore.NewWeighted(int64(concurrencyOrDefault(opts.Concurrency))),
+		copyObjectThreshold:  opts.CopyObjectThreshold,
 	}, nil
 }
 
@@ -3084,6 +3096,9 @@ func (p *Posix) UploadPartCopy(ctx context.Context, upi *s3.UploadPartCopyInput)
 	if err != nil {
 		return s3response.CopyPartResult{}, err
 	}
+	if length > p.copyObjectThreshold {
+		return s3response.CopyPartResult{}, s3err.GetCopySourceObjectTooLargeErr(p.copyObjectThreshold)
+	}
 
 	srcf, err := os.Open(objPath)
 	if errors.Is(err, fs.ErrNotExist) {
@@ -4706,6 +4721,9 @@ func (p *Posix) CopyObject(ctx context.Context, input s3response.CopyObjectInput
 	if !strings.HasSuffix(srcObject, "/") && fi.IsDir() {
 		return s3response.CopyObjectOutput{}, s3err.GetAPIError(s3err.ErrNoSuchKey)
 	}
+	if fi.Size() > p.copyObjectThreshold {
+		return s3response.CopyObjectOutput{}, s3err.GetCopySourceObjectTooLargeErr(p.copyObjectThreshold)
+	}
 
 	b, err := p.meta.RetrieveAttribute(f, srcBucket, srcObject, etagkey)
 	srcEtag := string(b)

backend/scoutfs/scoutfs.go

@@ -45,6 +45,11 @@ type ScoutfsOpts struct {
 	// Concurrency sets the maximum number of concurrently running POSIX actions.
 	// Defaults to 5000 when unset or non-positive.
 	Concurrency int
+	// CopyObjectThreshold sets the maximum allowed source object size (in bytes)
+	// for CopyObject and UploadPartCopy operations. Requests exceeding this
+	// threshold are rejected with an 'InvalidRequest' error. Defaults to the
+	// S3 specification limit of 5 GiB.
+	CopyObjectThreshold int64
 }
 
 var _ backend.Backend = &ScoutFS{}

backend/scoutfs/scoutfs_compat.go

@@ -82,6 +82,7 @@ func New(rootdir string, opts ScoutfsOpts) (*ScoutFS, error) {
 		VersioningDir:       opts.VersioningDir,
 		ValidateBucketNames: opts.ValidateBucketNames,
 		Concurrency:         opts.Concurrency,
+		CopyObjectThreshold: opts.CopyObjectThreshold,
 	})
 	if err != nil {
 		return nil, err