diff --git a/test/tus/tus_integration_test.go b/test/tus/tus_integration_test.go index c923577e3..96f4bd505 100644 --- a/test/tus/tus_integration_test.go +++ b/test/tus/tus_integration_test.go @@ -685,6 +685,135 @@ func TestTusCreationWithUpload(t *testing.T) { assert.Equal(t, testData, body) } +// TestTusConcatenation tests the concatenation extension: partial uploads +// assembled into a final upload +func TestTusConcatenation(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second) + defer cancel() + + cluster, err := startTestCluster(t, ctx) + require.NoError(t, err) + defer func() { + cluster.Stop() + os.RemoveAll(cluster.dataDir) + }() + + client := &http.Client{} + + // The extension is announced + optionsReq, err := http.NewRequest(http.MethodOptions, cluster.TusURL()+"/", nil) + require.NoError(t, err) + optionsReq.Header.Set("Tus-Resumable", TusVersion) + optionsResp, err := client.Do(optionsReq) + require.NoError(t, err) + optionsResp.Body.Close() + assert.Contains(t, optionsResp.Header.Get("Tus-Extension"), "concatenation") + + partA := []byte("Hello, ") + partB := []byte("concatenated TUS world!") + targetPath := "/concat/final.txt" + + createPartial := func(size int) string { + req, err := http.NewRequest(http.MethodPost, cluster.TusURL()+targetPath, nil) + require.NoError(t, err) + req.Header.Set("Tus-Resumable", TusVersion) + req.Header.Set("Upload-Length", strconv.Itoa(size)) + req.Header.Set("Upload-Concat", "partial") + resp, err := client.Do(req) + require.NoError(t, err) + defer resp.Body.Close() + require.Equal(t, http.StatusCreated, resp.StatusCode) + location := resp.Header.Get("Location") + require.NotEmpty(t, location) + return location + } + + patchAll := func(location string, data []byte) { + req, err := http.NewRequest(http.MethodPatch, cluster.FullURL(location), bytes.NewReader(data)) + require.NoError(t, err) + req.Header.Set("Tus-Resumable", TusVersion) + req.Header.Set("Upload-Offset", "0") + req.Header.Set("Content-Type", "application/offset+octet-stream") + resp, err := client.Do(req) + require.NoError(t, err) + resp.Body.Close() + require.Equal(t, http.StatusNoContent, resp.StatusCode) + } + + locationA := createPartial(len(partA)) + locationB := createPartial(len(partB)) + concatHeader := "final;" + locationA + " " + locationB + + patchAll(locationA, partA) + + // Concatenation is rejected while a listed partial is unfinished + prematureReq, err := http.NewRequest(http.MethodPost, cluster.TusURL()+targetPath, nil) + require.NoError(t, err) + prematureReq.Header.Set("Tus-Resumable", TusVersion) + prematureReq.Header.Set("Upload-Concat", concatHeader) + prematureResp, err := client.Do(prematureReq) + require.NoError(t, err) + prematureResp.Body.Close() + require.Equal(t, http.StatusBadRequest, prematureResp.StatusCode) + + patchAll(locationB, partB) + + // A completed partial reports its status and does not land at the target + headReq, err := http.NewRequest(http.MethodHead, cluster.FullURL(locationA), nil) + require.NoError(t, err) + headReq.Header.Set("Tus-Resumable", TusVersion) + headResp, err := client.Do(headReq) + require.NoError(t, err) + headResp.Body.Close() + require.Equal(t, http.StatusOK, headResp.StatusCode) + assert.Equal(t, "partial", headResp.Header.Get("Upload-Concat")) + + getResp, err := client.Get(cluster.FilerURL() + targetPath) + require.NoError(t, err) + getResp.Body.Close() + require.Equal(t, http.StatusNotFound, getResp.StatusCode, + "completed partials should not land at the target path") + + // Concatenate into the final upload + finalReq, err := http.NewRequest(http.MethodPost, cluster.TusURL()+targetPath, nil) + require.NoError(t, err) + finalReq.Header.Set("Tus-Resumable", TusVersion) + finalReq.Header.Set("Upload-Concat", concatHeader) + finalReq.Header.Set("Upload-Metadata", encodeTusMetadata(map[string]string{ + "content-type": "text/plain", + })) + finalResp, err := client.Do(finalReq) + require.NoError(t, err) + finalResp.Body.Close() + require.Equal(t, http.StatusCreated, finalResp.StatusCode) + assert.NotEmpty(t, finalResp.Header.Get("Location")) + + // The target file holds both parts in order + expected := append(append([]byte{}, partA...), partB...) + getResp2, err := client.Get(cluster.FilerURL() + targetPath) + require.NoError(t, err) + defer getResp2.Body.Close() + require.Equal(t, http.StatusOK, getResp2.StatusCode) + body, err := io.ReadAll(getResp2.Body) + require.NoError(t, err) + assert.Equal(t, expected, body, "Concatenated file should hold both parts in order") + assert.Contains(t, getResp2.Header.Get("Content-Type"), "text/plain") + + // The consumed partials are gone + headReq2, err := http.NewRequest(http.MethodHead, cluster.FullURL(locationA), nil) + require.NoError(t, err) + headReq2.Header.Set("Tus-Resumable", TusVersion) + headResp2, err := client.Do(headReq2) + require.NoError(t, err) + headResp2.Body.Close() + require.Equal(t, http.StatusNotFound, headResp2.StatusCode, + "consumed partial should be removed after concatenation") +} + // TestTusResumeAfterInterruption simulates resuming an upload after failure func TestTusResumeAfterInterruption(t *testing.T) { if testing.Short() { diff --git a/weed/server/filer_server_tus_concat_test.go b/weed/server/filer_server_tus_concat_test.go new file mode 100644 index 000000000..a361f3d04 --- /dev/null +++ b/weed/server/filer_server_tus_concat_test.go @@ -0,0 +1,345 @@ +package weed_server + +import ( + "context" + "encoding/base64" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +const ( + tusTestPartialAID = "11111111-1111-1111-1111-111111111111" + tusTestPartialBID = "22222222-2222-2222-2222-222222222222" +) + +// seedTusChunk writes a chunk marker entry into a session directory. +func seedTusChunk(t *testing.T, fs *FilerServer, store *renameTestStore, uploadID string, offset, size int64, fileId string) { + t.Helper() + entry := &filer.Entry{ + FullPath: util.FullPath(fs.tusChunkPath(uploadID, offset, size, fileId)), + Attr: filer.Attr{Crtime: time.Unix(1700000000, 0)}, + } + if err := store.InsertEntry(context.Background(), entry); err != nil { + t.Fatalf("seed chunk for %s: %v", uploadID, err) + } +} + +func tusRequest(method, path string, headers map[string]string, body string) *http.Request { + var req *http.Request + if body == "" { + req = httptest.NewRequest(method, path, http.NoBody) + } else { + req = httptest.NewRequest(method, path, strings.NewReader(body)) + } + req.Header.Set("Tus-Resumable", TusVersion) + for k, v := range headers { + req.Header.Set(k, v) + } + return req +} + +// TestFilerServer_tusPatchHandler_PartialSkipsCompletion verifies a completed +// partial upload keeps its session and does not land at the target path, while +// the same flow without Upload-Concat completes normally. +func TestFilerServer_tusPatchHandler_PartialSkipsCompletion(t *testing.T) { + tests := []struct { + name string + concat string + expectEntry bool + }{ + {"plain upload completes", "", true}, + {"partial upload retained", TusConcatPartial, false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + fs, store := newTusTestServer(t, nil) + targetPath := "/buckets/data/file.bin" + seedTusSession(t, fs, store, TusSession{ID: tusTestUploadID, TargetPath: targetPath, Size: 0, Concat: tt.concat}) + + req := tusRequest(http.MethodPatch, "/.tus/.uploads/"+tusTestUploadID, map[string]string{ + "Authorization": "Bearer " + signFilerToken(t, tusTestWriteKey, nil, nil), + "Content-Type": "application/offset+octet-stream", + "Upload-Offset": "0", + }, "") + rec := httptest.NewRecorder() + fs.tusHandler(rec, req) + + if rec.Code != http.StatusNoContent { + t.Fatalf("PATCH = %d, want %d; body=%q", rec.Code, http.StatusNoContent, rec.Body.String()) + } + _, entryErr := store.FindEntry(context.Background(), util.FullPath(targetPath)) + _, sessionErr := store.FindEntry(context.Background(), util.FullPath(fs.tusSessionInfoPath(tusTestUploadID))) + if tt.expectEntry && (entryErr != nil || sessionErr == nil) { + t.Fatalf("plain upload: entry err=%v, session err=%v; want entry present and session removed", entryErr, sessionErr) + } + if !tt.expectEntry && (entryErr == nil || sessionErr != nil) { + t.Fatalf("partial upload: entry err=%v, session err=%v; want no entry and session retained", entryErr, sessionErr) + } + }) + } +} + +// TestFilerServer_tusHeadHandler_EchoesUploadConcat verifies HEAD reports the +// stored Upload-Concat value for partial sessions and omits it otherwise. +func TestFilerServer_tusHeadHandler_EchoesUploadConcat(t *testing.T) { + fs, store := newTusTestServer(t, map[string]string{tusTestUploadID: "/buckets/data/plain.bin"}) + seedTusSession(t, fs, store, TusSession{ID: tusTestPartialAID, TargetPath: "/buckets/data/part.bin", Size: 1, Concat: TusConcatPartial}) + + head := func(uploadID string) *httptest.ResponseRecorder { + req := tusRequest(http.MethodHead, "/.tus/.uploads/"+uploadID, map[string]string{ + "Authorization": "Bearer " + signFilerToken(t, tusTestReadKey, nil, nil), + }, "") + rec := httptest.NewRecorder() + fs.tusHandler(rec, req) + return rec + } + + partialRec := head(tusTestPartialAID) + if partialRec.Code != http.StatusOK || partialRec.Header().Get("Upload-Concat") != TusConcatPartial { + t.Fatalf("partial HEAD = %d, Upload-Concat=%q; want 200 with %q", partialRec.Code, partialRec.Header().Get("Upload-Concat"), TusConcatPartial) + } + plainRec := head(tusTestUploadID) + if plainRec.Code != http.StatusOK || plainRec.Header().Get("Upload-Concat") != "" { + t.Fatalf("plain HEAD = %d, Upload-Concat=%q; want 200 without the header", plainRec.Code, plainRec.Header().Get("Upload-Concat")) + } +} + +// TestFilerServer_tusConcatFinal_AssemblesPartials drives a full concatenation: +// two completed partials, referenced as a path and as an absolute URL, land as +// one entry with re-based chunks, and the consumed sessions are removed. +func TestFilerServer_tusConcatFinal_AssemblesPartials(t *testing.T) { + fs, store := newTusTestServer(t, nil) + targetPath := "/buckets/data/final.bin" + fidA, fidB := "3,01637037d6", "4,02637037d6" + seedTusSession(t, fs, store, TusSession{ID: tusTestPartialAID, TargetPath: targetPath, Size: 8, Concat: TusConcatPartial}) + seedTusChunk(t, fs, store, tusTestPartialAID, 0, 8, fidA) + seedTusSession(t, fs, store, TusSession{ID: tusTestPartialBID, TargetPath: targetPath, Size: 4, Concat: TusConcatPartial}) + seedTusChunk(t, fs, store, tusTestPartialBID, 0, 4, fidB) + + req := tusRequest(http.MethodPost, "/.tus"+targetPath, map[string]string{ + "Authorization": "Bearer " + signFilerToken(t, tusTestWriteKey, nil, nil), + "Upload-Concat": "final;/.tus/.uploads/" + tusTestPartialAID + " http://example.com/.tus/.uploads/" + tusTestPartialBID, + "Upload-Metadata": "content-type " + base64.StdEncoding.EncodeToString([]byte("text/plain")), + }, "") + rec := httptest.NewRecorder() + fs.tusHandler(rec, req) + + if rec.Code != http.StatusCreated { + t.Fatalf("final POST = %d, want %d; body=%q", rec.Code, http.StatusCreated, rec.Body.String()) + } + if location := rec.Header().Get("Location"); !strings.HasPrefix(location, "/.tus/.uploads/") { + t.Fatalf("Location = %q, want an upload URL", location) + } + + entry, err := store.FindEntry(context.Background(), util.FullPath(targetPath)) + if err != nil { + t.Fatalf("final entry not created: %v", err) + } + if entry.Mime != "text/plain" { + t.Errorf("entry mime = %q, want %q", entry.Mime, "text/plain") + } + chunks := entry.GetChunks() + if len(chunks) != 2 { + t.Fatalf("entry chunks = %d, want 2", len(chunks)) + } + if chunks[0].FileId != fidA || chunks[0].Offset != 0 || chunks[0].Size != 8 { + t.Errorf("chunk[0] = %s@%d+%d, want %s@0+8", chunks[0].FileId, chunks[0].Offset, chunks[0].Size, fidA) + } + if chunks[1].FileId != fidB || chunks[1].Offset != 8 || chunks[1].Size != 4 { + t.Errorf("chunk[1] = %s@%d+%d, want %s@8+4", chunks[1].FileId, chunks[1].Offset, chunks[1].Size, fidB) + } + + for _, uploadID := range []string{tusTestPartialAID, tusTestPartialBID} { + if _, err := store.FindEntry(context.Background(), util.FullPath(fs.tusSessionInfoPath(uploadID))); err == nil { + t.Errorf("partial session %s still present after concatenation", uploadID) + } + } +} + +// TestFilerServer_tusConcatFinal_Validation covers the final upload request +// guards: header misuse, unusable references, unfinished or foreign partials. +func TestFilerServer_tusConcatFinal_Validation(t *testing.T) { + uploadsRef := func(uploadID string) string { return "/.tus/.uploads/" + uploadID } + completePartial := func(fs *FilerServer, store *renameTestStore, t *testing.T, uploadID, targetPath string, size int64) { + seedTusSession(t, fs, store, TusSession{ID: uploadID, TargetPath: targetPath, Size: size, Concat: TusConcatPartial}) + seedTusChunk(t, fs, store, uploadID, 0, size, "3,01637037d6") + } + + tests := []struct { + name string + seed func(t *testing.T, fs *FilerServer, store *renameTestStore) + headers map[string]string + body string + prefixes []string + expectStatus int + }{ + { + name: "upload length rejected", + headers: map[string]string{"Upload-Concat": "final;" + uploadsRef(tusTestPartialAID), "Upload-Length": "12"}, + expectStatus: http.StatusBadRequest, + }, + { + name: "body rejected", + headers: map[string]string{"Upload-Concat": "final;" + uploadsRef(tusTestPartialAID)}, + body: "x", + expectStatus: http.StatusForbidden, + }, + { + name: "reference outside uploads prefix", + headers: map[string]string{"Upload-Concat": "final;/elsewhere/" + tusTestPartialAID}, + expectStatus: http.StatusBadRequest, + }, + { + name: "duplicate reference", + headers: map[string]string{"Upload-Concat": "final;" + uploadsRef(tusTestPartialAID) + " " + uploadsRef(tusTestPartialAID)}, + expectStatus: http.StatusBadRequest, + }, + { + name: "empty reference list", + headers: map[string]string{"Upload-Concat": "final;"}, + expectStatus: http.StatusBadRequest, + }, + { + name: "missing partial", + headers: map[string]string{"Upload-Concat": "final;" + uploadsRef(tusTestPartialAID)}, + expectStatus: http.StatusNotFound, + }, + { + name: "non-partial session rejected", + seed: func(t *testing.T, fs *FilerServer, store *renameTestStore) { + seedTusSession(t, fs, store, TusSession{ID: tusTestPartialAID, TargetPath: "/buckets/data/a.bin", Size: 1}) + }, + headers: map[string]string{"Upload-Concat": "final;" + uploadsRef(tusTestPartialAID)}, + expectStatus: http.StatusBadRequest, + }, + { + name: "unfinished partial rejected", + seed: func(t *testing.T, fs *FilerServer, store *renameTestStore) { + seedTusSession(t, fs, store, TusSession{ID: tusTestPartialAID, TargetPath: "/buckets/data/a.bin", Size: 4, Concat: TusConcatPartial}) + }, + headers: map[string]string{"Upload-Concat": "final;" + uploadsRef(tusTestPartialAID)}, + expectStatus: http.StatusBadRequest, + }, + { + name: "cross-prefix partial denied", + seed: func(t *testing.T, fs *FilerServer, store *renameTestStore) { + completePartial(fs, store, t, tusTestPartialAID, "/buckets/secret/victim.bin", 4) + }, + headers: map[string]string{"Upload-Concat": "final;" + uploadsRef(tusTestPartialAID)}, + prefixes: []string{"/buckets/data"}, + expectStatus: http.StatusUnauthorized, + }, + { + name: "combined size over maximum", + seed: func(t *testing.T, fs *FilerServer, store *renameTestStore) { + completePartial(fs, store, t, tusTestPartialAID, "/buckets/data/a.bin", TusDefaultMaxSize) + seedTusSession(t, fs, store, TusSession{ID: tusTestPartialBID, TargetPath: "/buckets/data/b.bin", Size: 1, Concat: TusConcatPartial}) + }, + headers: map[string]string{"Upload-Concat": "final;" + uploadsRef(tusTestPartialAID) + " " + uploadsRef(tusTestPartialBID)}, + expectStatus: http.StatusRequestEntityTooLarge, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + fs, store := newTusTestServer(t, nil) + if tt.seed != nil { + tt.seed(t, fs, store) + } + headers := map[string]string{ + "Authorization": "Bearer " + signFilerToken(t, tusTestWriteKey, tt.prefixes, nil), + } + for k, v := range tt.headers { + headers[k] = v + } + req := tusRequest(http.MethodPost, "/.tus/buckets/data/final.bin", headers, tt.body) + rec := httptest.NewRecorder() + fs.tusHandler(rec, req) + + if rec.Code != tt.expectStatus { + t.Fatalf("final POST = %d, want %d; body=%q", rec.Code, tt.expectStatus, rec.Body.String()) + } + if _, err := store.FindEntry(context.Background(), util.FullPath("/buckets/data/final.bin")); err == nil { + t.Fatal("rejected concatenation still created the target entry") + } + }) + } +} + +// TestFilerServer_tusConcatFinal_ClaimedPartialRejected verifies a partial +// already claimed by a concurrent final cannot be consumed again, and that the +// foreign claim is left in place. +func TestFilerServer_tusConcatFinal_ClaimedPartialRejected(t *testing.T) { + fs, store := newTusTestServer(t, nil) + seedTusSession(t, fs, store, TusSession{ID: tusTestPartialAID, TargetPath: "/buckets/data/a.bin", Size: 4, Concat: TusConcatPartial}) + seedTusChunk(t, fs, store, tusTestPartialAID, 0, 4, "3,01637037d6") + markerPath := util.FullPath(fs.tusSessionConsumedPath(tusTestPartialAID)) + if err := store.InsertEntry(context.Background(), &filer.Entry{FullPath: markerPath}); err != nil { + t.Fatalf("seed consumed marker: %v", err) + } + + req := tusRequest(http.MethodPost, "/.tus/buckets/data/final.bin", map[string]string{ + "Authorization": "Bearer " + signFilerToken(t, tusTestWriteKey, nil, nil), + "Upload-Concat": "final;/.tus/.uploads/" + tusTestPartialAID, + }, "") + rec := httptest.NewRecorder() + fs.tusHandler(rec, req) + + if rec.Code != http.StatusConflict { + t.Fatalf("final POST = %d, want %d; body=%q", rec.Code, http.StatusConflict, rec.Body.String()) + } + if _, err := store.FindEntry(context.Background(), markerPath); err != nil { + t.Fatal("foreign claim was released by the losing request") + } +} + +// TestFilerServer_tusDeleteHandler_ConsumedSessionKeepsChunks verifies deleting +// a consumed session removes its metadata without freeing its chunks. The test +// filer has no chunk deletion queue, so an attempt to free chunks would panic. +func TestFilerServer_tusDeleteHandler_ConsumedSessionKeepsChunks(t *testing.T) { + fs, store := newTusTestServer(t, nil) + seedTusSession(t, fs, store, TusSession{ID: tusTestPartialAID, TargetPath: "/buckets/data/a.bin", Size: 4, Concat: TusConcatPartial}) + seedTusChunk(t, fs, store, tusTestPartialAID, 0, 4, "3,01637037d6") + if err := store.InsertEntry(context.Background(), &filer.Entry{FullPath: util.FullPath(fs.tusSessionConsumedPath(tusTestPartialAID))}); err != nil { + t.Fatalf("seed consumed marker: %v", err) + } + + req := tusRequest(http.MethodDelete, "/.tus/.uploads/"+tusTestPartialAID, map[string]string{ + "Authorization": "Bearer " + signFilerToken(t, tusTestWriteKey, nil, nil), + }, "") + rec := httptest.NewRecorder() + fs.tusHandler(rec, req) + + if rec.Code != http.StatusNoContent { + t.Fatalf("DELETE = %d, want %d; body=%q", rec.Code, http.StatusNoContent, rec.Body.String()) + } + if _, err := store.FindEntry(context.Background(), util.FullPath(fs.tusSessionInfoPath(tusTestPartialAID))); err == nil { + t.Fatal("consumed session metadata still present after DELETE") + } +} + +// TestFilerServer_tusPatchHandler_FinalRejected verifies PATCH against a final +// upload URL is refused, per the concatenation extension. +func TestFilerServer_tusPatchHandler_FinalRejected(t *testing.T) { + fs, store := newTusTestServer(t, nil) + seedTusSession(t, fs, store, TusSession{ID: tusTestUploadID, TargetPath: "/buckets/data/final.bin", Size: 4, Concat: "final;/.tus/.uploads/" + tusTestPartialAID}) + + req := tusRequest(http.MethodPatch, "/.tus/.uploads/"+tusTestUploadID, map[string]string{ + "Authorization": "Bearer " + signFilerToken(t, tusTestWriteKey, nil, nil), + "Content-Type": "application/offset+octet-stream", + "Upload-Offset": "0", + }, "") + rec := httptest.NewRecorder() + fs.tusHandler(rec, req) + + if rec.Code != http.StatusForbidden { + t.Fatalf("PATCH final = %d, want %d; body=%q", rec.Code, http.StatusForbidden, rec.Body.String()) + } +} diff --git a/weed/server/filer_server_tus_handlers.go b/weed/server/filer_server_tus_handlers.go index 5b72e45a8..598e45e0b 100644 --- a/weed/server/filer_server_tus_handlers.go +++ b/weed/server/filer_server_tus_handlers.go @@ -8,6 +8,7 @@ import ( "fmt" "io" "net/http" + "net/url" "path" "strconv" "strings" @@ -112,7 +113,7 @@ func (fs *FilerServer) tusHandler(w http.ResponseWriter, r *http.Request) { writeJsonError(w, r, http.StatusUnauthorized, errors.New("wrong jwt")) return } - fs.tusCreateHandler(w, r) + fs.tusCreateHandler(w, r, claims) default: w.WriteHeader(http.StatusMethodNotAllowed) } @@ -180,12 +181,22 @@ func (fs *FilerServer) tusOptionsHandler(w http.ResponseWriter, r *http.Request) } // tusCreateHandler handles POST requests to create new uploads -func (fs *FilerServer) tusCreateHandler(w http.ResponseWriter, r *http.Request) { +func (fs *FilerServer) tusCreateHandler(w http.ResponseWriter, r *http.Request, claims *security.SeaweedFilerClaims) { // Use a context that ignores cancellation from the request context. // Internal operations (creating TUS session, writing data, completing uploads) // may exceed the filer's client connection inactivity timeout. ctx := context.WithoutCancel(r.Context()) + concat := r.Header.Get("Upload-Concat") + if strings.HasPrefix(concat, TusConcatFinalPrefix) { + fs.tusConcatFinalHandler(w, r, claims, concat) + return + } + if concat != "" && concat != TusConcatPartial { + http.Error(w, "Invalid Upload-Concat", http.StatusBadRequest) + return + } + // Parse Upload-Length header (required) uploadLengthStr := r.Header.Get("Upload-Length") if uploadLengthStr == "" { @@ -205,9 +216,6 @@ func (fs *FilerServer) tusCreateHandler(w http.ResponseWriter, r *http.Request) // Parse Upload-Metadata header (optional) metadata := parseTusMetadata(r.Header.Get("Upload-Metadata")) - // TusBasePath is pre-normalized in filer_server.go (leading slash, no trailing slash) - tusPrefix := fs.option.TusBasePath - // Determine target path from request URL (leading slash guaranteed) targetPath := fs.tusTargetPath(r) if targetPath == "" || targetPath == "/" { @@ -226,18 +234,14 @@ func (fs *FilerServer) tusCreateHandler(w http.ResponseWriter, r *http.Request) uploadID := uuid.New().String() // Create upload session - session, err := fs.createTusSession(ctx, uploadID, targetPath, uploadLength, metadata) + session, err := fs.createTusSession(ctx, uploadID, targetPath, uploadLength, metadata, concat) if err != nil { glog.Errorf("Failed to create TUS session: %v", err) http.Error(w, "Failed to create upload", http.StatusInternalServerError) return } - // Build upload location URL (ensure it starts with single /) - uploadLocation := path.Clean(fmt.Sprintf("%s/.uploads/%s", tusPrefix, uploadID)) - if !strings.HasPrefix(uploadLocation, "/") { - uploadLocation = "/" + uploadLocation - } + uploadLocation := fs.tusUploadLocation(uploadID) // Handle creation-with-upload extension // TUS requires Content-Length for uploads; reject chunked encoding @@ -265,8 +269,9 @@ func (fs *FilerServer) tusCreateHandler(w http.ResponseWriter, r *http.Request) // Update offset in response header w.Header().Set("Upload-Offset", strconv.FormatInt(bytesWritten, 10)) - // Check if upload is complete - if bytesWritten == session.Size { + // Check if upload is complete; a partial upload keeps its chunks for + // a later concatenation instead of landing at the target path. + if bytesWritten == session.Size && !session.isPartial() { // Ensure the pinned session still exists, then refresh its chunks. if err = fs.refreshTusSessionChunks(ctx, session); err != nil { glog.Errorf("Failed to get updated TUS session: %v", err) @@ -287,8 +292,185 @@ func (fs *FilerServer) tusCreateHandler(w http.ResponseWriter, r *http.Request) w.WriteHeader(http.StatusCreated) } +// tusUploadLocation builds the upload URL path for a session id (single leading /) +func (fs *FilerServer) tusUploadLocation(uploadID string) string { + // TusBasePath is pre-normalized in filer_server.go (leading slash, no trailing slash) + uploadLocation := path.Clean(fmt.Sprintf("%s/.uploads/%s", fs.option.TusBasePath, uploadID)) + if !strings.HasPrefix(uploadLocation, "/") { + uploadLocation = "/" + uploadLocation + } + return uploadLocation +} + +// parseTusConcatFinal extracts the partial upload ids from an Upload-Concat +// final header. Each reference may be an absolute URL or a path, must point at +// this server's upload location, and may appear only once since concatenation +// consumes the partial. +func (fs *FilerServer) parseTusConcatFinal(concat string) ([]string, error) { + uploadsPrefix := fs.option.TusBasePath + "/.uploads/" + var partialIDs []string + seen := make(map[string]bool) + for _, ref := range strings.Fields(strings.TrimPrefix(concat, TusConcatFinalPrefix)) { + refURL, err := url.Parse(ref) + if err != nil { + return nil, fmt.Errorf("invalid partial upload reference %q", ref) + } + partialID := strings.TrimPrefix(refURL.Path, uploadsPrefix) + if partialID == refURL.Path || !isCanonicalTusUploadID(partialID) { + return nil, fmt.Errorf("invalid partial upload reference %q", ref) + } + if seen[partialID] { + return nil, fmt.Errorf("duplicate partial upload reference %q", ref) + } + seen[partialID] = true + partialIDs = append(partialIDs, partialID) + } + if len(partialIDs) == 0 { + return nil, errors.New("no partial uploads listed") + } + return partialIDs, nil +} + +// tusConcatFinalHandler handles POST requests carrying Upload-Concat final. It +// stitches the listed completed partial uploads, in order, into one entry at +// the request's target path. +func (fs *FilerServer) tusConcatFinalHandler(w http.ResponseWriter, r *http.Request, claims *security.SeaweedFilerClaims, concat string) { + ctx := context.WithoutCancel(r.Context()) + + // The final upload's length is the sum of the partial lengths. + if r.Header.Get("Upload-Length") != "" { + http.Error(w, "Upload-Length not allowed for a final upload", http.StatusBadRequest) + return + } + if r.ContentLength > 0 { + http.Error(w, "Cannot upload data to a final upload", http.StatusForbidden) + return + } + + partialIDs, err := fs.parseTusConcatFinal(concat) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + targetPath := fs.tusTargetPath(r) + if targetPath == "" || targetPath == "/" { + http.Error(w, "Target path required", http.StatusBadRequest) + return + } + if fs.filer.FilerConf.MatchStorageRule(targetPath).ReadOnly { + http.Error(w, ErrReadOnly.Error(), http.StatusInsufficientStorage) + return + } + + // Every partial must exist, be authorized for this credential, and be + // complete; concatenation-unfinished is not offered. Each one is claimed + // with an exclusive consumed marker before its chunks are read, so a + // concurrent final cannot consume the same partial and a concurrent DELETE + // or expiry cleanup cannot free chunks that move to the final entry. + var partials []*TusSession + var claimedIDs []string + var totalSize int64 + releaseClaims := func() { + for _, claimedID := range claimedIDs { + fs.rollbackTusSessionConsumed(ctx, claimedID) + } + } + for _, partialID := range partialIDs { + partial, err := fs.readTusSessionInfo(ctx, partialID) + if err != nil { + glog.V(1).Infof("TUS partial %s not resolved: %v", partialID, err) + releaseClaims() + http.Error(w, "Partial upload not found", http.StatusNotFound) + return + } + if !authorizeFilerJwtPaths(r, claims, []string{partial.TargetPath}) { + releaseClaims() + writeJsonError(w, r, http.StatusUnauthorized, errors.New("wrong jwt")) + return + } + if !partial.isPartial() { + releaseClaims() + http.Error(w, "Not a partial upload: "+partialID, http.StatusBadRequest) + return + } + totalSize += partial.Size + if totalSize > fs.option.TusMaxSize { + releaseClaims() + http.Error(w, "Combined upload size exceeds maximum", http.StatusRequestEntityTooLarge) + return + } + if err := fs.claimTusPartial(ctx, partial); err != nil { + releaseClaims() + switch { + case errors.Is(err, filer_pb.ErrEntryAlreadyExists): + http.Error(w, "Partial upload already consumed: "+partialID, http.StatusConflict) + case errors.Is(err, filer_pb.ErrNotFound): + glog.V(1).Infof("TUS partial %s changed before concatenation: %v", partialID, err) + http.Error(w, "Partial upload not found", http.StatusNotFound) + default: + glog.Errorf("Failed to claim TUS partial %s: %v", partialID, err) + http.Error(w, "Failed to create upload", http.StatusInternalServerError) + } + return + } + claimedIDs = append(claimedIDs, partialID) + if partial.Offset != partial.Size { + releaseClaims() + http.Error(w, "Partial upload not finished: "+partialID, http.StatusBadRequest) + return + } + partials = append(partials, partial) + } + + metadata := parseTusMetadata(r.Header.Get("Upload-Metadata")) + uploadID := uuid.New().String() + session, err := fs.createTusSession(ctx, uploadID, targetPath, totalSize, metadata, concat) + if err != nil { + glog.Errorf("Failed to create TUS session: %v", err) + releaseClaims() + http.Error(w, "Failed to create upload", http.StatusInternalServerError) + return + } + + // Re-base each partial's chunks onto the final upload's offset space. + for _, partial := range partials { + for _, chunk := range partial.Chunks { + session.Chunks = append(session.Chunks, &TusChunkInfo{ + Offset: session.Offset + chunk.Offset, + Size: chunk.Size, + FileId: chunk.FileId, + UploadAt: chunk.UploadAt, + }) + } + session.Offset += partial.Size + } + + if err := fs.completeTusUpload(ctx, session); err != nil { + fs.deleteTusSession(ctx, uploadID) + releaseClaims() + glog.Errorf("Failed to complete TUS concatenation: %v", err) + writeTusCompleteError(w, err) + return + } + + // The chunks now belong to the final entry, so remove only the partials' + // session metadata. + for _, partial := range partials { + if err := fs.filer.DeleteEntryMetaAndData(ctx, util.FullPath(fs.tusSessionPath(partial.ID)), true, false, false, false, nil, 0); err != nil { + glog.V(1).Infof("Failed to cleanup TUS partial session %s: %v", partial.ID, err) + } + } + + w.Header().Set("Location", fs.tusUploadLocation(uploadID)) + w.WriteHeader(http.StatusCreated) +} + // tusHeadHandler handles HEAD requests to get current upload offset func (fs *FilerServer) tusHeadHandler(w http.ResponseWriter, session *TusSession) { + if session.Concat != "" { + w.Header().Set("Upload-Concat", session.Concat) + } w.Header().Set("Upload-Offset", strconv.FormatInt(session.Offset, 10)) w.Header().Set("Upload-Length", strconv.FormatInt(session.Size, 10)) w.Header().Set("Cache-Control", "no-store") @@ -297,6 +479,11 @@ func (fs *FilerServer) tusHeadHandler(w http.ResponseWriter, session *TusSession // tusPatchHandler handles PATCH requests to upload data func (fs *FilerServer) tusPatchHandler(w http.ResponseWriter, r *http.Request, session *TusSession) { + if session.isFinal() { + http.Error(w, "Cannot PATCH a final upload", http.StatusForbidden) + return + } + // Use a context that ignores cancellation from the request context. // The filer's connection has an inactivity timeout: after the request body is fully read, // internal operations (assigning file IDs, uploading to volume servers, completing uploads) @@ -348,8 +535,9 @@ func (fs *FilerServer) tusPatchHandler(w http.ResponseWriter, r *http.Request, s newOffset := uploadOffset + bytesWritten - // Check if upload is complete - if newOffset == session.Size { + // Check if upload is complete; a partial upload keeps its chunks for a later + // concatenation instead of landing at the target path. + if newOffset == session.Size && !session.isPartial() { // Ensure the authorized session still exists, then refresh its chunks. if err = fs.refreshTusSessionChunks(ctx, session); err != nil { glog.Errorf("Failed to get updated TUS session: %v", err) diff --git a/weed/server/filer_server_tus_jwt_test.go b/weed/server/filer_server_tus_jwt_test.go index c8291ed2b..875a3ef84 100644 --- a/weed/server/filer_server_tus_jwt_test.go +++ b/weed/server/filer_server_tus_jwt_test.go @@ -46,9 +46,10 @@ func newTusTestServer(t *testing.T, sessions map[string]string) (*FilerServer, * t.Helper() store := newRenameTestStore() fs := &FilerServer{ - filer: newRenameTestFiler(t, store), - filerGuard: security.NewGuard(nil, tusTestWriteKey, 0, tusTestReadKey, 0), - option: &FilerOption{TusBasePath: "/.tus", TusMaxSize: TusDefaultMaxSize}, + filer: newRenameTestFiler(t, store), + filerGuard: security.NewGuard(nil, tusTestWriteKey, 0, tusTestReadKey, 0), + option: &FilerOption{TusBasePath: "/.tus", TusMaxSize: TusDefaultMaxSize}, + entryLockTable: util.NewLockTable[util.FullPath](), } for uploadID, targetPath := range sessions { seedTusSession(t, fs, store, TusSession{ID: uploadID, TargetPath: targetPath, Size: 1}) diff --git a/weed/server/filer_server_tus_session.go b/weed/server/filer_server_tus_session.go index 5c0494d44..e530ed7d8 100644 --- a/weed/server/filer_server_tus_session.go +++ b/weed/server/filer_server_tus_session.go @@ -25,8 +25,11 @@ const ( TusDefaultSessionExpiry = 24 * time.Hour TusUploadsFolder = ".uploads.tus" TusInfoFileName = ".info" + TusConsumedFileName = ".consumed" TusChunkExt = ".chunk" - TusExtensions = "creation,creation-with-upload,termination" + TusExtensions = "creation,creation-with-upload,termination,concatenation" + TusConcatPartial = "partial" + TusConcatFinalPrefix = "final;" ) // ErrWormEnforced marks a TUS completion rejected because the target entry is @@ -43,9 +46,20 @@ type TusSession struct { Metadata map[string]string `json:"metadata,omitempty"` CreatedAt time.Time `json:"created_at"` ExpiresAt time.Time `json:"expires_at,omitempty"` + Concat string `json:"concat,omitempty"` Chunks []*TusChunkInfo `json:"chunks,omitempty"` } +// isPartial reports whether the session is a concatenation partial upload: it +// holds chunks for a later final upload instead of landing at its target path. +func (session *TusSession) isPartial() bool { + return session.Concat == TusConcatPartial +} + +func (session *TusSession) isFinal() bool { + return strings.HasPrefix(session.Concat, TusConcatFinalPrefix) +} + // TusChunkInfo tracks individual chunk uploads within a session type TusChunkInfo struct { Offset int64 `json:"offset"` @@ -69,6 +83,68 @@ func (fs *FilerServer) tusSessionInfoPath(uploadID string) string { return fmt.Sprintf("/%s/%s/%s", TusUploadsFolder, uploadID, TusInfoFileName) } +// tusSessionConsumedPath returns the path of the marker recording that a +// session's chunks belong to a completed upload and must not be freed with it. +func (fs *FilerServer) tusSessionConsumedPath(uploadID string) string { + return fmt.Sprintf("/%s/%s/%s", TusUploadsFolder, uploadID, TusConsumedFileName) +} + +// markTusSessionConsumed claims a session's chunks for a completed upload. With +// exclusive set, a session already claimed by a concurrent request fails with +// filer_pb.ErrEntryAlreadyExists so one partial cannot be consumed twice. +func (fs *FilerServer) markTusSessionConsumed(ctx context.Context, uploadID string, exclusive bool) error { + return fs.filer.CreateEntry(ctx, &filer.Entry{ + FullPath: util.FullPath(fs.tusSessionConsumedPath(uploadID)), + Attr: filer.Attr{ + Mode: 0644, + Crtime: time.Now(), + Mtime: time.Now(), + Uid: OS_UID, + Gid: OS_GID, + }, + }, nil, exclusive, false, nil, true, fs.filer.MaxFilenameLength) +} + +// isTusSessionConsumed fails closed: when the marker cannot be looked up, the +// caller must not treat the session's chunks as free. +// claimTusPartial claims a partial for one final upload, serialized per session +// on this filer, and re-verifies the pinned session under the claim. A failed +// verification releases the claim before returning. +func (fs *FilerServer) claimTusPartial(ctx context.Context, partial *TusSession) error { + sessionPath := util.FullPath(fs.tusSessionPath(partial.ID)) + pathLock := fs.entryLockTable.AcquireLock("tusClaim", sessionPath, util.ExclusiveLock) + defer fs.entryLockTable.ReleaseLock(sessionPath, pathLock) + + if err := fs.markTusSessionConsumed(ctx, partial.ID, true); err != nil { + return err + } + if err := fs.refreshTusSessionChunks(ctx, partial); err != nil { + fs.rollbackTusSessionConsumed(ctx, partial.ID) + return err + } + return nil +} + +// rollbackTusSessionConsumed releases a consumed marker after a failed claim or +// completion. A failed rollback wedges the session as consumed: DELETE and +// expiry then preserve its chunks, which leak until removed by fsck. +func (fs *FilerServer) rollbackTusSessionConsumed(ctx context.Context, uploadID string) { + if err := fs.filer.DeleteEntryMetaAndData(ctx, util.FullPath(fs.tusSessionConsumedPath(uploadID)), false, false, false, false, nil, 0); err != nil && !errors.Is(err, filer_pb.ErrNotFound) { + glog.Errorf("TUS session %s wedged as consumed, marker rollback failed: %v", uploadID, err) + } +} + +func (fs *FilerServer) isTusSessionConsumed(ctx context.Context, uploadID string) (bool, error) { + _, err := fs.filer.FindEntry(ctx, util.FullPath(fs.tusSessionConsumedPath(uploadID))) + if err == nil { + return true, nil + } + if errors.Is(err, filer_pb.ErrNotFound) { + return false, nil + } + return false, err +} + // tusChunkPath returns the path to store a chunk info file // Format: /{TusUploadsFolder}/{uploadID}/chunk_{offset}_{size}_{encodedFileId} func (fs *FilerServer) tusChunkPath(uploadID string, offset, size int64, fileId string) string { @@ -117,7 +193,7 @@ func parseTusChunkPath(entry *filer.Entry) (*TusChunkInfo, error) { } // createTusSession creates a new TUS upload session -func (fs *FilerServer) createTusSession(ctx context.Context, uploadID, targetPath string, size int64, metadata map[string]string) (*TusSession, error) { +func (fs *FilerServer) createTusSession(ctx context.Context, uploadID, targetPath string, size int64, metadata map[string]string, concat string) (*TusSession, error) { session := &TusSession{ ID: uploadID, TargetPath: targetPath, @@ -126,6 +202,7 @@ func (fs *FilerServer) createTusSession(ctx context.Context, uploadID, targetPat Metadata: metadata, CreatedAt: time.Now(), ExpiresAt: time.Now().Add(fs.option.TusSessionExpiry), + Concat: concat, Chunks: []*TusChunkInfo{}, } @@ -288,11 +365,9 @@ func (fs *FilerServer) loadTusSessionChunks(ctx context.Context, session *TusSes return nil } -// refreshTusSessionChunks verifies the pinned session still exists and still -// identifies the same upload before refreshing its chunk state, so a PATCH -// cannot complete after a concurrent DELETE or metadata replacement and land at -// a TargetPath other than the one that was authorized. -func (fs *FilerServer) refreshTusSessionChunks(ctx context.Context, session *TusSession) error { +// verifyTusSessionUnchanged confirms the stored .info still identifies the same +// pinned session. +func (fs *FilerServer) verifyTusSessionUnchanged(ctx context.Context, session *TusSession) error { stored, err := fs.readTusSessionInfo(ctx, session.ID) if err != nil { return err @@ -300,6 +375,17 @@ func (fs *FilerServer) refreshTusSessionChunks(ctx context.Context, session *Tus if stored.TargetPath != session.TargetPath || stored.Size != session.Size || !stored.CreatedAt.Equal(session.CreatedAt) { return fmt.Errorf("TUS session identity changed: %s", session.ID) } + return nil +} + +// refreshTusSessionChunks verifies the pinned session still exists and still +// identifies the same upload before refreshing its chunk state, so a PATCH +// cannot complete after a concurrent DELETE or metadata replacement and land at +// a TargetPath other than the one that was authorized. +func (fs *FilerServer) refreshTusSessionChunks(ctx context.Context, session *TusSession) error { + if err := fs.verifyTusSessionUnchanged(ctx, session); err != nil { + return err + } return fs.loadTusSessionChunks(ctx, session) } @@ -333,6 +419,9 @@ func (fs *FilerServer) saveTusChunk(ctx context.Context, uploadID string, chunk // deleteTusSession removes a TUS upload session and all its data func (fs *FilerServer) deleteTusSession(ctx context.Context, uploadID string) error { + sessionPath := util.FullPath(fs.tusSessionPath(uploadID)) + pathLock := fs.entryLockTable.AcquireLock("tusDelete", sessionPath, util.ExclusiveLock) + defer fs.entryLockTable.ReleaseLock(sessionPath, pathLock) session, err := fs.getTusSession(ctx, uploadID) if err != nil { @@ -341,8 +430,21 @@ func (fs *FilerServer) deleteTusSession(ctx context.Context, uploadID string) er return nil } - // Batch delete all uploaded chunks from volume servers - if len(session.Chunks) > 0 { + // Remove the .info before deciding about chunk data: a final request claims + // its consumed marker before re-verifying the .info, so once the .info is + // gone no new claim can pass verification, and a claim that did pass was + // created earlier and is visible at the check below. + if err := fs.filer.DeleteEntryMetaAndData(ctx, util.FullPath(fs.tusSessionInfoPath(uploadID)), false, false, false, false, nil, 0); err != nil && !errors.Is(err, filer_pb.ErrNotFound) { + return fmt.Errorf("delete session info: %w", err) + } + + // Batch delete all uploaded chunks from volume servers, unless the session + // was consumed: then the chunks belong to a completed upload's entry. + consumed, err := fs.isTusSessionConsumed(ctx, uploadID) + if err != nil { + return fmt.Errorf("check consumed marker: %w", err) + } + if len(session.Chunks) > 0 && !consumed { var chunksToDelete []*filer_pb.FileChunk for _, chunk := range session.Chunks { if chunk.FileId != "" { @@ -370,6 +472,12 @@ func (fs *FilerServer) completeTusUpload(ctx context.Context, session *TusSessio return fmt.Errorf("upload incomplete: offset=%d, expected=%d", session.Offset, session.Size) } + // Serialize the ownership transition with deleteTusSession on this filer; + // the marker/.info handshake below covers a delete served by another filer. + sessionPath := util.FullPath(fs.tusSessionPath(session.ID)) + pathLock := fs.entryLockTable.AcquireLock("tusComplete", sessionPath, util.ExclusiveLock) + defer fs.entryLockTable.ReleaseLock(sessionPath, pathLock) + // Sort chunks by offset to ensure correct order sort.Slice(session.Chunks, func(i, j int) bool { return session.Chunks[i].Offset < session.Chunks[j].Offset @@ -428,6 +536,23 @@ func (fs *FilerServer) completeTusUpload(ctx context.Context, session *TusSessio return ErrWormEnforced } + // Claim the chunks for the entry before creating it: once the marker is + // durable, a failed cleanup below cannot lead DELETE or expiry to free the + // entry's chunks. If the entry creation then fails, the client retry re-runs + // completion; an abandoned marked session leaks its chunks instead of + // corrupting a live entry. + if err := fs.markTusSessionConsumed(ctx, session.ID, false); err != nil { + return fmt.Errorf("mark session consumed: %w", err) + } + + // Handshake with deleteTusSession, which removes the .info before checking + // the marker: a session whose .info is still present here cannot have its + // chunks freed by a delete that missed the marker just made durable. + if err := fs.verifyTusSessionUnchanged(ctx, session); err != nil { + fs.rollbackTusSessionConsumed(ctx, session.ID) + return fmt.Errorf("session deleted before completion: %w", err) + } + entry := &filer.Entry{ FullPath: targetPath, Attr: filer.Attr{