Add support for UploadPartCopy on multipart part uploads

gofakes3.go

@@ -2,6 +2,7 @@ package gofakes3
 
 import (
 	"bytes"
+	"context"
 	"encoding/base64"
 	"encoding/hex"
 	"errors"
@@ -957,6 +958,18 @@ func (g *GoFakeS3) putMultipartUploadPart(bucket, object string, uploadID Upload
 		return ErrInvalidPart
 	}
 
+	// An UploadPartCopy request carries no body: the part is copied
+	// server-side from an existing object instead. Handle it before any
+	// body/size parsing, which would otherwise fail on a request without a
+	// Content-Length header.
+	if r.Header.Get("X-Amz-Copy-Source") != "" {
+		upload, err := g.uploader.Get(bucket, object, uploadID)
+		if err != nil {
+			return err
+		}
+		return g.copyMultipartUploadPart(upload, partNumber, r, w)
+	}
+
 	size, err := strconv.ParseInt(r.Header.Get("Content-Length"), 10, 64)
 	if err != nil {
 		return ErrMissingContentLength
@@ -1018,34 +1031,121 @@ func (g *GoFakeS3) putMultipartUploadPart(bucket, object string, uploadID Upload
 		}
 	}
 
+	etag, err := g.storeMultipartUploadPart(r.Context(), upload, int(partNumber), size, rdr)
+	if err != nil {
+		return err
+	}
+
+	w.Header().Add("ETag", etag)
+	return nil
+}
+
+// copyMultipartUploadPart implements UploadPartCopy: the part is copied
+// server-side from an existing object (optionally a range of it) into the
+// given multipart upload, instead of being streamed by the client.
+func (g *GoFakeS3) copyMultipartUploadPart(upload *multipartUpload, partNumber int64, r *http.Request, w http.ResponseWriter) (err error) {
+	source := r.Header.Get("X-Amz-Copy-Source")
+	g.log.Print(LogInfo, "copy multipart upload part", source, "TO", upload.Bucket, upload.Object)
+
+	srcBucket, srcKey, err := splitCopySource(source)
+	if err != nil {
+		return err
+	}
+
+	srcObj, err := g.storage.HeadObject(r.Context(), srcBucket, srcKey)
+	if err != nil {
+		return err
+	}
+
+	copyRange, err := copySourceRange(r.Header.Get("X-Amz-Copy-Source-Range"), srcObj.Size)
+	if err != nil {
+		return err
+	}
+
+	srcObj, err = g.storage.GetObject(r.Context(), srcBucket, srcKey, nil)
+	if err != nil {
+		return err
+	}
+	defer CheckClose(srcObj.Contents, &err)
+
+	size := srcObj.Size
+	rdr := io.Reader(srcObj.Contents)
+	if copyRange != nil {
+		size = copyRange.Length
+		if seeker, ok := srcObj.Contents.(io.Seeker); ok {
+			if _, err := seeker.Seek(copyRange.Start, io.SeekStart); err != nil {
+				return err
+			}
+		} else {
+			if _, err := io.CopyN(io.Discard, srcObj.Contents, copyRange.Start); err != nil {
+				return err
+			}
+		}
+	}
+	rdr = io.LimitReader(rdr, size)
+
+	etag, err := g.storeMultipartUploadPart(r.Context(), upload, int(partNumber), size, rdr)
+	if err != nil {
+		return err
+	}
+
+	return g.xmlEncoder(w).Encode(&CopyPartResult{
+		ETag:         etag,
+		LastModified: NewContentTime(g.timeSource.Now()),
+	})
+}
+
+// storeMultipartUploadPart stores a part either into the MultipartBackend
+// (streaming) or the in-memory uploader, returning the ETag of the stored part.
+func (g *GoFakeS3) storeMultipartUploadPart(ctx context.Context, upload *multipartUpload, partNumber int, size int64, rdr io.Reader) (string, error) {
 	if upload.streaming {
-		etag, err := g.multipart.UploadPart(r.Context(), bucket, object, uploadID, int(partNumber), size, rdr)
+		etag, err := g.multipart.UploadPart(ctx, upload.Bucket, upload.Object, upload.ID, partNumber, size, rdr)
 		if err != nil {
-			return err
+			return "", err
 		}
-		if err := upload.AddStreamingPart(int(partNumber), g.timeSource.Now(), size, etag); err != nil {
-			return err
+		if err := upload.AddStreamingPart(partNumber, g.timeSource.Now(), size, etag); err != nil {
+			return "", err
 		}
-		w.Header().Add("ETag", etag)
-		return nil
+		return etag, nil
 	}
 
 	body, err := ReadAll(rdr, size)
 	if err != nil {
-		return err
+		return "", err
 	}
-
 	if int64(len(body)) != size {
-		return ErrIncompleteBody
+		return "", ErrIncompleteBody
 	}
+	return upload.AddPart(partNumber, g.timeSource.Now(), body)
+}
 
-	etag, err := upload.AddPart(int(partNumber), g.timeSource.Now(), body)
+// splitCopySource parses an X-Amz-Copy-Source header into a bucket and key,
+// handling the leading slash and any versionId query subresource.
+func splitCopySource(source string) (bucket, key string, err error) {
+	parts := strings.SplitN(strings.TrimPrefix(source, "/"), "/", 2)
+	if len(parts) != 2 {
+		return "", "", ErrorMessage(ErrInvalidArgument, fmt.Sprintf("invalid copy source %q", source))
+	}
+	bucket = parts[0]
+	key = strings.SplitN(parts[1], "?", 2)[0]
+	key, err = url.QueryUnescape(key)
 	if err != nil {
-		return err
+		return "", "", err
 	}
+	return bucket, key, nil
+}
 
-	w.Header().Add("ETag", etag)
-	return nil
+// copySourceRange parses the optional X-Amz-Copy-Source-Range header and
+// resolves it against the source object size.
+func copySourceRange(header string, size int64) (*ObjectRange, error) {
+	if header == "" {
+		return nil, nil
+	}
+	req, err := parseRangeHeader(header)
+	if err != nil {
+		return nil, err
+	}
+	return req.Range(size)
 }
 
 // isChunkedStreamingPayload reports whether x-amz-content-sha256 denotes aws-chunked transfer encoding.

gofakes3_test.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"io"
 	"mime/multipart"
+	"net"
 	"net/http"
 	"net/http/httptest"
 	"net/http/httputil"
@@ -1400,3 +1401,159 @@ func TestGetObjectResponseOverride(t *testing.T) {
 		}
 	})
 }
