s3: add the RenameObject endpoint (#10659)

* s3: add the RenameObject endpoint

PUT /{bucket}/{key}?renameObject with x-amz-rename-source moves an object
through the filer's AtomicRenameEntry, so no bytes are read or rewritten and
the ETag, tags and SSE keys travel with the entry.

Only unversioned buckets: a versioned rename would have to rebuild the
.versions chain, and AWS offers RenameObject on directory buckets, which
cannot be versioned. The source arrives in a header, so it is authorized
separately for read and delete; both keys are locked, in key order, across the
precondition checks and the move.

* s3: let a matched source ETag precondition settle its date precondition

RFC 7232 has an ETag precondition outrank the date precondition on its own
side, and AWS documents the same for CopyObject: a matching
x-amz-copy-source-if-match with a failing x-amz-copy-source-if-unmodified-since
copies rather than returning 412. The source check evaluated all four headers in
sequence, so the date header could still veto a decided ETag match.

validateConditionalHeadersForReads already applies this precedence; the source
path now matches it.

* s3: cover a rename source named as a directory without a trailing slash

Renaming a directory would move a whole subtree, so it has to stay a missing
key whether or not the caller wrote the trailing slash.

* s3: accept a bare object key as the RenameObject source

AWS spells x-amz-rename-source both ways. Its CLI, Java and Rust examples pass
the bare source key, and only a second CLI example and the boto3 conditional
example pass bucket/key; the API reference's own example is a bare key too. The
header was read as bucket/key only, so the form AWS leads with was rejected with
InvalidArgument and the endpoint was unusable as documented.

A value is now read as a literal key first — the only reading that can never
name the wrong object — and as bucket-qualified second, when it carries the
request's own bucket and the literal key does not exist. That costs one extra
lookup only for a source that starts with the bucket's own name.

Another bucket's name in the source is no longer a distinct error: RenameObject
moves within one bucket, so it is simply part of a key this bucket does not
hold, and it reports NoSuchKey.

* s3: only a proven absence picks the other reading of a rename source

A source that resolves to a directory is not a miss to fall through on: the
literal path is still what the caller named, so answering for it beats renaming
a different object under the bucket-qualified reading. With a directory at
bucket/source.txt and an object at source.txt, a rename naming the former moved
the latter.

A failed lookup is not a proof of absence either, so a blip can no longer
redirect a rename to the other reading.

test/s3/copying/s3_rename_test.go

@@ -0,0 +1,331 @@
+package copying_test
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/url"
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/aws-sdk-go-v2/service/s3/types"
+	smithyhttp "github.com/aws/smithy-go/transport/http"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// createRenameSource builds the x-amz-rename-source value the way the AWS CLI,
+// Java and Rust examples do: the bare source key, URL encoded.
+func createRenameSource(key string) string {
+	return url.PathEscape(key)
+}
+
+// createQualifiedRenameSource builds the alternate bucket/key form the second
+// AWS CLI example and the boto3 conditional example use.
+func createQualifiedRenameSource(bucketName, key string) string {
+	return fmt.Sprintf("%s/%s", bucketName, url.PathEscape(key))
+}
+
+func renameObject(t *testing.T, client *s3.Client, bucketName, srcKey, dstKey string) {
+	t.Helper()
+	_, err := client.RenameObject(context.TODO(), &s3.RenameObjectInput{
+		Bucket:       aws.String(bucketName),
+		Key:          aws.String(dstKey),
+		RenameSource: aws.String(createRenameSource(srcKey)),
+	})
+	require.NoError(t, err)
+}
+
+// requireRenameStatus asserts err carries the given HTTP status code.
+func requireRenameStatus(t *testing.T, err error, status int) {
+	t.Helper()
+	require.Error(t, err)
+	var respErr *smithyhttp.ResponseError
+	require.True(t, errors.As(err, &respErr), "expected an HTTP response error, got %v", err)
+	assert.Equal(t, status, respErr.HTTPStatusCode(), "unexpected error: %v", err)
+}
+
+func objectExists(t *testing.T, client *s3.Client, bucketName, key string) bool {
+	t.Helper()
+	_, err := client.HeadObject(context.TODO(), &s3.HeadObjectInput{
+		Bucket: aws.String(bucketName),
+		Key:    aws.String(key),
+	})
+	return err == nil
+}
+
+// TestRenameObject renames an object and checks the bytes, metadata and ETag
+// arrive under the new key while the old key disappears.
+func TestRenameObject(t *testing.T) {
+	client := getS3Client(t)
+	bucketName := getNewBucketName()
+	createBucket(t, client, bucketName)
+	defer deleteBucket(t, client, bucketName)
+
+	content := "rename me"
+	put := putObjectWithMetadata(t, client, bucketName, "source.txt", content,
+		map[string]string{"origin": "source"}, "text/plain")
+
+	renameObject(t, client, bucketName, "source.txt", "renamed/target.txt")
+
+	assert.False(t, objectExists(t, client, bucketName, "source.txt"), "source should be gone")
+
+	resp := getObject(t, client, bucketName, "renamed/target.txt")
+	assert.Equal(t, content, getObjectBody(t, resp))
+	assert.Equal(t, "text/plain", aws.ToString(resp.ContentType))
+	assert.Equal(t, "source", resp.Metadata["origin"])
+	assert.Equal(t, aws.ToString(put.ETag), aws.ToString(resp.ETag), "ETag must survive the rename")
+}
+
+// TestRenameObjectOverwritesDestination: without a conditional header a rename
+// replaces whatever the destination key held.
+func TestRenameObjectOverwritesDestination(t *testing.T) {
+	client := getS3Client(t)
+	bucketName := getNewBucketName()
+	createBucket(t, client, bucketName)
+	defer deleteBucket(t, client, bucketName)
+
+	putObject(t, client, bucketName, "source.txt", "new content")
+	putObject(t, client, bucketName, "target.txt", "old content")
+
+	renameObject(t, client, bucketName, "source.txt", "target.txt")
+
+	resp := getObject(t, client, bucketName, "target.txt")
+	assert.Equal(t, "new content", getObjectBody(t, resp))
+}
+
+// TestRenameObjectIfNoneMatch: If-None-Match: * protects an existing destination.
+func TestRenameObjectIfNoneMatch(t *testing.T) {
+	client := getS3Client(t)
+	bucketName := getNewBucketName()
+	createBucket(t, client, bucketName)
+	defer deleteBucket(t, client, bucketName)
+
+	putObject(t, client, bucketName, "source.txt", "new content")
+	putObject(t, client, bucketName, "target.txt", "old content")
+
+	_, err := client.RenameObject(context.TODO(), &s3.RenameObjectInput{
+		Bucket:                 aws.String(bucketName),
+		Key:                    aws.String("target.txt"),
+		RenameSource:           aws.String(createRenameSource("source.txt")),
+		DestinationIfNoneMatch: aws.String("*"),
+	})
+	requireRenameStatus(t, err, 412)
+
+	resp := getObject(t, client, bucketName, "target.txt")
+	assert.Equal(t, "old content", getObjectBody(t, resp))
+	assert.True(t, objectExists(t, client, bucketName, "source.txt"), "a failed rename must leave the source alone")
+
+	// The same rename onto a free key succeeds.
+	_, err = client.RenameObject(context.TODO(), &s3.RenameObjectInput{
+		Bucket:                 aws.String(bucketName),
+		Key:                    aws.String("free.txt"),
+		RenameSource:           aws.String(createRenameSource("source.txt")),
+		DestinationIfNoneMatch: aws.String("*"),
+	})
+	require.NoError(t, err)
+}
+
+// TestRenameObjectSourceIfMatch gates the rename on the source's ETag.
+func TestRenameObjectSourceIfMatch(t *testing.T) {
+	client := getS3Client(t)
+	bucketName := getNewBucketName()
+	createBucket(t, client, bucketName)
+	defer deleteBucket(t, client, bucketName)
+
+	put := putObject(t, client, bucketName, "source.txt", "content")
+
+	_, err := client.RenameObject(context.TODO(), &s3.RenameObjectInput{
+		Bucket:        aws.String(bucketName),
+		Key:           aws.String("target.txt"),
+		RenameSource:  aws.String(createRenameSource("source.txt")),
+		SourceIfMatch: aws.String("\"00000000000000000000000000000000\""),
+	})
+	requireRenameStatus(t, err, 412)
+	assert.True(t, objectExists(t, client, bucketName, "source.txt"))
+
+	_, err = client.RenameObject(context.TODO(), &s3.RenameObjectInput{
+		Bucket:        aws.String(bucketName),
+		Key:           aws.String("target.txt"),
+		RenameSource:  aws.String(createRenameSource("source.txt")),
+		SourceIfMatch: put.ETag,
+	})
+	require.NoError(t, err)
+	assert.False(t, objectExists(t, client, bucketName, "source.txt"))
+	assert.True(t, objectExists(t, client, bucketName, "target.txt"))
+}
+
+// TestRenameObjectOntoDirectory: a key that already holds other objects is a
+// directory, and an object must not be allowed to replace one.
+func TestRenameObjectOntoDirectory(t *testing.T) {
+	client := getS3Client(t)
+	bucketName := getNewBucketName()
+	createBucket(t, client, bucketName)
+	defer deleteBucket(t, client, bucketName)
+
+	putObject(t, client, bucketName, "source.txt", "content")
+	putObject(t, client, bucketName, "target/child.txt", "child")
+
+	_, err := client.RenameObject(context.TODO(), &s3.RenameObjectInput{
+		Bucket:       aws.String(bucketName),
+		Key:          aws.String("target"),
+		RenameSource: aws.String(createRenameSource("source.txt")),
+	})
+	requireRenameStatus(t, err, 409)
+	assert.True(t, objectExists(t, client, bucketName, "source.txt"))
+	assert.True(t, objectExists(t, client, bucketName, "target/child.txt"))
+}
+
+// TestRenameObjectDirectorySource: a directory can be named without a trailing
+// slash, and renaming one would move a whole subtree. It is not an object, so it
+// is a missing key.
+func TestRenameObjectDirectorySource(t *testing.T) {
+	client := getS3Client(t)
+	bucketName := getNewBucketName()
+	createBucket(t, client, bucketName)
+	defer deleteBucket(t, client, bucketName)
+
+	putObject(t, client, bucketName, "source/child.txt", "child")
+
+	for _, src := range []string{"source", "source/"} {
+		_, err := client.RenameObject(context.TODO(), &s3.RenameObjectInput{
+			Bucket:       aws.String(bucketName),
+			Key:          aws.String("target.txt"),
+			RenameSource: aws.String(createRenameSource(src)),
+		})
+		requireRenameStatus(t, err, 404)
+	}
+	assert.True(t, objectExists(t, client, bucketName, "source/child.txt"))
+	assert.False(t, objectExists(t, client, bucketName, "target.txt"))
+}
+
+// TestRenameObjectMissingSource reports a missing source as NoSuchKey.
+func TestRenameObjectMissingSource(t *testing.T) {
+	client := getS3Client(t)
+	bucketName := getNewBucketName()
+	createBucket(t, client, bucketName)
+	defer deleteBucket(t, client, bucketName)
+
+	_, err := client.RenameObject(context.TODO(), &s3.RenameObjectInput{
+		Bucket:       aws.String(bucketName),
+		Key:          aws.String("target.txt"),
+		RenameSource: aws.String(createRenameSource("absent.txt")),
+	})
+	requireRenameStatus(t, err, 404)
+}
+
+// TestRenameObjectQualifiedSource: the bucket/key form AWS's second CLI example
+// and the boto3 conditional example use resolves to the same object as the bare
+// key.
+func TestRenameObjectQualifiedSource(t *testing.T) {
+	client := getS3Client(t)
+	bucketName := getNewBucketName()
+	createBucket(t, client, bucketName)
+	defer deleteBucket(t, client, bucketName)
+
+	putObject(t, client, bucketName, "dir/source.txt", "content")
+
+	_, err := client.RenameObject(context.TODO(), &s3.RenameObjectInput{
+		Bucket:       aws.String(bucketName),
+		Key:          aws.String("target.txt"),
+		RenameSource: aws.String(createQualifiedRenameSource(bucketName, "dir/source.txt")),
+	})
+	require.NoError(t, err)
+	assert.False(t, objectExists(t, client, bucketName, "dir/source.txt"))
+	assert.Equal(t, "content", getObjectBody(t, getObject(t, client, bucketName, "target.txt")))
+}
+
+// TestRenameObjectSourceShadowingTheBucketName: a key whose own first segment is
+// the bucket name is a real key, and must win over reading the same value as a
+// bucket-qualified source.
+func TestRenameObjectSourceShadowingTheBucketName(t *testing.T) {
+	client := getS3Client(t)
+	bucketName := getNewBucketName()
+	createBucket(t, client, bucketName)
+	defer deleteBucket(t, client, bucketName)
+
+	shadowed := bucketName + "/source.txt"
+	putObject(t, client, bucketName, shadowed, "shadowed")
+	putObject(t, client, bucketName, "source.txt", "bare")
+
+	_, err := client.RenameObject(context.TODO(), &s3.RenameObjectInput{
+		Bucket:       aws.String(bucketName),
+		Key:          aws.String("target.txt"),
+		RenameSource: aws.String(createQualifiedRenameSource(bucketName, "source.txt")),
+	})
+	require.NoError(t, err)
+	assert.Equal(t, "shadowed", getObjectBody(t, getObject(t, client, bucketName, "target.txt")))
+	assert.False(t, objectExists(t, client, bucketName, shadowed))
+	assert.True(t, objectExists(t, client, bucketName, "source.txt"), "the bare key must be left alone")
+}
+
+// TestRenameObjectSourceShadowedByADirectory: the literal reading of the source
+// names a directory here, and the bucket-qualified reading names a live object.
+// Naming a directory is an error about that directory — falling through to the
+// other reading would rename a different object than the one asked for.
+func TestRenameObjectSourceShadowedByADirectory(t *testing.T) {
+	client := getS3Client(t)
+	bucketName := getNewBucketName()
+	createBucket(t, client, bucketName)
+	defer deleteBucket(t, client, bucketName)
+
+	putObject(t, client, bucketName, bucketName+"/source.txt/child.txt", "child")
+	putObject(t, client, bucketName, "source.txt", "bare")
+
+	_, err := client.RenameObject(context.TODO(), &s3.RenameObjectInput{
+		Bucket:       aws.String(bucketName),
+		Key:          aws.String("target.txt"),
+		RenameSource: aws.String(createQualifiedRenameSource(bucketName, "source.txt")),
+	})
+	requireRenameStatus(t, err, 404)
+	assert.True(t, objectExists(t, client, bucketName, "source.txt"), "the bare key must be left alone")
+	assert.False(t, objectExists(t, client, bucketName, "target.txt"))
+}
+
+// TestRenameObjectCrossBucket: RenameObject moves within one bucket, so another
+// bucket's name in the source is just part of a key this bucket does not hold.
+func TestRenameObjectCrossBucket(t *testing.T) {
+	client := getS3Client(t)
+	srcBucket := getNewBucketName()
+	createBucket(t, client, srcBucket)
+	defer deleteBucket(t, client, srcBucket)
+	dstBucket := getNewBucketName()
+	createBucket(t, client, dstBucket)
+	defer deleteBucket(t, client, dstBucket)
+
+	putObject(t, client, srcBucket, "source.txt", "content")
+
+	_, err := client.RenameObject(context.TODO(), &s3.RenameObjectInput{
+		Bucket:       aws.String(dstBucket),
+		Key:          aws.String("target.txt"),
+		RenameSource: aws.String(createQualifiedRenameSource(srcBucket, "source.txt")),
+	})
+	requireRenameStatus(t, err, 404)
+	assert.True(t, objectExists(t, client, srcBucket, "source.txt"))
+}
+
+// TestRenameObjectVersionedBucket: versioned buckets are not supported yet, and
+// must say so rather than silently dropping versions.
+func TestRenameObjectVersionedBucket(t *testing.T) {
+	client := getS3Client(t)
+	bucketName := getNewBucketName()
+	createBucket(t, client, bucketName)
+	defer deleteBucket(t, client, bucketName)
+
+	_, err := client.PutBucketVersioning(context.TODO(), &s3.PutBucketVersioningInput{
+		Bucket:                  aws.String(bucketName),
+		VersioningConfiguration: &types.VersioningConfiguration{Status: types.BucketVersioningStatusEnabled},
+	})
+	require.NoError(t, err)
+
+	putObject(t, client, bucketName, "source.txt", "content")
+
+	_, err = client.RenameObject(context.TODO(), &s3.RenameObjectInput{
+		Bucket:       aws.String(bucketName),
+		Key:          aws.String("target.txt"),
+		RenameSource: aws.String(createRenameSource("source.txt")),
+	})
+	requireRenameStatus(t, err, 501)
+	assert.True(t, objectExists(t, client, bucketName, "source.txt"))
+}