diff --git a/internal/multipart/session.go b/internal/multipart/session.go index e25d1d1c7e..23eb002cfd 100644 --- a/internal/multipart/session.go +++ b/internal/multipart/session.go @@ -79,6 +79,7 @@ type Session struct { Mimetype string Modified time.Time Hashes map[*utils.HashType]string + Identity string // opaque client fingerprint, see qualifiesResume Creator *model.User mu sync.Mutex @@ -160,6 +161,7 @@ type InitReq struct { Mimetype string Modified time.Time Hashes map[*utils.HashType]string + Identity string } // Manager owns all live sessions. Sessions are in-memory only (aligned with @@ -193,11 +195,61 @@ func sessionKey(userID uint, path string, size int64) string { return fmt.Sprintf("%d|%s|%d", userID, path, size) } +// qualifiesResume reports whether the client proved it is re-uploading the +// file a live session already holds data for. Path+size alone is NOT enough — +// buffered chunks of a different same-sized file would be silently mixed into +// the result. Note this guards against accidents, not against an adversary: +// sessions are per-user (see sessionKey and Manager.get), so the worst a +// client can do is splice its own upload, which it could equally do by +// uploading the spliced file directly. +func qualifiesResume(s *Session, req InitReq) bool { + // unlike the file hashes, the identity fingerprint is sent regardless of + // whether rapid upload was requested. It is opaque and may cover only part + // of the content (the web client hashes the first and last MiB), so a hash + // both sides sent must still agree + if s.Identity != "" || req.Identity != "" { + return s.Identity != "" && s.Identity == req.Identity && !hashesConflict(s.Hashes, req.Hashes) + } + return hashesQualifyResume(s.Hashes, req.Hashes) +} + +// resumes reports whether an init for req picks up s, in state st, instead of +// superseding it. +func resumes(s *Session, st State, req InitReq) bool { + return st == StateFailedRetriable || (st == StateReceiving && qualifiesResume(s, req)) +} + +// ResumesSameFile reports whether Init(req) would resume a session that req +// proves to be the same file by identity or hashes. That upload owns what sits +// at the destination (its own partial output on in-place drivers, or a file it +// was allowed to replace), so "file exists" must not stop the resume. Picking +// up a failed attempt is no proof: any init for the same path and size does. +func (m *Manager) ResumesSameFile(req InitReq) bool { + m.mu.Lock() + defer m.mu.Unlock() + s, ok := m.byID[m.byKey[sessionKey(req.User.ID, req.Path, req.Size)]] + if !ok { + return false + } + s.mu.Lock() + st := s.state + s.mu.Unlock() + return resumes(s, st, req) && qualifiesResume(s, req) +} + +// hashesConflict reports whether a hash type sent on both inits differs. +func hashesConflict(old, new map[*utils.HashType]string) bool { + for t, ov := range old { + if nv, ok := new[t]; ok && nv != ov { + return true + } + } + return false +} + // hashesQualifyResume reports whether two hash sets prove the client is // re-uploading the same file: they must share at least one hash type and -// agree on every shared one. Path+size alone is NOT enough to resume into a -// receiving session — buffered chunks of a different same-sized file would be -// silently mixed into the result. +// agree on every shared one. func hashesQualifyResume(old, new map[*utils.HashType]string) bool { shared := false for t, ov := range old { @@ -317,9 +369,8 @@ func (m *Manager) Init(req InitReq) (SessionSnapshot, bool, error) { // failed_retriable resumes unconditionally: nothing of the old // attempt survives except CRCs, and the re-fill CRC check catches // a changed file. A receiving session still holds data, so it only - // resumes when hashes prove it is the same file. - if st == StateFailedRetriable || - (st == StateReceiving && hashesQualifyResume(s.Hashes, req.Hashes)) { + // resumes when the client proves it is the same file. + if resumes(s, st, req) { m.mu.Unlock() s.touch() return s.Snapshot(), true, nil @@ -344,6 +395,7 @@ func (m *Manager) Init(req InitReq) (SessionSnapshot, bool, error) { Mimetype: req.Mimetype, Modified: req.Modified, Hashes: req.Hashes, + Identity: req.Identity, Creator: req.User, state: StateReceiving, } diff --git a/internal/multipart/session_test.go b/internal/multipart/session_test.go index daec7f8062..d2c9005067 100644 --- a/internal/multipart/session_test.go +++ b/internal/multipart/session_test.go @@ -667,3 +667,170 @@ func TestInitRejectsBadSize(t *testing.T) { t.Fatal("Init with size 0: expected error") } } + +func withIdentity(req InitReq, identity string) InitReq { + req.Identity = identity + return req +} + +func withMD5(req InitReq, md5 string) InitReq { + if md5 != "" { + req.Hashes = map[*utils.HashType]string{utils.MD5: md5} + } + return req +} + +// The identity fingerprint is what lets a reload resume: it is sent whether or +// not the user asked for a rapid upload, unlike the file hashes. +func TestResumeReceivingByIdentity(t *testing.T) { + cases := []struct { + name string + first, second string + // MD5 sent with the first and the second init; empty sends none + firstMD5, secondMD5 string + wantSameResume bool + }{ + {name: "matching identity resumes", first: "fp-a", second: "fp-a", wantSameResume: true}, + {name: "different identity starts fresh", first: "fp-a", second: "fp-b"}, + {name: "identity dropped starts fresh", first: "fp-a", second: ""}, + {name: "identity added starts fresh", first: "", second: "fp-b"}, + {name: "no identity falls back to hashes", firstMD5: "same-file", secondMD5: "same-file", wantSameResume: true}, + // the identity covers only part of the content: hashes both sides + // sent must still agree, or a file changed in the middle is spliced + {name: "matching identity with conflicting hashes starts fresh", first: "fp-a", second: "fp-a", firstMD5: "file-v1", secondMD5: "file-v2"}, + {name: "matching identity with matching hashes resumes", first: "fp-a", second: "fp-a", firstMD5: "same-file", secondMD5: "same-file", wantSameResume: true}, + {name: "no identity and no hashes never resumes", first: "", second: ""}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + m := setupSessionTest(t) + const chunkSize = 1024 + totalSize := int64(4 * chunkSize) + data := genData(totalSize) + user := testUser() + + started := make(chan struct{}) + var once sync.Once // a superseding Init starts a second pipeline + stubPut(t, func(ctx context.Context, dst string, fs *stream.FileStream, up driver.UpdateProgress) error { + defer fs.Close() + once.Do(func() { close(started) }) + <-ctx.Done() + return ctx.Err() + }) + + base := initReq(user, totalSize, chunkSize) + snap, _, err := m.Init(withMD5(withIdentity(base, tc.first), tc.firstMD5)) + if err != nil { + t.Fatalf("Init: %v", err) + } + <-started + sendChunk(t, m, user, snap.ID, data, 0, chunkSize) + + again, resumed, err := m.Init(withMD5(withIdentity(base, tc.second), tc.secondMD5)) + if err != nil { + t.Fatalf("re-Init: %v", err) + } + if tc.wantSameResume { + if !resumed || again.ID != snap.ID { + t.Fatalf("re-Init = (id=%s, resumed=%v), want the same session resumed", again.ID, resumed) + } + if len(again.Received) == 0 { + t.Fatal("resumed session reports nothing received, so the client re-uploads everything") + } + } else if resumed || again.ID == snap.ID { + t.Fatalf("re-Init = (id=%s, resumed=%v), want a fresh session", again.ID, resumed) + } + }) + } +} + +// An interrupted upload must be able to resume even though an in-place driver +// has already written its partial output to the destination, but only an init +// that proves it is the same file may skip the "file exists" check: a failed +// attempt is picked up by any init for the same path and size. +func TestResumesSameFile(t *testing.T) { + const md5A = "0cc175b9c0f1b6a831c399e269772661" + cases := []struct { + name string + failed bool // the first attempt fails after one chunk + firstIdentity string // sent with the first init + firstMD5 string + againIdentity string // sent with the second init + againMD5 string + want bool + }{ + {name: "receiving, same identity", firstIdentity: "fp-a", againIdentity: "fp-a", want: true}, + {name: "receiving, different identity", firstIdentity: "fp-a", againIdentity: "fp-b"}, + {name: "receiving, identity dropped", firstIdentity: "fp-a"}, + {name: "receiving, same hash", firstMD5: md5A, againMD5: md5A, want: true}, + {name: "failed, same identity", failed: true, firstIdentity: "fp-a", againIdentity: "fp-a", want: true}, + {name: "failed, different identity", failed: true, firstIdentity: "fp-a", againIdentity: "fp-b"}, + {name: "failed, nothing proves the file", failed: true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + m := setupSessionTest(t) + const chunkSize = 1024 + totalSize := int64(4 * chunkSize) + data := genData(totalSize) + user := testUser() + + started := make(chan struct{}) + var once sync.Once + stubPut(t, func(ctx context.Context, dst string, fs *stream.FileStream, up driver.UpdateProgress) error { + defer fs.Close() + once.Do(func() { close(started) }) + if tc.failed { + if _, err := io.ReadFull(fs, make([]byte, chunkSize)); err != nil { + return err + } + return errors.New("transient storage hiccup") + } + <-ctx.Done() + return ctx.Err() + }) + + first := withMD5(withIdentity(initReq(user, totalSize, chunkSize), tc.firstIdentity), tc.firstMD5) + snap, _, err := m.Init(first) + if err != nil { + t.Fatalf("Init: %v", err) + } + <-started + sendChunk(t, m, user, snap.ID, data, 0, chunkSize) + if tc.failed { + waitState(t, m, user, snap.ID, StateFailedRetriable) + } + + again := withMD5(withIdentity(initReq(user, totalSize, chunkSize), tc.againIdentity), tc.againMD5) + if got := m.ResumesSameFile(again); got != tc.want { + t.Fatalf("ResumesSameFile = %v, want %v", got, tc.want) + } + }) + } + t.Run("no session", func(t *testing.T) { + m := setupSessionTest(t) + if m.ResumesSameFile(withIdentity(initReq(testUser(), 4096, 1024), "fp-a")) { + t.Fatal("ResumesSameFile = true without any session") + } + }) + t.Run("failed permanently", func(t *testing.T) { + m := setupSessionTest(t) + user := testUser() + stubPut(t, func(ctx context.Context, dst string, fs *stream.FileStream, up driver.UpdateProgress) error { + defer fs.Close() + return fmt.Errorf("denied: %w", errs.PermissionDenied) + }) + + req := withIdentity(initReq(user, 4096, 1024), "fp-a") + snap, _, err := m.Init(req) + if err != nil { + t.Fatalf("Init: %v", err) + } + waitState(t, m, user, snap.ID, StateFailedPermanent) + // Init supersedes such a session with a fresh upload, which must + // still meet "file exists" + if m.ResumesSameFile(req) { + t.Fatal("ResumesSameFile = true for a session Init would supersede") + } + }) +} diff --git a/server/handles/multipart.go b/server/handles/multipart.go index 991a102bcf..9eba0b37ed 100644 --- a/server/handles/multipart.go +++ b/server/handles/multipart.go @@ -19,7 +19,11 @@ import ( "github.com/gin-gonic/gin" ) -const multipartMinChunkSize = int64(1) << 20 // 1MB +const ( + multipartMinChunkSize = int64(1) << 20 // 1MB + // X-File-Identity is stored per session, so bound what a client can pin + multipartMaxIdentityLen = 256 +) // multipartChunkSize resolves the effective chunk size. The admin setting is // the ceiling: a client may suggest a smaller chunk via X-Chunk-Size but never @@ -80,12 +84,6 @@ func MultipartInit(c *gin.Context) { } } overwrite := c.GetHeader("Overwrite") != "false" - if !overwrite { - if res, _ := fs.Get(c.Request.Context(), path, &fs.GetArgs{NoLog: true}); res != nil { - common.ErrorStrResp(c, "file exists", 403) - return - } - } dir, name := stdpath.Split(path) if shouldIgnoreSystemFile(name) { common.ErrorStrResp(c, errs.IgnoredSystemFile.Error(), 403) @@ -115,7 +113,14 @@ func MultipartInit(c *gin.Context) { if len(mimetype) == 0 { mimetype = utils.GetMimeType(name) } - snap, resumed, err := multipart.DefaultManager.Init(multipart.InitReq{ + // opaque to the server: it only ever compares it with the value the + // session was created with, to decide whether resuming is safe + identity := c.GetHeader("X-File-Identity") + if len(identity) > multipartMaxIdentityLen { + common.ErrorStrResp(c, "X-File-Identity is too long", 400) + return + } + req := multipart.InitReq{ User: user, Path: path, Size: size, @@ -123,7 +128,17 @@ func MultipartInit(c *gin.Context) { Mimetype: mimetype, Modified: getLastModified(c), Hashes: h, - }) + Identity: identity, + } + // an interrupted upload may resume even though in-place drivers have + // already written its partial output to the destination + if !overwrite && !multipart.DefaultManager.ResumesSameFile(req) { + if res, _ := fs.Get(c.Request.Context(), path, &fs.GetArgs{NoLog: true}); res != nil { + common.ErrorStrResp(c, "file exists", 403) + return + } + } + snap, resumed, err := multipart.DefaultManager.Init(req) if err != nil { common.ErrorResp(c, err, 500) return