From ea6e9d670a97551eb95695f407d85604244d3605 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 11 Nov 2026 08:44:06 -0800 Subject: [PATCH] volume: forward no Content-MD5 for a JPEG the orientation fix rewrote (#12557) * volume: forward no Content-MD5 for a JPEG the orientation fix rewrote With -images.fix.orientation the primary stores and forwards the rotated bytes, but sent replicas the client's digest of the original upload. The replica checks Content-MD5 against what it receives and rejects the write, so replicated JPEG uploads with a digest failed. The Rust volume server already drops the digest once the body is rewritten. Co-Authored-By: Claude Opus 5.5 * volume: test the forwarded Content-MD5 through PostHandler and a replica The test now posts each upload to a primary whose fake master names a real replica volume server, so it covers the JPEG gate, ReplicatedWrite and the replica's own Content-MD5 check, and reads the stored bytes back. Co-Authored-By: Claude Opus 5.5 * volume: fail the replica forwarding test instead of hanging when nothing is forwarded Co-Authored-By: Claude Opus 5.5 * volume: compare the primary and replica bytes in the replica forwarding test Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- weed/server/volume_server_handlers_write.go | 19 +++- .../volume_server_handlers_write_test.go | 104 ++++++++++++++++++ 2 files changed, 122 insertions(+), 1 deletion(-) diff --git a/weed/server/volume_server_handlers_write.go b/weed/server/volume_server_handlers_write.go index 564d4a8029..be2bef2db3 100644 --- a/weed/server/volume_server_handlers_write.go +++ b/weed/server/volume_server_handlers_write.go @@ -1,7 +1,10 @@ package weed_server import ( + "bytes" "context" + "crypto/md5" + "encoding/base64" "errors" "fmt" "net/http" @@ -18,6 +21,20 @@ import ( "github.com/seaweedfs/seaweedfs/weed/util/buffer_pool" ) +var jpegStart = []byte{0xFF, 0xD8, 0xFF} + +// replicaContentMd5 drops the client's digest once the orientation fix has +// rewritten the JPEG: replicas check it against the bytes they are sent. +func (vs *VolumeServer) replicaContentMd5(n *needle.Needle, contentMd5 string) string { + if !vs.FixJpgOrientation || contentMd5 == "" || !bytes.HasPrefix(n.Data, jpegStart) { + return contentMd5 + } + if sum := md5.Sum(n.Data); base64.StdEncoding.EncodeToString(sum[:]) != contentMd5 { + return "" + } + return contentMd5 +} + func (vs *VolumeServer) PostHandler(w http.ResponseWriter, r *http.Request) { ctx := r.Context() if e := r.ParseForm(); e != nil { @@ -52,7 +69,7 @@ func (vs *VolumeServer) PostHandler(w http.ResponseWriter, r *http.Request) { ret := operation.UploadResult{} // use context.WithoutCancel to avoid context cancellation when the client connection is closed - isUnchanged, writeError := topology.ReplicatedWrite(context.WithoutCancel(ctx), vs.GetMaster, vs.grpcDialOption, vs.store, volumeId, reqNeedle, r, contentMd5) + isUnchanged, writeError := topology.ReplicatedWrite(context.WithoutCancel(ctx), vs.GetMaster, vs.grpcDialOption, vs.store, volumeId, reqNeedle, r, vs.replicaContentMd5(reqNeedle, contentMd5)) if writeError != nil { stats.VolumeServerFileWriteFailures.Inc() status := http.StatusInternalServerError diff --git a/weed/server/volume_server_handlers_write_test.go b/weed/server/volume_server_handlers_write_test.go index 8ff91c5aaa..3c715be156 100644 --- a/weed/server/volume_server_handlers_write_test.go +++ b/weed/server/volume_server_handlers_write_test.go @@ -3,13 +3,21 @@ package weed_server import ( "bytes" "context" + "crypto/md5" + "encoding/base64" + "image" + "image/jpeg" + "mime/multipart" "net/http" "net/http/httptest" + "net/textproto" "runtime" + "strings" "sync" "testing" "time" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" "github.com/seaweedfs/seaweedfs/weed/security" "github.com/seaweedfs/seaweedfs/weed/stats" @@ -139,3 +147,99 @@ func TestBatchDeleteCancelsAWriteInItsFsyncWindow(t *testing.T) { }) } } + +// A replica rejects a Content-MD5 that does not match the bytes it is sent. +func TestPostHandlerForwardsAContentMd5TheReplicaAccepts(t *testing.T) { + var encoded bytes.Buffer + require.NoError(t, jpeg.Encode(&encoded, image.NewRGBA(image.Rect(0, 0, 3, 2)), nil)) + plain := encoded.Bytes() + exif := []byte("Exif\x00\x00MM\x00\x2a\x00\x00\x00\x08\x00\x01\x01\x12\x00\x03\x00\x00\x00\x01\x00\x03\x00\x00\x00\x00\x00\x00") + tagged := append([]byte{0xFF, 0xD8, 0xFF, 0xE1, 0, byte(len(exif) + 2)}, exif...) + tagged = append(tagged, plain[2:]...) + + vid := needle.VolumeId(849300201) + allocate := func(vs *VolumeServer) { + vs.store = newTraversalTestStore(t.TempDir()) + t.Cleanup(vs.store.Close) + _, err := vs.AllocateVolume(context.Background(), &volume_server_pb.AllocateVolumeRequest{VolumeId: uint32(vid), Replication: "001", Version: uint32(needle.GetCurrentVersion())}) + require.NoError(t, err) + } + replicaVs := &VolumeServer{fileSizeLimitBytes: 1 << 20} + allocate(replicaVs) + replicaVs.guard = security.NewGuard(nil, "", 0, "", 0) + replicaResp := make(chan *httptest.ResponseRecorder, 1) + replica := newProxyTestHTTPServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + rec := httptest.NewRecorder() + replicaVs.PostHandler(rec, r) + replicaResp <- rec + w.WriteHeader(rec.Code) + _, _ = w.Write(rec.Body.Bytes()) + })) + + primary := &master_pb.Location{} + vs := newProxyTestVolumeServer(t, vid.String(), []*master_pb.Location{primary, {Url: strings.TrimPrefix(replica.URL, "http://")}}, nil) + allocate(vs) + primary.Url = util.JoinHostPort(vs.store.Ip, vs.store.Port) + vs.guard = security.NewGuard(nil, "", 0, "", 0) + vs.fileSizeLimitBytes = 1 << 20 + + for i, tc := range []struct { + name string + fix bool + data []byte + rewritten bool + }{ + {"rotated", true, tagged, true}, + {"unchanged jpeg", true, plain, false}, + {"not a jpeg", true, []byte("plain text"), false}, + {"fix off", false, tagged, false}, + } { + t.Run(tc.name, func(t *testing.T) { + vs.FixJpgOrientation = tc.fix + sum := md5.Sum(tc.data) + clientMd5 := base64.StdEncoding.EncodeToString(sum[:]) + + var body bytes.Buffer + form := multipart.NewWriter(&body) + part, err := form.CreatePart(textproto.MIMEHeader{ + "Content-Disposition": {`form-data; name="file"; filename="photo.jpg"`}, + "Content-Type": {"image/jpeg"}, + "Content-Md5": {clientMd5}, + }) + require.NoError(t, err) + _, err = part.Write(tc.data) + require.NoError(t, err) + require.NoError(t, form.Close()) + fid := needle.NewFileId(vid, uint64(i+1), 0x1234) + req := httptest.NewRequest(http.MethodPost, "/"+fid.String(), &body) + req.Header.Set("Content-Type", form.FormDataContentType()) + + resp := httptest.NewRecorder() + vs.PostHandler(resp, req) + var forwarded *httptest.ResponseRecorder + select { + case forwarded = <-replicaResp: + default: + t.Fatalf("primary did not forward the write: %d %s", resp.Code, resp.Body.String()) + } + require.Equal(t, http.StatusCreated, forwarded.Code, forwarded.Body.String()) + require.Equal(t, http.StatusCreated, resp.Code, resp.Body.String()) + require.Equal(t, clientMd5, resp.Header().Get("Content-MD5")) + if tc.rewritten { + require.Empty(t, forwarded.Header().Get("Content-MD5")) + } else { + require.Equal(t, clientMd5, forwarded.Header().Get("Content-MD5")) + } + + stored := func(s *storage.Store) []byte { + n := &needle.Needle{Id: fid.Key, Cookie: fid.Cookie} + _, err := s.ReadVolumeNeedle(vid, n, nil, nil) + require.NoError(t, err) + return n.Data + } + replicaData := stored(replicaVs.store) + require.Equal(t, stored(vs.store), replicaData) + require.Equal(t, tc.rewritten, !bytes.Equal(replicaData, tc.data)) + }) + } +}