remote_storage/s3: send only real object tags to the remote (#12039)

* remote_storage/s3: send only real object tags to the remote

WriteFile and toTagging copied every entry.Extended attribute into the
remote object's tag set, including Seaweed-internal keys such as
Seaweed-X-Amz-Acl whose values are not legal S3 tag values. Remotes
that validate x-amz-tagging now reject these uploads with InvalidTag.

Only X-Amz-Tagging-* attributes are S3 tags: filter to that prefix,
strip it, and encode through s3TaggingHeader so keys and values are
properly escaped.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* remote_storage/s3: skip remote tagging call when the tag set is unchanged

UpdateFileMetadata fired PutObjectTagging on any Extended diff, but most
attribute changes are not tag changes. Compare the effective tag sets and
return early when they match.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* remote_storage/s3: assert WriteFile errors in tag-header tests

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

weed/remote_storage/s3/s3_storage_client.go

@@ -9,6 +9,7 @@ import (
 	"net/url"
 	"os"
 	"reflect"
+	"sort"
 	"strings"
 
 	"github.com/aws/aws-sdk-go/aws"
@@ -25,6 +26,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/remote_pb"
 	"github.com/seaweedfs/seaweedfs/weed/remote_storage"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 	"github.com/seaweedfs/seaweedfs/weed/util/version"
 )
@@ -567,18 +569,12 @@ func (s *s3RemoteStorageClient) WriteFile(loc *remote_pb.RemoteStorageLocation,
 		u.Concurrency = s.uploadConcurrency()
 	})
 
