mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-12 06:48:45 +01:00
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 <noreply@anthropic.com> * 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 <noreply@anthropic.com> * volume: fail the replica forwarding test instead of hanging when nothing is forwarded Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * volume: compare the primary and replica bytes in the replica forwarding test Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
1202c5ba08
commit
ea6e9d670a
2 files changed
+122
-1
No files matched your search
@@ -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
|
||||
|
||||
@@ -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))
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user