From 199b2e2dd9342c4f38767d7271b2796f5f4be5c5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nguy=E1=BB=85n=20L=E1=BB=99c=20Ph=C3=BAc?= <112325531+omninguyen@users.noreply.github.com> Date: Tue, 3 Nov 2026 13:08:21 +0700 Subject: [PATCH] fix(s3api): reject an invalid storage class before uploading the body (#12354) * fix(s3api): reject an invalid storage class before uploading the body * fix(s3api): reject an invalid storage class on CreateMultipartUpload * fix(s3api): validate storage class and tags before reading the PUT body * fix(s3api): validate storage class and tags with the other PUT headers Checking them next to ValidateRequestEncryption keeps the check ahead of any later hook that may answer the PUT without reaching putToFiler. Co-Authored-By: Claude Opus 5.5 * refactor(s3api): reuse validatePutHeaders in putToFiler Co-Authored-By: Claude Opus 5.5 * fix(s3api): store storage class and tags on directory-marker PUTs The marker path validated x-amz-storage-class and x-amz-tagging but its mkdir callback never wrote them, so a valid value was accepted and then dropped and GetObjectTagging on the marker came back empty. Store both the way putToFiler does. Co-Authored-By: Claude Opus 5.5 * test(s3api): read directory-marker tags back through GetObjectTagging Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Chris Lu Co-authored-by: Claude Opus 5.5 --- weed/s3api/s3_metadata_util.go | 4 + weed/s3api/s3api_object_handlers_multipart.go | 5 + weed/s3api/s3api_object_handlers_put.go | 62 +++-- .../s3api_storage_class_validation_test.go | 262 ++++++++++++++++++ 4 files changed, 312 insertions(+), 21 deletions(-) create mode 100644 weed/s3api/s3api_storage_class_validation_test.go diff --git a/weed/s3api/s3_metadata_util.go b/weed/s3api/s3_metadata_util.go index 01e0adc194..fbbfec9425 100644 --- a/weed/s3api/s3_metadata_util.go +++ b/weed/s3api/s3_metadata_util.go @@ -25,6 +25,10 @@ func ParseS3Metadata(r *http.Request, existing map[string][]byte, isReplace bool // Storage class if sc := r.Header.Get(s3_constants.AmzStorageClass); sc != "" { + if !validateStorageClass(sc) { + glog.Warningf("Invalid storage class %q", sc) + return nil, s3err.ErrInvalidStorageClass + } metadata[s3_constants.AmzStorageClass] = []byte(sc) } diff --git a/weed/s3api/s3api_object_handlers_multipart.go b/weed/s3api/s3api_object_handlers_multipart.go index 780041f5cf..ca39881a12 100644 --- a/weed/s3api/s3api_object_handlers_multipart.go +++ b/weed/s3api/s3api_object_handlers_multipart.go @@ -389,6 +389,11 @@ func (s3a *S3ApiServer) PutObjectPartHandler(w http.ResponseWriter, r *http.Requ return } + if _, errCode := validatePutHeaders(r); errCode != s3err.ErrNone { + s3err.WriteErrorResponse(w, r, errCode) + return + } + dataReader, s3ErrCode := getRequestDataReader(s3a, r) if s3ErrCode != s3err.ErrNone { glog.Errorf("PutObjectPartHandler: getRequestDataReader failed with code %v", s3ErrCode) diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go index 8e1f3b720c..f06360883d 100644 --- a/weed/s3api/s3api_object_handlers_put.go +++ b/weed/s3api/s3api_object_handlers_put.go @@ -134,6 +134,12 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request) return } + tagSet, headerCode := validatePutHeaders(r) + if headerCode != s3err.ErrNone { + s3err.WriteErrorResponse(w, r, headerCode) + return + } + if r.Header.Get("Cache-Control") != "" { if _, err = cacheobject.ParseRequestCacheControl(r.Header.Get("Cache-Control")); err != nil { s3err.WriteErrorResponse(w, r, s3err.ErrInvalidDigest) @@ -279,6 +285,12 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request) entry.Extended = make(map[string][]byte) } entry.Extended[s3_constants.ExtETagKey] = []byte(dirEtag) + if sc := r.Header.Get(s3_constants.AmzStorageClass); sc != "" { + entry.Extended[s3_constants.AmzStorageClass] = []byte(sc) + } + for key, value := range tagSet { + entry.Extended[S3TAG_PREFIX+key] = []byte(value) + } storeSSEMetadata(entry, sseResult) @@ -524,6 +536,28 @@ func withMultipartPart(r *http.Request) *http.Request { return r.WithContext(context.WithValue(r.Context(), multipartPartContextKey{}, true)) } +// validatePutHeaders checks x-amz-storage-class and x-amz-tagging, so the +// handlers can reject them before reading the body and putToFiler before +// uploading any chunk. It returns the parsed tag set. +func validatePutHeaders(r *http.Request) (map[string]string, s3err.ErrorCode) { + if sc := r.Header.Get(s3_constants.AmzStorageClass); sc != "" && !validateStorageClass(sc) { + return nil, s3err.ErrInvalidStorageClass + } + tagging := r.Header.Get(s3_constants.AmzObjectTagging) + if tagging == "" { + return nil, s3err.ErrNone + } + tagSet, err := parseTagsHeader(tagging) + if err == nil { + err = ValidateTags(tagSet) + } + if err != nil { + glog.V(1).Infof("invalid tagging header %q: %v", tagging, err) + return nil, s3err.ErrInvalidTag + } + return tagSet, s3err.ErrNone +} + func isMultipartPartRequest(r *http.Request) bool { marked, _ := r.Context().Value(multipartPartContextKey{}).(bool) return marked @@ -546,21 +580,11 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader return "", s3err.ErrInvalidRequest, SSEResponseMetadata{} } - // Reject invalid tags before any chunks are uploaded so a failed request - // leaves nothing behind on the volume servers. - var tagSet map[string]string - if tagging := r.Header.Get(s3_constants.AmzObjectTagging); tagging != "" { - parsed, err := parseTagsHeader(tagging) - if err != nil { - glog.Warningf("putToFiler: invalid tagging header for %s: %v", filePath, err) - return "", s3err.ErrInvalidTag, SSEResponseMetadata{} - } - if err := ValidateTags(parsed); err != nil { - glog.Warningf("putToFiler: invalid tags (%d) for %s: %v", len(parsed), filePath, err) - return "", s3err.ErrInvalidTag, SSEResponseMetadata{} - } - tagSet = parsed + tagSet, headerCode := validatePutHeaders(r) + if headerCode != s3err.ErrNone { + return "", headerCode, SSEResponseMetadata{} } + storageClass := r.Header.Get(s3_constants.AmzStorageClass) // NEW OPTIMIZATION: Write directly to volume servers, bypassing filer proxy // This eliminates the filer proxy overhead for PUT operations // Note: filePath is now passed directly instead of URL (no parsing needed) @@ -963,13 +987,9 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader } } - // Store the storage class from header - if sc := r.Header.Get(s3_constants.AmzStorageClass); sc != "" { - if !validateStorageClass(sc) { - glog.Warningf("putToFiler: Invalid storage class '%s' for %s", sc, filePath) - return "", s3err.ErrInvalidStorageClass, SSEResponseMetadata{} - } - entry.Extended[s3_constants.AmzStorageClass] = []byte(sc) + // Store the storage class validated before the chunk upload + if storageClass != "" { + entry.Extended[s3_constants.AmzStorageClass] = []byte(storageClass) } // Store the tags validated before the chunk upload diff --git a/weed/s3api/s3api_storage_class_validation_test.go b/weed/s3api/s3api_storage_class_validation_test.go new file mode 100644 index 0000000000..d6d51d1693 --- /dev/null +++ b/weed/s3api/s3api_storage_class_validation_test.go @@ -0,0 +1,262 @@ +package s3api + +import ( + "encoding/xml" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/gorilla/mux" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" + util_http "github.com/seaweedfs/seaweedfs/weed/util/http" + "github.com/stretchr/testify/require" +) + +func (f *fakeVolumeServer) storedCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.stored) +} + +// An invalid x-amz-storage-class must be rejected before the body is +// uploaded, so a failed PUT or UploadPart leaves no chunks behind. +func TestPutToFilerStorageClass(t *testing.T) { + for _, tt := range []struct { + name string + class string + part bool + wantCode s3err.ErrorCode + wantStored int + }{ + {name: "invalid", class: "BOGUS", wantCode: s3err.ErrInvalidStorageClass}, + {name: "invalid on a part", class: "BOGUS", part: true, wantCode: s3err.ErrInvalidStorageClass}, + {name: "valid", class: "STANDARD_IA", wantCode: s3err.ErrNone, wantStored: 1}, + {name: "absent", wantCode: s3err.ErrNone, wantStored: 1}, + } { + t.Run(tt.name, func(t *testing.T) { + volume := startFakeVolumeServer(t) + filer := &ambiguousPutFiler{volume: volume, apply: true, entries: map[string]*filer_pb.Entry{}} + s3a := newPutTestServer(t, startFakeFiler(t, filer)) + + r := httptest.NewRequest(http.MethodPut, "/b/o", nil) + if tt.class != "" { + r.Header.Set(s3_constants.AmzStorageClass, tt.class) + } + filePath, object, dest := "/buckets/b/o", "o", "" + if tt.part { + r = withMultipartPart(r) + filePath, object, dest = "/buckets/b/.uploads/u/0001.part", "", "/buckets/b/o" + } + _, code, _ := s3a.putToFiler(r, filePath, strings.NewReader("hello world"), "b", object, 1, 0, nil, false, dest) + + require.Equal(t, tt.wantCode, code) + require.Equal(t, tt.wantStored, volume.storedCount(), "chunks stored on the volume server") + if tt.wantCode != s3err.ErrNone { + require.Empty(t, filer.entries) + return + } + entry := filer.entries["/buckets/b/o"] + require.NotNil(t, entry) + require.Equal(t, tt.class, string(entry.Extended[s3_constants.AmzStorageClass])) + }) + } +} + +// countingBody counts the reads a handler makes of the request body. +type countingBody struct { + r io.Reader + reads atomic.Int32 +} + +func (b *countingBody) Read(p []byte) (int, error) { b.reads.Add(1); return b.r.Read(p) } +func (b *countingBody) Close() error { return nil } + +// drainProbe is the ResponseWriter of these tests. WriteErrorResponse drains +// the body before it answers, which would count as reads, so the probe notes +// how many the handler had made when that drain starts. +type drainProbe struct { + *httptest.ResponseRecorder + body *countingBody + readsAtDrain int32 +} + +func (p *drainProbe) SetReadDeadline(time.Time) error { + if p.readsAtDrain < 0 { + p.readsAtDrain = p.body.reads.Load() + } + return nil +} + +func newHeaderTestServer(t *testing.T, versioning string) (*S3ApiServer, *ambiguousPutFiler, *fakeVolumeServer) { + t.Helper() + util_http.InitGlobalHttpClient() + volume := startFakeVolumeServer(t) + filer := &ambiguousPutFiler{volume: volume, apply: true, entries: map[string]*filer_pb.Entry{}} + s3a := newPutTestServer(t, startFakeFiler(t, filer)) + s3a.iam = &IdentityAccessManagement{} + s3a.bucketConfigCache = NewBucketConfigCache(time.Minute) + s3a.bucketConfigCache.Set("b", &BucketConfig{Name: "b", Versioning: versioning}) + return s3a, filer, volume +} + +// newHeaderTestRequest builds a request without a Content-Type, so a handler +// has to sniff one from the body. +func newHeaderTestRequest(target, object, payload, header, value string) (*http.Request, *drainProbe) { + body := &countingBody{r: strings.NewReader(payload)} + req := httptest.NewRequest(http.MethodPut, target, body) + req.ContentLength = int64(len(payload)) + req.Header.Set(header, value) + req = mux.SetURLVars(req, map[string]string{"bucket": "b", "object": object}) + return req, &drainProbe{ResponseRecorder: httptest.NewRecorder(), body: body, readsAtDrain: -1} +} + +func tagHeader(n int) string { + tags := make([]string, n) + for i := range tags { + tags[i] = fmt.Sprintf("k%d=v", i) + } + return strings.Join(tags, "&") +} + +var invalidPutHeaders = []struct { + name, header, value string + want s3err.ErrorCode +}{ + {"storage class", s3_constants.AmzStorageClass, "BOGUS", s3err.ErrInvalidStorageClass}, + {"11 tags", s3_constants.AmzObjectTagging, tagHeader(11), s3err.ErrInvalidTag}, +} + +func requireRejected(t *testing.T, probe *drainProbe, volume *fakeVolumeServer, want s3err.ErrorCode) { + t.Helper() + apiErr := s3err.GetAPIError(want) + require.Equal(t, apiErr.HTTPStatusCode, probe.Code, probe.Body.String()) + require.Contains(t, probe.Body.String(), ""+apiErr.Code+"") + require.Zero(t, probe.readsAtDrain, "body reads before the error response") + require.Zero(t, volume.storedCount(), "chunks stored on the volume server") +} + +// An invalid storage class or tag set is answered before the handler touches +// the body, whichever way the bucket versions: the content-type sniff of a PUT +// without Content-Type blocks on the first read. +func TestPutObjectRejectsInvalidHeadersBeforeReadingBody(t *testing.T) { + for _, state := range []struct{ name, versioning string }{ + {"unversioned", ""}, + {"versioning enabled", s3_constants.VersioningEnabled}, + {"versioning suspended", s3_constants.VersioningSuspended}, + } { + for _, h := range invalidPutHeaders { + t.Run(state.name+"/"+h.name, func(t *testing.T) { + s3a, filer, volume := newHeaderTestServer(t, state.versioning) + req, probe := newHeaderTestRequest("/b/o", "o", "hello world", h.header, h.value) + + s3a.PutObjectHandler(probe, req) + + requireRejected(t, probe, volume, h.want) + filer.mu.Lock() + defer filer.mu.Unlock() + require.Empty(t, filer.entries) + }) + } + } +} + +func TestPutObjectPartRejectsInvalidHeadersBeforeReadingBody(t *testing.T) { + for _, h := range invalidPutHeaders { + t.Run(h.name, func(t *testing.T) { + s3a, filer, volume := newHeaderTestServer(t, "") + uploadID := s3a.generateUploadID("o") + filer.mu.Lock() + filer.entries["/buckets/b/.uploads/"+uploadID] = &filer_pb.Entry{ + Name: uploadID, IsDirectory: true, + Extended: map[string][]byte{s3_constants.ExtMultipartObjectKey: []byte("o")}, + } + filer.mu.Unlock() + req, probe := newHeaderTestRequest("/b/o?partNumber=1&uploadId="+uploadID, "o", "hello world", h.header, h.value) + + s3a.PutObjectPartHandler(probe, req) + + requireRejected(t, probe, volume, h.want) + }) + } +} + +// A directory marker is validated like any other PUT. +func TestPutObjectDirectoryMarkerRejectsInvalidHeaders(t *testing.T) { + for _, h := range invalidPutHeaders { + t.Run(h.name, func(t *testing.T) { + s3a, filer, volume := newHeaderTestServer(t, "") + req, probe := newHeaderTestRequest("/b/dir/", "dir/", "", h.header, h.value) + + s3a.PutObjectHandler(probe, req) + + requireRejected(t, probe, volume, h.want) + filer.mu.Lock() + defer filer.mu.Unlock() + require.Nil(t, filer.entries["/buckets/b/dir"], "directory marker created") + }) + } +} + +func TestPutObjectDirectoryMarkerStoresHeaders(t *testing.T) { + s3a, filer, _ := newHeaderTestServer(t, "") + req, probe := newHeaderTestRequest("/b/dir/", "dir/", "", s3_constants.AmzStorageClass, "STANDARD_IA") + req.Header.Set(s3_constants.AmzObjectTagging, "k1=v1&k2=v2") + + s3a.PutObjectHandler(probe, req) + + require.Equal(t, http.StatusOK, probe.Code, probe.Body.String()) + filer.mu.Lock() + entry := filer.entries["/buckets/b/dir"] + require.NotNil(t, entry) + require.Equal(t, "STANDARD_IA", string(entry.Extended[s3_constants.AmzStorageClass])) + require.Equal(t, "v1", string(entry.Extended[S3TAG_PREFIX+"k1"])) + require.Equal(t, "v2", string(entry.Extended[S3TAG_PREFIX+"k2"])) + filer.mu.Unlock() + + getReq := httptest.NewRequest(http.MethodGet, "/b/dir/?tagging", nil) + getReq = mux.SetURLVars(getReq, map[string]string{"bucket": "b", "object": "dir/"}) + rec := httptest.NewRecorder() + s3a.GetObjectTaggingHandler(rec, getReq) + require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) + var tagging Tagging + require.NoError(t, xml.Unmarshal(rec.Body.Bytes(), &tagging)) + require.ElementsMatch(t, []Tag{{Key: "k1", Value: "v1"}, {Key: "k2", Value: "v2"}}, tagging.TagSet.Tag) +} + +// A valid storage class still stores, and the content type is still sniffed. +func TestPutObjectValidStorageClassSniffsContentType(t *testing.T) { + s3a, filer, volume := newHeaderTestServer(t, "") + req, probe := newHeaderTestRequest("/b/o", "o", "hello world", s3_constants.AmzStorageClass, "STANDARD_IA") + + s3a.PutObjectHandler(probe, req) + + require.Equal(t, http.StatusOK, probe.Code, probe.Body.String()) + require.Equal(t, 1, volume.storedCount()) + filer.mu.Lock() + defer filer.mu.Unlock() + entry := filer.entries["/buckets/b/o"] + require.NotNil(t, entry) + require.Equal(t, "STANDARD_IA", string(entry.Extended[s3_constants.AmzStorageClass])) + require.Equal(t, "text/plain; charset=utf-8", entry.Attributes.Mime) +} + +// CreateMultipartUpload stores the storage class for the completed object, +// so it must reject an invalid one as PutObject and CopyObject do. +func TestParseS3MetadataRejectsInvalidStorageClass(t *testing.T) { + req := httptest.NewRequest(http.MethodPost, "http://s3/b/o?uploads", nil) + req.Header.Set(s3_constants.AmzStorageClass, "BOGUS") + _, code := ParseS3Metadata(req, nil, false) + require.Equal(t, s3err.ErrInvalidStorageClass, code) + + req.Header.Set(s3_constants.AmzStorageClass, "GLACIER") + metadata, code := ParseS3Metadata(req, nil, false) + require.Equal(t, s3err.ErrNone, code) + require.Equal(t, "GLACIER", string(metadata[s3_constants.AmzStorageClass])) +}