fix: S3 sink puts all entry.Extended into Tagging header instead of only object tags (#8930)

* test: add failing tests for S3 sink buildTaggingString

* fix: S3 sink should only put object tags into Tagging header

* fix: avoid sending empty x-amz-tagging header

weed/replication/sink/s3sink/s3_sink.go

@@ -3,6 +3,7 @@ package S3Sink
 import (
 	"encoding/base64"
 	"fmt"
+	"net/url"
 	"strconv"
 	"strings"
 
@@ -188,20 +189,16 @@ func (s3sink *S3Sink) CreateEntry(key string, entry *filer_pb.Entry, signatures
 		entry.Extended[s3_constants.AmzUserMetaMtime] = []byte(strconv.FormatInt(entry.Attributes.Mtime, 10))
 	}
 	// process tagging
-	tags := ""
-	for k, v := range entry.Extended {
-		if len(tags) > 0 {
-			tags = tags + "&"
-		}
-		tags = tags + k + "=" + string(v)
-	}
+	tags := buildTaggingString(entry.Extended)
 
 	// Upload the file to S3.
 	uploadInput := s3manager.UploadInput{
-		Bucket:  aws.String(s3sink.bucket),
-		Key:     aws.String(key),
-		Body:    reader,
-		Tagging: aws.String(tags),
+		Bucket: aws.String(s3sink.bucket),
+		Key:    aws.String(key),
+		Body:   reader,
+	}
+	if tags != "" {
+		uploadInput.Tagging = aws.String(tags)
 	}
 	if len(entry.Attributes.Md5) > 0 {
 		uploadInput.ContentMD5 = aws.String(base64.StdEncoding.EncodeToString([]byte(entry.Attributes.Md5)))
@@ -223,3 +220,18 @@ func cleanKey(key string) string {
 	}
 	return key
 }
+
+// buildTaggingString builds the S3 Tagging header value from entry extended metadata.
+// Only keys with the AmzObjectTaggingPrefix ("X-Amz-Tagging-") are included as object
+// tags. The prefix is stripped and values are URL-encoded to produce a valid S3 tagging
+// query string.
+func buildTaggingString(extended map[string][]byte) string {
+	tagValues := url.Values{}
+	for k, v := range extended {
+		if strings.HasPrefix(k, s3_constants.AmzObjectTaggingPrefix) {
+			tagKey := k[len(s3_constants.AmzObjectTaggingPrefix):]
+			tagValues.Set(tagKey, string(v))
+		}
+	}
+	return tagValues.Encode()
+}

weed/replication/sink/s3sink/s3_sink_test.go

@@ -0,0 +1,59 @@
+package S3Sink
+
+import (
+	"net/url"
+	"strings"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+)
+
+func TestBuildTaggingString_ShouldStripTagPrefix(t *testing.T) {
+	extended := map[string][]byte{
+		s3_constants.AmzObjectTaggingPrefix + "env": []byte("production"),
+	}
+
+	tagging := buildTaggingString(extended)
+
+	if strings.Contains(tagging, s3_constants.AmzObjectTaggingPrefix) {
+		t.Errorf("tagging should not contain storage prefix %q, got %q", s3_constants.AmzObjectTaggingPrefix, tagging)
+	}
+
+	parsed, err := url.ParseQuery(tagging)
+	if err != nil {
+		t.Fatalf("tagging should be valid URL query: %v", err)
+	}
+	if v := parsed.Get("env"); v != "production" {
+		t.Errorf("expected tag env=production, got %q", v)
+	}
+}
+
+func TestBuildTaggingString_ShouldURLEncodeValues(t *testing.T) {
+	extended := map[string][]byte{
+		s3_constants.AmzObjectTaggingPrefix + "path": []byte("/a/b=c&d"),
+	}
+
+	tagging := buildTaggingString(extended)
+
+	parsed, err := url.ParseQuery(tagging)
+	if err != nil {
+		t.Fatalf("tagging should be valid URL query: %v", err)
+	}
+	if v := parsed.Get("path"); v != "/a/b=c&d" {
+		t.Errorf("expected tag value /a/b=c&d after decoding, got %q", v)
+	}
+}
+
+func TestBuildTaggingString_EmptyWhenNoTags(t *testing.T) {
+	extended := map[string][]byte{
+		"Content-Encoding":            []byte("gzip"),
+		s3_constants.AmzUserMetaMtime: []byte("12345"),
+		s3_constants.SeaweedFSSSES3Key: []byte(`{"algorithm":"AES256","encryptedDEK":"abc"}`),
+	}
+
+	tagging := buildTaggingString(extended)
+
+	if tagging != "" {
+		t.Errorf("expected empty tagging when no tag keys present, got %q", tagging)
+	}
+}