From f71c0fe5230de22d308cb0f3d9bb2a999d79c397 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: Sat, 7 Nov 2026 00:32:13 +0700 Subject: [PATCH] fix(s3tables): route REST RenameTable and refuse unrouted S3 Tables requests (#12446) * s3tables: route REST RenameTable and refuse unrouted S3 Tables requests PUT /tables////rename had no route, so it fell through to the /{bucket} routers and ran as a PutObject into a bucket named "tables", answering 2xx while the table kept its name. - Answer any s3tables-signed request no S3 Tables route takes with 404 UnknownOperationException, before the health, domain and bucket routers. - Route the AWS RenameTable REST call (newNamespaceName, newName, both optional; versionToken). - Honor versionToken on RenameTable, checked against the snapshot the claim asserts so a commit between the two reads cannot slip a stale token through. * s3tables: match the credential scope's service, not a region named s3tables The unrouted-request fallback widens what a false match costs: an S3 client whose region is named s3tables had every request answered as an unknown S3 Tables operation. Co-Authored-By: Claude Opus 5.5 * s3tables: refuse an empty newName or newNamespaceName on REST rename An empty field was read as omitted, so {"newNamespaceName":"other", "newName":""} moved the table under its old name instead of failing validation the way AWS does. Co-Authored-By: Claude Opus 5.5 * s3tables: check the rename version token after the snapshot comparison Once the separately read metadata is known to equal the claimed snapshot, the token compares against the parsed metadata; no second decode of the snapshot bytes is needed. Co-Authored-By: Claude Opus 5.5 * s3tables test: cover policy and tag changes in the rename snapshot compare Co-Authored-By: Claude Opus 5.5 * s3tables test: give each REST rename subtest its own source table The token subtests renamed the table the first subtest produced, so run alone they failed on a missing table instead of testing the token. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Chris Lu Co-authored-by: Claude Opus 5.5 --- test/s3tables/table-buckets/client.go | 38 ++ .../table-buckets/rename_rest_test.go | 116 +++++ weed/s3api/s3api_tables.go | 80 +++- weed/s3api/s3api_tables_rename_rest_test.go | 148 +++++++ weed/s3api/s3api_tables_routing_test.go | 415 ++++++++++++++++++ weed/s3api/s3tables/handler_rename_test.go | 191 ++++++++ weed/s3api/s3tables/handler_table.go | 42 ++ weed/s3api/s3tables/types.go | 3 + 8 files changed, 1027 insertions(+), 6 deletions(-) create mode 100644 test/s3tables/table-buckets/rename_rest_test.go create mode 100644 weed/s3api/s3api_tables_rename_rest_test.go diff --git a/test/s3tables/table-buckets/client.go b/test/s3tables/table-buckets/client.go index 3bde4eeb0a..1ffd5d5cca 100644 --- a/test/s3tables/table-buckets/client.go +++ b/test/s3tables/table-buckets/client.go @@ -336,6 +336,44 @@ func (c *S3TablesClient) DeleteTable(bucketARN string, namespace []string, name return c.doRestRequestAndDecode("DeleteTable", http.MethodDelete, path, nil, nil) } +// RenameTable renames a table through the AWS REST shape: the path names the +// source and the body the destination, where an empty newNamespace or newName +// keeps the source's. Any 2xx is success, as for the SDK (AWS answers 204). +func (c *S3TablesClient) RenameTable(bucketARN string, namespace []string, name, newNamespace, newName, versionToken string) error { + nameSpace, err := getFirstNamespace(namespace) + if err != nil { + return fmt.Errorf("RenameTable requires namespace: %w", err) + } + body := map[string]string{} + if newNamespace != "" { + body["newNamespaceName"] = newNamespace + } + if newName != "" { + body["newName"] = newName + } + if versionToken != "" { + body["versionToken"] = versionToken + } + path := "/tables/" + url.PathEscape(bucketARN) + "/" + url.PathEscape(nameSpace) + "/" + url.PathEscape(name) + "/rename" + resp, err := c.doRestRequest(http.MethodPut, path, body) + if err != nil { + return err + } + defer resp.Body.Close() + bodyBytes, readErr := io.ReadAll(resp.Body) + if resp.StatusCode >= 200 && resp.StatusCode < 300 { + return nil + } + if readErr != nil { + return fmt.Errorf("RenameTable failed with status %d and could not read error response body: %v", resp.StatusCode, readErr) + } + var errResp s3tables.S3TablesError + if err := json.Unmarshal(bodyBytes, &errResp); err != nil { + return fmt.Errorf("RenameTable failed with status %d, could not decode error response: %v. Body: %s", resp.StatusCode, err, string(bodyBytes)) + } + return fmt.Errorf("RenameTable failed: %s - %s", errResp.Type, errResp.Message) +} + // Policy operations func (c *S3TablesClient) PutTableBucketPolicy(bucketARN, policy string) error { diff --git a/test/s3tables/table-buckets/rename_rest_test.go b/test/s3tables/table-buckets/rename_rest_test.go new file mode 100644 index 0000000000..4f7146d240 --- /dev/null +++ b/test/s3tables/table-buckets/rename_rest_test.go @@ -0,0 +1,116 @@ +package s3tables + +import ( + "context" + "testing" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" +) + +// emptyObjectBucket deletes every object in bucket. +func emptyObjectBucket(t *testing.T, ctx context.Context, client *s3.Client, bucket string) { + t.Helper() + pages := s3.NewListObjectsV2Paginator(client, &s3.ListObjectsV2Input{Bucket: aws.String(bucket)}) + for pages.HasMorePages() { + page, err := pages.NextPage(ctx) + require.NoError(t, err) + for _, o := range page.Contents { + _, err := client.DeleteObject(ctx, &s3.DeleteObjectInput{Bucket: aws.String(bucket), Key: o.Key}) + require.NoError(t, err) + } + } +} + +// The AWS SDK sends RenameTable as PUT /tables////rename. With +// no route for it the gateway ran it as a PutObject into an S3 bucket named +// "tables" and answered 2xx, so the rename reported success while the table +// kept its old name. The bucket is created up front so that failure shows as +// an object landing in it. +func TestRenameTableREST(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + ctx := context.Background() + tables := NewS3TablesClient(sharedCluster.s3Endpoint, testRegion, testAccessKey, testSecretKey) + objects := newObjectClient(t) + + const decoyBucket = "tables" + if _, err := objects.CreateBucket(ctx, &s3.CreateBucketInput{Bucket: aws.String(decoyBucket)}); err != nil { + require.Equal(t, "BucketAlreadyOwnedByYou", apiErrorCode(err), "create the %q bucket: %v", decoyBucket, err) + } + emptyObjectBucket(t, ctx, objects, decoyBucket) + defer func() { + emptyObjectBucket(t, ctx, objects, decoyBucket) + _, _ = objects.DeleteBucket(ctx, &s3.DeleteBucketInput{Bucket: aws.String(decoyBucket)}) + }() + + created, err := tables.CreateTableBucket("rename-rest-"+randomString(8), nil) + require.NoError(t, err) + defer tables.DeleteTableBucket(created.ARN) + bucketARN := created.ARN + + const namespace, otherNamespace = "rename_ns", "rename_other" + for _, ns := range []string{namespace, otherNamespace} { + _, err := tables.CreateNamespace(bucketARN, []string{ns}) + require.NoError(t, err) + defer tables.DeleteNamespace(bucketARN, []string{ns}) + } + + // Each subtest owns its source table, so any one of them runs alone. + createSource := func(t *testing.T, name string) string { + resp, err := tables.CreateTable(bucketARN, []string{namespace}, name, "ICEBERG", nil, nil) + require.NoError(t, err) + return resp.VersionToken + } + + t.Run("newName only", func(t *testing.T) { + createSource(t, "src") + defer tables.DeleteTable(bucketARN, []string{namespace}, "src") + defer tables.DeleteTable(bucketARN, []string{namespace}, "dst") + require.NoError(t, tables.RenameTable(bucketARN, []string{namespace}, "src", "", "dst", "")) + + got, err := tables.GetTable(bucketARN, []string{namespace}, "dst") + require.NoError(t, err, "the renamed table must resolve under its new name") + assert.Equal(t, "dst", got.Name) + + _, err = tables.GetTable(bucketARN, []string{namespace}, "src") + require.Error(t, err) + assert.Contains(t, err.Error(), s3tables.ErrCodeNoSuchTable) + + list, err := objects.ListObjectsV2(ctx, &s3.ListObjectsV2Input{Bucket: aws.String(decoyBucket)}) + require.NoError(t, err) + assert.Empty(t, list.Contents, "the rename must not store anything in the %q bucket", decoyBucket) + }) + + t.Run("stale versionToken", func(t *testing.T) { + token := createSource(t, "stale_src") + defer tables.DeleteTable(bucketARN, []string{namespace}, "stale_src") + defer tables.DeleteTable(bucketARN, []string{otherNamespace}, "stale_moved") + err := tables.RenameTable(bucketARN, []string{namespace}, "stale_src", otherNamespace, "stale_moved", token+"-stale") + require.Error(t, err) + assert.Contains(t, err.Error(), s3tables.ErrCodeConflict) + + _, err = tables.GetTable(bucketARN, []string{namespace}, "stale_src") + require.NoError(t, err, "a refused rename must leave the source in place") + }) + + t.Run("newNamespaceName and newName with the current versionToken", func(t *testing.T) { + token := createSource(t, "tok_src") + defer tables.DeleteTable(bucketARN, []string{namespace}, "tok_src") + defer tables.DeleteTable(bucketARN, []string{otherNamespace}, "moved") + require.NoError(t, tables.RenameTable(bucketARN, []string{namespace}, "tok_src", otherNamespace, "moved", token)) + + got, err := tables.GetTable(bucketARN, []string{otherNamespace}, "moved") + require.NoError(t, err) + assert.Equal(t, "moved", got.Name) + + _, err = tables.GetTable(bucketARN, []string{namespace}, "tok_src") + require.Error(t, err) + assert.Contains(t, err.Error(), s3tables.ErrCodeNoSuchTable) + }) +} diff --git a/weed/s3api/s3api_tables.go b/weed/s3api/s3api_tables.go index 6b50017197..cd2143fc31 100644 --- a/weed/s3api/s3api_tables.go +++ b/weed/s3api/s3api_tables.go @@ -141,6 +141,8 @@ func (s3a *S3ApiServer) registerS3TablesRoutes(router *mux.Router) { HandlerFunc(track(s3a.authenticateS3Tables(s3TablesApi.handleRestOperation("ListTables", buildListTablesRequest)), "S3Tables-ListTables")) router.Methods(http.MethodDelete).Path("/tables/{tableBucketARN:" + tableBucketARNRegex + "}/{namespace}/{name}").MatcherFunc(serviceMatcher). HandlerFunc(track(s3a.authenticateS3Tables(s3TablesApi.handleRestOperation("DeleteTable", buildDeleteTableRequest)), "S3Tables-DeleteTable")) + router.Methods(http.MethodPut).Path("/tables/{tableBucketARN:" + tableBucketARNRegex + "}/{namespace}/{name}/rename").MatcherFunc(serviceMatcher). + HandlerFunc(track(s3a.authenticateS3Tables(s3TablesApi.handleRestOperation("RenameTable", buildRenameTableRequest)), "S3Tables-RenameTable")) router.Methods(http.MethodPut).Path("/tables/{tableBucketARN:" + tableBucketARNRegex + "}/{namespace}/{name}/policy").MatcherFunc(serviceMatcher). HandlerFunc(track(s3a.authenticateS3Tables(s3TablesApi.handleRestOperation("PutTablePolicy", buildPutTablePolicyRequest)), "S3Tables-PutTablePolicy")) @@ -170,9 +172,25 @@ func (s3a *S3ApiServer) registerS3TablesRoutes(router *mux.Router) { router.Methods(http.MethodGet).Path("/get-table").MatcherFunc(serviceMatcher). HandlerFunc(track(s3a.authenticateS3Tables(s3TablesApi.handleRestOperation("GetTable", buildGetTableRequest)), "S3Tables-GetTable")) + // Every S3 Tables operation this gateway serves has a route above. A request + // signed for s3tables that none of them took is answered here: the /{bucket} + // routers registered later would run it as an S3 request on a bucket named + // after its first path segment, where a PUT stores the body as an object (or + // creates a bucket) and answers 2xx, which the SDK reports as success. + router.MatcherFunc(serviceMatcher). + HandlerFunc(track(handleUnroutedS3TablesRequest, "S3Tables-Unrouted")) + glog.V(1).Infof("S3 Tables API enabled") } +const s3TablesErrUnknownOperation = "UnknownOperationException" + +func handleUnroutedS3TablesRequest(w http.ResponseWriter, r *http.Request) { + glog.V(1).Infof("S3Tables: no route for %s %s", r.Method, r.URL.Path) + writeS3TablesError(w, http.StatusNotFound, s3TablesErrUnknownOperation, + "S3 Tables operation not supported: "+r.Method+" "+r.URL.Path) +} + type s3tablesRequestBuilder func(r *http.Request) (interface{}, error) func (st *S3TablesApiServer) handleRestOperation(operation string, builder s3tablesRequestBuilder) http.HandlerFunc { @@ -531,6 +549,58 @@ func buildDeleteTableRequest(r *http.Request) (interface{}, error) { }, nil } +// renameTableRESTBody is the AWS RenameTable body: the path names the source, +// the body only the destination, and both destination fields are optional. +type renameTableRESTBody struct { + NewNamespaceName *string `json:"newNamespaceName"` + NewName *string `json:"newName"` + VersionToken string `json:"versionToken"` +} + +func buildRenameTableRequest(r *http.Request) (interface{}, error) { + var body renameTableRESTBody + if err := readS3TablesJSONBody(r, &body); err != nil { + return nil, err + } + tableBucketARN, err := getDecodedPathParam(r, "tableBucketARN") + if err != nil { + return nil, err + } + namespace, err := parseRequiredNamespacePathParam(r, "namespace") + if err != nil { + return nil, err + } + name, err := getDecodedPathParam(r, "name") + if err != nil { + return nil, err + } + if name == "" { + return nil, fmt.Errorf("name is required") + } + if _, err := s3tables.ValidateTableName(name); err != nil { + return nil, err + } + if body.NewNamespaceName == nil && body.NewName == nil { + return nil, fmt.Errorf("newNamespaceName or newName is required") + } + destNamespace, destName := namespace, name + if body.NewNamespaceName != nil { + if destNamespace, err = s3tables.ParseNamespace(*body.NewNamespaceName); err != nil { + return nil, fmt.Errorf("invalid newNamespaceName: %w", err) + } + } + if body.NewName != nil { + if _, err := s3tables.ValidateTableName(*body.NewName); err != nil { + return nil, fmt.Errorf("invalid newName: %w", err) + } + destName = *body.NewName + } + return &s3tables.RenameTableRequest{ + TableBucketARN: tableBucketARN, SourceNamespace: namespace, SourceName: name, + DestNamespace: destNamespace, DestName: destName, VersionToken: body.VersionToken, + }, nil +} + func buildPutTablePolicyRequest(r *http.Request) (interface{}, error) { var req s3tables.PutTablePolicyRequest if err := readS3TablesJSONBody(r, &req); err != nil { @@ -674,12 +744,10 @@ func buildUntagResourceRequest(r *http.Request) (interface{}, error) { // internal test harnesses running against a default-allow server — // must sign with SERVICE=s3tables. func isS3TablesSignedRequest(r *http.Request) bool { - scope := extractCredentialScope(r) - // Credential scope is AK/DATE/REGION/SERVICE/aws4_request. Slashes - // do not appear inside any other component (access keys are - // alphanumeric), so /s3tables/ matches iff SERVICE is exactly - // s3tables. - return scope != "" && strings.Contains(scope, "/s3tables/") + // AK/DATE/REGION/SERVICE/aws4_request, split the way parseCredentialHeader + // does: a region named s3tables must not match. + parts := strings.Split(extractCredentialScope(r), "/") + return len(parts) == 5 && parts[3] == "s3tables" } // extractCredentialScope returns the raw credential value from either the diff --git a/weed/s3api/s3api_tables_rename_rest_test.go b/weed/s3api/s3api_tables_rename_rest_test.go new file mode 100644 index 0000000000..8b686ccd3f --- /dev/null +++ b/weed/s3api/s3api_tables_rename_rest_test.go @@ -0,0 +1,148 @@ +package s3api + +import ( + "net/http" + "net/http/httptest" + "reflect" + "strings" + "testing" + + "github.com/gorilla/mux" + + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" +) + +// TestBuildRenameTableRequest covers the AWS RenameTable REST shape: the path +// names the source and the body only the destination, either half of which +// may be omitted to keep the source's. +func TestBuildRenameTableRequest(t *testing.T) { + const encodedARN = "arn%3Aaws%3As3tables%3Aus-east-1%3A123456789012%3Abucket%2Ftest-bucket" + cases := []struct { + name string + arn string + namespace string + table string + body string + want *s3tables.RenameTableRequest + wantErr string + }{ + { + name: "newName only keeps the namespace", + body: `{"newName":"t2"}`, + want: &s3tables.RenameTableRequest{DestNamespace: []string{"ns"}, DestName: "t2"}, + }, + { + name: "newNamespaceName only keeps the name", + body: `{"newNamespaceName":"other"}`, + want: &s3tables.RenameTableRequest{DestNamespace: []string{"other"}, DestName: "t"}, + }, + { + name: "both destination fields", + body: `{"newNamespaceName":"other","newName":"t2"}`, + want: &s3tables.RenameTableRequest{DestNamespace: []string{"other"}, DestName: "t2"}, + }, + { + name: "percent-encoded ARN is decoded", + arn: encodedARN, + body: `{"newName":"t2"}`, + want: &s3tables.RenameTableRequest{DestNamespace: []string{"ns"}, DestName: "t2"}, + }, + { + name: "versionToken is passed through", + body: `{"newName":"t2","versionToken":"tok-1"}`, + want: &s3tables.RenameTableRequest{DestNamespace: []string{"ns"}, DestName: "t2", VersionToken: "tok-1"}, + }, + { + name: "empty body", + body: "", + wantErr: "newNamespaceName or newName is required", + }, + { + name: "empty object", + body: `{}`, + wantErr: "newNamespaceName or newName is required", + }, + { + // The X-Amz-Target field names must not be read as the REST ones: + // a body naming only destName renames nothing it asked for. + name: "SeaweedFS-shaped body", + body: `{"destNamespace":["other"],"destName":"t2"}`, + wantErr: "newNamespaceName or newName is required", + }, + { + name: "invalid newName", + body: `{"newName":"bad/name"}`, + wantErr: "invalid newName", + }, + { + name: "invalid newNamespaceName", + body: `{"newNamespaceName":".."}`, + wantErr: "invalid newNamespaceName", + }, + { + name: "empty newName beside a valid newNamespaceName", + body: `{"newNamespaceName":"other","newName":""}`, + wantErr: "invalid newName", + }, + { + name: "empty newNamespaceName beside a valid newName", + body: `{"newNamespaceName":"","newName":"t2"}`, + wantErr: "invalid newNamespaceName", + }, + { + name: "source namespace with ..", + namespace: "..", + body: `{"newName":"t2"}`, + wantErr: "invalid path parameter namespace", + }, + { + name: "invalid source name", + table: "..", + body: `{"newName":"t2"}`, + wantErr: "invalid path parameter name", + }, + { + name: "malformed body", + body: `{"newName":`, + wantErr: "unexpected end of JSON input", + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + arn, namespace, table := testTableBucketARN, "ns", "t" + if tc.arn != "" { + arn = tc.arn + } + if tc.namespace != "" { + namespace = tc.namespace + } + if tc.table != "" { + table = tc.table + } + req := httptest.NewRequest(http.MethodPut, "/tables/x/ns/t/rename", strings.NewReader(tc.body)) + req = mux.SetURLVars(req, map[string]string{"tableBucketARN": arn, "namespace": namespace, "name": table}) + + got, err := buildRenameTableRequest(req) + if tc.wantErr != "" { + if err == nil { + t.Fatalf("got %+v, want an error containing %q", got, tc.wantErr) + } + if !strings.Contains(err.Error(), tc.wantErr) { + t.Fatalf("error = %q, want it to contain %q", err.Error(), tc.wantErr) + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + want := *tc.want + want.TableBucketARN = testTableBucketARN + want.SourceNamespace = []string{"ns"} + want.SourceName = "t" + if !reflect.DeepEqual(got, &want) { + t.Fatalf("got %+v, want %+v", got, &want) + } + }) + } +} diff --git a/weed/s3api/s3api_tables_routing_test.go b/weed/s3api/s3api_tables_routing_test.go index 349ccbe595..4ffbfb1e6b 100644 --- a/weed/s3api/s3api_tables_routing_test.go +++ b/weed/s3api/s3api_tables_routing_test.go @@ -1,11 +1,29 @@ package s3api import ( + "context" + "encoding/json" + "fmt" + "net" "net/http" "net/http/httptest" + "path" + "strings" + "sync" "testing" + "time" + "github.com/aws/aws-sdk-go/aws/credentials" + v4 "github.com/aws/aws-sdk-go/aws/signer/v4" "github.com/gorilla/mux" + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" + "github.com/seaweedfs/seaweedfs/weed/wdclient" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" ) // TestIsS3TablesSignedRequest covers the credential-scope parser used to @@ -52,6 +70,16 @@ func TestIsS3TablesSignedRequest(t *testing.T) { auth: "AWS4-HMAC-SHA256 Credential=s3tablesAKIA/20260101/us-east-1/s3/aws4_request, Signature=zzz", want: false, }, + { + name: "region named s3tables on a regular S3 request", + auth: "AWS4-HMAC-SHA256 Credential=AKIA/20260101/s3tables/s3/aws4_request, Signature=zzz", + want: false, + }, + { + name: "region named s3tables on a presigned S3 request", + cred: "AKIA/20260101/s3tables/s3/aws4_request", + want: false, + }, } for _, tc := range cases { @@ -156,3 +184,390 @@ func TestRouting_GetObjectOnBucketNamedGetTable(t *testing.T) { t.Fatalf("GET /get-table signed for service=s3 matched the S3 Tables GetTable route (Path=%q); it must fall through to the bucket subrouter", tmpl) } } + +// recordingFiler answers every lookup with not-found and records each call the +// gateway makes, with the path it names, so a test can tell whether a request +// reached a handler that talks to the filer at all. +type recordingFiler struct { + filer_pb.UnimplementedSeaweedFilerServer + mu sync.Mutex + calls []string +} + +func (f *recordingFiler) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { + return nil, status.Errorf(codes.NotFound, "%s: %s/%s", filer_pb.ErrNotFound.Error(), req.Directory, req.Name) +} + +func (f *recordingFiler) record(method string, req interface{}) { + call := method + if d, ok := req.(interface{ GetDirectory() string }); ok { + name := "" + if n, ok := req.(interface{ GetName() string }); ok { + name = n.GetName() + } else if e, ok := req.(interface{ GetEntry() *filer_pb.Entry }); ok && e.GetEntry() != nil { + name = e.GetEntry().Name + } + call += " " + path.Join(d.GetDirectory(), name) + } + f.mu.Lock() + f.calls = append(f.calls, call) + f.mu.Unlock() +} + +func (f *recordingFiler) Calls() []string { + f.mu.Lock() + defer f.mu.Unlock() + return append([]string(nil), f.calls...) +} + +func (f *recordingFiler) reset() { + f.mu.Lock() + f.calls = nil + f.mu.Unlock() +} + +type recordingServerStream struct { + grpc.ServerStream + f *recordingFiler + method string +} + +func (s *recordingServerStream) RecvMsg(m interface{}) error { + err := s.ServerStream.RecvMsg(m) + if err == nil { + s.f.record(s.method, m) + } + return err +} + +// startRecordingFiler serves f on a random localhost port, recording every +// unary and streaming call before it is handled. +func startRecordingFiler(t *testing.T, f *recordingFiler) pb.ServerAddress { + t.Helper() + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + srv := grpc.NewServer( + grpc.UnaryInterceptor(func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) { + f.record(info.FullMethod, req) + return handler(ctx, req) + }), + grpc.StreamInterceptor(func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error { + return handler(srv, &recordingServerStream{ServerStream: ss, f: f, method: info.FullMethod}) + }), + ) + filer_pb.RegisterSeaweedFilerServer(srv, f) + go srv.Serve(lis) + t.Cleanup(srv.Stop) + return pb.ServerAddress(fmt.Sprintf("127.0.0.1:1.%d", lis.Addr().(*net.TCPAddr).Port)) +} + +// newS3TablesServeTestServer builds the production router over a gateway whose +// filer is a recordingFiler, wired far enough (identity account, circuit +// breaker, bucket config cache) that a request the S3 routes take runs its +// handler instead of panicking on a nil dependency. domainName is the +// -domainName option, empty for none. +func newS3TablesServeTestServer(t *testing.T, domainName string) (*mux.Router, *recordingFiler) { + t.Helper() + s3a := setupRoutingTestServer(t) + s3a.iam.m.Lock() + s3a.iam.accessKeyIdent[routingTestAccessKey].Account = &AccountAdmin + s3a.iam.m.Unlock() + // Built before the filer is wired so loading its config is not recorded. + s3a.cb = NewCircuitBreaker(s3a.option) + s3a.bucketConfigCache = NewBucketConfigCache(time.Minute) + + f := &recordingFiler{} + dialOption := grpc.WithTransportCredentials(insecure.NewCredentials()) + s3a.option.Filers = []pb.ServerAddress{startRecordingFiler(t, f)} + s3a.option.GrpcDialOption = dialOption + s3a.option.BucketsPath = "/buckets" + s3a.option.DomainName = domainName + s3a.filerClient = wdclient.NewFilerClient(s3a.option.Filers, dialOption, "") + + router := mux.NewRouter().SkipClean(true) + s3a.registerRouter(router) + f.reset() + return router, f +} + +// routingTestTableBucketARN is a table bucket ARN %-encoded the way the AWS +// SDK puts it in a REST path. +const routingTestTableBucketARN = "arn%3Aaws%3As3tables%3Aus-east-1%3A123456789012%3Abucket%2Ftable-bucket" + +// unroutedS3TablesOperations are S3 Tables REST shapes this gateway has no +// route for. Every one of them also parses as an object key in a bucket named +// after the first path segment. +var unroutedS3TablesOperations = []struct { + name string + method string + path string + body string +}{ + {"UpdateTableMetadataLocation", http.MethodPut, "/tables/" + routingTestTableBucketARN + "/ns/t/metadata-location", `{"versionToken":"v","metadataLocation":"s3://b/m.json"}`}, + {"GetTableMetadataLocation", http.MethodGet, "/tables/" + routingTestTableBucketARN + "/ns/t/metadata-location", ""}, + {"PutTableBucketEncryption", http.MethodPut, "/buckets/" + routingTestTableBucketARN + "/encryption", `{"encryptionConfiguration":{"sseAlgorithm":"AES256"}}`}, + {"GetTableBucketEncryption", http.MethodGet, "/buckets/" + routingTestTableBucketARN + "/encryption", ""}, + {"DeleteTableBucketEncryption", http.MethodDelete, "/buckets/" + routingTestTableBucketARN + "/encryption", ""}, + {"GetTableEncryption", http.MethodGet, "/tables/" + routingTestTableBucketARN + "/ns/t/encryption", ""}, + // A served path under a method it does not serve: gorilla drops the + // method mismatch once a later route matches, so these take the fallback + // rather than answering 405. + {"table policy under POST", http.MethodPost, "/tables/" + routingTestTableBucketARN + "/ns/t/policy", `{"resourcePolicy":"{}"}`}, + {"bucket policy under POST", http.MethodPost, "/buckets/" + routingTestTableBucketARN + "/policy", `{"resourcePolicy":"{}"}`}, + {"rename under GET", http.MethodGet, "/tables/" + routingTestTableBucketARN + "/ns/t/rename", ""}, +} + +func newRoutingTestRequest(t *testing.T, method, host, target, body, service string) *http.Request { + t.Helper() + req, err := http.NewRequest(method, "http://"+host+target, strings.NewReader(body)) + if err != nil { + t.Fatal(err) + } + if body != "" { + req.Header.Set("Content-Type", "application/x-amz-json-1.1") + } + if service != "" { + signRoutingTestRequest(t, req, body, service) + } + return req +} + +// isBucketOrObjectRoute reports whether a match landed on one of the S3 +// bucket routers, which bind the bucket from the path or the host. +func isBucketOrObjectRoute(match *mux.RouteMatch) bool { + if _, ok := match.Vars["bucket"]; ok { + return true + } + _, ok := match.Vars["object"] + return ok +} + +// assertUnknownOperation checks rr is the fallback's answer. +func assertUnknownOperation(t *testing.T, rr *httptest.ResponseRecorder) { + t.Helper() + if rr.Code != http.StatusNotFound { + t.Fatalf("status = %d, want 404; body %s", rr.Code, rr.Body.String()) + } + if ct := rr.Header().Get("Content-Type"); ct != "application/x-amz-json-1.1" { + t.Fatalf("Content-Type = %q, want application/x-amz-json-1.1", ct) + } + var resp struct { + Type string `json:"__type"` + } + if err := json.Unmarshal(rr.Body.Bytes(), &resp); err != nil { + t.Fatalf("decode body %q: %v", rr.Body.String(), err) + } + if resp.Type != s3TablesErrUnknownOperation { + t.Fatalf("__type = %q, want %q", resp.Type, s3TablesErrUnknownOperation) + } +} + +// TestRouting_UnroutedS3TablesOperationsNeverReachS3Handlers covers S3 Tables +// operations the gateway does not serve. Without a route of their own they +// fell through to the /{bucket} routers and ran as S3 requests — a PUT stored +// its body as an object in a bucket named "tables" and answered 2xx, which the +// SDK reports as success. They must be refused before any filer call, on the +// path-style, path-style-domain and virtual-host routers alike. +func TestRouting_UnroutedS3TablesOperationsNeverReachS3Handlers(t *testing.T) { + routers := []struct { + name string + domainName string + host string + }{ + {"no domain", "", "localhost:8333"}, + {"path-style domain", "s3.example.com,example.com", "s3.example.com"}, + {"virtual-host domain", "s3.example.com,example.com", "tables.example.com"}, + } + for _, rt := range routers { + t.Run(rt.name, func(t *testing.T) { + router, filer := newS3TablesServeTestServer(t, rt.domainName) + for _, tc := range unroutedS3TablesOperations { + t.Run(tc.name, func(t *testing.T) { + filer.reset() + req := newRoutingTestRequest(t, tc.method, rt.host, tc.path, tc.body, "s3tables") + + var match mux.RouteMatch + if !router.Match(req, &match) { + t.Fatalf("%s %s matched no route: %v", tc.method, tc.path, match.MatchErr) + } + tmpl, _ := match.Route.GetPathTemplate() + if tmpl == "/{bucket}" || tmpl == "/{bucket}/{object:(?s).+}" || isBucketOrObjectRoute(&match) { + t.Fatalf("%s %s signed for s3tables matched the S3 route %q (vars %v)", tc.method, tc.path, tmpl, match.Vars) + } + + rr := httptest.NewRecorder() + router.ServeHTTP(rr, newRoutingTestRequest(t, tc.method, rt.host, tc.path, tc.body, "s3tables")) + assertUnknownOperation(t, rr) + if calls := filer.Calls(); len(calls) != 0 { + t.Fatalf("an unrouted S3 Tables request reached the filer: %v", calls) + } + }) + } + }) + } +} + +// TestRouting_S3SignedArnShapedKeysStayObjectRequests is the other side of the +// fallback: the same paths signed for s3 are object keys shaped like ARNs in a +// bucket named "tables" or "buckets", and must keep reaching the S3 routes. +func TestRouting_S3SignedArnShapedKeysStayObjectRequests(t *testing.T) { + router, _ := newS3TablesServeTestServer(t, "") + for _, tc := range unroutedS3TablesOperations { + // A bare POST on an object path is no S3 operation, so the bucket + // routers never took it; it is not what this test guards. + if tc.method == http.MethodPost { + continue + } + t.Run(tc.name, func(t *testing.T) { + req := newRoutingTestRequest(t, tc.method, "localhost:8333", tc.path, tc.body, "s3") + var match mux.RouteMatch + if !router.Match(req, &match) { + t.Fatalf("%s %s matched no route: %v", tc.method, tc.path, match.MatchErr) + } + wantBucket := strings.SplitN(strings.TrimPrefix(tc.path, "/"), "/", 2)[0] + if got := match.Vars["bucket"]; got != wantBucket { + tmpl, _ := match.Route.GetPathTemplate() + t.Fatalf("%s %s signed for s3 matched %q with bucket %q; want the S3 route for bucket %q", tc.method, tc.path, tmpl, got, wantBucket) + } + }) + } +} + +// An S3 client configured with a region named s3tables still signs for +// service s3, so its requests must stay on the bucket routers. +func TestRouting_S3RegionNamedS3TablesStaysObjectRequest(t *testing.T) { + router, _ := newS3TablesServeTestServer(t, "") + const body = "data" + req, err := http.NewRequest(http.MethodPut, "http://localhost:8333/tables/"+routingTestTableBucketARN+"/ns/t/metadata-location", strings.NewReader(body)) + if err != nil { + t.Fatal(err) + } + signer := v4.NewSigner(credentials.NewStaticCredentials(routingTestAccessKey, routingTestSecretKey, "")) + if _, err := signer.Sign(req, strings.NewReader(body), "s3", "s3tables", time.Now()); err != nil { + t.Fatalf("sign request: %v", err) + } + var match mux.RouteMatch + if !router.Match(req, &match) { + t.Fatalf("PUT matched no route: %v", match.MatchErr) + } + if got := match.Vars["bucket"]; got != "tables" { + tmpl, _ := match.Route.GetPathTemplate() + t.Fatalf("S3 request in region s3tables matched %q with bucket %q; want the S3 route for bucket \"tables\"", tmpl, got) + } +} + +// TestRouting_RenameTableRESTReachesS3Tables covers the AWS RenameTable REST +// call. It used to fall through to the bucket router as a PutObject into a +// bucket named "tables"; it must reach the S3 Tables rename handler, answer in +// JSON, and never touch anything under that bucket. +func TestRouting_RenameTableRESTReachesS3Tables(t *testing.T) { + router, filer := newS3TablesServeTestServer(t, "") + const target = "/tables/" + routingTestTableBucketARN + "/ns/t/rename" + const body = `{"newName":"t2"}` + + req := newRoutingTestRequest(t, http.MethodPut, "localhost:8333", target, body, "s3tables") + var match mux.RouteMatch + if !router.Match(req, &match) { + t.Fatalf("RenameTable matched no route: %v", match.MatchErr) + } + tmpl, _ := match.Route.GetPathTemplate() + if !strings.HasPrefix(tmpl, "/tables/{tableBucketARN:") || !strings.HasSuffix(tmpl, "}/{namespace}/{name}/rename") { + t.Fatalf("RenameTable matched %q, want the rename route", tmpl) + } + + rr := httptest.NewRecorder() + router.ServeHTTP(rr, newRoutingTestRequest(t, http.MethodPut, "localhost:8333", target, body, "s3tables")) + if ct := rr.Header().Get("Content-Type"); ct != "application/x-amz-json-1.1" { + t.Fatalf("Content-Type = %q, want application/x-amz-json-1.1; body %s", ct, rr.Body.String()) + } + var resp struct { + Type string `json:"__type"` + } + if err := json.Unmarshal(rr.Body.Bytes(), &resp); err != nil { + t.Fatalf("decode body %q: %v", rr.Body.String(), err) + } + // The recording filer holds no tables, so the handler reports the source + // missing — an answer only the rename handler gives. + if rr.Code != http.StatusNotFound || resp.Type != s3tables.ErrCodeNoSuchTable { + t.Fatalf("status %d __type %q, want 404 %s from the rename handler; body %s", rr.Code, resp.Type, s3tables.ErrCodeNoSuchTable, rr.Body.String()) + } + + calls := filer.Calls() + if len(calls) == 0 { + t.Fatal("the rename handler never looked the source up") + } + for _, call := range calls { + if fields := strings.Fields(call); len(fields) == 2 && (fields[1] == "/buckets/tables" || strings.HasPrefix(fields[1], "/buckets/tables/")) { + t.Fatalf("RenameTable touched the S3 bucket \"tables\": %v", calls) + } + } +} + +// TestRouting_S3TablesFallbackDoesNotShadowOtherRoutes checks that the +// fallback only takes requests no other route claims: unsigned CORS +// preflights, health probes and STS keep their routes, and a served S3 Tables +// operation keeps its own. +func TestRouting_S3TablesFallbackDoesNotShadowOtherRoutes(t *testing.T) { + router, _ := newS3TablesServeTestServer(t, "") + const host = "localhost:8333" + + t.Run("bucket CORS preflight", func(t *testing.T) { + req := newRoutingTestRequest(t, http.MethodOptions, host, "/tables/"+routingTestTableBucketARN+"/ns/t", "", "") + req.Header.Set("Origin", "https://example.com") + req.Header.Set("Access-Control-Request-Method", http.MethodPut) + var match mux.RouteMatch + if !router.Match(req, &match) || match.Vars["bucket"] != "tables" { + t.Fatalf("OPTIONS preflight matched vars %v, want the bucket OPTIONS route", match.Vars) + } + }) + + t.Run("health probe preflight", func(t *testing.T) { + req := newRoutingTestRequest(t, http.MethodOptions, host, "/healthz", "", "") + req.Header.Set("Origin", "https://example.com") + var match mux.RouteMatch + if !router.Match(req, &match) { + t.Fatalf("OPTIONS /healthz matched no route: %v", match.MatchErr) + } + if tmpl, _ := match.Route.GetPathTemplate(); tmpl != "/{probe:status|healthz|readyz}" { + t.Fatalf("OPTIONS /healthz matched %q", tmpl) + } + }) + + t.Run("health probe", func(t *testing.T) { + req := newRoutingTestRequest(t, http.MethodGet, host, "/healthz", "", "") + var match mux.RouteMatch + if !router.Match(req, &match) { + t.Fatalf("GET /healthz matched no route: %v", match.MatchErr) + } + if tmpl, _ := match.Route.GetPathTemplate(); tmpl != "/healthz" { + t.Fatalf("GET /healthz matched %q", tmpl) + } + }) + + t.Run("STS", func(t *testing.T) { + req := newRoutingTestRequest(t, http.MethodPost, host, "/?Action=AssumeRoleWithWebIdentity&WebIdentityToken=test-token&RoleArn=arn:aws:iam::123:role/test&RoleSessionName=test-session", "", "") + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + rr := httptest.NewRecorder() + router.ServeHTTP(rr, req) + if rr.Code != http.StatusBadRequest && rr.Code != http.StatusServiceUnavailable { + t.Fatalf("STS request answered %d (%s), want the STS handler's 400/503", rr.Code, rr.Body.String()) + } + if ct := rr.Header().Get("Content-Type"); ct == "application/x-amz-json-1.1" { + t.Fatalf("STS request answered with the S3 Tables fallback: %s", rr.Body.String()) + } + }) + + t.Run("served S3 Tables operation", func(t *testing.T) { + req := newRoutingTestRequest(t, http.MethodDelete, host, "/tables/"+routingTestTableBucketARN+"/ns/t", "", "s3tables") + var match mux.RouteMatch + if !router.Match(req, &match) { + t.Fatalf("DeleteTable matched no route: %v", match.MatchErr) + } + tmpl, _ := match.Route.GetPathTemplate() + if !strings.HasPrefix(tmpl, "/tables/{tableBucketARN:") || !strings.HasSuffix(tmpl, "}/{namespace}/{name}") { + t.Fatalf("DeleteTable matched %q, want its own route", tmpl) + } + }) +} diff --git a/weed/s3api/s3tables/handler_rename_test.go b/weed/s3api/s3tables/handler_rename_test.go index e21ec2ede1..1506385a7b 100644 --- a/weed/s3api/s3tables/handler_rename_test.go +++ b/weed/s3api/s3tables/handler_rename_test.go @@ -3,7 +3,9 @@ package s3tables import ( "context" "encoding/json" + "net/http" "path" + "sync" "testing" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" @@ -350,3 +352,192 @@ func TestRenameTableRejectsConcurrentMaintenanceWrite(t *testing.T) { assert.ErrorIs(t, err, ErrConcurrentUpdate) assert.Equal(t, landedLate, src.Extended[ExtendedKeyMaintenance], "the late write must survive") } + +// setRenameSourceVersionToken stamps the seeded source table with token. +func setRenameSourceVersionToken(t *testing.T, fs *s3tablestest.MemFiler, token string) { + t.Helper() + src := fs.Get(GetNamespacePath(renameTestBucket, "ns"), "t") + require.NotNil(t, src) + var meta tableMetadataInternal + require.NoError(t, json.Unmarshal(src.Extended[ExtendedKeyMetadata], &meta)) + meta.VersionToken = token + raw, err := json.Marshal(&meta) + require.NoError(t, err) + src.Extended[ExtendedKeyMetadata] = raw +} + +// assertRenameRefused checks a refused rename left the source resolving, the +// destination absent, and no claim behind on the source. +func assertRenameRefused(t *testing.T, m *Manager, fs *s3tablestest.MemFiler) { + t.Helper() + _, err := runGetTable(t, m, fs, "ns", "t") + assert.NoError(t, err, "the source must still resolve") + + _, err = runGetTable(t, m, fs, "ns", "t2") + var s3Err *S3TablesError + require.ErrorAs(t, err, &s3Err, "the destination must not resolve") + assert.Equal(t, ErrCodeNoSuchTable, s3Err.Type) + assert.Nil(t, fs.Get(GetNamespacePath(renameTestBucket, "ns"), "t2"), "the destination must not be written") + + src := fs.Get(GetNamespacePath(renameTestBucket, "ns"), "t") + require.NotNil(t, src) + _, claimed := src.Extended[ExtendedKeyDeleting] + assert.False(t, claimed, "a refused rename must not leave its claim on the source") +} + +func renameTToT2(t *testing.T, token string) *RenameTableRequest { + return &RenameTableRequest{ + TableBucketARN: mustBucketARN(t), + SourceNamespace: []string{"ns"}, + SourceName: "t", + DestNamespace: []string{"ns"}, + DestName: "t2", + VersionToken: token, + } +} + +func TestRenameTableVersionTokenMatches(t *testing.T) { + fs, m := startRenameManager(t) + setRenameSourceVersionToken(t, fs, "tok-1") + + status, resp := runHandler(t, m, fs.Client, "RenameTable", renameTToT2(t, "tok-1")) + require.Equal(t, http.StatusOK, status, "%+v", resp) + + got, err := runGetTable(t, m, fs, "ns", "t2") + require.NoError(t, err) + assert.Equal(t, "t2", got.Name) +} + +func TestRenameTableVersionTokenMismatch(t *testing.T) { + fs, m := startRenameManager(t) + setRenameSourceVersionToken(t, fs, "tok-1") + + status, resp := runHandler(t, m, fs.Client, "RenameTable", renameTToT2(t, "stale")) + assert.Equal(t, http.StatusConflict, status, "%+v", resp) + assert.Equal(t, ErrCodeConflict, resp.Type) + assertRenameRefused(t, m, fs) +} + +// The token must be checked against the snapshot the claim asserts, not the +// metadata read at the start of the rename. An UpdateTable landing between +// the two moves the token on; checking the first read would accept the stale +// token, and the claim would then succeed on the newer snapshot. +func TestRenameTableVersionTokenCheckedAgainstClaimedSnapshot(t *testing.T) { + fs, m := startRenameManager(t) + setRenameSourceVersionToken(t, fs, "tok-1") + + status, resp := renameDuringUpdate(t, fs, m, "tok-1") + + assert.Equal(t, http.StatusConflict, status, "a rename carrying the pre-update token must be refused: %+v", resp) + assert.Equal(t, ErrCodeConflict, resp.Type) + assertRenameRefused(t, m, fs) + + got, err := runGetTable(t, m, fs, "ns", "t") + require.NoError(t, err) + assert.NotEqual(t, "tok-1", got.VersionToken, "the update must have moved the token on") +} + +// Without a token the same race must still not publish the metadata read +// before the update: the claim succeeds on the newer snapshot, so a rename +// built from the first read would move the table back to v3 and drop the +// update's commit. +func TestRenameTableRefusesSourceChangedBetweenReads(t *testing.T) { + fs, m := startRenameManager(t) + setRenameSourceVersionToken(t, fs, "tok-1") + + status, resp := renameDuringUpdate(t, fs, m, "") + + assert.Equal(t, http.StatusConflict, status, "a rename built from a torn read must be refused: %+v", resp) + assert.Equal(t, ErrCodeConflict, resp.Type) + assertRenameRefused(t, m, fs) + + got, err := runGetTable(t, m, fs, "ns", "t") + require.NoError(t, err) + assert.Equal(t, "s3://"+renameTestBucket+"/ns/t/metadata/v4.metadata.json", got.MetadataLocation, "the update's commit must survive") +} + +// renameDuringUpdate renames t to t2 carrying token while an UpdateTable to +// v4 lands between the rename's first read of the source and the lookup its +// claim snapshots, and returns the rename's answer. +func renameDuringUpdate(t *testing.T, fs *s3tablestest.MemFiler, m *Manager, token string) (int, S3TablesError) { + t.Helper() + nsPath := GetNamespacePath(renameTestBucket, "ns") + + // The rename's first lookup of the source reads its metadata; the second + // starts the reads that end in the lookupEntry the claim snapshots. The + // update lands on that second lookup. Its own lookups re-enter this hook, + // so it fires once under a mutex and a flag rather than a sync.Once, + // which would deadlock on the nested call. + var ( + mu sync.Mutex + lookups int + fired bool + updateStatus int + updateResp S3TablesError + ) + fs.OnLookup = func(dir, name string) { + if dir != nsPath || name != "t" { + return + } + mu.Lock() + if fired { + mu.Unlock() + return + } + lookups++ + if lookups < 2 { + mu.Unlock() + return + } + fired = true + mu.Unlock() + updateStatus, updateResp = runHandler(t, m, fs.Client, "UpdateTable", &UpdateTableRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Name: "t", + VersionToken: "tok-1", + MetadataLocation: "s3://" + renameTestBucket + "/ns/t/metadata/v4.metadata.json", + }) + } + + status, resp := runHandler(t, m, fs.Client, "RenameTable", renameTToT2(t, token)) + fs.OnLookup = nil + + mu.Lock() + require.True(t, fired, "the competing update never ran") + mu.Unlock() + require.Equal(t, http.StatusOK, updateStatus, "competing update: %+v", updateResp) + return status, resp +} + +func TestRenameReadsMatchSnapshot(t *testing.T) { + snapshot := map[string][]byte{ + ExtendedKeyMetadata: []byte(`{"name":"t"}`), + ExtendedKeyPolicy: []byte(`{"Version":"2012-10-17"}`), + ExtendedKeyTags: []byte(`{"team":"a"}`), + } + reads := func(policy string) map[string][]byte { + return map[string][]byte{ + ExtendedKeyMetadata: snapshot[ExtendedKeyMetadata], + ExtendedKeyPolicy: []byte(policy), + ExtendedKeyMaintenance: nil, + } + } + cases := []struct { + name string + policy string + tags map[string]string + want bool + }{ + {"unchanged", `{"Version":"2012-10-17"}`, map[string]string{"team": "a"}, true}, + {"policy changed", `{"Version":"2008-10-17"}`, map[string]string{"team": "a"}, false}, + {"policy removed", "", map[string]string{"team": "a"}, false}, + {"tag value changed", `{"Version":"2012-10-17"}`, map[string]string{"team": "b"}, false}, + {"tags removed", `{"Version":"2012-10-17"}`, nil, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.want, renameReadsMatchSnapshot(snapshot, tc.tags, reads(tc.policy))) + }) + } +} diff --git a/weed/s3api/s3tables/handler_table.go b/weed/s3api/s3tables/handler_table.go index 6db8fab282..3d036df2f6 100644 --- a/weed/s3api/s3tables/handler_table.go +++ b/weed/s3api/s3tables/handler_table.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "io" + "maps" "net/http" "path" "strconv" @@ -1395,6 +1396,24 @@ func (h *S3TablesHandler) handleRenameView(w http.ResponseWriter, r *http.Reques return h.renameCatalogEntry(w, r, filerClient, viewEntryKind) } +// renameReadsMatchSnapshot reports whether the source attributes a rename read +// separately (reads, plus the parsed tags) are the ones in snapshot, the entry +// its claim asserts. An absent attribute matches an empty read. +func renameReadsMatchSnapshot(snapshot map[string][]byte, tags map[string]string, reads map[string][]byte) bool { + for key, value := range reads { + if !bytes.Equal(snapshot[key], value) { + return false + } + } + var snapshotTags map[string]string + if raw := snapshot[ExtendedKeyTags]; len(raw) > 0 { + if err := json.Unmarshal(raw, &snapshotTags); err != nil { + return false + } + } + return maps.Equal(snapshotTags, tags) +} + // markSourceGone tags a missing source path so the rename reports it under the // source name instead of the closed-parent mapping naming the destination. func markSourceGone(err error) error { @@ -1447,6 +1466,7 @@ func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Requ destPath := GetTablePath(bucketName, destNamespace, destName) var metadata tableMetadataInternal + var metadataXattr []byte var metadataVersionXattr []byte var maintenanceXattr []byte var maintenanceStatusXattr []byte @@ -1461,6 +1481,7 @@ func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Requ if err != nil { return err } + metadataXattr = data if err := json.Unmarshal(data, &metadata); err != nil { return fmt.Errorf("failed to unmarshal table metadata: %w", err) } @@ -1636,6 +1657,27 @@ func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Requ return NewAuthError(kind.renameOp, principal, "not authorized to create "+kind.noun+" in the destination namespace") } + // The destination is built from attributes read one lookup at a time, but + // the claim and the source clear assert srcExtended. A write landing between + // those reads would let the claim succeed while the destination publishes + // the older values, dropping the write; refuse it so the caller retries. + // Past this check metadata is the claimed snapshot, so the token compares + // against it. + if !renameReadsMatchSnapshot(srcExtended, tableTags, map[string][]byte{ + ExtendedKeyMetadata: metadataXattr, + ExtendedKeyMetadataVersion: metadataVersionXattr, + ExtendedKeyPolicy: []byte(tablePolicy), + ExtendedKeyMaintenance: maintenanceXattr, + ExtendedKeyMaintenanceStatus: maintenanceStatusXattr, + }) { + h.writeError(w, http.StatusConflict, ErrCodeConflict, kind.noun+" changed during rename, retry the request") + return ErrConcurrentUpdate + } + if req.VersionToken != "" && metadata.VersionToken != req.VersionToken { + h.writeError(w, http.StatusConflict, ErrCodeConflict, "version token mismatch") + return ErrVersionTokenMismatch + } + if destExists { h.writeError(w, http.StatusConflict, kind.existsCode, fmt.Sprintf("%s %s already exists", kind.noun, destName)) return ErrTableAlreadyExists diff --git a/weed/s3api/s3tables/types.go b/weed/s3api/s3tables/types.go index 76af6d38bb..6d8808ce43 100644 --- a/weed/s3api/s3tables/types.go +++ b/weed/s3api/s3tables/types.go @@ -301,6 +301,9 @@ type RenameTableRequest struct { SourceName string `json:"sourceName"` DestNamespace []string `json:"destNamespace"` DestName string `json:"destName"` + // VersionToken, when set, makes the rename conditional on the source still + // carrying this version token. + VersionToken string `json:"versionToken,omitempty"` } type RenameTableResponse struct {