mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-07 04:27:49 +01:00
fix(s3tables): route REST RenameTable and refuse unrouted S3 Tables requests (#12446)
* s3tables: route REST RenameTable and refuse unrouted S3 Tables requests
PUT /tables/<arn>/<ns>/<name>/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 <noreply@anthropic.com>
* 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 <noreply@anthropic.com>
* 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 <noreply@anthropic.com>
* s3tables test: cover policy and tag changes in the rename snapshot compare
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
* 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 <noreply@anthropic.com>
---------
Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
8 files changed
+1027
-6
No files matched your search
@@ -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 {
|
||||
|
||||
@@ -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/<arn>/<ns>/<name>/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)
|
||||
})
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -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)))
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in new issue
Block a user