-	// process tagging
-	tags := ""
 	var awsTags *string
 	// openstack swift doesn't support s3 object tagging
 	if s.conf.S3SupportTagging {
-		for k, v := range entry.Extended {
-			if len(tags) > 0 {
-				tags = tags + "&"
-			}
-			tags = tags + k + "=" + string(v)
+		if tagging := toTagging(entry.Extended); len(tagging.TagSet) > 0 {
+			awsTags = aws.String(s3TaggingHeader(tagging.TagSet))
 		}
-		awsTags = aws.String(tags)
 	}
 
 	// Upload the file to S3.
@@ -609,14 +605,23 @@ func (s *s3RemoteStorageClient) WriteFile(loc *remote_pb.RemoteStorageLocation,
 
 }
 
+// toTagging maps the entry's extended attributes to the remote object's tag
+// set. Only X-Amz-Tagging-* attributes are S3 tags; the rest of Extended is
+// Seaweed-internal metadata whose values are not legal tag values.
 func toTagging(attributes map[string][]byte) *s3.Tagging {
 	tagging := &s3.Tagging{}
 	for k, v := range attributes {
+		if !strings.HasPrefix(k, s3_constants.AmzObjectTaggingPrefix) {
+			continue
+		}
 		tagging.TagSet = append(tagging.TagSet, &s3.Tag{
-			Key:   aws.String(k),
+			Key:   aws.String(k[len(s3_constants.AmzObjectTaggingPrefix):]),
 			Value: aws.String(string(v)),
 		})
 	}
+	sort.Slice(tagging.TagSet, func(i, j int) bool {
+		return aws.StringValue(tagging.TagSet[i].Key) < aws.StringValue(tagging.TagSet[j].Key)
+	})
 	return tagging
 }
 
@@ -707,6 +712,9 @@ func (s *s3RemoteStorageClient) UpdateFileMetadata(loc *remote_pb.RemoteStorageL
 	}
 
 	tagging := toTagging(newEntry.Extended)
+	if reflect.DeepEqual(tagging.TagSet, toTagging(oldEntry.Extended).TagSet) {
+		return
+	}
 	if len(tagging.TagSet) > 0 {
 		_, err = s.conn.PutObjectTagging(&s3.PutObjectTaggingInput{
 			Bucket:  aws.String(loc.Bucket),

weed/remote_storage/s3/s3_storage_client_test.go

@@ -364,7 +364,7 @@ func TestS3UpdateFileMetadataSkipsTaggingWhenUnsupported(t *testing.T) {
 	client, rt := newRecordingS3Client(t, false)
 	loc := &remote_pb.RemoteStorageLocation{Name: "test", Bucket: "bucket", Path: "/dir/file.bin"}
 	plain := &filer_pb.Entry{}
-	tagged := &filer_pb.Entry{Extended: map[string][]byte{"k1": []byte("v1")}}
+	tagged := &filer_pb.Entry{Extended: map[string][]byte{"X-Amz-Tagging-k1": []byte("v1")}}
 
 	_, err := client.UpdateFileMetadata(loc, plain, tagged)
 	require.NoError(t, err)
@@ -375,11 +375,22 @@ func TestS3UpdateFileMetadataSkipsTaggingWhenUnsupported(t *testing.T) {
 	require.Empty(t, rt.requests, "removing attributes must not send DeleteObjectTagging")
 }
 
+func TestS3UpdateFileMetadataIgnoresNonTagAttributes(t *testing.T) {
+	client, rt := newRecordingS3Client(t, true)
+	loc := &remote_pb.RemoteStorageLocation{Name: "test", Bucket: "bucket", Path: "/dir/file.bin"}
+	old := &filer_pb.Entry{Extended: map[string][]byte{"k1": []byte("v1")}}
+	updated := &filer_pb.Entry{Extended: map[string][]byte{"k1": []byte("v2")}}
+
+	_, err := client.UpdateFileMetadata(loc, old, updated)
+	require.NoError(t, err)
+	require.Empty(t, rt.requests, "non-tag attribute changes must not send tagging calls")
+}
+
 func TestS3UpdateFileMetadataSendsTaggingWhenSupported(t *testing.T) {
 	client, rt := newRecordingS3Client(t, true)
 	loc := &remote_pb.RemoteStorageLocation{Name: "test", Bucket: "bucket", Path: "/dir/file.bin"}
 	plain := &filer_pb.Entry{}
-	tagged := &filer_pb.Entry{Extended: map[string][]byte{"k1": []byte("v1")}}
+	tagged := &filer_pb.Entry{Extended: map[string][]byte{"X-Amz-Tagging-k1": []byte("v1")}}
 
 	_, err := client.UpdateFileMetadata(loc, plain, tagged)
 	require.NoError(t, err)
@@ -395,6 +406,49 @@ func TestS3UpdateFileMetadataSendsTaggingWhenSupported(t *testing.T) {
 	require.Contains(t, rt.requests[0].URL.RawQuery, "tagging")
 }
 
+func TestS3WriteFileSendsOnlyTagAttributes(t *testing.T) {
+	rt := &captureRoundTripper{}
+	client := newS3ClientWithTransport(t, rt, true)
+	loc := &remote_pb.RemoteStorageLocation{
+		Name:   "test",
+		Bucket: "bucket",
+		Path:   "/dir/test.bin",
+	}
+	entry := &filer_pb.Entry{
+		Extended: map[string][]byte{
+			"X-Amz-Tagging-a b":  []byte("x=y"),
+			"Seaweed-X-Amz-Acl":  []byte(`[{"Permission":"FULL_CONTROL"}]`),
+			"Seaweed-X-Amz-ETag": []byte("etag-1"),
+			"Cache-Control":      []byte("no-cache"),
+		},
+	}
+
+	_, err := client.WriteFile(loc, entry, bytes.NewReader([]byte("data")))
+	require.NoError(t, err)
+
+	require.NotNil(t, rt.uploadReq, "uploader should have issued a PUT")
+	require.Equal(t, "a%20b=x%3Dy", rt.uploadReq.Header.Get("X-Amz-Tagging"))
+}
+
+func TestS3WriteFileOmitsTaggingWhenNoTagAttributes(t *testing.T) {
+	rt := &captureRoundTripper{}
+	client := newS3ClientWithTransport(t, rt, true)
+	loc := &remote_pb.RemoteStorageLocation{
+		Name:   "test",
+		Bucket: "bucket",
+		Path:   "/dir/test.bin",
+	}
+	entry := &filer_pb.Entry{
+		Extended: map[string][]byte{"Seaweed-X-Amz-Acl": []byte(`[{"Permission":"FULL_CONTROL"}]`)},
+	}
+
+	_, err := client.WriteFile(loc, entry, bytes.NewReader([]byte("data")))
+	require.NoError(t, err)
+
+	require.NotNil(t, rt.uploadReq, "uploader should have issued a PUT")
+	require.Empty(t, rt.uploadReq.Header.Get("X-Amz-Tagging"))
+}
+
 func TestS3WriteFileOmitsContentTypeWhenMimeMissing(t *testing.T) {
 	client, rt := newCapturingS3Client(t)
 	loc := &remote_pb.RemoteStorageLocation{