+
+func TestUploadPartCopy(t *testing.T) {
+	ts := newTestServer(t)
+	defer ts.Close()
+	ctx := context.Background()
+	svc := ts.s3Client()
+
+	content := "source data for the multipart part copy\n"
+	ts.backendPutString(defaultBucket, "src-key", nil, content)
+
+	uploadID := ts.createMultipartUpload(defaultBucket, "dst-key", nil)
+
+	part, err := svc.UploadPartCopy(ctx, &s3.UploadPartCopyInput{
+		Bucket:     aws.String(defaultBucket),
+		Key:        aws.String("dst-key"),
+		UploadId:   aws.String(uploadID),
+		PartNumber: aws.Int32(1),
+		CopySource: aws.String("/" + defaultBucket + "/src-key"),
+	})
+	ts.OK(err)
+	if part.CopyPartResult == nil || part.CopyPartResult.ETag == nil || *part.CopyPartResult.ETag == "" {
+		t.Fatal("UploadPartCopy did not return a CopyPartResult with an ETag")
+	}
+
+	ts.assertCompleteUpload(defaultBucket, "dst-key", uploadID, []types.CompletedPart{
+		{PartNumber: aws.Int32(1), ETag: part.CopyPartResult.ETag},
+	}, content)
+}
+
+func TestUploadPartCopyStreaming(t *testing.T) {
+	be := newStreamingBackend(s3mem.New())
+	ts := newTestServer(t, withBackend(be))
+	defer ts.Close()
+	ctx := context.Background()
+	svc := ts.s3Client()
+
+	content := "streamed source data for the multipart part copy\n"
+	ts.backendPutString(defaultBucket, "src-key", nil, content)
+
+	uploadID := ts.createMultipartUpload(defaultBucket, "dst-key", nil)
+
+	part, err := svc.UploadPartCopy(ctx, &s3.UploadPartCopyInput{
+		Bucket:     aws.String(defaultBucket),
+		Key:        aws.String("dst-key"),
+		UploadId:   aws.String(uploadID),
+		PartNumber: aws.Int32(1),
+		CopySource: aws.String("/" + defaultBucket + "/src-key"),
+	})
+	ts.OK(err)
+	if part.CopyPartResult == nil || part.CopyPartResult.ETag == nil || *part.CopyPartResult.ETag == "" {
+		t.Fatal("UploadPartCopy did not return a CopyPartResult with an ETag")
+	}
+
+	ts.assertCompleteUpload(defaultBucket, "dst-key", uploadID, []types.CompletedPart{
+		{PartNumber: aws.Int32(1), ETag: part.CopyPartResult.ETag},
+	}, content)
+
+	if be.createCalls != 1 || be.completeCalls != 1 || be.uploadCalls != 1 {
+		t.Fatalf("expected one create+upload+complete through the streaming path, got create=%d upload=%d complete=%d", be.createCalls, be.uploadCalls, be.completeCalls)
+	}
+}
+
+func TestUploadPartCopyRange(t *testing.T) {
+	ts := newTestServer(t)
+	defer ts.Close()
+	ctx := context.Background()
+	svc := ts.s3Client()
+
+	content := "0123456789abcdefgh"
+	ts.backendPutString(defaultBucket, "src-key", nil, content)
+
+	uploadID := ts.createMultipartUpload(defaultBucket, "dst-key", nil)
+
+	part, err := svc.UploadPartCopy(ctx, &s3.UploadPartCopyInput{
+		Bucket:          aws.String(defaultBucket),
+		Key:             aws.String("dst-key"),
+		UploadId:        aws.String(uploadID),
+		PartNumber:      aws.Int32(1),
+		CopySource:      aws.String("/" + defaultBucket + "/src-key"),
+		CopySourceRange: aws.String("bytes=4-9"),
+	})
+	ts.OK(err)
+	if part.CopyPartResult == nil || part.CopyPartResult.ETag == nil || *part.CopyPartResult.ETag == "" {
+		t.Fatal("UploadPartCopy did not return a CopyPartResult with an ETag")
+	}
+
+	ts.assertCompleteUpload(defaultBucket, "dst-key", uploadID, []types.CompletedPart{
+		{PartNumber: aws.Int32(1), ETag: part.CopyPartResult.ETag},
+	}, "456789")
+}
+
+// TestUploadPartCopyWithoutContentLength checks that an UploadPartCopy
+// request is accepted even when it carries no Content-Length header. Such a
+// request is valid: the part is copied server-side and has no body.
+func TestUploadPartCopyWithoutContentLength(t *testing.T) {
+	ts := newTestServer(t)
+	defer ts.Close()
+
+	content := "source data for the multipart part copy\n"
+	ts.backendPutString(defaultBucket, "src-key", nil, content)
+	uploadID := ts.createMultipartUpload(defaultBucket, "dst-key", nil)
+
+	host := ts.server.Listener.Addr().String()
+	conn, err := net.Dial("tcp", host)
+	ts.OK(err)
+	defer conn.Close()
+
+	// A raw request with no Content-Length header; the part is copied
+	// server-side so there is no body to size.
+	req := fmt.Sprintf("PUT /%s/%s?uploadId=%s&partNumber=1 HTTP/1.1\r\nHost: %s\r\nX-Amz-Copy-Source: /%s/%s\r\n\r\n",
+		defaultBucket, "dst-key", uploadID, host, defaultBucket, "src-key")
+	if err := conn.SetWriteDeadline(time.Now().Add(5 * time.Second)); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := io.WriteString(conn, req); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := conn.SetReadDeadline(time.Now().Add(5 * time.Second)); err != nil {
+		t.Fatal(err)
+	}
+	resp, err := http.ReadResponse(bufio.NewReader(conn), nil)
+	ts.OK(err)
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("expected 200, got %d", resp.StatusCode)
+	}
+}
+
+// TestUploadPartCopyRangeStatusOK checks that a ranged UploadPartCopy answers
+// 200 with a CopyPartResult, matching S3. The range is an input parameter,
+// not a partial response.
+func TestUploadPartCopyRangeStatusOK(t *testing.T) {
+	ts := newTestServer(t)
+	defer ts.Close()
+
+	content := "0123456789abcdefgh"
+	ts.backendPutString(defaultBucket, "src-key", nil, content)
+	uploadID := ts.createMultipartUpload(defaultBucket, "dst-key", nil)
+
+	u := fmt.Sprintf("%s/%s/%s?uploadId=%s&partNumber=1", ts.server.URL, defaultBucket, "dst-key", uploadID)
+	req, err := http.NewRequest(http.MethodPut, u, nil)
+	ts.OK(err)
+	req.Header.Set("X-Amz-Copy-Source", "/"+defaultBucket+"/src-key")
+	req.Header.Set("X-Amz-Copy-Source-Range", "bytes=4-9")
+
+	resp, err := ts.server.Client().Do(req)
+	ts.OK(err)
+	defer resp.Body.Close()
+	body, _ := io.ReadAll(resp.Body)
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("expected 200, got %d: %s", resp.StatusCode, body)
+	}
+	if !bytes.Contains(body, []byte("CopyPartResult")) {
+		t.Fatalf("expected a CopyPartResult body, got %s", body)
+	}
+}

messages.go

@@ -440,6 +440,13 @@ type CopyObjectResult struct {
 	LastModified ContentTime `xml:"LastModified,omitempty"`
 }
 
+// CopyPartResult contains the response from an UploadPartCopy operation.
+type CopyPartResult struct {
+	XMLName      xml.Name    `xml:"CopyPartResult"`
+	ETag         string      `xml:"ETag"`
+	LastModified ContentTime `xml:"LastModified"`
+}
+
 // MFADeleteStatus is used by VersioningConfiguration.
 type MFADeleteStatus string